Skip to main content

xbp_cli/commands/
worktree_watch.rs

1use crate::codetime::GithubRepoRecord;
2use crate::commands::cli_session::{cli_request_client, resolve_cli_access_token};
3use crate::commands::terminal_table::{render_table, TableStyle};
4use crate::config::{
5    ensure_global_xbp_paths, map_windows_path_to_wsl_mnt, map_wsl_mnt_path_to_windows,
6    resolve_device_identity, resolve_global_xbp_root_dir, resolve_worktree_watch_config, ApiConfig,
7    SshConfig, WorktreeWatchConfig,
8};
9use chrono::{DateTime, Utc};
10use colored::Colorize;
11use notify::event::{CreateKind, ModifyKind, RemoveKind, RenameMode};
12use notify::{Config, Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
13use reqwest::StatusCode;
14use serde::{Deserialize, Serialize};
15use serde_json::{json, Value};
16use sha2::{Digest, Sha256};
17use std::collections::{BTreeMap, BTreeSet};
18use std::fs::{self, File, OpenOptions};
19use std::io::{BufRead, BufReader, Write};
20use std::path::{Path, PathBuf};
21use std::process::{Command, Stdio};
22use std::sync::{mpsc, OnceLock};
23use std::time::{Duration, Instant};
24use sysinfo::{Pid, ProcessRefreshKind, ProcessesToUpdate, System, UpdateKind};
25use uuid::Uuid;
26
27const BACKGROUND_CHILD_ENV: &str = "XBP_WORKTREE_WATCH_BACKGROUND_CHILD";
28const DEFAULT_REMOTE: &str = "origin";
29const EVENT_DEDUPE_WINDOW_SECONDS: i64 = 5;
30const WORKTREE_MUTATION_SYNC_BATCH_RECORD_LIMIT: usize = 250;
31const WORKTREE_MUTATION_SYNC_BATCH_JSON_BYTES: usize = 768 * 1024;
32const WORKTREE_MUTATION_SYNC_MAX_ATTEMPTS: usize = 3;
33const WORKTREE_MUTATION_SYNC_RETRY_DELAY_SECONDS: u64 = 2;
34
35/// Client-side ignore rules: built-in VCS/generated dirs + global config filters
36/// + `~/.xbp/.xbp-worktree-ignore` + project `.xbp/.xbpignore` / `watch_ignore_paths`.
37///
38/// Project `ignore_paths` (service discovery) intentionally do **not** apply here so
39/// ignored package roots still count for worktree-watch activity.
40/// Never uploaded; applied before events are spooled and again before sync.
41#[derive(Debug, Clone, Default)]
42struct WatchIgnoreRules {
43    /// Normalized lowercase prefixes without leading `./` (e.g. `secrets`, `apps/web/.env`).
44    forbidden_path_prefixes: Vec<String>,
45    /// Lowercase path segment names blocked anywhere (includes built-ins + config).
46    forbidden_folders: BTreeSet<String>,
47    /// Lowercase substrings matched against the full relative path.
48    banned_words: Vec<String>,
49    /// Machine-wide gitignore-style rules (`~/.xbp/.xbp-worktree-ignore`).
50    global_ignore: crate::utils::XbpIgnoreSet,
51    /// Project-local gitignore-style rules (`.xbp/.xbpignore`, yaml `watch_ignore_paths`).
52    project_ignore: crate::utils::XbpIgnoreSet,
53}
54
55impl WatchIgnoreRules {
56    fn load() -> Self {
57        Self::from_config(&resolve_worktree_watch_config())
58    }
59
60    fn from_config(config: &WorktreeWatchConfig) -> Self {
61        Self::from_config_and_project(config, None)
62    }
63
64    fn from_config_and_project(config: &WorktreeWatchConfig, project_root: Option<&Path>) -> Self {
65        let mut forbidden_folders = built_in_forbidden_folders();
66        for folder in &config.forbidden_folders {
67            let normalized = normalize_folder_token(folder);
68            if !normalized.is_empty() {
69                forbidden_folders.insert(normalized);
70            }
71        }
72
73        let forbidden_path_prefixes = config
74            .forbidden_paths
75            .iter()
76            .map(|path| normalize_path_prefix(path))
77            .filter(|path| !path.is_empty())
78            .collect::<Vec<_>>();
79
80        let banned_words = config
81            .banned_words
82            .iter()
83            .map(|word| word.trim().to_ascii_lowercase())
84            .filter(|word| !word.is_empty())
85            .collect::<Vec<_>>();
86
87        let project_ignore = project_root
88            .map(|root| load_project_watch_ignore(root))
89            .unwrap_or_default();
90
91        Self {
92            forbidden_path_prefixes,
93            forbidden_folders,
94            banned_words,
95            global_ignore: crate::config::load_global_worktree_ignore(),
96            project_ignore,
97        }
98    }
99
100    fn with_project_root(mut self, project_root: &Path) -> Self {
101        self.project_ignore = load_project_watch_ignore(project_root);
102        self
103    }
104
105    fn ignores_relative(&self, relative: &str) -> bool {
106        let normalized = normalize_relative_watch_path(relative);
107        if normalized.is_empty() {
108            return false;
109        }
110
111        if self.global_ignore.is_ignored_relative(&normalized, false)
112            || self.global_ignore.is_ignored_relative(&normalized, true)
113        {
114            return true;
115        }
116
117        if self.project_ignore.is_ignored_relative(&normalized, false)
118            || self.project_ignore.is_ignored_relative(&normalized, true)
119        {
120            return true;
121        }
122
123        let lower = normalized.to_ascii_lowercase();
124        let components = lower
125            .split('/')
126            .filter(|component| !component.is_empty())
127            .collect::<Vec<_>>();
128
129        if components
130            .iter()
131            .any(|component| self.forbidden_folders.contains(*component))
132        {
133            return true;
134        }
135
136        for prefix in &self.forbidden_path_prefixes {
137            if lower == *prefix || lower.starts_with(&format!("{prefix}/")) {
138                return true;
139            }
140            // Also treat a single-segment prefix as a folder ban anywhere.
141            if !prefix.contains('/') && components.iter().any(|component| *component == prefix) {
142                return true;
143            }
144        }
145
146        for word in &self.banned_words {
147            if lower.contains(word) {
148                return true;
149            }
150        }
151
152        false
153    }
154
155    fn ignores_path(&self, root: &Path, path: &Path) -> bool {
156        if let Some(relative) = relative_watch_path(root, path) {
157            return self.ignores_relative(&relative);
158        }
159        false
160    }
161
162    fn skips_discovery_dir(&self, path: &Path) -> bool {
163        let Some(name) = path.file_name().and_then(|value| value.to_str()) else {
164            return false;
165        };
166        let trimmed = name.trim();
167        if self
168            .forbidden_folders
169            .contains(&trimmed.to_ascii_lowercase())
170        {
171            return true;
172        }
173        if self.global_ignore.skips_dir_name(trimmed) {
174            return true;
175        }
176        self.project_ignore.skips_dir_name(trimmed)
177    }
178}
179
180fn load_project_watch_ignore(project_root: &Path) -> crate::utils::XbpIgnoreSet {
181    use crate::strategies::XbpConfig;
182    use crate::utils::{load_project_xbp_ignore, parse_config_with_auto_heal};
183
184    let mut extra = Vec::new();
185    // Watch-only patterns from xbp.yaml (`watch_ignore_paths` / `watch_ignore`).
186    // Service-discovery `ignore_paths` are intentionally excluded so worktree-watch
187    // still records activity under packages that are not registered as services.
188    if let Some(found) = crate::utils::find_xbp_config_upwards(project_root) {
189        if let Ok(content) = fs::read_to_string(&found.config_path) {
190            if let Ok((config, _)) = parse_config_with_auto_heal::<XbpConfig>(&content, found.kind)
191            {
192                extra = config.watch_ignore_paths;
193            }
194        }
195        return load_project_xbp_ignore(&found.project_root, &extra);
196    }
197    load_project_xbp_ignore(project_root, &extra)
198}
199
200fn built_in_forbidden_folders() -> BTreeSet<String> {
201    use crate::utils::DEFAULT_NOISE_DIR_NAMES;
202
203    DEFAULT_NOISE_DIR_NAMES
204        .iter()
205        .map(|name| name.to_ascii_lowercase())
206        .collect()
207}
208
209fn normalize_folder_token(raw: &str) -> String {
210    raw.trim()
211        .trim_matches(|ch| ch == '/' || ch == '\\')
212        .to_ascii_lowercase()
213}
214
215fn normalize_path_prefix(raw: &str) -> String {
216    let trimmed = raw.trim().replace('\\', "/");
217    let without_dot = trimmed
218        .trim_start_matches("./")
219        .trim_matches(|ch| ch == '/' || ch == '\\');
220    without_dot.to_ascii_lowercase()
221}
222
223fn normalize_relative_watch_path(raw: &str) -> String {
224    raw.trim()
225        .replace('\\', "/")
226        .trim_start_matches("./")
227        .trim_matches('/')
228        .to_string()
229}
230
231fn watch_ignore_rules() -> &'static WatchIgnoreRules {
232    static RULES: OnceLock<WatchIgnoreRules> = OnceLock::new();
233    RULES.get_or_init(WatchIgnoreRules::load)
234}
235
236/// Global config rules merged with per-repo `.xbp/.xbpignore` / `ignore_paths`.
237fn watch_ignore_rules_for_root(root: &Path) -> WatchIgnoreRules {
238    watch_ignore_rules().clone().with_project_root(root)
239}
240
241#[cfg(windows)]
242const CREATE_NO_WINDOW: u32 = 0x08000000;
243
244#[derive(Debug, Clone, Default)]
245pub struct WorktreeWatchTargetOptions {
246    pub repo: Option<PathBuf>,
247    pub parent: Option<PathBuf>,
248    /// Explicit multi-repo list (tray / advanced targeting). Ignored when `parent` is set.
249    pub repos: Vec<PathBuf>,
250}
251
252/// Prefer a running parent watcher when `status` is invoked with no target.
253///
254/// Lets `xbp worktree-watch status` from any cwd show the active parent cover
255/// (e.g. Documents/GitHub) instead of failing outside a git repo or only
256/// reporting the current checkout as stopped.
257pub fn resolve_default_worktree_watch_status_target() -> Result<WorktreeWatchTargetOptions, String> {
258    if let Some(parent) = prefer_running_parent_root() {
259        return Ok(WorktreeWatchTargetOptions {
260            repo: None,
261            parent: Some(parent),
262            repos: Vec::new(),
263        });
264    }
265    // Fall back to the most recent parent state even if the process just died,
266    // so bare `status` still shows the cover matrix instead of a git cwd error.
267    if let Some(parent) = prefer_any_parent_root() {
268        return Ok(WorktreeWatchTargetOptions {
269            repo: None,
270            parent: Some(parent),
271            repos: Vec::new(),
272        });
273    }
274    normalize_worktree_watch_target(WorktreeWatchTargetOptions {
275        repo: None,
276        parent: None,
277        repos: Vec::new(),
278    })
279}
280
281/// Resolve inventory names and pin absolute paths so detached tray/watcher children
282/// do not depend on the caller's cwd (or a bare short name like `athena`).
283pub fn normalize_worktree_watch_target(
284    target: WorktreeWatchTargetOptions,
285) -> Result<WorktreeWatchTargetOptions, String> {
286    if target.parent.is_some() && target.repo.is_some() {
287        return Err("Pass either `--repo` or `--parent`, not both.".to_string());
288    }
289    if target.parent.is_some() && !target.repos.is_empty() {
290        return Err("Pass either `--repos` or `--parent`, not both.".to_string());
291    }
292    if target.repo.is_some() && !target.repos.is_empty() {
293        return Err("Pass either `--repo` or `--repos`, not both.".to_string());
294    }
295
296    if let Some(parent) = target.parent.as_ref() {
297        let resolved = if parent.is_dir() {
298            normalize_windows_verbatim_path(
299                fs::canonicalize(parent).unwrap_or_else(|_| parent.clone()),
300            )
301        } else {
302            // Allow bare folder names under common clone roots.
303            if let Some(found) = resolve_parent_folder_hint(parent)? {
304                found
305            } else {
306                return Err(format!(
307                    "Parent folder not found: {}. Pass an absolute path or a folder under Documents/GitHub.",
308                    parent.display()
309                ));
310            }
311        };
312        return Ok(WorktreeWatchTargetOptions {
313            repo: None,
314            parent: Some(resolved),
315            repos: Vec::new(),
316        });
317    }
318
319    if !target.repos.is_empty() {
320        let mut repos = Vec::new();
321        let mut seen = BTreeSet::new();
322        for repo in &target.repos {
323            let path = resolve_repo_path_input(Some(repo.as_path()))?;
324            let absolute = normalize_windows_verbatim_path(
325                fs::canonicalize(&path).unwrap_or(path),
326            );
327            let key = path_identity_key(&absolute);
328            if seen.insert(key) {
329                repos.push(absolute);
330            }
331        }
332        return Ok(WorktreeWatchTargetOptions {
333            repo: None,
334            parent: None,
335            repos,
336        });
337    }
338
339    // Single repo (explicit or default to cwd) — always pin absolute root.
340    let identity = resolve_repo_identity(target.repo.as_deref())?;
341    Ok(WorktreeWatchTargetOptions {
342        repo: Some(identity.root),
343        parent: None,
344        repos: Vec::new(),
345    })
346}
347
348fn resolve_parent_folder_hint(parent: &Path) -> Result<Option<PathBuf>, String> {
349    if parent.is_absolute() {
350        return Ok(None);
351    }
352    let name = parent
353        .file_name()
354        .and_then(|value| value.to_str())
355        .unwrap_or_default();
356    if name.is_empty() {
357        return Ok(None);
358    }
359
360    let mut candidates: Vec<PathBuf> = Vec::new();
361    if let Some(home) = dirs::home_dir() {
362        candidates.push(home.join("Documents").join("GitHub").join(name));
363        candidates.push(home.join("Documents").join("github").join(name));
364        candidates.push(home.join("Documents").join(name));
365        candidates.push(home.join("src").join(name));
366        candidates.push(home.join(name));
367    }
368    // If the hint itself is "GitHub", resolve the common clone root.
369    if name.eq_ignore_ascii_case("github") {
370        if let Some(home) = dirs::home_dir() {
371            candidates.insert(0, home.join("Documents").join("GitHub"));
372            candidates.insert(1, home.join("Documents").join("github"));
373        }
374    }
375
376    for candidate in candidates {
377        if candidate.is_dir() {
378            return Ok(Some(normalize_windows_verbatim_path(
379                fs::canonicalize(&candidate).unwrap_or(candidate),
380            )));
381        }
382    }
383    Ok(None)
384}
385
386/// Live snapshot for the worktree-watch system tray UI.
387#[derive(Debug, Clone, Serialize)]
388#[serde(rename_all = "camelCase")]
389pub struct WorktreeWatchTraySnapshot {
390    pub target_label: String,
391    pub mode: String,
392    pub parent: Option<String>,
393    pub repository_count: usize,
394    pub running_watchers: usize,
395    pub any_running: bool,
396    pub all_running: bool,
397    pub parent_watcher_running: bool,
398    pub parent_watcher_pid: Option<u32>,
399    pub total_unsynced_files: u64,
400    pub total_unsynced_records: u64,
401    pub repositories: Vec<WorktreeWatchTrayRepoStatus>,
402    pub stats_line: Option<String>,
403}
404
405#[derive(Debug, Clone, Serialize)]
406#[serde(rename_all = "camelCase")]
407pub struct WorktreeWatchTrayRepoStatus {
408    pub root: String,
409    pub owner: String,
410    pub name: String,
411    pub branch: String,
412    pub running: bool,
413    pub pid: Option<u32>,
414    pub unsynced_files: u64,
415    pub unsynced_records: u64,
416}
417
418#[derive(Debug, Clone)]
419pub struct WorktreeWatchStartOptions {
420    pub target: WorktreeWatchTargetOptions,
421    pub detach: bool,
422    pub sync_interval_seconds: u64,
423    pub once: bool,
424}
425
426#[derive(Debug, Clone)]
427pub struct WorktreeWatchSyncOptions {
428    pub target: WorktreeWatchTargetOptions,
429    pub dry_run: bool,
430    pub resync: bool,
431}
432
433#[derive(Debug, Clone)]
434pub struct WorktreeWatchStopOptions {
435    pub target: WorktreeWatchTargetOptions,
436    pub force: bool,
437}
438
439#[derive(Debug, Clone)]
440pub struct WorktreeWatchStatusOptions {
441    pub target: WorktreeWatchTargetOptions,
442    pub json: bool,
443    pub records: bool,
444    pub record_limit: usize,
445    pub stats: bool,
446    pub repo_activity: bool,
447    pub stats_gap_minutes: u64,
448}
449
450#[derive(Debug, Clone)]
451struct RepoIdentity {
452    owner: String,
453    name: String,
454    branch: String,
455    root: PathBuf,
456    head_sha: Option<String>,
457}
458
459#[derive(Debug, Clone)]
460struct SpoolLayout {
461    root: PathBuf,
462    events_file: PathBuf,
463    commits_file: PathBuf,
464    stats_file: PathBuf,
465    watcher_state_file: PathBuf,
466    sync_log_file: PathBuf,
467    event_dedupe_file: PathBuf,
468    event_dedupe_dir: PathBuf,
469    commit_dedupe_dir: PathBuf,
470}
471
472#[derive(Debug, Clone)]
473struct ParentSpoolLayout {
474    watcher_state_file: PathBuf,
475}
476
477#[derive(Debug)]
478struct WatchedRepo {
479    identity: RepoIdentity,
480    layout: SpoolLayout,
481    last_head: Option<String>,
482    event_dedupe: RecentEventDedupe,
483}
484
485#[derive(Debug, Serialize, Deserialize, Clone)]
486#[serde(rename_all = "camelCase")]
487struct WorktreeMutationEvent {
488    id: String,
489    #[serde(default, skip_serializing_if = "Option::is_none")]
490    fingerprint: Option<String>,
491    repo_owner: String,
492    repo_name: String,
493    branch_name: String,
494    repo_root: String,
495    head_sha: Option<String>,
496    event_kind: String,
497    paths: Vec<String>,
498    primary_path: Option<String>,
499    old_path: Option<String>,
500    new_path: Option<String>,
501    added_lines: Option<u64>,
502    removed_lines: Option<u64>,
503    total_lines: Option<u64>,
504    file_created: bool,
505    file_removed: bool,
506    folder_created: bool,
507    folder_removed: bool,
508    renamed_or_moved: bool,
509    raw_kind: String,
510    occurred_at: DateTime<Utc>,
511}
512
513#[derive(Debug, Serialize, Deserialize, Clone)]
514#[serde(rename_all = "camelCase")]
515struct WorktreeCommitEvent {
516    id: String,
517    #[serde(default, skip_serializing_if = "Option::is_none")]
518    fingerprint: Option<String>,
519    repo_owner: String,
520    repo_name: String,
521    branch_name: String,
522    repo_root: String,
523    previous_head_sha: Option<String>,
524    head_sha: String,
525    subject: Option<String>,
526    author_name: Option<String>,
527    author_email: Option<String>,
528    committed_at: Option<String>,
529    occurred_at: DateTime<Utc>,
530}
531
532#[derive(Debug, Serialize)]
533#[serde(rename_all = "camelCase")]
534struct WorktreeMutationIngestPayload {
535    device: WorktreeMutationDevicePayload,
536    repository: WorktreeMutationRepositoryPayload,
537    events: Vec<WorktreeMutationEvent>,
538    commits: Vec<WorktreeCommitEvent>,
539}
540
541#[derive(Debug, Serialize)]
542#[serde(rename_all = "camelCase")]
543struct WorktreeMutationDevicePayload {
544    hardware_id: String,
545    device_name: Option<String>,
546    hostname: Option<String>,
547    platform: String,
548    /// Stable runtime isolation key (e.g. `windows`, `wsl-ubuntu`, `linux`).
549    /// Lets Windows + WSL watchers coexist as distinct upload sources.
550    #[serde(skip_serializing_if = "Option::is_none")]
551    runtime_key: Option<String>,
552}
553
554#[derive(Debug, Serialize)]
555#[serde(rename_all = "camelCase")]
556struct WorktreeMutationRepositoryPayload {
557    owner: String,
558    name: String,
559    branch_name: String,
560    repo_root: String,
561}
562
563#[derive(Debug)]
564struct SyncCandidate {
565    path: PathBuf,
566    records: SyncRecords,
567}
568
569#[derive(Debug)]
570struct WorktreeMutationUploadBatch {
571    events: Vec<WorktreeMutationEvent>,
572    commits: Vec<WorktreeCommitEvent>,
573    estimated_json_bytes: usize,
574}
575
576#[derive(Debug, Clone, Copy, PartialEq, Eq)]
577enum WorktreeMutationSyncErrorKind {
578    Retryable,
579    NonRetryable,
580}
581
582#[derive(Debug)]
583struct WorktreeMutationSyncError {
584    kind: WorktreeMutationSyncErrorKind,
585    message: String,
586}
587
588struct SyncLogAppend<'a> {
589    identity: &'a RepoIdentity,
590    layout: &'a SpoolLayout,
591    endpoint: &'a str,
592    status_code: u16,
593    event_count: usize,
594    commit_count: usize,
595    candidates: &'a [SyncCandidate],
596    resync: bool,
597}
598
599#[derive(Debug, Clone, Copy)]
600enum SyncFileKind {
601    Events,
602    Commits,
603}
604
605#[derive(Debug)]
606enum SyncRecords {
607    Events(Vec<WorktreeMutationEvent>),
608    Commits(Vec<WorktreeCommitEvent>),
609}
610
611impl SyncRecords {
612    fn len(&self) -> usize {
613        match self {
614            Self::Events(records) => records.len(),
615            Self::Commits(records) => records.len(),
616        }
617    }
618
619    fn is_empty(&self) -> bool {
620        self.len() == 0
621    }
622}
623
624#[derive(Debug, Serialize, Deserialize, Clone)]
625#[serde(rename_all = "camelCase")]
626struct WorktreeWatchStatsSummary {
627    generated_at: DateTime<Utc>,
628    repository_owner: String,
629    repository_name: String,
630    branch_name: String,
631    repo_root: String,
632    spool_root: String,
633    session_gap_minutes: u64,
634    total_events: u64,
635    total_commits: u64,
636    first_event_at: Option<DateTime<Utc>>,
637    last_event_at: Option<DateTime<Utc>>,
638    session_count: u64,
639    observed_span_seconds: u64,
640    estimated_coding_seconds: u64,
641    added_lines: u64,
642    removed_lines: u64,
643    by_file: Vec<WorktreeWatchFileStats>,
644    by_filetype: Vec<WorktreeWatchBucketStats>,
645    by_event_kind: Vec<WorktreeWatchBucketStats>,
646}
647
648#[derive(Debug, Serialize, Clone)]
649#[serde(rename_all = "camelCase")]
650struct WorktreeWatchRepoActivitySummary {
651    generated_at: DateTime<Utc>,
652    repository_owner: String,
653    repository_name: String,
654    repo_root: String,
655    repository_spool_root: String,
656    stats_file: String,
657    session_gap_minutes: u64,
658    branch_count: u64,
659    total_events: u64,
660    total_commits: u64,
661    first_event_at: Option<DateTime<Utc>>,
662    last_event_at: Option<DateTime<Utc>>,
663    session_count: u64,
664    observed_span_seconds: u64,
665    estimated_coding_seconds: u64,
666    added_lines: u64,
667    removed_lines: u64,
668    branches: Vec<WorktreeWatchBranchActivityStats>,
669}
670
671#[derive(Debug, Serialize, Clone)]
672#[serde(rename_all = "camelCase")]
673struct WorktreeWatchBranchActivityStats {
674    branch_name: String,
675    spool_root: String,
676    total_events: u64,
677    total_commits: u64,
678    first_event_at: Option<DateTime<Utc>>,
679    last_event_at: Option<DateTime<Utc>>,
680    session_count: u64,
681    observed_span_seconds: u64,
682    estimated_coding_seconds: u64,
683    added_lines: u64,
684    removed_lines: u64,
685}
686
687#[derive(Debug, Serialize, Deserialize, Clone, Default)]
688#[serde(rename_all = "camelCase")]
689struct WorktreeWatchFileStats {
690    path: String,
691    filetype: String,
692    event_count: u64,
693    estimated_coding_seconds: u64,
694    added_lines: u64,
695    removed_lines: u64,
696}
697
698#[derive(Debug, Serialize, Deserialize, Clone, Default)]
699#[serde(rename_all = "camelCase")]
700struct WorktreeWatchBucketStats {
701    name: String,
702    event_count: u64,
703    estimated_coding_seconds: u64,
704    added_lines: u64,
705    removed_lines: u64,
706}
707
708#[derive(Debug, Serialize, Deserialize, Clone)]
709#[serde(rename_all = "camelCase")]
710struct WorktreeWatcherState {
711    pid: u32,
712    repo_root: String,
713    repository_owner: String,
714    repository_name: String,
715    branch_name: String,
716    executable: String,
717    started_at: DateTime<Utc>,
718    /// Present on modern state files so Windows/WSL/Linux watchers never clobber each other.
719    #[serde(default, skip_serializing_if = "Option::is_none")]
720    runtime_key: Option<String>,
721    #[serde(default, skip_serializing_if = "Option::is_none")]
722    platform: Option<String>,
723}
724
725#[derive(Debug, Serialize, Deserialize, Clone)]
726#[serde(rename_all = "camelCase")]
727struct ParentWorktreeWatcherState {
728    pid: u32,
729    parent_root: String,
730    executable: String,
731    started_at: DateTime<Utc>,
732    #[serde(default, skip_serializing_if = "Option::is_none")]
733    runtime_key: Option<String>,
734    #[serde(default, skip_serializing_if = "Option::is_none")]
735    platform: Option<String>,
736}
737
738#[derive(Debug, Serialize, Deserialize, Clone)]
739#[serde(rename_all = "camelCase")]
740struct WorktreeWatchSyncLogEntry {
741    id: String,
742    synced_at: DateTime<Utc>,
743    endpoint: String,
744    repository_owner: String,
745    repository_name: String,
746    branch_name: String,
747    repo_root: String,
748    resync: bool,
749    status_code: u16,
750    event_count: usize,
751    commit_count: usize,
752    spool_file_count: usize,
753    spool_files: Vec<String>,
754}
755
756#[derive(Debug, Clone, Copy, PartialEq, Eq)]
757enum StopWatcherOutcome {
758    Stopped(u32),
759    RemovedStaleState,
760    NoState,
761}
762
763#[derive(Debug, Serialize, Deserialize, Clone)]
764#[serde(rename_all = "camelCase")]
765struct WorktreeWatchEventDedupeRecord {
766    fingerprint: String,
767    first_seen_at: DateTime<Utc>,
768    event_kind: String,
769    primary_path: Option<String>,
770}
771
772#[derive(Debug, Default)]
773struct RecentEventDedupe {
774    fingerprints: BTreeSet<String>,
775}
776
777pub async fn run_worktree_watch_start(options: WorktreeWatchStartOptions) -> Result<(), String> {
778    if options.detach {
779        // Keep detached start free of the foreground watch future so the local
780        // HTTP API can call this without creating an async type cycle.
781        return run_worktree_watch_start_detached(options);
782    }
783
784    if let Some(parent) = options.target.parent.as_deref() {
785        let identities = resolve_target_identities(&options.target)?;
786        if identities.is_empty() {
787            println!("No git repositories found under parent folder.");
788            return Ok(());
789        }
790
791        if options.once {
792            for identity in &identities {
793                let layout = prepare_spool_layout(identity)?;
794                record_commit_snapshot(identity, &layout, None)?;
795                println!("Recorded current git state for {}", identity.root.display());
796            }
797            return Ok(());
798        }
799
800        let parent = canonical_parent_path(parent)?;
801        println!(
802            "Watching {} recursively and routing mutations for {} repo(s).",
803            parent.display(),
804            identities.len()
805        );
806        return watch_parent_foreground(parent, identities, options.sync_interval_seconds).await;
807    }
808
809    let identity = resolve_repo_identity(options.target.repo.as_deref())?;
810    let layout = prepare_spool_layout(&identity)?;
811    println!(
812        "Watching {} and spooling mutations to {} [runtime={}]",
813        identity.root.display(),
814        layout.root.display(),
815        worktree_runtime_key()
816    );
817
818    if options.once {
819        record_commit_snapshot(&identity, &layout, None)?;
820        return Ok(());
821    }
822
823    watch_foreground(identity, layout, options.sync_interval_seconds).await
824}
825
826/// Spawn detached watcher process(es) only (no foreground watch loop).
827///
828/// Used by the local HTTP API and CLI `--detach` so handlers never type-depend
829/// on the infinite `watch_foreground` future.
830pub fn run_worktree_watch_start_detached(
831    options: WorktreeWatchStartOptions,
832) -> Result<(), String> {
833    if let Some(parent) = options.target.parent.as_deref() {
834        // Fast .git root count only — full identity resolve is deferred to the child.
835        let parent_path = canonical_parent_path(parent)?;
836        let repo_count = count_git_roots_under(&parent_path)?;
837        if repo_count == 0 {
838            println!("No git repositories found under parent folder.");
839            return Ok(());
840        }
841        let path = spawn_detached_parent_worktree_watch(&parent_path)?;
842        println!(
843            "Started one background worktree watcher for {} covering {} repo(s).",
844            path.display(),
845            repo_count
846        );
847        if let Some(port) =
848            crate::commands::worktree_watch_api::resolve_worktree_watch_api_port(None)
849        {
850            println!(
851                "Local API (when free): http://127.0.0.1:{port}  — dashboard: apps/worktree-watch-app/"
852            );
853        }
854        return Ok(());
855    }
856
857    if !options.target.repos.is_empty() {
858        for repo in &options.target.repos {
859            let path = spawn_detached_worktree_watch(Some(repo.as_path()))?;
860            println!("Started background worktree watcher for {}", path.display());
861        }
862        if let Some(port) =
863            crate::commands::worktree_watch_api::resolve_worktree_watch_api_port(None)
864        {
865            println!(
866                "Local API (when free): http://127.0.0.1:{port}  — dashboard: apps/worktree-watch-app/"
867            );
868        }
869        return Ok(());
870    }
871
872    spawn_detached_worktree_watch(options.target.repo.as_deref()).map(|path| {
873        println!("Started background worktree watcher for {}", path.display());
874        if let Some(port) =
875            crate::commands::worktree_watch_api::resolve_worktree_watch_api_port(None)
876        {
877            println!(
878                "Local API (when free): http://127.0.0.1:{port}  — dashboard: apps/worktree-watch-app/"
879            );
880        }
881    })
882}
883
884pub async fn run_worktree_watch_sync(options: WorktreeWatchSyncOptions) -> Result<(), String> {
885    let identities = resolve_target_identities(&options.target)?;
886    let repo_count = identities.len();
887    let mut plans = Vec::new();
888    let mut total_files = 0usize;
889    let mut total_records = 0usize;
890
891    for identity in identities {
892        let layout = prepare_spool_layout(&identity)?;
893        let candidates = collect_spool_candidates(&layout, options.resync)?;
894        total_files += candidates.len();
895        total_records += candidates
896            .iter()
897            .map(|candidate| candidate.records.len())
898            .sum::<usize>();
899        if !candidates.is_empty() {
900            plans.push((identity, candidates));
901        }
902    }
903
904    if options.dry_run {
905        println!(
906            "Would sync {} record(s) from {} spool file(s) across {} repo(s).",
907            total_records, total_files, repo_count
908        );
909        if options.resync {
910            println!("Resync mode includes files already marked `.synced.*.jsonl`.");
911        }
912        return Ok(());
913    }
914
915    if plans.is_empty() {
916        print_no_sync_candidates_message(&options.target, options.resync)?;
917        return Ok(());
918    }
919
920    let upload_repo_count = plans.len();
921    for (identity, candidates) in plans {
922        sync_candidates(&identity, candidates, options.resync).await?;
923    }
924    println!(
925        "Synced {} worktree mutation record(s) across {} repo(s).",
926        total_records, upload_repo_count
927    );
928    Ok(())
929}
930
931pub fn run_worktree_watch_stop(options: WorktreeWatchStopOptions) -> Result<(), String> {
932    if let Some(parent) = options.target.parent.as_deref() {
933        let parent = canonical_parent_path(parent)?;
934        match stop_existing_parent_watcher(&parent, options.force)? {
935            StopWatcherOutcome::Stopped(pid) => {
936                println!(
937                    "Stopped background parent worktree watcher {} for {}",
938                    pid,
939                    parent.display()
940                );
941            }
942            StopWatcherOutcome::RemovedStaleState => {
943                println!(
944                    "Removed stale parent worktree watcher state for {}",
945                    parent.display()
946                );
947            }
948            StopWatcherOutcome::NoState => {
949                println!("No parent worktree watcher state for {}", parent.display());
950            }
951        }
952    }
953
954    let identities = resolve_target_identities(&options.target)?;
955    if identities.is_empty() {
956        println!("No git repositories found for worktree watcher stop.");
957        return Ok(());
958    }
959
960    let mut stopped = 0usize;
961    let mut stale = 0usize;
962    let mut missing = 0usize;
963    for identity in identities {
964        let layout = prepare_spool_layout(&identity)?;
965        match stop_existing_watcher(&identity, &layout, options.force)? {
966            StopWatcherOutcome::Stopped(pid) => {
967                stopped += 1;
968                println!(
969                    "Stopped background worktree watcher {} for {}",
970                    pid,
971                    identity.root.display()
972                );
973            }
974            StopWatcherOutcome::RemovedStaleState => {
975                stale += 1;
976                println!(
977                    "Removed stale worktree watcher state for {}",
978                    identity.root.display()
979                );
980            }
981            StopWatcherOutcome::NoState => {
982                missing += 1;
983                println!(
984                    "No background worktree watcher state for {}",
985                    identity.root.display()
986                );
987            }
988        }
989    }
990
991    println!(
992        "Worktree watcher stop complete: {stopped} stopped, {stale} stale state file(s) removed, {missing} without state."
993    );
994    Ok(())
995}
996
997pub async fn run_worktree_watch_status(options: WorktreeWatchStatusOptions) -> Result<(), String> {
998    // Lightweight overview (running detection + spool counts + last sync age).
999    // Full JSONL decode only when --records / --stats / --repo-activity is set.
1000    let need_heavy = options.records || options.stats || options.repo_activity;
1001    let snapshot = collect_worktree_watch_tray_snapshot(&options.target)?;
1002
1003    let mut payloads = Vec::new();
1004    let mut total_files = 0usize;
1005    let mut total_records = 0usize;
1006
1007    for tray_repo in &snapshot.repositories {
1008        let identity = identity_from_tray_repo(tray_repo);
1009        let mut status = if need_heavy {
1010            worktree_watch_status_payload(&identity, options.records, options.record_limit)?
1011        } else {
1012            // Reuse tray counts; only open sync-log for age.
1013            worktree_watch_status_from_tray_repo(tray_repo)?
1014        };
1015
1016        if let Some(object) = status.as_object_mut() {
1017            object.insert("running".to_string(), json!(tray_repo.running));
1018            object.insert("pid".to_string(), json!(tray_repo.pid));
1019            if let Some(last_sync) = object.get("lastSync") {
1020                if let Some(synced_at) = last_sync.get("syncedAt").and_then(Value::as_str) {
1021                    if let Ok(dt) = DateTime::parse_from_rfc3339(synced_at) {
1022                        let age = (Utc::now() - dt.with_timezone(&Utc))
1023                            .num_seconds()
1024                            .max(0) as u64;
1025                        object.insert("lastSyncSecondsAgo".to_string(), json!(age));
1026                    }
1027                }
1028            }
1029        }
1030
1031        let stats = if options.stats {
1032            Some(generate_and_store_stats(
1033                &identity,
1034                options.stats_gap_minutes,
1035            )?)
1036        } else {
1037            None
1038        };
1039        let repo_activity = if options.repo_activity {
1040            Some(generate_and_store_repo_activity(
1041                &identity,
1042                options.stats_gap_minutes,
1043            )?)
1044        } else {
1045            None
1046        };
1047        // Refresh TODO→issue coding effort from this repo's mutation spool when present.
1048        if options.stats || options.repo_activity {
1049            if let Err(err) = crate::commands::todos::effort::recompute_effort(
1050                &identity.root,
1051                options.stats_gap_minutes,
1052                true,
1053            ) {
1054                eprintln!("note: todo effort recompute: {err}");
1055            }
1056        }
1057        let status = attach_optional_stats(status, stats, repo_activity)?;
1058        total_files += status
1059            .get("unsyncedFiles")
1060            .and_then(Value::as_u64)
1061            .unwrap_or(0) as usize;
1062        total_records += status
1063            .get("unsyncedRecords")
1064            .and_then(Value::as_u64)
1065            .unwrap_or(0) as usize;
1066        payloads.push(status);
1067    }
1068
1069    if options.json {
1070        let payload = if options.target.parent.is_some() || payloads.len() > 1 {
1071            json!({
1072                "targetLabel": snapshot.target_label,
1073                "mode": snapshot.mode,
1074                "parent": snapshot.parent,
1075                "parentWatcherRunning": snapshot.parent_watcher_running,
1076                "parentWatcherPid": snapshot.parent_watcher_pid,
1077                "anyRunning": snapshot.any_running,
1078                "allRunning": snapshot.all_running,
1079                "runningWatchers": snapshot.running_watchers,
1080                "repositories": payloads,
1081                "repositoryCount": payloads.len(),
1082                "unsyncedFiles": total_files,
1083                "unsyncedRecords": total_records,
1084            })
1085        } else {
1086            payloads.into_iter().next().unwrap_or_else(|| json!({}))
1087        };
1088        println!(
1089            "{}",
1090            serde_json::to_string_pretty(&payload)
1091                .map_err(|error| format!("Failed to render status JSON: {error}"))?
1092        );
1093    } else {
1094        print_status_overview(&snapshot, total_files, total_records);
1095        print_status_matrix(&payloads, &snapshot);
1096        if need_heavy {
1097            for (index, payload) in payloads.iter().enumerate() {
1098                if index > 0 {
1099                    println!();
1100                }
1101                if options.records {
1102                    println!(
1103                        "{} {}/{}",
1104                        "records".bright_black(),
1105                        payload
1106                            .get("repositoryOwner")
1107                            .and_then(Value::as_str)
1108                            .unwrap_or("unknown"),
1109                        payload
1110                            .get("repositoryName")
1111                            .and_then(Value::as_str)
1112                            .unwrap_or("repository")
1113                    );
1114                    print_status_records(payload);
1115                }
1116                if options.stats {
1117                    print_status_stats(payload);
1118                }
1119                if options.repo_activity {
1120                    print_repo_activity(payload);
1121                }
1122            }
1123        }
1124    }
1125
1126    Ok(())
1127}
1128
1129fn print_status_overview(
1130    snapshot: &WorktreeWatchTraySnapshot,
1131    total_files: usize,
1132    total_records: usize,
1133) {
1134    let running_label = if snapshot.any_running {
1135        "RUNNING".green().bold()
1136    } else {
1137        "STOPPED".red().bold()
1138    };
1139    println!(
1140        "{} {}  {}",
1141        "worktree-watch".bright_magenta().bold(),
1142        running_label,
1143        snapshot.target_label.bright_white()
1144    );
1145
1146    if snapshot.mode == "parent" {
1147        let parent_label = if snapshot.parent_watcher_running {
1148            format!(
1149                "parent watcher {}",
1150                snapshot
1151                    .parent_watcher_pid
1152                    .map(|pid| format!("pid {pid}"))
1153                    .unwrap_or_else(|| "active".to_string())
1154            )
1155            .green()
1156            .to_string()
1157        } else {
1158            "parent watcher not detected".red().to_string()
1159        };
1160        println!(
1161            "  {}  {}  {} repo(s) covered",
1162            parent_label,
1163            "·".bright_black(),
1164            snapshot.repository_count
1165        );
1166    } else {
1167        println!(
1168            "  {} watching · {} repo(s)",
1169            snapshot.running_watchers, snapshot.repository_count
1170        );
1171    }
1172
1173    println!(
1174        "  {} unsynced record(s) in {} file(s)",
1175        total_records.to_string().yellow(),
1176        total_files
1177    );
1178    if let Some(line) = snapshot.stats_line.as_ref() {
1179        println!("  {}", line.bright_black());
1180    }
1181    println!();
1182}
1183
1184fn print_status_matrix(payloads: &[Value], snapshot: &WorktreeWatchTraySnapshot) {
1185    if payloads.is_empty() {
1186        println!("{}", "No repositories found for this target.".yellow());
1187        return;
1188    }
1189
1190    let headers = [
1191        "REPO",
1192        "BRANCH",
1193        "STATE",
1194        "PID",
1195        "UNSYNCED",
1196        "LAST SYNC",
1197        "AGE",
1198    ];
1199    let mut rows = Vec::with_capacity(payloads.len());
1200
1201    for payload in payloads {
1202        let owner = payload
1203            .get("repositoryOwner")
1204            .or_else(|| payload.get("owner"))
1205            .and_then(Value::as_str)
1206            .unwrap_or("unknown");
1207        let name = payload
1208            .get("repositoryName")
1209            .or_else(|| payload.get("name"))
1210            .and_then(Value::as_str)
1211            .unwrap_or("repository");
1212        let branch = payload
1213            .get("branchName")
1214            .or_else(|| payload.get("branch"))
1215            .and_then(Value::as_str)
1216            .unwrap_or("unknown");
1217        let running = payload
1218            .get("running")
1219            .and_then(Value::as_bool)
1220            .unwrap_or(false);
1221        let pid = payload
1222            .get("pid")
1223            .and_then(Value::as_u64)
1224            .map(|value| value.to_string())
1225            .unwrap_or_else(|| {
1226                if running {
1227                    snapshot
1228                        .parent_watcher_pid
1229                        .map(|value| value.to_string())
1230                        .unwrap_or_else(|| "—".to_string())
1231                } else {
1232                    "—".to_string()
1233                }
1234            });
1235        let unsynced = payload
1236            .get("unsyncedRecords")
1237            .and_then(Value::as_u64)
1238            .unwrap_or(0);
1239        let (last_sync_at, age) = last_sync_display(payload);
1240        let state = if running {
1241            "RUN".green().bold().to_string()
1242        } else {
1243            "OFF".red().to_string()
1244        };
1245        let unsynced_cell = if unsynced > 0 {
1246            unsynced.to_string().yellow().to_string()
1247        } else {
1248            "0".bright_black().to_string()
1249        };
1250        let age_cell = if age == "never" {
1251            age.bright_black().to_string()
1252        } else {
1253            age.cyan().to_string()
1254        };
1255
1256        rows.push(vec![
1257            format!("{owner}/{name}"),
1258            branch.to_string(),
1259            state,
1260            pid,
1261            unsynced_cell,
1262            last_sync_at,
1263            age_cell,
1264        ]);
1265    }
1266
1267    print!(
1268        "{}",
1269        render_table(&headers, &rows, TableStyle::Box, "")
1270    );
1271
1272    let running = snapshot.running_watchers;
1273    let stopped = snapshot
1274        .repository_count
1275        .saturating_sub(snapshot.running_watchers);
1276    println!();
1277    println!(
1278        "{} {} running · {} stopped · {} total",
1279        "summary".bright_black(),
1280        running.to_string().green().bold(),
1281        stopped.to_string().red(),
1282        snapshot.repository_count
1283    );
1284}
1285
1286fn last_sync_display(payload: &Value) -> (String, String) {
1287    let Some(last_sync) = payload.get("lastSync") else {
1288        return ("—".to_string(), "never".to_string());
1289    };
1290    let synced_at = last_sync
1291        .get("syncedAt")
1292        .and_then(Value::as_str)
1293        .unwrap_or("unknown");
1294    let age = if let Some(seconds) = payload
1295        .get("lastSyncSecondsAgo")
1296        .and_then(Value::as_u64)
1297    {
1298        format_age_seconds(seconds)
1299    } else if let Ok(dt) = DateTime::parse_from_rfc3339(synced_at) {
1300        let seconds = (Utc::now() - dt.with_timezone(&Utc)).num_seconds().max(0) as u64;
1301        format_age_seconds(seconds)
1302    } else {
1303        "unknown".to_string()
1304    };
1305    // Compact timestamp for the matrix column.
1306    let compact = if synced_at.len() >= 19 {
1307        synced_at[11..19].to_string()
1308    } else {
1309        synced_at.to_string()
1310    };
1311    (compact, age)
1312}
1313
1314fn format_age_seconds(seconds: u64) -> String {
1315    if seconds < 60 {
1316        format!("{seconds}s ago")
1317    } else if seconds < 3600 {
1318        format!("{}m ago", seconds / 60)
1319    } else if seconds < 86400 {
1320        format!("{}h ago", seconds / 3600)
1321    } else {
1322        format!("{}d ago", seconds / 86400)
1323    }
1324}
1325
1326fn attach_optional_stats(
1327    mut payload: Value,
1328    stats: Option<WorktreeWatchStatsSummary>,
1329    repo_activity: Option<WorktreeWatchRepoActivitySummary>,
1330) -> Result<Value, String> {
1331    if let Some(object) = payload.as_object_mut() {
1332        if let Some(stats) = stats {
1333            object.insert(
1334                "stats".to_string(),
1335                serde_json::to_value(stats)
1336                    .map_err(|error| format!("Failed to render stats payload: {error}"))?,
1337            );
1338        }
1339        if let Some(repo_activity) = repo_activity {
1340            object.insert(
1341                "repositoryActivity".to_string(),
1342                serde_json::to_value(repo_activity)
1343                    .map_err(|error| format!("Failed to render repo activity payload: {error}"))?,
1344            );
1345        }
1346    }
1347    Ok(payload)
1348}
1349
1350pub fn spawn_detached_worktree_watch_for_current_repo() -> Result<Option<PathBuf>, String> {
1351    if is_background_child() {
1352        return Ok(None);
1353    }
1354
1355    let identity = match resolve_repo_identity(None) {
1356        Ok(identity) => identity,
1357        Err(_) => return Ok(None),
1358    };
1359    spawn_detached_worktree_watch_for_identity(&identity).map(Some)
1360}
1361
1362/// Snapshot of worktree-watch storage and watcher state for `xbp diag`.
1363#[derive(Debug, Clone, Serialize, Deserialize)]
1364#[serde(rename_all = "camelCase")]
1365pub struct WorktreeWatchDiagnostic {
1366    pub global_xbp_root: String,
1367    pub mutations_root: String,
1368    pub runtime_key: String,
1369    pub platform: String,
1370    pub is_wsl: bool,
1371    pub repo_owner: Option<String>,
1372    pub repo_name: Option<String>,
1373    pub branch: Option<String>,
1374    pub branch_spool: Option<String>,
1375    pub branch_spool_exists: bool,
1376    pub watcher_state_path: Option<String>,
1377    pub watcher_state_present: bool,
1378    pub watcher_running: bool,
1379    pub event_jsonl_files: usize,
1380    pub commit_jsonl_files: usize,
1381    pub alternate_spool_roots: Vec<String>,
1382    pub notes: Vec<String>,
1383}
1384
1385/// Lightweight mutation activity for TODO→issue effort attribution.
1386#[derive(Debug, Clone)]
1387pub struct MutationActivityEvent {
1388    pub paths: Vec<String>,
1389    pub primary_path: Option<String>,
1390    pub occurred_at: DateTime<Utc>,
1391    pub added_lines: Option<u64>,
1392    pub removed_lines: Option<u64>,
1393    pub head_sha: Option<String>,
1394    pub branch_name: String,
1395}
1396
1397/// Lightweight commit activity for TODO→issue effort attribution.
1398#[derive(Debug, Clone)]
1399pub struct CommitActivityEvent {
1400    pub occurred_at: DateTime<Utc>,
1401    pub head_sha: String,
1402    pub subject: Option<String>,
1403    pub branch_name: String,
1404    pub repo_root: String,
1405}
1406
1407/// Load all spooled mutation events for a GitHub `owner/name` across branch folders.
1408pub fn load_repo_mutation_activity(
1409    owner: &str,
1410    name: &str,
1411) -> Result<Vec<MutationActivityEvent>, String> {
1412    let identity = RepoIdentity {
1413        owner: owner.to_string(),
1414        name: name.to_string(),
1415        branch: "main".to_string(),
1416        root: PathBuf::from("."),
1417        head_sha: None,
1418    };
1419    let mut events = Vec::new();
1420    for repo_root in repository_spool_search_roots(&identity)? {
1421        if !repo_root.is_dir() {
1422            continue;
1423        }
1424        for entry in fs::read_dir(&repo_root).map_err(|e| {
1425            format!(
1426                "Failed to read mutation spool {}: {e}",
1427                repo_root.display()
1428            )
1429        })? {
1430            let entry = entry.map_err(|e| format!("Failed to read spool entry: {e}"))?;
1431            if !entry.file_type().map(|t| t.is_dir()).unwrap_or(false) {
1432                continue;
1433            }
1434            let branch_layout = spool_layout_for_existing_root(&entry.path());
1435            let candidates = collect_spool_candidates(&branch_layout, true)?;
1436            for candidate in candidates {
1437                if let SyncRecords::Events(records) = candidate.records {
1438                    for record in records {
1439                        events.push(MutationActivityEvent {
1440                            paths: record.paths,
1441                            primary_path: record.primary_path,
1442                            occurred_at: record.occurred_at,
1443                            added_lines: record.added_lines,
1444                            removed_lines: record.removed_lines,
1445                            head_sha: record.head_sha,
1446                            branch_name: record.branch_name,
1447                        });
1448                    }
1449                }
1450            }
1451        }
1452    }
1453    events.sort_by_key(|e| e.occurred_at);
1454    Ok(events)
1455}
1456
1457/// Load all spooled commit events for a GitHub `owner/name`.
1458pub fn load_repo_commit_activity(
1459    owner: &str,
1460    name: &str,
1461) -> Result<Vec<CommitActivityEvent>, String> {
1462    let identity = RepoIdentity {
1463        owner: owner.to_string(),
1464        name: name.to_string(),
1465        branch: "main".to_string(),
1466        root: PathBuf::from("."),
1467        head_sha: None,
1468    };
1469    let mut commits = Vec::new();
1470    for repo_root in repository_spool_search_roots(&identity)? {
1471        if !repo_root.is_dir() {
1472            continue;
1473        }
1474        for entry in fs::read_dir(&repo_root).map_err(|e| {
1475            format!(
1476                "Failed to read mutation spool {}: {e}",
1477                repo_root.display()
1478            )
1479        })? {
1480            let entry = entry.map_err(|e| format!("Failed to read spool entry: {e}"))?;
1481            if !entry.file_type().map(|t| t.is_dir()).unwrap_or(false) {
1482                continue;
1483            }
1484            let branch_layout = spool_layout_for_existing_root(&entry.path());
1485            let candidates = collect_spool_candidates(&branch_layout, true)?;
1486            for candidate in candidates {
1487                if let SyncRecords::Commits(records) = candidate.records {
1488                    for record in records {
1489                        commits.push(CommitActivityEvent {
1490                            occurred_at: record.occurred_at,
1491                            head_sha: record.head_sha,
1492                            subject: record.subject,
1493                            branch_name: record.branch_name,
1494                            repo_root: record.repo_root,
1495                        });
1496                    }
1497                }
1498            }
1499        }
1500    }
1501    commits.sort_by_key(|c| c.occurred_at);
1502    Ok(commits)
1503}
1504
1505/// Collect home-level worktree-watch paths and optional current-repo spool status.
1506pub fn collect_worktree_watch_diagnostic() -> WorktreeWatchDiagnostic {
1507    let runtime_key = worktree_runtime_key();
1508    let platform = std::env::consts::OS.to_string();
1509    let is_wsl = is_wsl_runtime();
1510    let mut notes = Vec::new();
1511
1512    let global_xbp_root = match ensure_global_xbp_paths() {
1513        Ok(paths) => paths.root_dir,
1514        Err(error) => {
1515            notes.push(format!("Failed to resolve global XBP home: {error}"));
1516            resolve_global_xbp_root_dir()
1517        }
1518    };
1519    let mutations_root = global_xbp_root.join("mutations");
1520
1521    let identity = resolve_repo_identity(None).ok();
1522    let (
1523        repo_owner,
1524        repo_name,
1525        branch,
1526        branch_spool,
1527        branch_spool_exists,
1528        watcher_state_path,
1529        watcher_state_present,
1530        watcher_running,
1531        event_jsonl_files,
1532        commit_jsonl_files,
1533        alternate_spool_roots,
1534    ) = if let Some(identity) = identity.as_ref() {
1535        let layout = prepare_spool_layout(identity).ok();
1536        let branch_spool = layout
1537            .as_ref()
1538            .map(|value| value.root.display().to_string());
1539        let branch_spool_exists = layout
1540            .as_ref()
1541            .map(|value| value.root.exists())
1542            .unwrap_or(false);
1543        let watcher_state_path = layout
1544            .as_ref()
1545            .map(|value| value.watcher_state_file.display().to_string());
1546        let watcher_state_present = layout
1547            .as_ref()
1548            .map(|value| {
1549                watcher_state_read_candidates(&value.watcher_state_file)
1550                    .iter()
1551                    .any(|candidate| candidate.exists())
1552            })
1553            .unwrap_or(false);
1554        let watcher_running = layout
1555            .as_ref()
1556            .and_then(|value| read_watcher_state(&value.watcher_state_file).ok().flatten())
1557            .map(|state| is_matching_watcher_process(&state))
1558            .unwrap_or(false);
1559
1560        let mut event_jsonl_files = 0usize;
1561        let mut commit_jsonl_files = 0usize;
1562        if let Some(layout) = layout.as_ref() {
1563            if let Ok(entries) = fs::read_dir(&layout.root) {
1564                for entry in entries.flatten() {
1565                    let name = entry.file_name().to_string_lossy().to_string();
1566                    if name.starts_with("events-") && name.ends_with(".jsonl") {
1567                        event_jsonl_files += 1;
1568                    } else if name.starts_with("commits-") && name.ends_with(".jsonl") {
1569                        commit_jsonl_files += 1;
1570                    }
1571                }
1572            }
1573        }
1574
1575        let alternate_spool_roots = repository_spool_search_roots(identity)
1576            .unwrap_or_default()
1577            .into_iter()
1578            .map(|path| {
1579                path.join(sanitize_path_component(&identity.branch))
1580                    .display()
1581                    .to_string()
1582            })
1583            .filter(|path| {
1584                layout
1585                    .as_ref()
1586                    .map(|value| path_identity_key(Path::new(path)) != path_identity_key(&value.root))
1587                    .unwrap_or(true)
1588            })
1589            .filter(|path| Path::new(path).exists())
1590            .collect::<Vec<_>>();
1591
1592        let root_display = global_xbp_root
1593            .display()
1594            .to_string()
1595            .replace('\\', "/")
1596            .to_ascii_lowercase();
1597        if is_wsl && !root_display.starts_with("/mnt/") {
1598            notes.push(
1599                "WSL global XBP home is not on a Windows drive mount. If Windows has mutations under C:\\Users\\…\\.xbp, set XBP_HOME=/mnt/c/Users/<you>/.xbp so status/sync share that spool."
1600                    .to_string(),
1601            );
1602        }
1603        if is_wsl && root_display.starts_with("/mnt/") {
1604            notes.push(
1605                "Using a Windows-profile-backed XBP home from WSL; spoils are reconcilable with native Windows."
1606                    .to_string(),
1607            );
1608        }
1609        if !alternate_spool_roots.is_empty() {
1610            notes.push(format!(
1611                "Found {} alternate spool location(s) with historical data (legacy homes / runtime layouts).",
1612                alternate_spool_roots.len()
1613            ));
1614        }
1615
1616        (
1617            Some(identity.owner.clone()),
1618            Some(identity.name.clone()),
1619            Some(identity.branch.clone()),
1620            branch_spool,
1621            branch_spool_exists,
1622            watcher_state_path,
1623            watcher_state_present,
1624            watcher_running,
1625            event_jsonl_files,
1626            commit_jsonl_files,
1627            alternate_spool_roots,
1628        )
1629    } else {
1630        notes.push(
1631            "Not inside a git repository with a resolvable remote; showing global XBP home only."
1632                .to_string(),
1633        );
1634        (
1635            None,
1636            None,
1637            None,
1638            None,
1639            false,
1640            None,
1641            false,
1642            false,
1643            0,
1644            0,
1645            Vec::new(),
1646        )
1647    };
1648
1649    WorktreeWatchDiagnostic {
1650        global_xbp_root: global_xbp_root.display().to_string(),
1651        mutations_root: mutations_root.display().to_string(),
1652        runtime_key,
1653        platform,
1654        is_wsl,
1655        repo_owner,
1656        repo_name,
1657        branch,
1658        branch_spool,
1659        branch_spool_exists,
1660        watcher_state_path,
1661        watcher_state_present,
1662        watcher_running,
1663        event_jsonl_files,
1664        commit_jsonl_files,
1665        alternate_spool_roots,
1666        notes,
1667    }
1668}
1669
1670fn resolve_target_identities(
1671    target: &WorktreeWatchTargetOptions,
1672) -> Result<Vec<RepoIdentity>, String> {
1673    if target.repo.is_some() && target.parent.is_some() {
1674        return Err("Pass either `--repo` or `--parent`, not both.".to_string());
1675    }
1676    if !target.repos.is_empty() && target.parent.is_some() {
1677        return Err("Pass either `--repos` or `--parent`, not both.".to_string());
1678    }
1679    if target.repo.is_some() && !target.repos.is_empty() {
1680        return Err("Pass either `--repo` or `--repos`, not both.".to_string());
1681    }
1682
1683    if let Some(parent) = target.parent.as_deref() {
1684        return discover_repo_identities_under(parent);
1685    }
1686
1687    if !target.repos.is_empty() {
1688        let mut identities = Vec::new();
1689        let mut seen = BTreeSet::new();
1690        for repo in &target.repos {
1691            let identity = resolve_repo_identity(Some(repo.as_path()))?;
1692            let key = path_identity_key(&identity.root);
1693            if seen.insert(key) {
1694                identities.push(identity);
1695            }
1696        }
1697        return Ok(identities);
1698    }
1699
1700    resolve_repo_identity(target.repo.as_deref()).map(|identity| vec![identity])
1701}
1702
1703/// Collect a tray-friendly snapshot of watcher state + unsynced backlog.
1704pub fn collect_worktree_watch_tray_snapshot(
1705    target: &WorktreeWatchTargetOptions,
1706) -> Result<WorktreeWatchTraySnapshot, String> {
1707    let identities = resolve_target_identities(target)?;
1708    let mode = if target.parent.is_some() {
1709        "parent"
1710    } else if !target.repos.is_empty() {
1711        "repos"
1712    } else {
1713        "repo"
1714    }
1715    .to_string();
1716
1717    let parent_label = target
1718        .parent
1719        .as_ref()
1720        .map(|path| path.display().to_string());
1721    let target_label = if let Some(parent) = parent_label.as_ref() {
1722        format!("parent:{parent}")
1723    } else if !target.repos.is_empty() {
1724        format!("{} repo(s)", target.repos.len())
1725    } else if let Some(repo) = target.repo.as_ref() {
1726        repo.display().to_string()
1727    } else {
1728        identities
1729            .first()
1730            .map(|identity| identity.root.display().to_string())
1731            .unwrap_or_else(|| "current repo".to_string())
1732    };
1733
1734    // Collect PIDs first so we only refresh those processes once (System::new_all
1735    // per repo was freezing status for large parent folders on Windows).
1736    let mut parent_state: Option<ParentWorktreeWatcherState> = None;
1737    if let Some(parent) = target.parent.as_deref() {
1738        if let Ok(parent) = canonical_parent_path(parent) {
1739            if let Ok(layout) = prepare_parent_spool_layout(&parent) {
1740                parent_state = read_parent_watcher_state(&layout.watcher_state_file)?;
1741            }
1742        }
1743    }
1744
1745    let mut repo_states: Vec<(usize, WorktreeWatcherState)> = Vec::new();
1746    let mut layouts: Vec<SpoolLayout> = Vec::with_capacity(identities.len());
1747    for (index, identity) in identities.iter().enumerate() {
1748        let layout = prepare_spool_layout(identity)?;
1749        if let Some(state) = read_watcher_state(&layout.watcher_state_file)? {
1750            if same_repo_watcher_state(identity, &state) {
1751                repo_states.push((index, state));
1752            }
1753        }
1754        layouts.push(layout);
1755    }
1756
1757    let mut pids: Vec<u32> = repo_states.iter().map(|(_, state)| state.pid).collect();
1758    if let Some(state) = parent_state.as_ref() {
1759        pids.push(state.pid);
1760    }
1761    let process_system = process_system_for_pids(&pids);
1762
1763    let mut parent_watcher_running = false;
1764    let mut parent_watcher_pid = None;
1765    if let Some(state) = parent_state.as_ref() {
1766        // Include self: the loopback API runs inside the watcher process, so
1767        // `state.pid == std::process::id()` must count as running (the
1768        // "other process" matcher intentionally excludes self for stop/replace).
1769        if is_live_parent_watcher_process_with_system(&process_system, state) {
1770            parent_watcher_running = true;
1771            parent_watcher_pid = Some(state.pid);
1772        }
1773    }
1774
1775    let mut repositories = Vec::new();
1776    let mut running_watchers = 0usize;
1777    let mut total_unsynced_files = 0u64;
1778    let mut total_unsynced_records = 0u64;
1779
1780    for (index, identity) in identities.iter().enumerate() {
1781        let layout = &layouts[index];
1782        // Line counts only — never fully parse/sanitize JSONL on the hot path.
1783        let counts = count_spool_summary(layout)?;
1784        let unsynced_files = counts.unsynced_files;
1785        let unsynced_records = counts.unsynced_records;
1786        total_unsynced_files += unsynced_files;
1787        total_unsynced_records += unsynced_records;
1788
1789        let (running, pid) = match repo_states
1790            .iter()
1791            .find(|(repo_index, _)| *repo_index == index)
1792            .map(|(_, state)| state)
1793        {
1794            Some(state) if is_live_watcher_process_with_system(&process_system, state) => {
1795                (true, Some(state.pid))
1796            }
1797            _ => (false, None),
1798        };
1799        // Parent watcher covers repos under parent even without per-repo state.
1800        let running = running || parent_watcher_running;
1801        let pid = pid.or(if running {
1802            parent_watcher_pid
1803        } else {
1804            None
1805        });
1806        if running {
1807            running_watchers += 1;
1808        }
1809
1810        repositories.push(WorktreeWatchTrayRepoStatus {
1811            root: normalize_repo_root_for_payload(&identity.root),
1812            owner: identity.owner.clone(),
1813            name: identity.name.clone(),
1814            branch: identity.branch.clone(),
1815            running,
1816            pid,
1817            unsynced_files,
1818            unsynced_records,
1819        });
1820    }
1821
1822    let repository_count = repositories.len();
1823    let any_running = parent_watcher_running || running_watchers > 0;
1824    let all_running = repository_count > 0 && running_watchers >= repository_count;
1825
1826    // Lightweight summary for tooltips; full stats are computed on demand via the tray menu.
1827    let stats_line = Some(format!(
1828        "{repository_count} repo(s) · {running_watchers} watching · {total_unsynced_records} unsynced · {total_unsynced_files} file(s)"
1829    ));
1830
1831    Ok(WorktreeWatchTraySnapshot {
1832        target_label,
1833        mode,
1834        parent: parent_label,
1835        repository_count,
1836        running_watchers,
1837        any_running,
1838        all_running,
1839        parent_watcher_running,
1840        parent_watcher_pid,
1841        total_unsynced_files,
1842        total_unsynced_records,
1843        repositories,
1844        stats_line,
1845    })
1846}
1847
1848/// Default session gap used when the local HTTP API recomputes coding stats.
1849pub const WORKTREE_WATCH_API_DEFAULT_STATS_GAP_MINUTES: u64 = 30;
1850
1851/// Build the JSON body for `GET /api/worktree-watch/status` (tray snapshot + per-repo status).
1852pub fn collect_worktree_watch_api_status(
1853    target: &WorktreeWatchTargetOptions,
1854    last_error: Option<&str>,
1855) -> Result<Value, String> {
1856    let snapshot = collect_worktree_watch_tray_snapshot(target)?;
1857    let mut repositories = Vec::new();
1858
1859    for tray_repo in &snapshot.repositories {
1860        // Reuse tray counts; attach lastSync without re-scanning spool JSONL bodies.
1861        let mut status = worktree_watch_status_from_tray_repo(tray_repo)?;
1862        if let Some(object) = status.as_object_mut() {
1863            object.insert("running".to_string(), json!(tray_repo.running));
1864            object.insert("pid".to_string(), json!(tray_repo.pid));
1865            object.insert("owner".to_string(), json!(tray_repo.owner));
1866            object.insert("name".to_string(), json!(tray_repo.name));
1867            object.insert("branch".to_string(), json!(tray_repo.branch));
1868            object.insert("root".to_string(), json!(tray_repo.root));
1869            if let Some(spool) = object.get("spoolRoot").cloned() {
1870                object.insert("spoolPath".to_string(), spool);
1871            }
1872            object.insert("lastError".to_string(), Value::Null);
1873            if let Some(last_sync) = object.get("lastSync") {
1874                if let Some(synced_at) = last_sync.get("syncedAt").and_then(Value::as_str) {
1875                    if let Ok(dt) = DateTime::parse_from_rfc3339(synced_at) {
1876                        let age = (Utc::now() - dt.with_timezone(&Utc))
1877                            .num_seconds()
1878                            .max(0) as u64;
1879                        object.insert("lastSyncSecondsAgo".to_string(), json!(age));
1880                    }
1881                }
1882            }
1883        }
1884        repositories.push(status);
1885    }
1886
1887    Ok(json!({
1888        "targetLabel": snapshot.target_label,
1889        "mode": snapshot.mode,
1890        "parent": snapshot.parent,
1891        "repositoryCount": snapshot.repository_count,
1892        "runningWatchers": snapshot.running_watchers,
1893        "anyRunning": snapshot.any_running,
1894        "allRunning": snapshot.all_running,
1895        "parentWatcherRunning": snapshot.parent_watcher_running,
1896        "parentWatcherPid": snapshot.parent_watcher_pid,
1897        "totalUnsyncedFiles": snapshot.total_unsynced_files,
1898        "totalUnsyncedRecords": snapshot.total_unsynced_records,
1899        "unsyncedFiles": snapshot.total_unsynced_files,
1900        "unsyncedRecords": snapshot.total_unsynced_records,
1901        "statsLine": snapshot.stats_line,
1902        "lastError": last_error,
1903        "repositories": repositories,
1904    }))
1905}
1906
1907fn identity_from_tray_repo(repo: &WorktreeWatchTrayRepoStatus) -> RepoIdentity {
1908    RepoIdentity {
1909        owner: repo.owner.clone(),
1910        name: repo.name.clone(),
1911        branch: repo.branch.clone(),
1912        root: parent_root_as_local_path(&repo.root),
1913        head_sha: None,
1914    }
1915}
1916
1917fn worktree_watch_status_from_tray_repo(
1918    repo: &WorktreeWatchTrayRepoStatus,
1919) -> Result<Value, String> {
1920    let identity = identity_from_tray_repo(repo);
1921    let layout = prepare_spool_layout(&identity)?;
1922    let mut payload = json!({
1923        "repoRoot": repo.root,
1924        "repositoryOwner": repo.owner,
1925        "repositoryName": repo.name,
1926        "branchName": repo.branch,
1927        "platform": std::env::consts::OS,
1928        "runtimeKey": worktree_runtime_key(),
1929        "spoolRoot": layout.root.display().to_string(),
1930        "unsyncedFiles": repo.unsynced_files,
1931        "unsyncedRecords": repo.unsynced_records,
1932        "syncedFiles": 0,
1933        "syncedRecords": 0,
1934        "localSpoolFiles": repo.unsynced_files,
1935        "localSpoolRecords": repo.unsynced_records,
1936        "running": repo.running,
1937        "pid": repo.pid,
1938    });
1939
1940    if let Some(last_sync) = read_last_sync_log_entry(&layout.sync_log_file)? {
1941        if let Some(object) = payload.as_object_mut() {
1942            object.insert(
1943                "lastSync".to_string(),
1944                serde_json::to_value(last_sync)
1945                    .map_err(|error| format!("Failed to render sync log payload: {error}"))?,
1946            );
1947        }
1948    }
1949
1950    Ok(payload)
1951}
1952
1953/// Build the JSON body for `GET /api/worktree-watch/repositories`.
1954pub fn collect_worktree_watch_api_repositories(
1955    target: &WorktreeWatchTargetOptions,
1956) -> Result<Value, String> {
1957    let status = collect_worktree_watch_api_status(target, None)?;
1958    let repositories = status
1959        .get("repositories")
1960        .cloned()
1961        .unwrap_or_else(|| json!([]));
1962    Ok(json!({ "repositories": repositories }))
1963}
1964
1965/// Build the JSON body for `GET /api/worktree-watch/stats`.
1966///
1967/// Prefers cached `stats.json` when fresh so the dashboard poll stays cheap;
1968/// recomputes on miss/stale (same shape as `xbp worktree-watch status --stats`).
1969pub fn collect_worktree_watch_api_stats(
1970    target: &WorktreeWatchTargetOptions,
1971    session_gap_minutes: u64,
1972) -> Result<Value, String> {
1973    let identities = resolve_target_identities(target)?;
1974    if identities.is_empty() {
1975        return Ok(json!({
1976            "generatedAt": null,
1977            "message": "no stats yet",
1978            "repositories": [],
1979        }));
1980    }
1981
1982    let gap = if session_gap_minutes == 0 {
1983        WORKTREE_WATCH_API_DEFAULT_STATS_GAP_MINUTES
1984    } else {
1985        session_gap_minutes
1986    };
1987
1988    let mut repositories = Vec::new();
1989    for identity in &identities {
1990        let stats = read_or_generate_stats(identity, gap, Duration::from_secs(120))?;
1991        repositories.push(json!({
1992            "owner": identity.owner,
1993            "name": identity.name,
1994            "branch": identity.branch,
1995            "repositoryOwner": identity.owner,
1996            "repositoryName": identity.name,
1997            "branchName": identity.branch,
1998            "repoRoot": normalize_repo_root_for_payload(&identity.root),
1999            "root": normalize_repo_root_for_payload(&identity.root),
2000            "stats": stats,
2001        }));
2002    }
2003
2004    if repositories.len() == 1 {
2005        let first = repositories.remove(0);
2006        let stats = first
2007            .get("stats")
2008            .cloned()
2009            .unwrap_or(Value::Null);
2010        // Flat WorktreeWatchStatsSummary for single-repo targets (dashboard accepts both).
2011        if let Some(mut object) = stats.as_object().cloned() {
2012            object.insert(
2013                "repositories".to_string(),
2014                json!([{
2015                    "owner": first.get("owner"),
2016                    "name": first.get("name"),
2017                    "branch": first.get("branch"),
2018                    "stats": first.get("stats"),
2019                }]),
2020            );
2021            return Ok(Value::Object(object));
2022        }
2023        return Ok(first);
2024    }
2025
2026    Ok(json!({
2027        "repositories": repositories,
2028        "repositoryCount": repositories.len(),
2029    }))
2030}
2031
2032/// Fast status payload: spool file/line counts + lastSync, no full JSONL decode.
2033fn worktree_watch_status_payload_light(identity: &RepoIdentity) -> Result<Value, String> {
2034    let layout = prepare_spool_layout(identity)?;
2035    let counts = count_spool_summary(&layout)?;
2036
2037    let mut payload = json!({
2038        "repoRoot": normalize_repo_root_for_payload(&identity.root),
2039        "repositoryOwner": identity.owner.clone(),
2040        "repositoryName": identity.name.clone(),
2041        "branchName": identity.branch.clone(),
2042        "platform": std::env::consts::OS,
2043        "runtimeKey": worktree_runtime_key(),
2044        "spoolRoot": layout.root.display().to_string(),
2045        "unsyncedFiles": counts.unsynced_files,
2046        "unsyncedRecords": counts.unsynced_records,
2047        "syncedFiles": counts.synced_files,
2048        "syncedRecords": counts.synced_records,
2049        "localSpoolFiles": counts.unsynced_files + counts.synced_files,
2050        "localSpoolRecords": counts.unsynced_records + counts.synced_records,
2051    });
2052
2053    if let Some(last_sync) = read_last_sync_log_entry(&layout.sync_log_file)? {
2054        if let Some(object) = payload.as_object_mut() {
2055            object.insert(
2056                "lastSync".to_string(),
2057                serde_json::to_value(last_sync)
2058                    .map_err(|error| format!("Failed to render sync log payload: {error}"))?,
2059            );
2060        }
2061    }
2062
2063    Ok(payload)
2064}
2065
2066fn worktree_watch_status_payload(
2067    identity: &RepoIdentity,
2068    include_records: bool,
2069    record_limit: usize,
2070) -> Result<Value, String> {
2071    // Counts are always available via the light path; only decode JSONL when
2072    // the caller needs individual records.
2073    if !include_records {
2074        return worktree_watch_status_payload_light(identity);
2075    }
2076
2077    let layout = prepare_spool_layout(identity)?;
2078    let candidates = collect_sync_candidates(&layout)?;
2079    let all_candidates = collect_spool_candidates(&layout, true)?;
2080    let record_count: usize = candidates
2081        .iter()
2082        .map(|candidate| candidate.records.len())
2083        .sum();
2084    let all_record_count: usize = all_candidates
2085        .iter()
2086        .map(|candidate| candidate.records.len())
2087        .sum();
2088    let synced_file_count = all_candidates.len().saturating_sub(candidates.len());
2089    let synced_record_count = all_record_count.saturating_sub(record_count);
2090
2091    let mut payload = json!({
2092        "repoRoot": normalize_repo_root_for_payload(&identity.root),
2093        "repositoryOwner": identity.owner.clone(),
2094        "repositoryName": identity.name.clone(),
2095        "branchName": identity.branch.clone(),
2096        "platform": std::env::consts::OS,
2097        "runtimeKey": worktree_runtime_key(),
2098        "spoolRoot": layout.root.display().to_string(),
2099        "unsyncedFiles": candidates.len(),
2100        "unsyncedRecords": record_count,
2101        "syncedFiles": synced_file_count,
2102        "syncedRecords": synced_record_count,
2103        "localSpoolFiles": all_candidates.len(),
2104        "localSpoolRecords": all_record_count,
2105    });
2106
2107    if let Some(last_sync) = read_last_sync_log_entry(&layout.sync_log_file)? {
2108        if let Some(object) = payload.as_object_mut() {
2109            object.insert(
2110                "lastSync".to_string(),
2111                serde_json::to_value(last_sync)
2112                    .map_err(|error| format!("Failed to render sync log payload: {error}"))?,
2113            );
2114        }
2115    }
2116
2117    if let Some(object) = payload.as_object_mut() {
2118        object.insert(
2119            "records".to_string(),
2120            collect_status_records(&candidates, record_limit)?,
2121        );
2122    }
2123
2124    Ok(payload)
2125}
2126
2127#[derive(Debug, Default, Clone, Copy)]
2128struct SpoolCountSummary {
2129    unsynced_files: u64,
2130    unsynced_records: u64,
2131    synced_files: u64,
2132    synced_records: u64,
2133}
2134
2135/// Count spool JSONL files and non-empty lines without parsing or rewriting.
2136fn count_spool_summary(layout: &SpoolLayout) -> Result<SpoolCountSummary, String> {
2137    let mut roots = vec![layout.root.clone()];
2138    if let Some(branch) = layout.root.file_name().and_then(|name| name.to_str()) {
2139        if let Some(repo_root) = layout.root.parent() {
2140            if let (Some(name), Some(owner)) = (
2141                repo_root.file_name().and_then(|n| n.to_str()),
2142                repo_root
2143                    .parent()
2144                    .and_then(|p| p.file_name())
2145                    .and_then(|n| n.to_str()),
2146            ) {
2147                let identity = RepoIdentity {
2148                    owner: owner.to_string(),
2149                    name: name.to_string(),
2150                    branch: branch.to_string(),
2151                    root: PathBuf::new(),
2152                    head_sha: None,
2153                };
2154                if let Ok(search_roots) = repository_spool_search_roots(&identity) {
2155                    for search_root in search_roots {
2156                        let branch_root = search_root.join(branch);
2157                        if !roots.iter().any(|existing| {
2158                            path_identity_key(existing) == path_identity_key(&branch_root)
2159                        }) {
2160                            roots.push(branch_root);
2161                        }
2162                    }
2163                }
2164            }
2165        }
2166    }
2167
2168    let mut summary = SpoolCountSummary::default();
2169    let mut seen_files = BTreeSet::new();
2170    for root in roots {
2171        if !root.exists() {
2172            continue;
2173        }
2174        for entry in fs::read_dir(&root).map_err(|error| {
2175            format!(
2176                "Failed to read spool directory {}: {error}",
2177                root.display()
2178            )
2179        })? {
2180            let path = entry
2181                .map_err(|error| format!("Failed to read spool entry: {error}"))?
2182                .path();
2183            let Some(file_name) = path.file_name().and_then(|value| value.to_str()) else {
2184                continue;
2185            };
2186            if !file_name.ends_with(".jsonl") {
2187                continue;
2188            }
2189            if !(file_name.starts_with("events-") || file_name.starts_with("commits-")) {
2190                continue;
2191            }
2192            let identity_key = path_identity_key(&path);
2193            if !seen_files.insert(identity_key) {
2194                continue;
2195            }
2196            let lines = count_nonempty_lines(&path)?;
2197            if file_name.contains(".synced.") {
2198                summary.synced_files += 1;
2199                summary.synced_records += lines;
2200            } else {
2201                summary.unsynced_files += 1;
2202                summary.unsynced_records += lines;
2203            }
2204        }
2205    }
2206    Ok(summary)
2207}
2208
2209fn count_nonempty_lines(path: &Path) -> Result<u64, String> {
2210    let file = File::open(path)
2211        .map_err(|error| format!("Failed to open spool file {}: {error}", path.display()))?;
2212    let reader = BufReader::new(file);
2213    let mut count = 0u64;
2214    for line in reader.lines() {
2215        let line =
2216            line.map_err(|error| format!("Failed to read spool file {}: {error}", path.display()))?;
2217        if !line.trim().is_empty() {
2218            count += 1;
2219        }
2220    }
2221    Ok(count)
2222}
2223
2224fn read_or_generate_stats(
2225    identity: &RepoIdentity,
2226    session_gap_minutes: u64,
2227    max_age: Duration,
2228) -> Result<WorktreeWatchStatsSummary, String> {
2229    let layout = prepare_spool_layout(identity)?;
2230    if layout.stats_file.exists() {
2231        if let Ok(raw) = fs::read_to_string(&layout.stats_file) {
2232            if let Ok(stats) = serde_json::from_str::<WorktreeWatchStatsSummary>(&raw) {
2233                let age = Utc::now()
2234                    .signed_duration_since(stats.generated_at)
2235                    .to_std()
2236                    .unwrap_or(Duration::from_secs(u64::MAX));
2237                if age <= max_age {
2238                    return Ok(stats);
2239                }
2240            }
2241        }
2242    }
2243    generate_and_store_stats(identity, session_gap_minutes)
2244}
2245
2246fn collect_status_records(candidates: &[SyncCandidate], limit: usize) -> Result<Value, String> {
2247    let available: usize = candidates
2248        .iter()
2249        .map(|candidate| candidate.records.len())
2250        .sum();
2251    let mut shown = 0usize;
2252    let mut events = Vec::new();
2253    let mut commits = Vec::new();
2254
2255    'candidates: for candidate in candidates {
2256        match &candidate.records {
2257            SyncRecords::Events(records) => {
2258                for event in records {
2259                    if shown >= limit {
2260                        break 'candidates;
2261                    }
2262                    events.push(json!({
2263                        "sourceFile": candidate.path.display().to_string(),
2264                        "occurredAt": event.occurred_at,
2265                        "eventKind": event.event_kind.clone(),
2266                        "primaryPath": event.primary_path.clone(),
2267                        "oldPath": event.old_path.clone(),
2268                        "newPath": event.new_path.clone(),
2269                        "paths": event.paths.clone(),
2270                        "addedLines": event.added_lines,
2271                        "removedLines": event.removed_lines,
2272                        "totalLines": event.total_lines,
2273                        "headSha": event.head_sha.clone(),
2274                        "rawKind": event.raw_kind.clone(),
2275                    }));
2276                    shown += 1;
2277                }
2278            }
2279            SyncRecords::Commits(records) => {
2280                for commit in records {
2281                    if shown >= limit {
2282                        break 'candidates;
2283                    }
2284                    commits.push(json!({
2285                        "sourceFile": candidate.path.display().to_string(),
2286                        "occurredAt": commit.occurred_at,
2287                        "headSha": commit.head_sha.clone(),
2288                        "previousHeadSha": commit.previous_head_sha.clone(),
2289                        "subject": commit.subject.clone(),
2290                        "authorName": commit.author_name.clone(),
2291                        "authorEmail": commit.author_email.clone(),
2292                        "committedAt": commit.committed_at.clone(),
2293                    }));
2294                    shown += 1;
2295                }
2296            }
2297        }
2298    }
2299
2300    Ok(json!({
2301        "available": available,
2302        "shown": shown,
2303        "truncated": shown < available,
2304        "events": events,
2305        "commits": commits,
2306    }))
2307}
2308
2309fn print_status_records(payload: &Value) {
2310    let Some(records) = payload.get("records") else {
2311        return;
2312    };
2313    let shown = records.get("shown").and_then(Value::as_u64).unwrap_or(0);
2314    if shown == 0 {
2315        println!("records: none");
2316        return;
2317    }
2318
2319    println!("records:");
2320    if let Some(events) = records.get("events").and_then(Value::as_array) {
2321        if !events.is_empty() {
2322            println!("  mutations:");
2323            for event in events {
2324                print_status_event_record(event);
2325            }
2326        }
2327    }
2328    if let Some(commits) = records.get("commits").and_then(Value::as_array) {
2329        if !commits.is_empty() {
2330            println!("  commits:");
2331            for commit in commits {
2332                print_status_commit_record(commit);
2333            }
2334        }
2335    }
2336    if records
2337        .get("truncated")
2338        .and_then(Value::as_bool)
2339        .unwrap_or(false)
2340    {
2341        let available = records
2342            .get("available")
2343            .and_then(Value::as_u64)
2344            .unwrap_or(shown);
2345        println!("  showing {shown} of {available} record(s)");
2346    }
2347}
2348
2349fn print_status_stats(payload: &Value) {
2350    let Some(stats) = payload.get("stats") else {
2351        return;
2352    };
2353    println!("stats:");
2354    println!(
2355        "  coding time: {}",
2356        format_duration(
2357            stats
2358                .get("estimatedCodingSeconds")
2359                .and_then(Value::as_u64)
2360                .unwrap_or(0)
2361        )
2362    );
2363    println!(
2364        "  sessions: {}",
2365        stats
2366            .get("sessionCount")
2367            .and_then(Value::as_u64)
2368            .unwrap_or(0)
2369    );
2370    println!(
2371        "  observed span: {}",
2372        format_duration(
2373            stats
2374                .get("observedSpanSeconds")
2375                .and_then(Value::as_u64)
2376                .unwrap_or(0)
2377        )
2378    );
2379    println!(
2380        "  events: {}",
2381        stats
2382            .get("totalEvents")
2383            .and_then(Value::as_u64)
2384            .unwrap_or(0)
2385    );
2386    println!(
2387        "  commits: {}",
2388        stats
2389            .get("totalCommits")
2390            .and_then(Value::as_u64)
2391            .unwrap_or(0)
2392    );
2393    println!(
2394        "  lines: +{} -{}",
2395        stats.get("addedLines").and_then(Value::as_u64).unwrap_or(0),
2396        stats
2397            .get("removedLines")
2398            .and_then(Value::as_u64)
2399            .unwrap_or(0)
2400    );
2401    if let Some(spool_root) = value_str(stats, "spoolRoot") {
2402        println!(
2403            "  stored: {}",
2404            Path::new(spool_root).join("stats.json").display()
2405        );
2406    }
2407    print_stats_bucket_section(stats, "byFiletype", "  by filetype:", 10);
2408    print_stats_bucket_section(stats, "byEventKind", "  by event kind:", 10);
2409    print_stats_file_section(stats, 10);
2410}
2411
2412fn print_repo_activity(payload: &Value) {
2413    let Some(activity) = payload.get("repositoryActivity") else {
2414        return;
2415    };
2416    println!("repo activity:");
2417    println!(
2418        "  coding time: {}",
2419        format_duration(
2420            activity
2421                .get("estimatedCodingSeconds")
2422                .and_then(Value::as_u64)
2423                .unwrap_or(0)
2424        )
2425    );
2426    println!(
2427        "  observed span: {}",
2428        format_duration(
2429            activity
2430                .get("observedSpanSeconds")
2431                .and_then(Value::as_u64)
2432                .unwrap_or(0)
2433        )
2434    );
2435    println!(
2436        "  branches: {}",
2437        activity
2438            .get("branchCount")
2439            .and_then(Value::as_u64)
2440            .unwrap_or(0)
2441    );
2442    println!(
2443        "  sessions: {}, events: {}, commits: {}, lines: +{} -{}",
2444        activity
2445            .get("sessionCount")
2446            .and_then(Value::as_u64)
2447            .unwrap_or(0),
2448        activity
2449            .get("totalEvents")
2450            .and_then(Value::as_u64)
2451            .unwrap_or(0),
2452        activity
2453            .get("totalCommits")
2454            .and_then(Value::as_u64)
2455            .unwrap_or(0),
2456        activity
2457            .get("addedLines")
2458            .and_then(Value::as_u64)
2459            .unwrap_or(0),
2460        activity
2461            .get("removedLines")
2462            .and_then(Value::as_u64)
2463            .unwrap_or(0)
2464    );
2465    if let Some(stats_file) = value_str(activity, "statsFile") {
2466        println!("  stored: {stats_file}");
2467    }
2468    let Some(branches) = activity.get("branches").and_then(Value::as_array) else {
2469        return;
2470    };
2471    if branches.is_empty() {
2472        return;
2473    }
2474    println!("  by branch:");
2475    for branch in branches.iter().take(20) {
2476        let name = value_str(branch, "branchName").unwrap_or("unknown");
2477        let coding_seconds = branch
2478            .get("estimatedCodingSeconds")
2479            .and_then(Value::as_u64)
2480            .unwrap_or(0);
2481        let observed_seconds = branch
2482            .get("observedSpanSeconds")
2483            .and_then(Value::as_u64)
2484            .unwrap_or(0);
2485        let sessions = branch
2486            .get("sessionCount")
2487            .and_then(Value::as_u64)
2488            .unwrap_or(0);
2489        let events = branch
2490            .get("totalEvents")
2491            .and_then(Value::as_u64)
2492            .unwrap_or(0);
2493        let commits = branch
2494            .get("totalCommits")
2495            .and_then(Value::as_u64)
2496            .unwrap_or(0);
2497        println!(
2498            "    {name}: {} coding, {} span, {} session(s), {} event(s), {} commit(s)",
2499            format_duration(coding_seconds),
2500            format_duration(observed_seconds),
2501            sessions,
2502            events,
2503            commits
2504        );
2505    }
2506}
2507
2508#[allow(dead_code)]
2509fn print_last_sync(payload: &Value) {
2510    let Some(last_sync) = payload.get("lastSync") else {
2511        return;
2512    };
2513    let synced_at = value_str(last_sync, "syncedAt").unwrap_or("unknown-time");
2514    let status = last_sync
2515        .get("statusCode")
2516        .and_then(Value::as_u64)
2517        .unwrap_or(0);
2518    let events = last_sync
2519        .get("eventCount")
2520        .and_then(Value::as_u64)
2521        .unwrap_or(0);
2522    let commits = last_sync
2523        .get("commitCount")
2524        .and_then(Value::as_u64)
2525        .unwrap_or(0);
2526    let files = last_sync
2527        .get("spoolFileCount")
2528        .and_then(Value::as_u64)
2529        .unwrap_or(0);
2530    let resync = last_sync
2531        .get("resync")
2532        .and_then(Value::as_bool)
2533        .unwrap_or(false);
2534    println!(
2535        "last sync: {synced_at}, HTTP {status}, {events} event(s), {commits} commit(s), {files} file(s), resync={resync}"
2536    );
2537}
2538
2539fn print_stats_bucket_section(stats: &Value, key: &str, title: &str, limit: usize) {
2540    let Some(items) = stats.get(key).and_then(Value::as_array) else {
2541        return;
2542    };
2543    if items.is_empty() {
2544        return;
2545    }
2546    println!("{title}");
2547    for item in items.iter().take(limit) {
2548        let name = value_str(item, "name").unwrap_or("unknown");
2549        let events = item.get("eventCount").and_then(Value::as_u64).unwrap_or(0);
2550        let seconds = item
2551            .get("estimatedCodingSeconds")
2552            .and_then(Value::as_u64)
2553            .unwrap_or(0);
2554        let added = item.get("addedLines").and_then(Value::as_u64).unwrap_or(0);
2555        let removed = item
2556            .get("removedLines")
2557            .and_then(Value::as_u64)
2558            .unwrap_or(0);
2559        println!(
2560            "    {name}: {}, {} event(s), +{} -{}",
2561            format_duration(seconds),
2562            events,
2563            added,
2564            removed
2565        );
2566    }
2567}
2568
2569fn print_stats_file_section(stats: &Value, limit: usize) {
2570    let Some(items) = stats.get("byFile").and_then(Value::as_array) else {
2571        return;
2572    };
2573    if items.is_empty() {
2574        return;
2575    }
2576    println!("  by file:");
2577    for item in items.iter().take(limit) {
2578        let path = value_str(item, "path").unwrap_or("(no path)");
2579        let filetype = value_str(item, "filetype").unwrap_or("(none)");
2580        let events = item.get("eventCount").and_then(Value::as_u64).unwrap_or(0);
2581        let seconds = item
2582            .get("estimatedCodingSeconds")
2583            .and_then(Value::as_u64)
2584            .unwrap_or(0);
2585        let added = item.get("addedLines").and_then(Value::as_u64).unwrap_or(0);
2586        let removed = item
2587            .get("removedLines")
2588            .and_then(Value::as_u64)
2589            .unwrap_or(0);
2590        println!(
2591            "    {path} [{filetype}]: {}, {} event(s), +{} -{}",
2592            format_duration(seconds),
2593            events,
2594            added,
2595            removed
2596        );
2597    }
2598}
2599
2600fn print_status_event_record(event: &Value) {
2601    let occurred_at = value_str(event, "occurredAt").unwrap_or("unknown-time");
2602    let kind = value_str(event, "eventKind").unwrap_or("event");
2603    let primary_path = event_display_path(event);
2604    let added = event.get("addedLines").and_then(Value::as_u64).unwrap_or(0);
2605    let removed = event
2606        .get("removedLines")
2607        .and_then(Value::as_u64)
2608        .unwrap_or(0);
2609    let total_lines = event.get("totalLines").and_then(Value::as_u64);
2610    if let Some(total_lines) = total_lines {
2611        println!(
2612            "    {occurred_at} {kind} {primary_path} (+{added} -{removed}, {total_lines} lines)"
2613        );
2614    } else {
2615        println!("    {occurred_at} {kind} {primary_path} (+{added} -{removed})");
2616    }
2617
2618    if let Some(head_sha) = value_str(event, "headSha") {
2619        println!("      head: {}", short_sha(head_sha));
2620    }
2621    if let Some(paths) = event.get("paths").and_then(Value::as_array) {
2622        if paths.len() > 1 {
2623            let rendered = paths
2624                .iter()
2625                .filter_map(Value::as_str)
2626                .collect::<Vec<_>>()
2627                .join(", ");
2628            println!("      paths: {rendered}");
2629        }
2630    }
2631}
2632
2633fn print_status_commit_record(commit: &Value) {
2634    let occurred_at = value_str(commit, "occurredAt").unwrap_or("unknown-time");
2635    let head_sha = value_str(commit, "headSha")
2636        .map(short_sha)
2637        .unwrap_or_else(|| "unknown".to_string());
2638    let subject = value_str(commit, "subject").unwrap_or("(no subject)");
2639    println!("    {occurred_at} {head_sha} {subject}");
2640
2641    if let Some(author) = value_str(commit, "authorName") {
2642        println!("      author: {author}");
2643    }
2644}
2645
2646fn event_display_path(event: &Value) -> String {
2647    let old_path = value_str(event, "oldPath");
2648    let new_path = value_str(event, "newPath");
2649    if let (Some(old_path), Some(new_path)) = (old_path, new_path) {
2650        return format!("{old_path} -> {new_path}");
2651    }
2652    value_str(event, "primaryPath")
2653        .map(str::to_string)
2654        .or_else(|| {
2655            event
2656                .get("paths")
2657                .and_then(Value::as_array)
2658                .and_then(|paths| paths.first())
2659                .and_then(Value::as_str)
2660                .map(str::to_string)
2661        })
2662        .unwrap_or_else(|| "(no path)".to_string())
2663}
2664
2665fn value_str<'a>(value: &'a Value, key: &str) -> Option<&'a str> {
2666    value.get(key).and_then(Value::as_str)
2667}
2668
2669fn short_sha(value: &str) -> String {
2670    value.chars().take(12).collect()
2671}
2672
2673fn generate_and_store_stats(
2674    identity: &RepoIdentity,
2675    session_gap_minutes: u64,
2676) -> Result<WorktreeWatchStatsSummary, String> {
2677    let layout = prepare_spool_layout(identity)?;
2678    let candidates = collect_spool_candidates(&layout, true)?;
2679    let summary = build_stats_summary(identity, &layout, &candidates, session_gap_minutes)?;
2680    let rendered = serde_json::to_string_pretty(&summary)
2681        .map_err(|error| format!("Failed to serialize stats: {error}"))?;
2682    fs::write(&layout.stats_file, format!("{rendered}\n")).map_err(|error| {
2683        format!(
2684            "Failed to write stats file {}: {error}",
2685            layout.stats_file.display()
2686        )
2687    })?;
2688    Ok(summary)
2689}
2690
2691fn generate_and_store_repo_activity(
2692    identity: &RepoIdentity,
2693    session_gap_minutes: u64,
2694) -> Result<WorktreeWatchRepoActivitySummary, String> {
2695    let repository_spool_root = repository_spool_root(identity)?;
2696    let summary =
2697        build_repo_activity_summary(identity, &repository_spool_root, session_gap_minutes)?;
2698    let rendered = serde_json::to_string_pretty(&summary)
2699        .map_err(|error| format!("Failed to serialize repo activity: {error}"))?;
2700    fs::create_dir_all(&repository_spool_root).map_err(|error| {
2701        format!(
2702            "Failed to create repository spool root {}: {error}",
2703            repository_spool_root.display()
2704        )
2705    })?;
2706    let stats_file = repository_spool_root.join("repo-activity.json");
2707    fs::write(&stats_file, format!("{rendered}\n")).map_err(|error| {
2708        format!(
2709            "Failed to write repo activity file {}: {error}",
2710            stats_file.display()
2711        )
2712    })?;
2713    Ok(summary)
2714}
2715
2716fn build_repo_activity_summary(
2717    identity: &RepoIdentity,
2718    repository_spool_root: &Path,
2719    session_gap_minutes: u64,
2720) -> Result<WorktreeWatchRepoActivitySummary, String> {
2721    let mut branches = Vec::new();
2722    if repository_spool_root.exists() {
2723        for entry in fs::read_dir(repository_spool_root).map_err(|error| {
2724            format!(
2725                "Failed to read repository spool root {}: {error}",
2726                repository_spool_root.display()
2727            )
2728        })? {
2729            let entry =
2730                entry.map_err(|error| format!("Failed to read repository spool entry: {error}"))?;
2731            let file_type = entry.file_type().map_err(|error| {
2732                format!("Failed to inspect {}: {error}", entry.path().display())
2733            })?;
2734            if !file_type.is_dir() {
2735                continue;
2736            }
2737            let branch_name = entry.file_name().to_string_lossy().to_string();
2738            let branch_root = entry.path();
2739            let branch_layout = spool_layout_for_existing_root(&branch_root);
2740            let branch_identity = RepoIdentity {
2741                branch: branch_name.clone(),
2742                ..identity.clone()
2743            };
2744            let candidates = collect_spool_candidates(&branch_layout, true)?;
2745            let stats = build_stats_summary(
2746                &branch_identity,
2747                &branch_layout,
2748                &candidates,
2749                session_gap_minutes,
2750            )?;
2751            if stats.total_events == 0 && stats.total_commits == 0 {
2752                continue;
2753            }
2754            branches.push(WorktreeWatchBranchActivityStats {
2755                branch_name,
2756                spool_root: branch_root.display().to_string(),
2757                total_events: stats.total_events,
2758                total_commits: stats.total_commits,
2759                first_event_at: stats.first_event_at,
2760                last_event_at: stats.last_event_at,
2761                session_count: stats.session_count,
2762                observed_span_seconds: stats.observed_span_seconds,
2763                estimated_coding_seconds: stats.estimated_coding_seconds,
2764                added_lines: stats.added_lines,
2765                removed_lines: stats.removed_lines,
2766            });
2767        }
2768    }
2769
2770    branches.sort_by(|left, right| {
2771        right
2772            .estimated_coding_seconds
2773            .cmp(&left.estimated_coding_seconds)
2774            .then_with(|| right.total_events.cmp(&left.total_events))
2775            .then_with(|| left.branch_name.cmp(&right.branch_name))
2776    });
2777
2778    let first_event_at = branches
2779        .iter()
2780        .filter_map(|branch| branch.first_event_at)
2781        .min();
2782    let last_event_at = branches
2783        .iter()
2784        .filter_map(|branch| branch.last_event_at)
2785        .max();
2786    let observed_span_seconds = observed_span_seconds(first_event_at, last_event_at);
2787    let stats_file = repository_spool_root.join("repo-activity.json");
2788
2789    Ok(WorktreeWatchRepoActivitySummary {
2790        generated_at: Utc::now(),
2791        repository_owner: identity.owner.clone(),
2792        repository_name: identity.name.clone(),
2793        repo_root: identity.root.display().to_string(),
2794        repository_spool_root: repository_spool_root.display().to_string(),
2795        stats_file: stats_file.display().to_string(),
2796        session_gap_minutes,
2797        branch_count: branches.len() as u64,
2798        total_events: branches.iter().map(|branch| branch.total_events).sum(),
2799        total_commits: branches.iter().map(|branch| branch.total_commits).sum(),
2800        first_event_at,
2801        last_event_at,
2802        session_count: branches.iter().map(|branch| branch.session_count).sum(),
2803        observed_span_seconds,
2804        estimated_coding_seconds: branches
2805            .iter()
2806            .map(|branch| branch.estimated_coding_seconds)
2807            .sum(),
2808        added_lines: branches.iter().map(|branch| branch.added_lines).sum(),
2809        removed_lines: branches.iter().map(|branch| branch.removed_lines).sum(),
2810        branches,
2811    })
2812}
2813
2814fn build_stats_summary(
2815    identity: &RepoIdentity,
2816    layout: &SpoolLayout,
2817    candidates: &[SyncCandidate],
2818    session_gap_minutes: u64,
2819) -> Result<WorktreeWatchStatsSummary, String> {
2820    let mut events = Vec::new();
2821    let mut total_commits = 0u64;
2822    for candidate in candidates {
2823        match &candidate.records {
2824            SyncRecords::Events(records) => events.extend(records.iter().cloned()),
2825            SyncRecords::Commits(records) => total_commits += records.len() as u64,
2826        }
2827    }
2828    events.sort_by_key(|event| event.occurred_at);
2829
2830    let gap_seconds = session_gap_minutes.saturating_mul(60).max(60);
2831    let mut previous_at: Option<DateTime<Utc>> = None;
2832    let mut session_count = 0u64;
2833    let mut total_seconds = 0u64;
2834    let mut total_added = 0u64;
2835    let mut total_removed = 0u64;
2836    let mut by_file: BTreeMap<String, WorktreeWatchFileStats> = BTreeMap::new();
2837    let mut by_filetype: BTreeMap<String, WorktreeWatchBucketStats> = BTreeMap::new();
2838    let mut by_event_kind: BTreeMap<String, WorktreeWatchBucketStats> = BTreeMap::new();
2839
2840    for event in &events {
2841        if starts_new_coding_session(previous_at, event.occurred_at, gap_seconds) {
2842            session_count += 1;
2843        }
2844        let seconds = coding_seconds_for_event(previous_at, event.occurred_at, gap_seconds);
2845        previous_at = Some(event.occurred_at);
2846        let added = event.added_lines.unwrap_or(0);
2847        let removed = event.removed_lines.unwrap_or(0);
2848        let path = stats_event_path(event);
2849        let filetype = filetype_for_path(&path);
2850
2851        total_seconds += seconds;
2852        total_added += added;
2853        total_removed += removed;
2854        update_file_stats(&mut by_file, &path, &filetype, seconds, added, removed);
2855        update_bucket_stats(&mut by_filetype, &filetype, seconds, added, removed);
2856        update_bucket_stats(
2857            &mut by_event_kind,
2858            &event.event_kind,
2859            seconds,
2860            added,
2861            removed,
2862        );
2863    }
2864
2865    Ok(WorktreeWatchStatsSummary {
2866        generated_at: Utc::now(),
2867        repository_owner: identity.owner.clone(),
2868        repository_name: identity.name.clone(),
2869        branch_name: identity.branch.clone(),
2870        repo_root: identity.root.display().to_string(),
2871        spool_root: layout.root.display().to_string(),
2872        session_gap_minutes,
2873        total_events: events.len() as u64,
2874        total_commits,
2875        first_event_at: events.first().map(|event| event.occurred_at),
2876        last_event_at: events.last().map(|event| event.occurred_at),
2877        session_count,
2878        observed_span_seconds: observed_span_seconds(
2879            events.first().map(|event| event.occurred_at),
2880            events.last().map(|event| event.occurred_at),
2881        ),
2882        estimated_coding_seconds: total_seconds,
2883        added_lines: total_added,
2884        removed_lines: total_removed,
2885        by_file: sorted_file_stats(by_file),
2886        by_filetype: sorted_bucket_stats(by_filetype),
2887        by_event_kind: sorted_bucket_stats(by_event_kind),
2888    })
2889}
2890
2891fn coding_seconds_for_event(
2892    previous_at: Option<DateTime<Utc>>,
2893    occurred_at: DateTime<Utc>,
2894    gap_seconds: u64,
2895) -> u64 {
2896    let Some(previous_at) = previous_at else {
2897        return 60;
2898    };
2899    let delta = occurred_at.signed_duration_since(previous_at).num_seconds();
2900    if delta <= 0 {
2901        0
2902    } else {
2903        (delta as u64).min(gap_seconds)
2904    }
2905}
2906
2907fn starts_new_coding_session(
2908    previous_at: Option<DateTime<Utc>>,
2909    occurred_at: DateTime<Utc>,
2910    gap_seconds: u64,
2911) -> bool {
2912    let Some(previous_at) = previous_at else {
2913        return true;
2914    };
2915    let delta = occurred_at.signed_duration_since(previous_at).num_seconds();
2916    delta > gap_seconds as i64
2917}
2918
2919fn observed_span_seconds(
2920    first_event_at: Option<DateTime<Utc>>,
2921    last_event_at: Option<DateTime<Utc>>,
2922) -> u64 {
2923    let (Some(first_event_at), Some(last_event_at)) = (first_event_at, last_event_at) else {
2924        return 0;
2925    };
2926    last_event_at
2927        .signed_duration_since(first_event_at)
2928        .num_seconds()
2929        .max(0) as u64
2930}
2931
2932fn update_file_stats(
2933    by_file: &mut BTreeMap<String, WorktreeWatchFileStats>,
2934    path: &str,
2935    filetype: &str,
2936    seconds: u64,
2937    added: u64,
2938    removed: u64,
2939) {
2940    let stats = by_file
2941        .entry(path.to_string())
2942        .or_insert_with(|| WorktreeWatchFileStats {
2943            path: path.to_string(),
2944            filetype: filetype.to_string(),
2945            ..WorktreeWatchFileStats::default()
2946        });
2947    stats.event_count += 1;
2948    stats.estimated_coding_seconds += seconds;
2949    stats.added_lines += added;
2950    stats.removed_lines += removed;
2951}
2952
2953fn update_bucket_stats(
2954    buckets: &mut BTreeMap<String, WorktreeWatchBucketStats>,
2955    name: &str,
2956    seconds: u64,
2957    added: u64,
2958    removed: u64,
2959) {
2960    let stats = buckets
2961        .entry(name.to_string())
2962        .or_insert_with(|| WorktreeWatchBucketStats {
2963            name: name.to_string(),
2964            ..WorktreeWatchBucketStats::default()
2965        });
2966    stats.event_count += 1;
2967    stats.estimated_coding_seconds += seconds;
2968    stats.added_lines += added;
2969    stats.removed_lines += removed;
2970}
2971
2972fn sorted_file_stats(
2973    by_file: BTreeMap<String, WorktreeWatchFileStats>,
2974) -> Vec<WorktreeWatchFileStats> {
2975    let mut stats = by_file.into_values().collect::<Vec<_>>();
2976    stats.sort_by(|left, right| {
2977        right
2978            .estimated_coding_seconds
2979            .cmp(&left.estimated_coding_seconds)
2980            .then_with(|| right.event_count.cmp(&left.event_count))
2981            .then_with(|| left.path.cmp(&right.path))
2982    });
2983    stats
2984}
2985
2986fn sorted_bucket_stats(
2987    buckets: BTreeMap<String, WorktreeWatchBucketStats>,
2988) -> Vec<WorktreeWatchBucketStats> {
2989    let mut stats = buckets.into_values().collect::<Vec<_>>();
2990    stats.sort_by(|left, right| {
2991        right
2992            .estimated_coding_seconds
2993            .cmp(&left.estimated_coding_seconds)
2994            .then_with(|| right.event_count.cmp(&left.event_count))
2995            .then_with(|| left.name.cmp(&right.name))
2996    });
2997    stats
2998}
2999
3000fn stats_event_path(event: &WorktreeMutationEvent) -> String {
3001    event
3002        .new_path
3003        .as_deref()
3004        .or(event.primary_path.as_deref())
3005        .or_else(|| event.paths.first().map(String::as_str))
3006        .unwrap_or("(no path)")
3007        .to_string()
3008}
3009
3010fn filetype_for_path(path: &str) -> String {
3011    Path::new(path)
3012        .extension()
3013        .and_then(|value| value.to_str())
3014        .map(|value| value.to_ascii_lowercase())
3015        .filter(|value| !value.is_empty())
3016        .unwrap_or_else(|| "(none)".to_string())
3017}
3018
3019fn format_duration(seconds: u64) -> String {
3020    let hours = seconds / 3600;
3021    let minutes = (seconds % 3600) / 60;
3022    let remaining_seconds = seconds % 60;
3023    if hours > 0 {
3024        format!("{hours}h {minutes}m")
3025    } else if minutes > 0 {
3026        format!("{minutes}m {remaining_seconds}s")
3027    } else {
3028        format!("{remaining_seconds}s")
3029    }
3030}
3031
3032async fn watch_foreground(
3033    identity: RepoIdentity,
3034    layout: SpoolLayout,
3035    sync_interval_seconds: u64,
3036) -> Result<(), String> {
3037    // Refresh watcher state so status sees this process (foreground + detach child).
3038    if let Ok(executable) = std::env::current_exe() {
3039        let _ = write_watcher_state(&identity, &layout, std::process::id(), &executable);
3040    }
3041
3042    // Local dashboard API (loopback). Background thread avoids an async type
3043    // cycle with POST /start handlers that call detached start.
3044    let api_target = WorktreeWatchTargetOptions {
3045        repo: Some(identity.root.clone()),
3046        parent: None,
3047        repos: Vec::new(),
3048    };
3049    if let Some(port) = crate::commands::worktree_watch_api::spawn_worktree_watch_api_in_background(
3050        api_target,
3051        sync_interval_seconds,
3052        None,
3053    ) {
3054        if !is_background_child() {
3055            crate::commands::worktree_watch_api::print_api_listen_hint(port);
3056        } else {
3057            tracing::info!(
3058                "worktree-watch local API listening on http://127.0.0.1:{port}"
3059            );
3060        }
3061    }
3062
3063    let (tx, rx) = mpsc::channel();
3064    let mut watcher = create_path_watcher(&identity.root, tx)?;
3065    watcher
3066        .watch(&identity.root, RecursiveMode::Recursive)
3067        .map_err(|error| {
3068            format!(
3069                "Failed to watch repository root {}: {error}",
3070                identity.root.display()
3071            )
3072        })?;
3073    log_watcher_backend(&identity.root, "repository");
3074
3075    let mut last_head = identity.head_sha.clone();
3076    let mut last_commit_check = Instant::now();
3077    let mut last_sync = Instant::now();
3078    let mut event_dedupe = RecentEventDedupe {
3079        fingerprints: load_event_dedupe_fingerprints(&layout.event_dedupe_file)?,
3080    };
3081
3082    loop {
3083        match rx.recv_timeout(Duration::from_millis(750)) {
3084            Ok(Ok(event)) => {
3085                if let Some(mutation) = mutation_event_from_notify(&identity, event) {
3086                    append_unique_mutation_event(&layout, &mutation, &mut event_dedupe)?;
3087                }
3088            }
3089            Ok(Err(error)) => {
3090                eprintln!("worktree watcher error: {error}");
3091            }
3092            Err(mpsc::RecvTimeoutError::Timeout) => {}
3093            Err(mpsc::RecvTimeoutError::Disconnected) => {
3094                return Err("Filesystem watcher disconnected.".to_string());
3095            }
3096        }
3097
3098        if last_commit_check.elapsed() >= Duration::from_secs(3) {
3099            last_head = record_commit_snapshot(&identity, &layout, last_head.as_deref())?;
3100            last_commit_check = Instant::now();
3101        }
3102
3103        if sync_interval_seconds > 0
3104            && last_sync.elapsed() >= Duration::from_secs(sync_interval_seconds)
3105        {
3106            if let Err(error) = sync_spool_for_identity(&identity, &layout).await {
3107                eprintln!("worktree mutation sync failed: {error}");
3108            }
3109            last_sync = Instant::now();
3110        }
3111    }
3112}
3113
3114async fn watch_parent_foreground(
3115    parent_root: PathBuf,
3116    identities: Vec<RepoIdentity>,
3117    sync_interval_seconds: u64,
3118) -> Result<(), String> {
3119    // Ensure status/CLI can detect this process even for non-detach starts
3120    // (and refresh state if the detached parent wrote a different PID).
3121    if let Ok(layout) = prepare_parent_spool_layout(&parent_root) {
3122        if let Ok(executable) = std::env::current_exe() {
3123            let _ = write_parent_watcher_state(
3124                &parent_root,
3125                &layout,
3126                std::process::id(),
3127                &executable,
3128            );
3129        }
3130    }
3131
3132    let api_target = WorktreeWatchTargetOptions {
3133        repo: None,
3134        parent: Some(parent_root.clone()),
3135        repos: Vec::new(),
3136    };
3137    if let Some(port) = crate::commands::worktree_watch_api::spawn_worktree_watch_api_in_background(
3138        api_target,
3139        sync_interval_seconds,
3140        None,
3141    ) {
3142        if !is_background_child() {
3143            crate::commands::worktree_watch_api::print_api_listen_hint(port);
3144        } else {
3145            tracing::info!(
3146                "worktree-watch local API listening on http://127.0.0.1:{port}"
3147            );
3148        }
3149    }
3150
3151    let (tx, rx) = mpsc::channel();
3152    let mut watcher = create_path_watcher(&parent_root, tx)?;
3153    watcher
3154        .watch(&parent_root, RecursiveMode::Recursive)
3155        .map_err(|error| {
3156            format!(
3157                "Failed to watch parent folder {}: {error}",
3158                parent_root.display()
3159            )
3160        })?;
3161    log_watcher_backend(&parent_root, "parent");
3162
3163    let mut repos = prepare_watched_repos(identities)?;
3164    let mut last_commit_check = Instant::now();
3165    let mut last_sync = Instant::now();
3166
3167    loop {
3168        match rx.recv_timeout(Duration::from_millis(750)) {
3169            Ok(Ok(event)) => {
3170                for (repo_index, routed_event) in route_parent_event(&repos, &event) {
3171                    let repo = &mut repos[repo_index];
3172                    if let Some(mutation) = mutation_event_from_notify(&repo.identity, routed_event)
3173                    {
3174                        append_unique_mutation_event(
3175                            &repo.layout,
3176                            &mutation,
3177                            &mut repo.event_dedupe,
3178                        )?;
3179                    }
3180                }
3181            }
3182            Ok(Err(error)) => {
3183                eprintln!("parent worktree watcher error: {error}");
3184            }
3185            Err(mpsc::RecvTimeoutError::Timeout) => {}
3186            Err(mpsc::RecvTimeoutError::Disconnected) => {
3187                return Err("Parent filesystem watcher disconnected.".to_string());
3188            }
3189        }
3190
3191        if last_commit_check.elapsed() >= Duration::from_secs(3) {
3192            for repo in &mut repos {
3193                repo.last_head = record_commit_snapshot(
3194                    &repo.identity,
3195                    &repo.layout,
3196                    repo.last_head.as_deref(),
3197                )?;
3198            }
3199            last_commit_check = Instant::now();
3200        }
3201
3202        if sync_interval_seconds > 0
3203            && last_sync.elapsed() >= Duration::from_secs(sync_interval_seconds)
3204        {
3205            for repo in &repos {
3206                if let Err(error) = sync_spool_for_identity(&repo.identity, &repo.layout).await {
3207                    eprintln!(
3208                        "worktree mutation sync failed for {}: {error}",
3209                        repo.identity.root.display()
3210                    );
3211                }
3212            }
3213            last_sync = Instant::now();
3214        }
3215    }
3216}
3217
3218/// Filesystem watcher handle that can be either native (inotify/FSEvents/ReadDirectoryChanges)
3219/// or polling (needed for WSL mounts of Windows drives where inotify is unreliable).
3220enum PathWatcher {
3221    Recommended(RecommendedWatcher),
3222    Poll(notify::PollWatcher),
3223}
3224
3225impl PathWatcher {
3226    fn watch(&mut self, path: &Path, mode: RecursiveMode) -> notify::Result<()> {
3227        match self {
3228            Self::Recommended(watcher) => watcher.watch(path, mode),
3229            Self::Poll(watcher) => watcher.watch(path, mode),
3230        }
3231    }
3232}
3233
3234fn create_path_watcher(
3235    path: &Path,
3236    tx: mpsc::Sender<notify::Result<Event>>,
3237) -> Result<PathWatcher, String> {
3238    let config = Config::default();
3239    if should_use_poll_watcher(path) {
3240        let poll_config = config.with_poll_interval(Duration::from_secs(2));
3241        notify::PollWatcher::new(
3242            move |result| {
3243                let _ = tx.send(result);
3244            },
3245            poll_config,
3246        )
3247        .map(PathWatcher::Poll)
3248        .map_err(|error| format!("Failed to create poll filesystem watcher: {error}"))
3249    } else {
3250        RecommendedWatcher::new(
3251            move |result| {
3252                let _ = tx.send(result);
3253            },
3254            config,
3255        )
3256        .map(PathWatcher::Recommended)
3257        .map_err(|error| format!("Failed to create filesystem watcher: {error}"))
3258    }
3259}
3260
3261fn should_use_poll_watcher(path: &Path) -> bool {
3262    // WSL inotify does not reliably cover 9p/drvfs mounts under /mnt/*.
3263    if !is_wsl_runtime() {
3264        return false;
3265    }
3266    let key = path_identity_key(path).replace('\\', "/");
3267    key.starts_with("/mnt/") || key.contains("/mnt/")
3268}
3269
3270fn log_watcher_backend(path: &Path, label: &str) {
3271    if should_use_poll_watcher(path) {
3272        eprintln!(
3273            "worktree-watch: using poll watcher for {label} {} (WSL mount; inotify is unreliable)",
3274            path.display()
3275        );
3276    }
3277}
3278
3279fn prepare_watched_repos(identities: Vec<RepoIdentity>) -> Result<Vec<WatchedRepo>, String> {
3280    let mut repos = Vec::with_capacity(identities.len());
3281    for identity in identities {
3282        let layout = prepare_spool_layout(&identity)?;
3283        let event_dedupe = RecentEventDedupe {
3284            fingerprints: load_event_dedupe_fingerprints(&layout.event_dedupe_file)?,
3285        };
3286        repos.push(WatchedRepo {
3287            last_head: identity.head_sha.clone(),
3288            identity,
3289            layout,
3290            event_dedupe,
3291        });
3292    }
3293    repos.sort_by_key(|repo| std::cmp::Reverse(repo.identity.root.components().count()));
3294    Ok(repos)
3295}
3296
3297fn route_parent_event(repos: &[WatchedRepo], event: &Event) -> Vec<(usize, Event)> {
3298    let mut paths_by_repo: BTreeMap<usize, Vec<PathBuf>> = BTreeMap::new();
3299    for path in &event.paths {
3300        for (index, repo) in repos.iter().enumerate() {
3301            if path_belongs_to_repo(path, &repo.identity.root) {
3302                let paths = paths_by_repo.entry(index).or_default();
3303                if !paths.iter().any(|existing| existing == path) {
3304                    paths.push(path.clone());
3305                }
3306                break;
3307            }
3308        }
3309    }
3310
3311    let mut routed = Vec::new();
3312    for (index, paths) in paths_by_repo {
3313        let mut repo_event = event.clone();
3314        repo_event.paths = paths;
3315        routed.push((index, repo_event));
3316    }
3317    routed
3318}
3319
3320fn path_belongs_to_repo(path: &Path, repo_root: &Path) -> bool {
3321    let path_key = path_identity_key(path).replace('\\', "/");
3322    let root_key = path_identity_key(repo_root).replace('\\', "/");
3323    path_key == root_key || path_key.starts_with(&format!("{root_key}/"))
3324}
3325
3326fn mutation_event_from_notify(
3327    identity: &RepoIdentity,
3328    event: Event,
3329) -> Option<WorktreeMutationEvent> {
3330    if event.kind.is_access() || event.kind.is_other() {
3331        return None;
3332    }
3333
3334    let paths: Vec<String> = event
3335        .paths
3336        .iter()
3337        .filter(|path| !is_ignored_path(&identity.root, path))
3338        .filter_map(|path| relative_slash_path(&identity.root, path))
3339        .fold(Vec::new(), |mut paths, path| {
3340            if !paths.iter().any(|existing| existing == &path) {
3341                paths.push(path);
3342            }
3343            paths
3344        });
3345
3346    if paths.is_empty() {
3347        return None;
3348    }
3349
3350    let primary_path = paths.first().cloned();
3351    let (old_path, new_path) = rename_paths(&event.kind, &paths);
3352    let line_counts = primary_path
3353        .as_deref()
3354        .and_then(|path| git_numstat_for_path(&identity.root, path).ok());
3355    let total_lines = primary_path
3356        .as_deref()
3357        .and_then(|path| file_line_count_for_path(&identity.root, path).ok());
3358    let event_kind = classify_event_kind(&event.kind);
3359    let is_dir = event
3360        .paths
3361        .first()
3362        .and_then(|path| fs::metadata(path).ok())
3363        .map(|metadata| metadata.is_dir())
3364        .unwrap_or(false);
3365
3366    let occurred_at = Utc::now();
3367    let mut mutation = WorktreeMutationEvent {
3368        id: String::new(),
3369        fingerprint: None,
3370        repo_owner: identity.owner.clone(),
3371        repo_name: identity.name.clone(),
3372        branch_name: identity.branch.clone(),
3373        repo_root: normalize_repo_root_for_payload(&identity.root),
3374        head_sha: current_head(&identity.root).ok().flatten(),
3375        event_kind: event_kind.to_string(),
3376        paths,
3377        primary_path,
3378        old_path,
3379        new_path,
3380        added_lines: line_counts.map(|counts| counts.0),
3381        removed_lines: line_counts.map(|counts| counts.1),
3382        total_lines,
3383        file_created: matches!(
3384            event.kind,
3385            EventKind::Create(CreateKind::File | CreateKind::Any)
3386        ) && !is_dir,
3387        file_removed: matches!(
3388            event.kind,
3389            EventKind::Remove(RemoveKind::File | RemoveKind::Any)
3390        ) && !is_dir,
3391        folder_created: matches!(
3392            event.kind,
3393            EventKind::Create(CreateKind::Folder | CreateKind::Any)
3394        ) && is_dir,
3395        folder_removed: matches!(
3396            event.kind,
3397            EventKind::Remove(RemoveKind::Folder | RemoveKind::Any)
3398        ) && is_dir,
3399        renamed_or_moved: is_rename_or_move(&event.kind),
3400        raw_kind: format!("{:?}", event.kind),
3401        occurred_at,
3402    };
3403    // Stable id/fingerprint for idempotent spool + upload (same logical mutation
3404    // within the dedupe window always maps to the same identity).
3405    let fingerprint = mutation_event_fingerprint(&mutation);
3406    mutation.fingerprint = Some(fingerprint.clone());
3407    mutation.id = stable_event_id(&fingerprint);
3408    Some(mutation)
3409}
3410
3411fn append_unique_mutation_event(
3412    layout: &SpoolLayout,
3413    mutation: &WorktreeMutationEvent,
3414    dedupe: &mut RecentEventDedupe,
3415) -> Result<bool, String> {
3416    let fingerprint = mutation
3417        .fingerprint
3418        .clone()
3419        .unwrap_or_else(|| mutation_event_fingerprint(mutation));
3420    if dedupe.fingerprints.contains(&fingerprint)
3421        || event_dedupe_file_contains(&layout.event_dedupe_file, &fingerprint)?
3422    {
3423        dedupe.fingerprints.insert(fingerprint);
3424        return Ok(false);
3425    }
3426
3427    if !claim_event_fingerprint(layout, &fingerprint)? {
3428        dedupe.fingerprints.insert(fingerprint);
3429        return Ok(false);
3430    }
3431
3432    let mut mutation = mutation.clone();
3433    mutation.fingerprint = Some(fingerprint.clone());
3434    if mutation.id.is_empty() {
3435        mutation.id = stable_event_id(&fingerprint);
3436    }
3437    if let Err(error) = append_json_line(&layout.events_file, &mutation) {
3438        release_event_fingerprint_claim(layout, &fingerprint);
3439        return Err(error);
3440    }
3441    append_json_line(
3442        &layout.event_dedupe_file,
3443        &WorktreeWatchEventDedupeRecord {
3444            fingerprint: fingerprint.clone(),
3445            first_seen_at: Utc::now(),
3446            event_kind: mutation.event_kind.clone(),
3447            primary_path: mutation.primary_path.clone(),
3448        },
3449    )?;
3450    dedupe.fingerprints.insert(fingerprint);
3451    Ok(true)
3452}
3453
3454fn claim_event_fingerprint(layout: &SpoolLayout, fingerprint: &str) -> Result<bool, String> {
3455    fs::create_dir_all(&layout.event_dedupe_dir).map_err(|error| {
3456        format!(
3457            "Failed to create event dedupe directory {}: {error}",
3458            layout.event_dedupe_dir.display()
3459        )
3460    })?;
3461    let marker = event_dedupe_marker_path(layout, fingerprint);
3462    match OpenOptions::new()
3463        .write(true)
3464        .create_new(true)
3465        .open(&marker)
3466    {
3467        Ok(mut file) => {
3468            writeln!(file, "{}", Utc::now().to_rfc3339()).map_err(|error| {
3469                format!(
3470                    "Failed to write event dedupe marker {}: {error}",
3471                    marker.display()
3472                )
3473            })?;
3474            Ok(true)
3475        }
3476        Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => Ok(false),
3477        Err(error) => Err(format!(
3478            "Failed to create event dedupe marker {}: {error}",
3479            marker.display()
3480        )),
3481    }
3482}
3483
3484fn release_event_fingerprint_claim(layout: &SpoolLayout, fingerprint: &str) {
3485    let _ = fs::remove_file(event_dedupe_marker_path(layout, fingerprint));
3486}
3487
3488fn event_dedupe_marker_path(layout: &SpoolLayout, fingerprint: &str) -> PathBuf {
3489    layout.event_dedupe_dir.join(format!("{fingerprint}.seen"))
3490}
3491
3492fn mutation_event_fingerprint(mutation: &WorktreeMutationEvent) -> String {
3493    let bucket = mutation.occurred_at.timestamp() / EVENT_DEDUPE_WINDOW_SECONDS;
3494    // Omit absolute `repoRoot` so Windows `C:\...` and WSL `/mnt/c/...` spellings
3495    // of the same tree share one fingerprint. Platform is kept only as telemetry
3496    // on the upload device payload — not in the local idempotency key — so both
3497    // environments can share one home-level spool safely.
3498    let payload = json!({
3499        "schema": 3,
3500        "repoOwner": mutation.repo_owner,
3501        "repoName": mutation.repo_name,
3502        "branchName": mutation.branch_name,
3503        "headSha": mutation.head_sha,
3504        "eventKind": mutation.event_kind,
3505        "paths": normalize_path_list_for_fingerprint(&mutation.paths),
3506        "primaryPath": mutation.primary_path.as_deref().map(normalize_relative_watch_path),
3507        "oldPath": mutation.old_path.as_deref().map(normalize_relative_watch_path),
3508        "newPath": mutation.new_path.as_deref().map(normalize_relative_watch_path),
3509        "addedLines": mutation.added_lines,
3510        "removedLines": mutation.removed_lines,
3511        "fileCreated": mutation.file_created,
3512        "fileRemoved": mutation.file_removed,
3513        "folderCreated": mutation.folder_created,
3514        "folderRemoved": mutation.folder_removed,
3515        "renamedOrMoved": mutation.renamed_or_moved,
3516        "timeBucket": bucket,
3517    });
3518    let encoded = serde_json::to_vec(&payload).unwrap_or_default();
3519    let mut hasher = Sha256::new();
3520    hasher.update(encoded);
3521    format!("{:x}", hasher.finalize())
3522}
3523
3524fn normalize_path_list_for_fingerprint(paths: &[String]) -> Vec<String> {
3525    let mut normalized = paths
3526        .iter()
3527        .map(|path| normalize_relative_watch_path(path))
3528        .filter(|path| !path.is_empty())
3529        .collect::<Vec<_>>();
3530    normalized.sort();
3531    normalized.dedup();
3532    normalized
3533}
3534
3535fn stable_event_id(fingerprint: &str) -> String {
3536    // Deterministic id derived from fingerprint so resync / multi-process claims
3537    // never invent a second identity for the same logical mutation.
3538    format!("wtmut_{fingerprint}")
3539}
3540
3541fn normalize_repo_root_for_payload(root: &Path) -> String {
3542    path_identity_key(root).replace('\\', "/")
3543}
3544
3545fn load_event_dedupe_fingerprints(path: &Path) -> Result<BTreeSet<String>, String> {
3546    let mut fingerprints = BTreeSet::new();
3547    if !path.exists() {
3548        return Ok(fingerprints);
3549    }
3550
3551    for value in read_jsonl_values(path)? {
3552        if let Some(fingerprint) = value.get("fingerprint").and_then(Value::as_str) {
3553            fingerprints.insert(fingerprint.to_string());
3554        }
3555    }
3556    Ok(fingerprints)
3557}
3558
3559fn event_dedupe_file_contains(path: &Path, fingerprint: &str) -> Result<bool, String> {
3560    if !path.exists() {
3561        return Ok(false);
3562    }
3563    for value in read_jsonl_values(path)? {
3564        if value
3565            .get("fingerprint")
3566            .and_then(Value::as_str)
3567            .map(|value| value == fingerprint)
3568            .unwrap_or(false)
3569        {
3570            return Ok(true);
3571        }
3572    }
3573    Ok(false)
3574}
3575
3576fn classify_event_kind(kind: &EventKind) -> &'static str {
3577    match kind {
3578        EventKind::Create(CreateKind::File) => "file_create",
3579        EventKind::Create(CreateKind::Folder) => "folder_create",
3580        EventKind::Create(_) => "create",
3581        EventKind::Remove(RemoveKind::File) => "file_remove",
3582        EventKind::Remove(RemoveKind::Folder) => "folder_remove",
3583        EventKind::Remove(_) => "remove",
3584        EventKind::Modify(ModifyKind::Name(
3585            RenameMode::From | RenameMode::To | RenameMode::Both,
3586        )) => "rename_or_move",
3587        EventKind::Modify(ModifyKind::Data(_)) => "file_modify",
3588        EventKind::Modify(ModifyKind::Metadata(_)) => "metadata_modify",
3589        EventKind::Modify(_) => "modify",
3590        _ => "other",
3591    }
3592}
3593
3594fn is_rename_or_move(kind: &EventKind) -> bool {
3595    matches!(
3596        kind,
3597        EventKind::Modify(ModifyKind::Name(
3598            RenameMode::From | RenameMode::To | RenameMode::Both
3599        ))
3600    )
3601}
3602
3603fn rename_paths(kind: &EventKind, paths: &[String]) -> (Option<String>, Option<String>) {
3604    if !is_rename_or_move(kind) {
3605        return (None, None);
3606    }
3607
3608    match paths {
3609        [old_path, new_path, ..] => (Some(old_path.clone()), Some(new_path.clone())),
3610        [path] => match kind {
3611            EventKind::Modify(ModifyKind::Name(RenameMode::From)) => (Some(path.clone()), None),
3612            EventKind::Modify(ModifyKind::Name(RenameMode::To)) => (None, Some(path.clone())),
3613            _ => (None, Some(path.clone())),
3614        },
3615        _ => (None, None),
3616    }
3617}
3618
3619fn record_commit_snapshot(
3620    identity: &RepoIdentity,
3621    layout: &SpoolLayout,
3622    previous_head: Option<&str>,
3623) -> Result<Option<String>, String> {
3624    let head = current_head(&identity.root)?;
3625    let Some(head_sha) = head else {
3626        return Ok(None);
3627    };
3628
3629    if previous_head == Some(head_sha.as_str()) {
3630        return Ok(Some(head_sha));
3631    }
3632
3633    let mut commit = WorktreeCommitEvent {
3634        id: String::new(),
3635        fingerprint: None,
3636        repo_owner: identity.owner.clone(),
3637        repo_name: identity.name.clone(),
3638        branch_name: identity.branch.clone(),
3639        repo_root: normalize_repo_root_for_payload(&identity.root),
3640        previous_head_sha: previous_head.map(str::to_string),
3641        head_sha: head_sha.clone(),
3642        subject: git_output(&identity.root, &["log", "-1", "--pretty=%s"]).ok(),
3643        author_name: git_output(&identity.root, &["log", "-1", "--pretty=%an"]).ok(),
3644        author_email: git_output(&identity.root, &["log", "-1", "--pretty=%ae"]).ok(),
3645        committed_at: git_output(&identity.root, &["log", "-1", "--pretty=%cI"]).ok(),
3646        occurred_at: Utc::now(),
3647    };
3648    let fingerprint = commit_event_fingerprint(&commit);
3649    commit.fingerprint = Some(fingerprint.clone());
3650    commit.id = stable_event_id(&fingerprint);
3651    append_unique_commit_event(layout, &commit)?;
3652    Ok(Some(head_sha))
3653}
3654
3655fn append_unique_commit_event(
3656    layout: &SpoolLayout,
3657    commit: &WorktreeCommitEvent,
3658) -> Result<bool, String> {
3659    let fingerprint = commit
3660        .fingerprint
3661        .clone()
3662        .unwrap_or_else(|| commit_event_fingerprint(commit));
3663    if !claim_commit_fingerprint(layout, &fingerprint)? {
3664        return Ok(false);
3665    }
3666
3667    let mut commit = commit.clone();
3668    commit.fingerprint = Some(fingerprint.clone());
3669    if commit.id.is_empty() {
3670        commit.id = stable_event_id(&fingerprint);
3671    }
3672    if let Err(error) = append_json_line(&layout.commits_file, &commit) {
3673        release_commit_fingerprint_claim(layout, &fingerprint);
3674        return Err(error);
3675    }
3676    Ok(true)
3677}
3678
3679fn commit_event_fingerprint(commit: &WorktreeCommitEvent) -> String {
3680    let payload = json!({
3681        "schema": 3,
3682        "repoOwner": commit.repo_owner,
3683        "repoName": commit.repo_name,
3684        "branchName": commit.branch_name,
3685        "previousHeadSha": commit.previous_head_sha,
3686        "headSha": commit.head_sha,
3687    });
3688    let encoded = serde_json::to_vec(&payload).unwrap_or_default();
3689    let mut hasher = Sha256::new();
3690    hasher.update(encoded);
3691    format!("{:x}", hasher.finalize())
3692}
3693
3694fn claim_commit_fingerprint(layout: &SpoolLayout, fingerprint: &str) -> Result<bool, String> {
3695    fs::create_dir_all(&layout.commit_dedupe_dir).map_err(|error| {
3696        format!(
3697            "Failed to create commit dedupe directory {}: {error}",
3698            layout.commit_dedupe_dir.display()
3699        )
3700    })?;
3701    let marker = commit_dedupe_marker_path(layout, fingerprint);
3702    match OpenOptions::new()
3703        .write(true)
3704        .create_new(true)
3705        .open(&marker)
3706    {
3707        Ok(mut file) => {
3708            writeln!(file, "{}", Utc::now().to_rfc3339()).map_err(|error| {
3709                format!(
3710                    "Failed to write commit dedupe marker {}: {error}",
3711                    marker.display()
3712                )
3713            })?;
3714            Ok(true)
3715        }
3716        Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => Ok(false),
3717        Err(error) => Err(format!(
3718            "Failed to create commit dedupe marker {}: {error}",
3719            marker.display()
3720        )),
3721    }
3722}
3723
3724fn release_commit_fingerprint_claim(layout: &SpoolLayout, fingerprint: &str) {
3725    let _ = fs::remove_file(commit_dedupe_marker_path(layout, fingerprint));
3726}
3727
3728fn commit_dedupe_marker_path(layout: &SpoolLayout, fingerprint: &str) -> PathBuf {
3729    layout.commit_dedupe_dir.join(format!("{fingerprint}.seen"))
3730}
3731
3732async fn sync_spool_for_identity(
3733    identity: &RepoIdentity,
3734    layout: &SpoolLayout,
3735) -> Result<(), String> {
3736    let candidates = collect_sync_candidates(layout)?;
3737    if candidates.is_empty() {
3738        return Ok(());
3739    }
3740
3741    sync_candidates(identity, candidates, false).await
3742}
3743
3744async fn sync_candidates(
3745    identity: &RepoIdentity,
3746    candidates: Vec<SyncCandidate>,
3747    resync: bool,
3748) -> Result<(), String> {
3749    let mut events = Vec::new();
3750    let mut commits = Vec::new();
3751    for candidate in &candidates {
3752        match &candidate.records {
3753            SyncRecords::Events(records) => events.extend(records.iter().cloned()),
3754            SyncRecords::Commits(records) => commits.extend(records.iter().cloned()),
3755        }
3756    }
3757
3758    let token = resolve_cli_access_token()?;
3759    let device = resolve_device_identity()?;
3760    let client = cli_request_client()?;
3761    let api = ApiConfig::from_env();
3762    let endpoint = api.cli_worktree_mutations_endpoint();
3763    let event_count = events.len();
3764    let commit_count = commits.len();
3765    let hardware_id = worktree_device_hardware_id(&device.hardware_id);
3766    let hostname = current_hostname();
3767    let platform = std::env::consts::OS.to_string();
3768    let runtime_key = worktree_runtime_key();
3769    let repo_root = normalize_repo_root_for_payload(&identity.root);
3770    // Ensure every record has a stable fingerprint/id before upload (legacy spool rows).
3771    let events = events
3772        .into_iter()
3773        .map(ensure_mutation_event_identity)
3774        .collect::<Vec<_>>();
3775    let commits = commits
3776        .into_iter()
3777        .map(ensure_commit_event_identity)
3778        .collect::<Vec<_>>();
3779    let batches = split_worktree_mutation_upload_batches(events.clone(), commits.clone());
3780    let batch_count = batches.len();
3781
3782    if batch_count > 1 {
3783        println!(
3784            "Uploading {} worktree mutation record(s) in {} batch(es) for {}/{} on {} [{}].",
3785            event_count + commit_count,
3786            batch_count,
3787            identity.owner,
3788            identity.name,
3789            identity.branch,
3790            runtime_key
3791        );
3792    }
3793
3794    let mut status_code = 202;
3795    for attempt in 1..=WORKTREE_MUTATION_SYNC_MAX_ATTEMPTS {
3796        let attempt_batches =
3797            split_worktree_mutation_upload_batches(events.clone(), commits.clone());
3798        match upload_worktree_mutation_batches(
3799            &client,
3800            &endpoint,
3801            &token,
3802            &hardware_id,
3803            hostname.as_deref(),
3804            &platform,
3805            &runtime_key,
3806            identity,
3807            &repo_root,
3808            attempt_batches,
3809        )
3810        .await
3811        {
3812            Ok(code) => {
3813                status_code = code;
3814                break;
3815            }
3816            Err(error)
3817                if error.kind == WorktreeMutationSyncErrorKind::Retryable
3818                    && attempt < WORKTREE_MUTATION_SYNC_MAX_ATTEMPTS =>
3819            {
3820                eprintln!(
3821                    "Worktree mutation sync attempt {attempt}/{} failed for {}/{} on {}: {} Restarting from the first batch in {} second(s).",
3822                    WORKTREE_MUTATION_SYNC_MAX_ATTEMPTS,
3823                    identity.owner,
3824                    identity.name,
3825                    identity.branch,
3826                    error.message,
3827                    WORKTREE_MUTATION_SYNC_RETRY_DELAY_SECONDS
3828                );
3829                tokio::time::sleep(Duration::from_secs(
3830                    WORKTREE_MUTATION_SYNC_RETRY_DELAY_SECONDS,
3831                ))
3832                .await;
3833            }
3834            Err(error) => return Err(error.message),
3835        }
3836    }
3837
3838    let layout = prepare_spool_layout(identity)?;
3839    append_sync_log(SyncLogAppend {
3840        identity,
3841        layout: &layout,
3842        endpoint: &endpoint,
3843        status_code,
3844        event_count,
3845        commit_count,
3846        candidates: &candidates,
3847        resync,
3848    })?;
3849
3850    for candidate in candidates {
3851        if !is_synced_spool_path(&candidate.path) {
3852            mark_synced(&candidate.path)?;
3853        }
3854    }
3855
3856    Ok(())
3857}
3858
3859async fn upload_worktree_mutation_batches(
3860    client: &reqwest::Client,
3861    endpoint: &str,
3862    token: &str,
3863    hardware_id: &str,
3864    hostname: Option<&str>,
3865    platform: &str,
3866    runtime_key: &str,
3867    identity: &RepoIdentity,
3868    repo_root: &str,
3869    batches: Vec<WorktreeMutationUploadBatch>,
3870) -> Result<u16, WorktreeMutationSyncError> {
3871    let batch_count = batches.len();
3872    let mut status_code = 202;
3873
3874    for (batch_index, batch) in batches.into_iter().enumerate() {
3875        let batch_event_count = batch.events.len();
3876        let batch_commit_count = batch.commits.len();
3877        let payload = WorktreeMutationIngestPayload {
3878            device: WorktreeMutationDevicePayload {
3879                hardware_id: hardware_id.to_string(),
3880                device_name: hostname.map(str::to_string),
3881                hostname: hostname.map(str::to_string),
3882                platform: platform.to_string(),
3883                runtime_key: Some(runtime_key.to_string()),
3884            },
3885            repository: WorktreeMutationRepositoryPayload {
3886                owner: identity.owner.clone(),
3887                name: identity.name.clone(),
3888                branch_name: identity.branch.clone(),
3889                repo_root: repo_root.to_string(),
3890            },
3891            events: batch.events,
3892            commits: batch.commits,
3893        };
3894
3895        let response = client
3896            .post(endpoint)
3897            .bearer_auth(token)
3898            .json(&payload)
3899            .send()
3900            .await
3901            .map_err(|error| WorktreeMutationSyncError {
3902                kind: WorktreeMutationSyncErrorKind::Retryable,
3903                message: format!(
3904                    "Failed to upload worktree mutation batch {}/{}: {error}",
3905                    batch_index + 1,
3906                    batch_count
3907                ),
3908            })?;
3909
3910        if response.status() == StatusCode::UNAUTHORIZED {
3911            return Err(WorktreeMutationSyncError {
3912                kind: WorktreeMutationSyncErrorKind::NonRetryable,
3913                message: "Your stored CLI session is no longer valid. Run `xbp login` again."
3914                    .to_string(),
3915            });
3916        }
3917
3918        if !response.status().is_success() {
3919            let status = response.status();
3920            let body = response.text().await.unwrap_or_default();
3921            return Err(WorktreeMutationSyncError {
3922                kind: if should_retry_worktree_mutation_sync_status(status) {
3923                    WorktreeMutationSyncErrorKind::Retryable
3924                } else {
3925                    WorktreeMutationSyncErrorKind::NonRetryable
3926                },
3927                message: format!(
3928                    "Worktree mutation upload batch {}/{} failed with {status}: {body}",
3929                    batch_index + 1,
3930                    batch_count
3931                ),
3932            });
3933        }
3934
3935        status_code = response.status().as_u16();
3936        if batch_count > 1 {
3937            println!(
3938                "Uploaded worktree mutation batch {}/{} ({} event(s), {} commit(s)).",
3939                batch_index + 1,
3940                batch_count,
3941                batch_event_count,
3942                batch_commit_count
3943            );
3944        }
3945    }
3946
3947    Ok(status_code)
3948}
3949
3950fn should_retry_worktree_mutation_sync_status(status: StatusCode) -> bool {
3951    status.is_server_error()
3952        || status == StatusCode::REQUEST_TIMEOUT
3953        || status == StatusCode::TOO_MANY_REQUESTS
3954}
3955
3956fn split_worktree_mutation_upload_batches(
3957    events: Vec<WorktreeMutationEvent>,
3958    commits: Vec<WorktreeCommitEvent>,
3959) -> Vec<WorktreeMutationUploadBatch> {
3960    let mut batches = Vec::new();
3961    let mut current = empty_worktree_mutation_upload_batch();
3962
3963    for event in events {
3964        let estimated_json_bytes = serialized_json_len(&event);
3965        if should_start_next_worktree_mutation_upload_batch(&current, estimated_json_bytes) {
3966            batches.push(current);
3967            current = empty_worktree_mutation_upload_batch();
3968        }
3969        current.estimated_json_bytes += estimated_json_bytes;
3970        current.events.push(event);
3971    }
3972
3973    for commit in commits {
3974        let estimated_json_bytes = serialized_json_len(&commit);
3975        if should_start_next_worktree_mutation_upload_batch(&current, estimated_json_bytes) {
3976            batches.push(current);
3977            current = empty_worktree_mutation_upload_batch();
3978        }
3979        current.estimated_json_bytes += estimated_json_bytes;
3980        current.commits.push(commit);
3981    }
3982
3983    if current_record_count(&current) > 0 {
3984        batches.push(current);
3985    }
3986
3987    batches
3988}
3989
3990fn empty_worktree_mutation_upload_batch() -> WorktreeMutationUploadBatch {
3991    WorktreeMutationUploadBatch {
3992        events: Vec::new(),
3993        commits: Vec::new(),
3994        estimated_json_bytes: 0,
3995    }
3996}
3997
3998fn should_start_next_worktree_mutation_upload_batch(
3999    batch: &WorktreeMutationUploadBatch,
4000    next_record_json_bytes: usize,
4001) -> bool {
4002    if current_record_count(batch) == 0 {
4003        return false;
4004    }
4005
4006    current_record_count(batch) >= WORKTREE_MUTATION_SYNC_BATCH_RECORD_LIMIT
4007        || batch.estimated_json_bytes + next_record_json_bytes
4008            > WORKTREE_MUTATION_SYNC_BATCH_JSON_BYTES
4009}
4010
4011fn current_record_count(batch: &WorktreeMutationUploadBatch) -> usize {
4012    batch.events.len() + batch.commits.len()
4013}
4014
4015fn serialized_json_len<T: Serialize>(value: &T) -> usize {
4016    serde_json::to_vec(value)
4017        .map(|encoded| encoded.len())
4018        .unwrap_or(0)
4019}
4020
4021fn collect_sync_candidates(layout: &SpoolLayout) -> Result<Vec<SyncCandidate>, String> {
4022    collect_spool_candidates(layout, false)
4023}
4024
4025fn print_no_sync_candidates_message(
4026    target: &WorktreeWatchTargetOptions,
4027    resync: bool,
4028) -> Result<(), String> {
4029    if resync {
4030        println!("No local worktree mutation JSONL files found.");
4031        return Ok(());
4032    }
4033
4034    let identities = resolve_target_identities(target)?;
4035    let mut synced_files = 0usize;
4036    let mut synced_records = 0usize;
4037    for identity in identities {
4038        let layout = prepare_spool_layout(&identity)?;
4039        let all_candidates = collect_spool_candidates(&layout, true)?;
4040        let unsynced_candidates = collect_sync_candidates(&layout)?;
4041        synced_files += all_candidates
4042            .len()
4043            .saturating_sub(unsynced_candidates.len());
4044        let all_records = all_candidates
4045            .iter()
4046            .map(|candidate| candidate.records.len())
4047            .sum::<usize>();
4048        let unsynced_records = unsynced_candidates
4049            .iter()
4050            .map(|candidate| candidate.records.len())
4051            .sum::<usize>();
4052        synced_records += all_records.saturating_sub(unsynced_records);
4053    }
4054
4055    if synced_files > 0 {
4056        println!(
4057            "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."
4058        );
4059    } else {
4060        println!("No unsynced worktree mutation files found.");
4061    }
4062    Ok(())
4063}
4064
4065fn collect_spool_candidates(
4066    layout: &SpoolLayout,
4067    include_synced: bool,
4068) -> Result<Vec<SyncCandidate>, String> {
4069    // When the layout root is a branch spool, also scan sibling legacy roots for
4070    // the same owner/repo/branch so WSL can see Windows-written files and vice versa.
4071    let mut roots = vec![layout.root.clone()];
4072    if let Some(branch) = layout.root.file_name().and_then(|name| name.to_str()) {
4073        if let Some(repo_root) = layout.root.parent() {
4074            if let (Some(name), Some(owner)) = (
4075                repo_root.file_name().and_then(|n| n.to_str()),
4076                repo_root.parent().and_then(|p| p.file_name()).and_then(|n| n.to_str()),
4077            ) {
4078                let identity = RepoIdentity {
4079                    owner: owner.to_string(),
4080                    name: name.to_string(),
4081                    branch: branch.to_string(),
4082                    root: PathBuf::new(),
4083                    head_sha: None,
4084                };
4085                if let Ok(search_roots) = repository_spool_search_roots(&identity) {
4086                    for search_root in search_roots {
4087                        let branch_root = search_root.join(branch);
4088                        if !roots.iter().any(|existing| {
4089                            path_identity_key(existing) == path_identity_key(&branch_root)
4090                        }) {
4091                            roots.push(branch_root);
4092                        }
4093                    }
4094                }
4095            }
4096        }
4097    }
4098
4099    let mut candidates = Vec::new();
4100    let mut seen_files = BTreeSet::new();
4101    for root in roots {
4102        collect_spool_candidates_from_root(&root, include_synced, &mut candidates, &mut seen_files)?;
4103    }
4104    Ok(candidates)
4105}
4106
4107fn collect_spool_candidates_from_root(
4108    root: &Path,
4109    include_synced: bool,
4110    candidates: &mut Vec<SyncCandidate>,
4111    seen_files: &mut BTreeSet<String>,
4112) -> Result<(), String> {
4113    if !root.exists() {
4114        return Ok(());
4115    }
4116
4117    for entry in fs::read_dir(root).map_err(|error| {
4118        format!(
4119            "Failed to read spool directory {}: {error}",
4120            root.display()
4121        )
4122    })? {
4123        let path = entry
4124            .map_err(|error| format!("Failed to read spool entry: {error}"))?
4125            .path();
4126        let Some(file_name) = path.file_name().and_then(|value| value.to_str()) else {
4127            continue;
4128        };
4129        if !file_name.ends_with(".jsonl") || (!include_synced && file_name.contains(".synced.")) {
4130            continue;
4131        }
4132
4133        let kind = if file_name.starts_with("events-") {
4134            SyncFileKind::Events
4135        } else if file_name.starts_with("commits-") {
4136            SyncFileKind::Commits
4137        } else {
4138            continue;
4139        };
4140
4141        let identity_key = path_identity_key(&path);
4142        if !seen_files.insert(identity_key) {
4143            continue;
4144        }
4145
4146        let records = sanitize_spool_records(&path, kind, read_spool_records(&path, kind)?)?;
4147        if records.is_empty() {
4148            continue;
4149        }
4150        candidates.push(SyncCandidate { path, records });
4151    }
4152
4153    Ok(())
4154}
4155
4156fn append_sync_log(request: SyncLogAppend<'_>) -> Result<(), String> {
4157    let entry = WorktreeWatchSyncLogEntry {
4158        id: Uuid::new_v4().to_string(),
4159        synced_at: Utc::now(),
4160        endpoint: request.endpoint.to_string(),
4161        repository_owner: request.identity.owner.clone(),
4162        repository_name: request.identity.name.clone(),
4163        branch_name: request.identity.branch.clone(),
4164        repo_root: request.identity.root.display().to_string(),
4165        resync: request.resync,
4166        status_code: request.status_code,
4167        event_count: request.event_count,
4168        commit_count: request.commit_count,
4169        spool_file_count: request.candidates.len(),
4170        spool_files: request
4171            .candidates
4172            .iter()
4173            .map(|candidate| candidate.path.display().to_string())
4174            .collect(),
4175    };
4176    append_json_line(&request.layout.sync_log_file, &entry)
4177}
4178
4179fn sanitize_spool_records(
4180    path: &Path,
4181    kind: SyncFileKind,
4182    records: SyncRecords,
4183) -> Result<SyncRecords, String> {
4184    match (kind, records) {
4185        (SyncFileKind::Events, SyncRecords::Events(records)) => {
4186            let mut sanitized = Vec::with_capacity(records.len());
4187            let mut changed = false;
4188
4189            for record in records {
4190                let original = serde_json::to_value(&record).ok();
4191                let Some(record) = sanitize_spooled_event(record) else {
4192                    changed = true;
4193                    continue;
4194                };
4195                if !changed {
4196                    changed = serde_json::to_value(&record).ok() != original;
4197                }
4198                sanitized.push(record);
4199            }
4200
4201            if changed {
4202                rewrite_jsonl_records(path, &sanitized)?;
4203            }
4204
4205            Ok(SyncRecords::Events(sanitized))
4206        }
4207        (_, records) => Ok(records),
4208    }
4209}
4210
4211fn sanitize_spooled_event(mut event: WorktreeMutationEvent) -> Option<WorktreeMutationEvent> {
4212    event.paths = dedupe_filtered_spool_paths(event.paths);
4213    event.primary_path = sanitize_optional_spool_path(event.primary_path);
4214    event.old_path = sanitize_optional_spool_path(event.old_path);
4215    event.new_path = sanitize_optional_spool_path(event.new_path);
4216
4217    if event.paths.is_empty() {
4218        if let Some(path) = event.primary_path.clone() {
4219            event.paths.push(path);
4220        }
4221        if let Some(path) = event.old_path.clone() {
4222            if !event.paths.iter().any(|existing| existing == &path) {
4223                event.paths.push(path);
4224            }
4225        }
4226        if let Some(path) = event.new_path.clone() {
4227            if !event.paths.iter().any(|existing| existing == &path) {
4228                event.paths.push(path);
4229            }
4230        }
4231    }
4232
4233    if event.primary_path.is_none() {
4234        event.primary_path = event.paths.first().cloned();
4235    }
4236
4237    if event.paths.is_empty() {
4238        return None;
4239    }
4240
4241    Some(event)
4242}
4243
4244fn sanitize_optional_spool_path(path: Option<String>) -> Option<String> {
4245    path.and_then(sanitize_spool_path)
4246}
4247
4248fn sanitize_spool_path(path: String) -> Option<String> {
4249    let trimmed = path.trim();
4250    if trimmed.is_empty() || watch_ignore_rules().ignores_relative(trimmed) {
4251        return None;
4252    }
4253    Some(trimmed.replace('\\', "/"))
4254}
4255
4256fn dedupe_filtered_spool_paths(paths: Vec<String>) -> Vec<String> {
4257    let mut sanitized = Vec::with_capacity(paths.len());
4258    for path in paths {
4259        let Some(path) = sanitize_spool_path(path) else {
4260            continue;
4261        };
4262        if !sanitized.iter().any(|existing| existing == &path) {
4263            sanitized.push(path);
4264        }
4265    }
4266    sanitized
4267}
4268
4269fn rewrite_jsonl_records<T: Serialize>(path: &Path, records: &[T]) -> Result<(), String> {
4270    if records.is_empty() {
4271        if path.exists() {
4272            fs::remove_file(path).map_err(|error| {
4273                format!(
4274                    "Failed to remove emptied spool file {}: {error}",
4275                    path.display()
4276                )
4277            })?;
4278        }
4279        return Ok(());
4280    }
4281
4282    let temp_path = path.with_extension(format!(
4283        "{}.{}.tmp",
4284        path.extension()
4285            .and_then(|value| value.to_str())
4286            .unwrap_or("jsonl"),
4287        Uuid::new_v4()
4288    ));
4289    {
4290        let mut file = File::create(&temp_path).map_err(|error| {
4291            format!(
4292                "Failed to create temporary spool file {}: {error}",
4293                temp_path.display()
4294            )
4295        })?;
4296        for record in records {
4297            let line = serde_json::to_string(record)
4298                .map_err(|error| format!("Failed to serialize worktree event: {error}"))?;
4299            writeln!(file, "{line}").map_err(|error| {
4300                format!(
4301                    "Failed to write temporary spool file {}: {error}",
4302                    temp_path.display()
4303                )
4304            })?;
4305        }
4306    }
4307
4308    if path.exists() {
4309        fs::remove_file(path).map_err(|error| {
4310            format!(
4311                "Failed to replace rewritten spool file {}: {error}",
4312                path.display()
4313            )
4314        })?;
4315    }
4316
4317    fs::rename(&temp_path, path)
4318        .map_err(|error| format!("Failed to rewrite spool file {}: {error}", path.display()))
4319}
4320
4321fn read_last_sync_log_entry(path: &Path) -> Result<Option<WorktreeWatchSyncLogEntry>, String> {
4322    if !path.exists() {
4323        return Ok(None);
4324    }
4325    let values = read_jsonl_values(path)?;
4326    let Some(value) = values.into_iter().last() else {
4327        return Ok(None);
4328    };
4329    serde_json::from_value(value)
4330        .map(Some)
4331        .map_err(|error| format!("Failed to decode sync log {}: {error}", path.display()))
4332}
4333
4334fn is_synced_spool_path(path: &Path) -> bool {
4335    path.file_name()
4336        .and_then(|value| value.to_str())
4337        .map(|value| value.contains(".synced."))
4338        .unwrap_or(false)
4339}
4340
4341fn mark_synced(path: &Path) -> Result<(), String> {
4342    let Some(file_name) = path.file_name().and_then(|value| value.to_str()) else {
4343        return Ok(());
4344    };
4345    let synced_name = file_name.replace(
4346        ".jsonl",
4347        &format!(".synced.{}.jsonl", Utc::now().timestamp()),
4348    );
4349    let synced_path = path.with_file_name(synced_name);
4350    fs::rename(path, &synced_path).map_err(|error| {
4351        format!(
4352            "Failed to mark spool file {} as synced: {error}",
4353            path.display()
4354        )
4355    })
4356}
4357
4358fn read_spool_records(path: &Path, kind: SyncFileKind) -> Result<SyncRecords, String> {
4359    match kind {
4360        SyncFileKind::Events => read_jsonl_records::<WorktreeMutationEvent>(path)
4361            .map(SyncRecords::Events)
4362            .map_err(|error| format!("Failed to decode event from {}: {error}", path.display())),
4363        SyncFileKind::Commits => read_jsonl_records::<WorktreeCommitEvent>(path)
4364            .map(SyncRecords::Commits)
4365            .map_err(|error| format!("Failed to decode commit from {}: {error}", path.display())),
4366    }
4367}
4368
4369/// Strip whitespace and sparse-file / torn-write NUL padding before JSON parse.
4370/// Windows concurrent appends have produced lines that start with long `\0` runs
4371/// then valid JSON; parsing the raw line yields "expected value at line 1 column 1".
4372fn sanitize_jsonl_line(line: &str) -> Option<&str> {
4373    let trimmed = line.trim().trim_matches('\0').trim();
4374    if trimmed.is_empty() {
4375        None
4376    } else {
4377        Some(trimmed)
4378    }
4379}
4380
4381fn read_jsonl_records<T>(path: &Path) -> Result<Vec<T>, String>
4382where
4383    T: for<'de> Deserialize<'de>,
4384{
4385    let file = File::open(path)
4386        .map_err(|error| format!("Failed to open spool file {}: {error}", path.display()))?;
4387    let reader = BufReader::new(file);
4388    let mut records = Vec::new();
4389    for (line_no, line) in reader.lines().enumerate() {
4390        let line =
4391            line.map_err(|error| format!("Failed to read spool file {}: {error}", path.display()))?;
4392        let Some(payload) = sanitize_jsonl_line(&line) else {
4393            continue;
4394        };
4395        match serde_json::from_str(payload) {
4396            Ok(record) => records.push(record),
4397            Err(error) => {
4398                // Skip unparseable lines (partial writes, encoding glitches) rather than
4399                // killing worktree-watch / dedupe load for the whole repo.
4400                eprintln!(
4401                    "warning: skipping unparseable JSONL record in {} at line {}: {error}",
4402                    path.display(),
4403                    line_no + 1
4404                );
4405            }
4406        }
4407    }
4408    Ok(records)
4409}
4410
4411fn read_jsonl_values(path: &Path) -> Result<Vec<Value>, String> {
4412    read_jsonl_records(path)
4413}
4414
4415fn spawn_detached_worktree_watch(repo: Option<&Path>) -> Result<PathBuf, String> {
4416    let identity = resolve_repo_identity(repo)?;
4417    spawn_detached_worktree_watch_for_identity(&identity)
4418}
4419
4420fn spawn_detached_parent_worktree_watch(parent: &Path) -> Result<PathBuf, String> {
4421    let parent = canonical_parent_path(parent)?;
4422    let layout = prepare_parent_spool_layout(&parent)?;
4423    replace_existing_parent_watcher(&parent, &layout)?;
4424    let executable = std::env::current_exe()
4425        .map_err(|error| format!("Failed to resolve current XBP executable: {error}"))?;
4426    let mut command = Command::new(&executable);
4427    command
4428        .arg("worktree-watch")
4429        .arg("start")
4430        .arg("--parent")
4431        .arg(&parent)
4432        .env(BACKGROUND_CHILD_ENV, "1")
4433        .current_dir(&parent)
4434        .stdin(Stdio::null())
4435        .stdout(Stdio::null())
4436        .stderr(Stdio::null());
4437    configure_detached_child_process(&mut command);
4438
4439    let child = command
4440        .spawn()
4441        .map_err(|error| format!("Failed to spawn background parent worktree watcher: {error}"))?;
4442    write_parent_watcher_state(&parent, &layout, child.id(), &executable)?;
4443    Ok(parent)
4444}
4445
4446fn spawn_detached_worktree_watch_for_identity(identity: &RepoIdentity) -> Result<PathBuf, String> {
4447    let layout = prepare_spool_layout(identity)?;
4448    replace_existing_watcher(identity, &layout)?;
4449    let executable = std::env::current_exe()
4450        .map_err(|error| format!("Failed to resolve current XBP executable: {error}"))?;
4451    let mut command = Command::new(&executable);
4452    command
4453        .arg("worktree-watch")
4454        .arg("start")
4455        .arg("--repo")
4456        .arg(&identity.root)
4457        .env(BACKGROUND_CHILD_ENV, "1")
4458        .current_dir(&identity.root)
4459        .stdin(Stdio::null())
4460        .stdout(Stdio::null())
4461        .stderr(Stdio::null());
4462    configure_detached_child_process(&mut command);
4463
4464    let child: std::process::Child = command
4465        .spawn()
4466        .map_err(|error| format!("Failed to spawn background worktree watcher: {error}"))?;
4467    write_watcher_state(identity, &layout, child.id(), &executable)?;
4468    Ok(identity.root.clone())
4469}
4470
4471fn configure_detached_child_process(command: &mut Command) {
4472    #[cfg(windows)]
4473    {
4474        use std::os::windows::process::CommandExt;
4475        command.creation_flags(CREATE_NO_WINDOW);
4476    }
4477
4478    // On Unix (including WSL), start a new session so the watcher survives the
4479    // launching shell and does not receive SIGHUP when the parent exits.
4480    #[cfg(unix)]
4481    {
4482        use std::os::unix::process::CommandExt;
4483        // SAFETY: setsid is called only in the child just after fork, before exec.
4484        unsafe {
4485            command.pre_exec(|| {
4486                extern "C" {
4487                    fn setsid() -> i32;
4488                }
4489                let _ = setsid();
4490                Ok(())
4491            });
4492        }
4493    }
4494}
4495
4496fn replace_existing_parent_watcher(
4497    parent: &Path,
4498    layout: &ParentSpoolLayout,
4499) -> Result<(), String> {
4500    let Some(state) = read_parent_watcher_state(&layout.watcher_state_file)? else {
4501        return Ok(());
4502    };
4503
4504    if path_identity_key(parent) != path_identity_key(Path::new(&state.parent_root)) {
4505        return Ok(());
4506    }
4507
4508    if is_matching_parent_watcher_process(&state) {
4509        stop_parent_process(&state, false)?;
4510        println!(
4511            "Replaced existing background parent worktree watcher {} for {}",
4512            state.pid,
4513            parent.display()
4514        );
4515        let _ = fs::remove_file(&layout.watcher_state_file);
4516        return Ok(());
4517    }
4518
4519    if parent_watcher_state_pid_exists_with_same_executable(&state) {
4520        return Err(format!(
4521            "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.",
4522            parent.display(),
4523            state.pid,
4524            parent.display()
4525        ));
4526    }
4527
4528    let _ = fs::remove_file(&layout.watcher_state_file);
4529    Ok(())
4530}
4531
4532fn replace_existing_watcher(identity: &RepoIdentity, layout: &SpoolLayout) -> Result<(), String> {
4533    let Some(state) = read_watcher_state(&layout.watcher_state_file)? else {
4534        return Ok(());
4535    };
4536
4537    if !same_repo_watcher_state(identity, &state) {
4538        return Ok(());
4539    }
4540
4541    if is_matching_watcher_process(&state) {
4542        stop_process(&state, false)?;
4543        println!(
4544            "Replaced existing background worktree watcher {} for {}",
4545            state.pid,
4546            identity.root.display()
4547        );
4548        let _ = fs::remove_file(&layout.watcher_state_file);
4549        return Ok(());
4550    }
4551
4552    if watcher_state_pid_exists_with_same_executable(&state) {
4553        return Err(format!(
4554            "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.",
4555            identity.root.display(),
4556            state.pid,
4557            identity.root.display()
4558        ));
4559    }
4560
4561    let _ = fs::remove_file(&layout.watcher_state_file);
4562    Ok(())
4563}
4564
4565fn stop_existing_watcher(
4566    identity: &RepoIdentity,
4567    layout: &SpoolLayout,
4568    force: bool,
4569) -> Result<StopWatcherOutcome, String> {
4570    let Some(state) = read_watcher_state(&layout.watcher_state_file)? else {
4571        return Ok(StopWatcherOutcome::NoState);
4572    };
4573
4574    if !same_repo_watcher_state(identity, &state) {
4575        return Ok(StopWatcherOutcome::NoState);
4576    }
4577
4578    if is_matching_watcher_process(&state) || force {
4579        let pid = state.pid;
4580        stop_process(&state, force)?;
4581        let _ = fs::remove_file(&layout.watcher_state_file);
4582        return Ok(StopWatcherOutcome::Stopped(pid));
4583    }
4584
4585    if watcher_state_pid_exists_with_same_executable(&state) {
4586        return Err(format!(
4587            "Watcher state for {} points to running XBP process {}, but its command line could not be verified. Re-run with `--force` to stop that PID.",
4588            identity.root.display(),
4589            state.pid
4590        ));
4591    }
4592
4593    let _ = fs::remove_file(&layout.watcher_state_file);
4594    Ok(StopWatcherOutcome::RemovedStaleState)
4595}
4596
4597fn stop_existing_parent_watcher(parent: &Path, force: bool) -> Result<StopWatcherOutcome, String> {
4598    let layout = prepare_parent_spool_layout(parent)?;
4599    let Some(state) = read_parent_watcher_state(&layout.watcher_state_file)? else {
4600        return Ok(StopWatcherOutcome::NoState);
4601    };
4602
4603    if path_identity_key(parent) != path_identity_key(Path::new(&state.parent_root)) {
4604        return Ok(StopWatcherOutcome::NoState);
4605    }
4606
4607    if is_matching_parent_watcher_process(&state) || force {
4608        let pid = state.pid;
4609        stop_parent_process(&state, force)?;
4610        let _ = fs::remove_file(&layout.watcher_state_file);
4611        return Ok(StopWatcherOutcome::Stopped(pid));
4612    }
4613
4614    if parent_watcher_state_pid_exists_with_same_executable(&state) {
4615        return Err(format!(
4616            "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.",
4617            parent.display(),
4618            state.pid
4619        ));
4620    }
4621
4622    let _ = fs::remove_file(&layout.watcher_state_file);
4623    Ok(StopWatcherOutcome::RemovedStaleState)
4624}
4625
4626fn read_watcher_state(path: &Path) -> Result<Option<WorktreeWatcherState>, String> {
4627    // Prefer runtime-scoped state; fall back to legacy shared watcher-state.json.
4628    for candidate in watcher_state_read_candidates(path) {
4629        if !candidate.exists() {
4630            continue;
4631        }
4632        let raw = fs::read_to_string(&candidate).map_err(|error| {
4633            format!(
4634                "Failed to read watcher state {}: {error}",
4635                candidate.display()
4636            )
4637        })?;
4638        return serde_json::from_str(&raw).map(Some).map_err(|error| {
4639            format!(
4640                "Failed to parse watcher state {}: {error}",
4641                candidate.display()
4642            )
4643        });
4644    }
4645    Ok(None)
4646}
4647
4648fn watcher_state_read_candidates(path: &Path) -> Vec<PathBuf> {
4649    let mut candidates = vec![path.to_path_buf()];
4650    if let Some(parent) = path.parent() {
4651        let legacy = parent.join("watcher-state.json");
4652        if legacy != path {
4653            candidates.push(legacy);
4654        }
4655    }
4656    candidates
4657}
4658
4659fn read_parent_watcher_state(path: &Path) -> Result<Option<ParentWorktreeWatcherState>, String> {
4660    if !path.exists() {
4661        return Ok(None);
4662    }
4663    let raw = fs::read_to_string(path).map_err(|error| {
4664        format!(
4665            "Failed to read parent watcher state {}: {error}",
4666            path.display()
4667        )
4668    })?;
4669    serde_json::from_str(&raw).map(Some).map_err(|error| {
4670        format!(
4671            "Failed to parse parent watcher state {}: {error}",
4672            path.display()
4673        )
4674    })
4675}
4676
4677fn write_watcher_state(
4678    identity: &RepoIdentity,
4679    layout: &SpoolLayout,
4680    pid: u32,
4681    executable: &Path,
4682) -> Result<(), String> {
4683    let state = WorktreeWatcherState {
4684        pid,
4685        repo_root: normalize_repo_root_for_payload(&identity.root),
4686        repository_owner: identity.owner.clone(),
4687        repository_name: identity.name.clone(),
4688        branch_name: identity.branch.clone(),
4689        executable: executable.display().to_string(),
4690        started_at: Utc::now(),
4691        runtime_key: Some(worktree_runtime_key()),
4692        platform: Some(std::env::consts::OS.to_string()),
4693    };
4694    let rendered = serde_json::to_string_pretty(&state)
4695        .map_err(|error| format!("Failed to serialize watcher state: {error}"))?;
4696    fs::write(&layout.watcher_state_file, format!("{rendered}\n")).map_err(|error| {
4697        format!(
4698            "Failed to write watcher state {}: {error}",
4699            layout.watcher_state_file.display()
4700        )
4701    })
4702}
4703
4704fn write_parent_watcher_state(
4705    parent: &Path,
4706    layout: &ParentSpoolLayout,
4707    pid: u32,
4708    executable: &Path,
4709) -> Result<(), String> {
4710    let state = ParentWorktreeWatcherState {
4711        pid,
4712        parent_root: normalize_repo_root_for_payload(parent),
4713        executable: executable.display().to_string(),
4714        started_at: Utc::now(),
4715        runtime_key: Some(worktree_runtime_key()),
4716        platform: Some(std::env::consts::OS.to_string()),
4717    };
4718    let rendered = serde_json::to_string_pretty(&state)
4719        .map_err(|error| format!("Failed to serialize parent watcher state: {error}"))?;
4720    fs::write(&layout.watcher_state_file, format!("{rendered}\n")).map_err(|error| {
4721        format!(
4722            "Failed to write parent watcher state {}: {error}",
4723            layout.watcher_state_file.display()
4724        )
4725    })
4726}
4727
4728fn same_repo_watcher_state(identity: &RepoIdentity, state: &WorktreeWatcherState) -> bool {
4729    if !runtime_keys_compatible(state.runtime_key.as_deref()) {
4730        return false;
4731    }
4732    path_identity_key(&identity.root) == path_identity_key(Path::new(&state.repo_root))
4733        && identity.owner == state.repository_owner
4734        && identity.name == state.repository_name
4735        && identity.branch == state.branch_name
4736}
4737
4738fn runtime_keys_compatible(state_runtime: Option<&str>) -> bool {
4739    let current = worktree_runtime_key();
4740    match state_runtime {
4741        Some(key) => key == current,
4742        // Legacy state files (pre-runtime isolation) only match native Windows/Linux
4743        // paths that still live outside the runtimes/ spool tree — never cross WSL.
4744        None => !is_wsl_runtime(),
4745    }
4746}
4747
4748fn process_system_for_pids(pids: &[u32]) -> System {
4749    let mut system = System::new();
4750    if pids.is_empty() {
4751        return system;
4752    }
4753    let pid_list: Vec<Pid> = pids
4754        .iter()
4755        .copied()
4756        .filter(|pid| *pid != 0 && *pid != std::process::id())
4757        .map(Pid::from_u32)
4758        .collect();
4759    if pid_list.is_empty() {
4760        return system;
4761    }
4762    // Refresh only the candidate watcher PIDs (not the entire process table).
4763    system.refresh_processes_specifics(
4764        ProcessesToUpdate::Some(&pid_list),
4765        true,
4766        ProcessRefreshKind::everything()
4767            .with_cmd(UpdateKind::Always)
4768            .with_exe(UpdateKind::Always),
4769    );
4770    system
4771}
4772
4773fn is_matching_watcher_process(state: &WorktreeWatcherState) -> bool {
4774    let system = process_system_for_pids(&[state.pid]);
4775    // "Matching other process" — used by stop/replace; never treats self as match.
4776    is_matching_watcher_process_with_system(&system, state)
4777}
4778
4779/// True when the state PID is a live watcher, including this process (status/API).
4780fn is_live_watcher_process_with_system(system: &System, state: &WorktreeWatcherState) -> bool {
4781    if state.pid == std::process::id() {
4782        return runtime_keys_compatible(state.runtime_key.as_deref());
4783    }
4784    is_matching_watcher_process_with_system(system, state)
4785}
4786
4787fn is_matching_watcher_process_with_system(
4788    system: &System,
4789    state: &WorktreeWatcherState,
4790) -> bool {
4791    if state.pid == std::process::id() {
4792        return false;
4793    }
4794    let Some(process) = system.process(Pid::from_u32(state.pid)) else {
4795        return false;
4796    };
4797    let command_line = process
4798        .cmd()
4799        .iter()
4800        .map(|part| part.to_string_lossy())
4801        .collect::<Vec<_>>()
4802        .join(" ");
4803    let command_matches = command_line_contains_ignore_case(&command_line, "worktree-watch")
4804        && command_line_contains_ignore_case(&command_line, "start")
4805        && command_line_mentions_path(&command_line, &state.repo_root);
4806    command_matches || watcher_state_process_identity_matches(process, state)
4807}
4808
4809fn is_matching_parent_watcher_process(state: &ParentWorktreeWatcherState) -> bool {
4810    let system = process_system_for_pids(&[state.pid]);
4811    is_matching_parent_watcher_process_with_system(&system, state)
4812}
4813
4814/// True when the parent state PID is a live watcher, including this process.
4815fn is_live_parent_watcher_process_with_system(
4816    system: &System,
4817    state: &ParentWorktreeWatcherState,
4818) -> bool {
4819    if state.pid == std::process::id() {
4820        return runtime_keys_compatible(state.runtime_key.as_deref());
4821    }
4822    is_matching_parent_watcher_process_with_system(system, state)
4823}
4824
4825fn is_matching_parent_watcher_process_with_system(
4826    system: &System,
4827    state: &ParentWorktreeWatcherState,
4828) -> bool {
4829    if state.pid == std::process::id() {
4830        return false;
4831    }
4832    if !runtime_keys_compatible(state.runtime_key.as_deref()) {
4833        return false;
4834    }
4835    let Some(process) = system.process(Pid::from_u32(state.pid)) else {
4836        return false;
4837    };
4838    let command_line = process
4839        .cmd()
4840        .iter()
4841        .map(|part| part.to_string_lossy())
4842        .collect::<Vec<_>>()
4843        .join(" ");
4844    let command_matches = command_line_contains_ignore_case(&command_line, "worktree-watch")
4845        && command_line_contains_ignore_case(&command_line, "start")
4846        && command_line_contains_ignore_case(&command_line, "--parent")
4847        && command_line_mentions_path(&command_line, &state.parent_root);
4848    // Accept: verified worktree-watch cmdline, or same executable at this PID
4849    // (Windows often omits args / start_time can be coarse after CREATE_NO_WINDOW).
4850    command_matches
4851        || parent_watcher_state_process_identity_matches(process, state)
4852        || (process_executable_matches_path(process, Path::new(&state.executable))
4853            && process_looks_like_worktree_watch(process))
4854}
4855
4856fn process_looks_like_worktree_watch(process: &sysinfo::Process) -> bool {
4857    let command_line = process
4858        .cmd()
4859        .iter()
4860        .map(|part| part.to_string_lossy())
4861        .collect::<Vec<_>>()
4862        .join(" ");
4863    if command_line_contains_ignore_case(&command_line, "worktree-watch") {
4864        return true;
4865    }
4866    // Empty cmd() is common for some Windows query modes — exe match + state file is enough.
4867    command_line.trim().is_empty()
4868}
4869
4870fn command_line_contains_ignore_case(haystack: &str, needle: &str) -> bool {
4871    haystack.to_ascii_lowercase().contains(&needle.to_ascii_lowercase())
4872}
4873
4874fn command_line_mentions_path(command_line: &str, path: &str) -> bool {
4875    if path.is_empty() {
4876        return false;
4877    }
4878    if command_line.contains(path) {
4879        return true;
4880    }
4881    let command_lower = command_line.to_ascii_lowercase().replace('\\', "/");
4882    let path_lower = path.to_ascii_lowercase().replace('\\', "/");
4883    if command_lower.contains(&path_lower) {
4884        return true;
4885    }
4886    // Cross-platform: C:\Users\… vs /mnt/c/users/…
4887    let path_key = path_identity_key(Path::new(path));
4888    if !path_key.is_empty() && command_lower.contains(&path_key) {
4889        return true;
4890    }
4891    for token in command_line.split_whitespace() {
4892        let cleaned = token.trim_matches('"').trim_matches('\'');
4893        if cleaned.is_empty() {
4894            continue;
4895        }
4896        if path_identity_key(Path::new(cleaned)) == path_key {
4897            return true;
4898        }
4899    }
4900    // Final path segment (GitHub / github) case-insensitive.
4901    Path::new(path)
4902        .file_name()
4903        .and_then(|name| name.to_str())
4904        .map(|name| command_line_contains_ignore_case(command_line, name))
4905        .unwrap_or(false)
4906}
4907
4908fn watcher_state_pid_exists_with_same_executable(state: &WorktreeWatcherState) -> bool {
4909    if state.pid == std::process::id() {
4910        return false;
4911    }
4912    let system = process_system_for_pids(&[state.pid]);
4913    let Some(process) = system.process(Pid::from_u32(state.pid)) else {
4914        return false;
4915    };
4916    process_executable_matches_state(process, state)
4917}
4918
4919fn parent_watcher_state_pid_exists_with_same_executable(
4920    state: &ParentWorktreeWatcherState,
4921) -> bool {
4922    if state.pid == std::process::id() {
4923        return false;
4924    }
4925    let system = process_system_for_pids(&[state.pid]);
4926    let Some(process) = system.process(Pid::from_u32(state.pid)) else {
4927        return false;
4928    };
4929    process_executable_matches_path(process, Path::new(&state.executable))
4930}
4931
4932fn watcher_state_process_identity_matches(
4933    process: &sysinfo::Process,
4934    state: &WorktreeWatcherState,
4935) -> bool {
4936    process_executable_matches_state(process, state)
4937        && process_start_time_matches(process.start_time(), state.started_at)
4938}
4939
4940fn parent_watcher_state_process_identity_matches(
4941    process: &sysinfo::Process,
4942    state: &ParentWorktreeWatcherState,
4943) -> bool {
4944    process_executable_matches_path(process, Path::new(&state.executable))
4945        && process_start_time_matches(process.start_time(), state.started_at)
4946}
4947
4948fn process_executable_matches_state(
4949    process: &sysinfo::Process,
4950    state: &WorktreeWatcherState,
4951) -> bool {
4952    process_executable_matches_path(process, Path::new(&state.executable))
4953}
4954
4955fn process_executable_matches_path(process: &sysinfo::Process, executable: &Path) -> bool {
4956    let expected = path_identity_key(executable);
4957    process
4958        .exe()
4959        .map(|path| path_identity_key(path) == expected)
4960        .unwrap_or(false)
4961}
4962
4963fn process_start_time_matches(process_started: u64, state_started: DateTime<Utc>) -> bool {
4964    // 0 means "unknown" on some platforms / query modes — don't reject the match.
4965    if process_started == 0 {
4966        return true;
4967    }
4968    let process_started = process_started as i64;
4969    let state_started = state_started.timestamp();
4970    // Wider window: spawn → write state can lag, and Windows clocks are coarse.
4971    (process_started - state_started).abs() <= 120
4972}
4973
4974fn stop_process(state: &WorktreeWatcherState, force: bool) -> Result<(), String> {
4975    let system = process_system_for_pids(&[state.pid]);
4976    let Some(process) = system.process(Pid::from_u32(state.pid)) else {
4977        return Ok(());
4978    };
4979    if !force && !is_matching_watcher_process_with_system(&system, state) {
4980        return Ok(());
4981    }
4982    if process.kill() {
4983        Ok(())
4984    } else {
4985        Err(format!(
4986            "Failed to stop existing watcher process {}",
4987            state.pid
4988        ))
4989    }
4990}
4991
4992fn stop_parent_process(state: &ParentWorktreeWatcherState, force: bool) -> Result<(), String> {
4993    let system = process_system_for_pids(&[state.pid]);
4994    let Some(process) = system.process(Pid::from_u32(state.pid)) else {
4995        return Ok(());
4996    };
4997    if !force && !is_matching_parent_watcher_process_with_system(&system, state) {
4998        return Ok(());
4999    }
5000    if process.kill() {
5001        Ok(())
5002    } else {
5003        Err(format!(
5004            "Failed to stop existing parent watcher process {}",
5005            state.pid
5006        ))
5007    }
5008}
5009
5010/// Prefer a live parent watcher so bare `status` reports the active cover.
5011fn prefer_running_parent_root() -> Option<PathBuf> {
5012    let mut candidates = list_parent_watcher_state_files().ok()?;
5013    // Newest state first.
5014    candidates.sort_by(|left, right| right.1.started_at.cmp(&left.1.started_at));
5015    for (_path, state) in candidates {
5016        let system = process_system_for_pids(&[state.pid]);
5017        if !is_live_parent_watcher_process_with_system(&system, &state) {
5018            continue;
5019        }
5020        let root = parent_root_as_local_path(&state.parent_root);
5021        if root.is_dir() {
5022            return Some(root);
5023        }
5024        return Some(root);
5025    }
5026    None
5027}
5028
5029/// Most recent parent cover path (even if the watcher process is not live).
5030fn prefer_any_parent_root() -> Option<PathBuf> {
5031    let mut candidates = list_parent_watcher_state_files().ok()?;
5032    candidates.sort_by(|left, right| right.1.started_at.cmp(&left.1.started_at));
5033    for (_path, state) in candidates {
5034        let root = parent_root_as_local_path(&state.parent_root);
5035        if root.is_dir() {
5036            return Some(root);
5037        }
5038    }
5039    None
5040}
5041
5042fn count_git_roots_under(parent: &Path) -> Result<usize, String> {
5043    let mut roots = Vec::new();
5044    let mut seen = BTreeSet::new();
5045    collect_git_roots_in_dir(parent, parent, &mut seen, &mut roots)?;
5046    Ok(roots.len())
5047}
5048
5049fn list_parent_watcher_state_files() -> Result<Vec<(PathBuf, ParentWorktreeWatcherState)>, String> {
5050    let parents_root = global_mutations_root()?.join("parents");
5051    if !parents_root.is_dir() {
5052        return Ok(Vec::new());
5053    }
5054    let mut out = Vec::new();
5055    for entry in fs::read_dir(&parents_root)
5056        .map_err(|error| format!("Failed to read {}: {error}", parents_root.display()))?
5057    {
5058        let entry = entry.map_err(|error| format!("Failed to read parent spool entry: {error}"))?;
5059        if !entry.file_type().map(|ft| ft.is_dir()).unwrap_or(false) {
5060            continue;
5061        }
5062        for file in fs::read_dir(entry.path()).map_err(|error| {
5063            format!(
5064                "Failed to read parent spool {}: {error}",
5065                entry.path().display()
5066            )
5067        })? {
5068            let file = file.map_err(|error| format!("Failed to read parent state entry: {error}"))?;
5069            let path = file.path();
5070            let Some(name) = path.file_name().and_then(|value| value.to_str()) else {
5071                continue;
5072            };
5073            if !(name.starts_with("watcher-state") && name.ends_with(".json")) {
5074                continue;
5075            }
5076            if let Ok(Some(state)) = read_parent_watcher_state(&path) {
5077                // Prefer runtime-scoped files for this runtime.
5078                if let Some(runtime) = state.runtime_key.as_deref() {
5079                    if runtime != worktree_runtime_key() {
5080                        // Still allow legacy files without runtime_key via runtime_keys_compatible.
5081                        if !runtime_keys_compatible(Some(runtime)) {
5082                            continue;
5083                        }
5084                    }
5085                } else if !runtime_keys_compatible(None) {
5086                    continue;
5087                }
5088                out.push((path, state));
5089            }
5090        }
5091    }
5092    Ok(out)
5093}
5094
5095fn parent_root_as_local_path(parent_root: &str) -> PathBuf {
5096    let trimmed = parent_root.trim();
5097    if trimmed.is_empty() {
5098        return PathBuf::new();
5099    }
5100    if let Some(windows) = map_wsl_mnt_path_to_windows(trimmed) {
5101        return PathBuf::from(windows);
5102    }
5103    let as_path = PathBuf::from(trimmed);
5104    if as_path.is_dir() {
5105        return as_path;
5106    }
5107    // path_identity_key form is lowercase /mnt/c/... even on Windows storage.
5108    if let Some(windows) = map_wsl_mnt_path_to_windows(&trimmed.replace('\\', "/")) {
5109        return PathBuf::from(windows);
5110    }
5111    as_path
5112}
5113
5114fn discover_repo_identities_under(parent: &Path) -> Result<Vec<RepoIdentity>, String> {
5115    let parent = canonical_parent_path(parent)?;
5116
5117    let mut roots = Vec::new();
5118    let mut seen = BTreeSet::new();
5119    collect_git_roots_in_dir(&parent, &parent, &mut seen, &mut roots)?;
5120
5121    // Resolve identities in parallel — sequential git spawns for 60+ repos
5122    // made status/API unusable on Windows.
5123    let identities = resolve_repo_identities_parallel(&roots);
5124    let mut identities = identities.into_iter().flatten().collect::<Vec<_>>();
5125    identities.sort_by(|left, right| left.root.cmp(&right.root));
5126    Ok(identities)
5127}
5128
5129fn collect_git_roots_in_dir(
5130    parent: &Path,
5131    dir: &Path,
5132    seen: &mut BTreeSet<String>,
5133    roots: &mut Vec<PathBuf>,
5134) -> Result<(), String> {
5135    if dir != parent && is_skipped_discovery_dir(dir) {
5136        return Ok(());
5137    }
5138
5139    if dir.join(".git").exists() {
5140        let key = path_identity_key(dir);
5141        if seen.insert(key) {
5142            roots.push(dir.to_path_buf());
5143        }
5144        return Ok(());
5145    }
5146
5147    let entries = fs::read_dir(dir)
5148        .map_err(|error| format!("Failed to read folder {}: {error}", dir.display()))?;
5149    for entry in entries {
5150        let entry = entry.map_err(|error| {
5151            format!(
5152                "Failed to read folder entry under {}: {error}",
5153                dir.display()
5154            )
5155        })?;
5156        let file_type = entry
5157            .file_type()
5158            .map_err(|error| format!("Failed to inspect {}: {error}", entry.path().display()))?;
5159        if file_type.is_dir() {
5160            collect_git_roots_in_dir(parent, &entry.path(), seen, roots)?;
5161        }
5162    }
5163
5164    Ok(())
5165}
5166
5167fn resolve_repo_identities_parallel(roots: &[PathBuf]) -> Vec<Option<RepoIdentity>> {
5168    if roots.is_empty() {
5169        return Vec::new();
5170    }
5171    if roots.len() == 1 {
5172        return vec![resolve_repo_identity_light(Some(&roots[0])).ok()];
5173    }
5174
5175    let thread_count = roots.len().min(8).max(1);
5176    let chunk_size = (roots.len() + thread_count - 1) / thread_count;
5177    let mut handles = Vec::new();
5178    for chunk in roots.chunks(chunk_size) {
5179        let chunk = chunk.to_vec();
5180        handles.push(std::thread::spawn(move || {
5181            chunk
5182                .iter()
5183                .map(|root| resolve_repo_identity_light(Some(root)).ok())
5184                .collect::<Vec<_>>()
5185        }));
5186    }
5187
5188    let mut out = Vec::with_capacity(roots.len());
5189    for handle in handles {
5190        match handle.join() {
5191            Ok(part) => out.extend(part),
5192            Err(_) => {
5193                // Fallback sequential if a worker panics.
5194            }
5195        }
5196    }
5197    // If a worker panicked, fill remaining via sequential so we never under-report silently.
5198    if out.len() < roots.len() {
5199        for root in roots.iter().skip(out.len()) {
5200            out.push(resolve_repo_identity_light(Some(root)).ok());
5201        }
5202    }
5203    out
5204}
5205
5206/// Status/tray identity resolve: skip HEAD SHA (one less git spawn per repo).
5207fn resolve_repo_identity_light(repo: Option<&Path>) -> Result<RepoIdentity, String> {
5208    let start = resolve_repo_path_input(repo)?;
5209    let root_raw = git_output(&start, &["rev-parse", "--show-toplevel"]).map_err(|error| {
5210        if repo.is_some() {
5211            format!(
5212                "Failed to run git rev-parse --show-toplevel in {}: {error}",
5213                start.display()
5214            )
5215        } else {
5216            format!("Failed to run git rev-parse --show-toplevel: {error}")
5217        }
5218    })?;
5219    let root = normalize_windows_verbatim_path(
5220        fs::canonicalize(root_raw.trim()).unwrap_or_else(|_| PathBuf::from(root_raw.trim())),
5221    );
5222    let branch = git_output(&root, &["rev-parse", "--abbrev-ref", "HEAD"])
5223        .unwrap_or_else(|_| "unknown".to_string());
5224    let remote = git_output(&root, &["remote", "get-url", DEFAULT_REMOTE]).unwrap_or_default();
5225    let (owner, name) = parse_remote_owner_repo(&remote).unwrap_or_else(|| {
5226        let name = root
5227            .file_name()
5228            .and_then(|value| value.to_str())
5229            .unwrap_or("repository")
5230            .to_string();
5231        ("unknown".to_string(), name)
5232    });
5233
5234    Ok(RepoIdentity {
5235        owner,
5236        name,
5237        branch: sanitize_path_component(branch.trim()),
5238        root,
5239        head_sha: None,
5240    })
5241}
5242
5243fn resolve_repo_identity(repo: Option<&Path>) -> Result<RepoIdentity, String> {
5244    let start = resolve_repo_path_input(repo)?;
5245    let root_raw = git_output(&start, &["rev-parse", "--show-toplevel"]).map_err(|error| {
5246        if repo.is_some() {
5247            format!(
5248                "Failed to run git rev-parse --show-toplevel in {}: {error}",
5249                start.display()
5250            )
5251        } else {
5252            format!("Failed to run git rev-parse --show-toplevel: {error}")
5253        }
5254    })?;
5255    let root = normalize_windows_verbatim_path(
5256        fs::canonicalize(root_raw.trim()).unwrap_or_else(|_| PathBuf::from(root_raw.trim())),
5257    );
5258    let branch = git_output(&root, &["rev-parse", "--abbrev-ref", "HEAD"])
5259        .unwrap_or_else(|_| "unknown".to_string());
5260    let remote = git_output(&root, &["remote", "get-url", DEFAULT_REMOTE]).unwrap_or_default();
5261    let (owner, name) = parse_remote_owner_repo(&remote).unwrap_or_else(|| {
5262        let name = root
5263            .file_name()
5264            .and_then(|value| value.to_str())
5265            .unwrap_or("repository")
5266            .to_string();
5267        ("unknown".to_string(), name)
5268    });
5269    let head_sha = current_head(&root)?;
5270
5271    Ok(RepoIdentity {
5272        owner,
5273        name,
5274        branch: sanitize_path_component(branch.trim()),
5275        root,
5276        head_sha,
5277    })
5278}
5279
5280/// Resolve `--repo` / `--repos` values to a filesystem path.
5281///
5282/// Accepts:
5283/// - absolute or relative git roots that exist on disk
5284/// - short repo names / `owner/repo` / full paths from
5285///   `~/.xbp/config.yaml` → `system_inventory.github_repos`
5286fn resolve_repo_path_input(repo: Option<&Path>) -> Result<PathBuf, String> {
5287    let Some(input) = repo else {
5288        return std::env::current_dir()
5289            .map_err(|error| format!("Failed to resolve current directory: {error}"));
5290    };
5291
5292    if path_looks_like_existing_dir(input) {
5293        return Ok(input.to_path_buf());
5294    }
5295
5296    let token = input.to_string_lossy();
5297    let token = token.trim();
5298    if token.is_empty() {
5299        return Err("--repo value is empty".to_string());
5300    }
5301
5302    // Bare names (athena) and owner/repo (xylex-group/athena) never contain path
5303    // separators on disk until inventory expands them; try inventory first.
5304    if let Some(path) = lookup_repo_path_from_system_inventory(token)? {
5305        return Ok(path);
5306    }
5307
5308    // Keep original so callers can surface a path-oriented git error for real paths.
5309    if input.components().count() > 1
5310        || token.contains('/')
5311        || token.contains('\\')
5312        || token.contains(':')
5313    {
5314        return Ok(input.to_path_buf());
5315    }
5316
5317    Err(format!(
5318        "Could not resolve `--repo {token}` as a filesystem path or inventory name. \
5319Refresh with `xbp diag --refresh-system-inventory`, pass a full path, or use `--parent`."
5320    ))
5321}
5322
5323fn path_looks_like_existing_dir(path: &Path) -> bool {
5324    path.is_dir()
5325        || fs::metadata(path)
5326            .map(|meta| meta.is_dir())
5327            .unwrap_or(false)
5328}
5329
5330fn lookup_repo_path_from_system_inventory(token: &str) -> Result<Option<PathBuf>, String> {
5331    let config = match SshConfig::load() {
5332        Ok(config) => config,
5333        Err(_) => return Ok(None),
5334    };
5335    let Some(inventory) = config.system_inventory.as_ref() else {
5336        return Ok(None);
5337    };
5338
5339    if let Some(path) = match_github_repo_inventory(&inventory.github_repos, token)? {
5340        return Ok(Some(path));
5341    }
5342
5343    // Fall back to discovered XBP project names (folder / project name).
5344    let needle = normalize_repo_lookup_token(token);
5345    let mut project_hits: Vec<PathBuf> = Vec::new();
5346    for project in &inventory.xbp_projects {
5347        let name = normalize_repo_lookup_token(&project.name);
5348        let root_name = Path::new(&project.root)
5349            .file_name()
5350            .and_then(|value| value.to_str())
5351            .map(normalize_repo_lookup_token)
5352            .unwrap_or_default();
5353        if name == needle || root_name == needle {
5354            let path = PathBuf::from(&project.root);
5355            if path_looks_like_existing_dir(&path)
5356                && !project_hits.iter().any(|existing| {
5357                    path_identity_key(existing.as_path()) == path_identity_key(path.as_path())
5358                })
5359            {
5360                project_hits.push(path);
5361            }
5362        }
5363    }
5364    match project_hits.len() {
5365        0 => Ok(None),
5366        1 => Ok(Some(project_hits.remove(0))),
5367        _ => Err(format_ambiguous_repo_matches(token, &project_hits)),
5368    }
5369}
5370
5371fn match_github_repo_inventory(
5372    repos: &[GithubRepoRecord],
5373    token: &str,
5374) -> Result<Option<PathBuf>, String> {
5375    let needle = normalize_repo_lookup_token(token);
5376    if needle.is_empty() {
5377        return Ok(None);
5378    }
5379
5380    let mut exact_path: Vec<PathBuf> = Vec::new();
5381    let mut full_name_hits: Vec<PathBuf> = Vec::new();
5382    let mut short_name_hits: Vec<PathBuf> = Vec::new();
5383
5384    for record in repos {
5385        let path = PathBuf::from(record.path.trim());
5386        if record.path.trim().is_empty() || !path_looks_like_existing_dir(&path) {
5387            continue;
5388        }
5389
5390        let path_key = path_identity_key(&path);
5391        let full_name = normalize_repo_lookup_token(&record.full_name);
5392        let short = normalize_repo_lookup_token(&record.repo);
5393        let owner_repo = normalize_repo_lookup_token(&format!("{}/{}", record.owner, record.repo));
5394        let folder = path
5395            .file_name()
5396            .and_then(|value| value.to_str())
5397            .map(normalize_repo_lookup_token)
5398            .unwrap_or_default();
5399
5400        let push_unique = |bucket: &mut Vec<PathBuf>| {
5401            if !bucket
5402                .iter()
5403                .any(|existing| path_identity_key(existing.as_path()) == path_key)
5404            {
5405                bucket.push(path.clone());
5406            }
5407        };
5408
5409        if path_identity_key(Path::new(record.path.trim())) == path_identity_key(Path::new(token))
5410            || normalize_repo_lookup_token(&record.path) == needle
5411        {
5412            push_unique(&mut exact_path);
5413        }
5414        if full_name == needle || owner_repo == needle {
5415            push_unique(&mut full_name_hits);
5416        }
5417        if short == needle || folder == needle {
5418            push_unique(&mut short_name_hits);
5419        }
5420    }
5421
5422    if let Some(path) = exact_path.into_iter().next() {
5423        return Ok(Some(path));
5424    }
5425    match full_name_hits.len() {
5426        0 => {}
5427        1 => return Ok(Some(full_name_hits.remove(0))),
5428        _ => return Err(format_ambiguous_repo_matches(token, &full_name_hits)),
5429    }
5430    match short_name_hits.len() {
5431        0 => Ok(None),
5432        1 => Ok(Some(short_name_hits.remove(0))),
5433        _ => Err(format_ambiguous_repo_matches(token, &short_name_hits)),
5434    }
5435}
5436
5437fn normalize_repo_lookup_token(value: &str) -> String {
5438    value
5439        .trim()
5440        .trim_end_matches('/')
5441        .trim_end_matches('\\')
5442        .trim_end_matches(".git")
5443        .replace('\\', "/")
5444        .to_ascii_lowercase()
5445}
5446
5447fn format_ambiguous_repo_matches(token: &str, paths: &[PathBuf]) -> String {
5448    let list = paths
5449        .iter()
5450        .map(|path| format!("  - {}", path.display()))
5451        .collect::<Vec<_>>()
5452        .join("\n");
5453    format!(
5454        "Ambiguous `--repo {token}` matches multiple inventory paths:\n{list}\n\
5455Use a full path or `owner/repo` (for example `xylex-group/{token}`)."
5456    )
5457}
5458
5459fn prepare_spool_layout(identity: &RepoIdentity) -> Result<SpoolLayout, String> {
5460    let root = repository_spool_root(identity)?.join(sanitize_path_component(&identity.branch));
5461    fs::create_dir_all(&root).map_err(|error| {
5462        format!(
5463            "Failed to create worktree mutation spool {}: {error}",
5464            root.display()
5465        )
5466    })?;
5467    let event_dedupe_dir = root.join("event-dedupe");
5468    fs::create_dir_all(&event_dedupe_dir).map_err(|error| {
5469        format!(
5470            "Failed to create worktree mutation dedupe dir {}: {error}",
5471            event_dedupe_dir.display()
5472        )
5473    })?;
5474    let commit_dedupe_dir = root.join("commit-dedupe");
5475    fs::create_dir_all(&commit_dedupe_dir).map_err(|error| {
5476        format!(
5477            "Failed to create worktree commit dedupe dir {}: {error}",
5478            commit_dedupe_dir.display()
5479        )
5480    })?;
5481    Ok(spool_layout_for_root(
5482        root,
5483        Some(Uuid::new_v4().to_string()),
5484    ))
5485}
5486
5487/// Canonical mutations root under the established global XBP home
5488/// (`XBP_HOME` / Windows profile `.xbp` / `~/.xbp`), not the repo.
5489fn global_mutations_root() -> Result<PathBuf, String> {
5490    let paths = ensure_global_xbp_paths()?;
5491    Ok(paths.root_dir.join("mutations"))
5492}
5493
5494fn repository_spool_root(identity: &RepoIdentity) -> Result<PathBuf, String> {
5495    Ok(global_mutations_root()?
5496        .join(sanitize_path_component(&identity.owner))
5497        .join(sanitize_path_component(&identity.name)))
5498}
5499
5500/// All mutation roots that may already hold data for this identity.
5501///
5502/// Primary root is the established global XBP home; legacy/alternate homes and
5503/// older `runtimes/*` layouts are included so WSL/Windows can still read each
5504/// other's historical spoils while new writes go to the primary path.
5505fn repository_spool_search_roots(identity: &RepoIdentity) -> Result<Vec<PathBuf>, String> {
5506    let owner = sanitize_path_component(&identity.owner);
5507    let name = sanitize_path_component(&identity.name);
5508    let mut roots: Vec<PathBuf> = Vec::new();
5509    let mut push = |path: PathBuf| {
5510        if roots.iter().any(|existing| {
5511            path_identity_key(existing.as_path()) == path_identity_key(path.as_path())
5512        }) {
5513            return;
5514        }
5515        roots.push(path);
5516    };
5517
5518    push(repository_spool_root(identity)?);
5519
5520    for base in legacy_mutations_base_candidates() {
5521        push(base.join(&owner).join(&name));
5522        // Previous runtime-isolated layout.
5523        let runtimes = base.join("runtimes");
5524        if runtimes.is_dir() {
5525            if let Ok(entries) = fs::read_dir(&runtimes) {
5526                for entry in entries.flatten() {
5527                    if entry.file_type().map(|ft| ft.is_dir()).unwrap_or(false) {
5528                        push(entry.path().join(&owner).join(&name));
5529                    }
5530                }
5531            }
5532        }
5533    }
5534
5535    Ok(roots)
5536}
5537
5538fn legacy_mutations_base_candidates() -> Vec<PathBuf> {
5539    let mut bases: Vec<PathBuf> = Vec::new();
5540    let mut push = |path: PathBuf| {
5541        if bases
5542            .iter()
5543            .any(|existing| path_identity_key(existing.as_path()) == path_identity_key(path.as_path()))
5544        {
5545            return;
5546        }
5547        bases.push(path);
5548    };
5549
5550    if let Ok(primary) = global_mutations_root() {
5551        push(primary);
5552    }
5553    // Also surface the raw resolved root without create-side effects for alternate spellings.
5554    push(resolve_global_xbp_root_dir().join("mutations"));
5555
5556    if let Some(home) = dirs::home_dir() {
5557        push(home.join(".xbp").join("mutations"));
5558        push(home.join(".config").join("xbp").join("mutations"));
5559    }
5560
5561    // Dual-map the primary mutations root across WSL/Windows path spellings.
5562    if let Ok(primary) = global_mutations_root() {
5563        let rendered = primary.to_string_lossy().replace('\\', "/");
5564        if let Some(mnt) = map_windows_path_to_wsl_mnt(&rendered) {
5565            push(PathBuf::from(mnt));
5566        }
5567        if let Some(windows) = map_wsl_mnt_path_to_windows(&rendered) {
5568            if let Some(mnt) = map_windows_path_to_wsl_mnt(&windows) {
5569                push(PathBuf::from(mnt));
5570            }
5571            push(PathBuf::from(windows));
5572        }
5573    }
5574
5575    bases
5576}
5577
5578fn spool_layout_for_existing_root(root: &Path) -> SpoolLayout {
5579    spool_layout_for_root(root.to_path_buf(), None)
5580}
5581
5582fn spool_layout_for_root(root: PathBuf, run_id: Option<String>) -> SpoolLayout {
5583    let run_id = run_id.unwrap_or_else(|| Uuid::new_v4().to_string());
5584    let event_dedupe_dir = root.join("event-dedupe");
5585    let commit_dedupe_dir = root.join("commit-dedupe");
5586    // Events/stats/dedupe are shared across Windows+WSL on the established home.
5587    // Watcher PID state is runtime-scoped so both can run without clobbering stop/start.
5588    let watcher_state_file = root.join(format!(
5589        "watcher-state-{}.json",
5590        sanitize_path_component(&worktree_runtime_key())
5591    ));
5592    SpoolLayout {
5593        events_file: root.join(format!("events-{run_id}.jsonl")),
5594        commits_file: root.join(format!("commits-{run_id}.jsonl")),
5595        stats_file: root.join("stats.json"),
5596        watcher_state_file,
5597        sync_log_file: root.join("sync-log.jsonl"),
5598        event_dedupe_file: root.join("event-dedupe.jsonl"),
5599        event_dedupe_dir,
5600        commit_dedupe_dir,
5601        root,
5602    }
5603}
5604
5605fn prepare_parent_spool_layout(parent: &Path) -> Result<ParentSpoolLayout, String> {
5606    let parent = canonical_parent_path(parent)?;
5607    // Prefer a path-identity that is stable across WSL `/mnt/c/...` and Windows `C:\...`.
5608    let parent_key = cross_platform_path_identity_key(&parent);
5609    let mut hasher = Sha256::new();
5610    hasher.update(parent_key.as_bytes());
5611    let digest = format!("{:x}", hasher.finalize());
5612    let label = parent
5613        .file_name()
5614        .and_then(|value| value.to_str())
5615        .map(sanitize_path_component)
5616        .filter(|value| !value.is_empty())
5617        .unwrap_or_else(|| "parent".to_string());
5618    let root = global_mutations_root()?
5619        .join("parents")
5620        .join(format!("{}-{}", label, &digest[..16]));
5621    fs::create_dir_all(&root).map_err(|error| {
5622        format!(
5623            "Failed to create parent worktree watcher state dir {}: {error}",
5624            root.display()
5625        )
5626    })?;
5627    Ok(ParentSpoolLayout {
5628        watcher_state_file: root.join(format!(
5629            "watcher-state-{}.json",
5630            sanitize_path_component(&worktree_runtime_key())
5631        )),
5632    })
5633}
5634
5635fn canonical_parent_path(parent: &Path) -> Result<PathBuf, String> {
5636    let normalized = normalize_windows_verbatim_path(
5637        fs::canonicalize(parent).unwrap_or_else(|_| parent.to_path_buf()),
5638    );
5639    if !normalized.is_dir() {
5640        return Err(format!(
5641            "Parent folder {} does not exist or is not a directory.",
5642            normalized.display()
5643        ));
5644    }
5645    Ok(normalized)
5646}
5647
5648fn append_json_line<T: Serialize>(path: &Path, value: &T) -> Result<(), String> {
5649    let mut file = OpenOptions::new()
5650        .create(true)
5651        .append(true)
5652        .open(path)
5653        .map_err(|error| format!("Failed to open spool file {}: {error}", path.display()))?;
5654    let line = serde_json::to_string(value)
5655        .map_err(|error| format!("Failed to serialize worktree event: {error}"))?;
5656    writeln!(file, "{line}")
5657        .map_err(|error| format!("Failed to append spool file {}: {error}", path.display()))
5658}
5659
5660fn git_numstat_for_path(repo_root: &Path, relative_path: &str) -> Result<(u64, u64), String> {
5661    let output = Command::new("git")
5662        .args(["diff", "--numstat", "--"])
5663        .arg(relative_path)
5664        .current_dir(repo_root)
5665        .output()
5666        .map_err(|error| format!("Failed to run git diff --numstat: {error}"))?;
5667    if !output.status.success() {
5668        return Ok((0, 0));
5669    }
5670    let stdout = String::from_utf8_lossy(&output.stdout);
5671    let mut added = 0;
5672    let mut removed = 0;
5673    for line in stdout.lines() {
5674        let mut parts = line.split_whitespace();
5675        added += parse_numstat_count(parts.next());
5676        removed += parse_numstat_count(parts.next());
5677    }
5678    Ok((added, removed))
5679}
5680
5681fn file_line_count_for_path(repo_root: &Path, relative_path: &str) -> Result<u64, String> {
5682    let path = repo_root.join(relative_path);
5683    let content = fs::read_to_string(&path)
5684        .map_err(|error| format!("Failed to read file {}: {error}", path.display()))?;
5685    Ok(content.lines().count() as u64)
5686}
5687
5688fn parse_numstat_count(value: Option<&str>) -> u64 {
5689    value.and_then(|raw| raw.parse::<u64>().ok()).unwrap_or(0)
5690}
5691
5692fn current_head(repo_root: &Path) -> Result<Option<String>, String> {
5693    match git_output(repo_root, &["rev-parse", "HEAD"]) {
5694        Ok(value) => Ok(Some(value)),
5695        Err(error)
5696            if error.contains("unknown revision") || error.contains("ambiguous argument") =>
5697        {
5698            Ok(None)
5699        }
5700        Err(error) => Err(error),
5701    }
5702}
5703
5704fn git_output(repo_root: &Path, args: &[&str]) -> Result<String, String> {
5705    let output = Command::new("git")
5706        .args(args)
5707        .current_dir(repo_root)
5708        .output()
5709        .map_err(|error| format!("Failed to run git {}: {error}", args.join(" ")))?;
5710    if !output.status.success() {
5711        return Err(String::from_utf8_lossy(&output.stderr).trim().to_string());
5712    }
5713    Ok(String::from_utf8_lossy(&output.stdout).trim().to_string())
5714}
5715
5716fn parse_remote_owner_repo(remote: &str) -> Option<(String, String)> {
5717    let trimmed = remote.trim().trim_end_matches(".git");
5718    if trimmed.is_empty() {
5719        return None;
5720    }
5721
5722    let path_part = if !trimmed.contains("://") {
5723        if let Some((_, path)) = trimmed.rsplit_once(':') {
5724            path
5725        } else {
5726            trimmed
5727        }
5728    } else {
5729        trimmed
5730            .trim_start_matches("https://")
5731            .trim_start_matches("http://")
5732            .trim_start_matches("ssh://")
5733            .split_once('/')
5734            .map(|(_, path)| path)
5735            .unwrap_or(trimmed)
5736    };
5737    let mut parts = path_part.rsplitn(2, '/');
5738    let name = parts.next()?.trim();
5739    let owner = parts.next()?.trim();
5740    if owner.is_empty() || name.is_empty() {
5741        return None;
5742    }
5743    Some((
5744        sanitize_path_component(owner),
5745        sanitize_path_component(name.trim_end_matches(".git")),
5746    ))
5747}
5748
5749fn sanitize_path_component(value: &str) -> String {
5750    let sanitized: String = value
5751        .chars()
5752        .map(|ch| {
5753            if ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.') {
5754                ch
5755            } else {
5756                '-'
5757            }
5758        })
5759        .collect();
5760    sanitized
5761        .trim_matches('-')
5762        .chars()
5763        .take(160)
5764        .collect::<String>()
5765}
5766
5767fn normalize_windows_verbatim_path(path: PathBuf) -> PathBuf {
5768    PathBuf::from(strip_windows_verbatim_prefix(&path.to_string_lossy()))
5769}
5770
5771fn strip_windows_verbatim_prefix(input: &str) -> &str {
5772    input.strip_prefix(r"\\?\").unwrap_or(input)
5773}
5774
5775fn path_identity_key(path: &Path) -> String {
5776    cross_platform_path_identity_key(path)
5777}
5778
5779/// Path identity that treats `C:\foo` and `/mnt/c/foo` as the same location.
5780fn cross_platform_path_identity_key(path: &Path) -> String {
5781    let mut value = path.to_string_lossy().replace('\\', "/");
5782    if let Some(mnt) = map_windows_path_to_wsl_mnt(&value) {
5783        value = mnt;
5784    } else if let Some(windows) = map_wsl_mnt_path_to_windows(&value) {
5785        if let Some(mnt) = map_windows_path_to_wsl_mnt(&windows) {
5786            value = mnt;
5787        } else {
5788            value = windows.replace('\\', "/");
5789        }
5790    }
5791    // Drive-backed paths are case-insensitive on Windows; always fold for stable keys.
5792    value.to_ascii_lowercase()
5793}
5794
5795/// Runtime isolation key for spoils + device identity.
5796///
5797/// Examples: `windows`, `linux`, `macos`, `wsl-ubuntu`, `wsl-debian`.
5798fn worktree_runtime_key() -> String {
5799    static KEY: OnceLock<String> = OnceLock::new();
5800    KEY.get_or_init(|| {
5801        if is_wsl_runtime() {
5802            let distro = std::env::var("WSL_DISTRO_NAME")
5803                .ok()
5804                .map(|value| value.trim().to_string())
5805                .filter(|value| !value.is_empty())
5806                .unwrap_or_else(|| "wsl".to_string());
5807            format!("wsl-{}", sanitize_path_component(&distro).to_ascii_lowercase())
5808        } else {
5809            std::env::consts::OS.to_string()
5810        }
5811    })
5812    .clone()
5813}
5814
5815fn is_wsl_runtime() -> bool {
5816    if !cfg!(target_os = "linux") {
5817        return false;
5818    }
5819    if std::env::var_os("WSL_DISTRO_NAME").is_some() || std::env::var_os("WSL_INTEROP").is_some() {
5820        return true;
5821    }
5822    // Fall back to /proc/version marker used by Microsoft WSL kernels.
5823    fs::read_to_string("/proc/version")
5824        .map(|content| {
5825            let lower = content.to_ascii_lowercase();
5826            lower.contains("microsoft") || lower.contains("wsl")
5827        })
5828        .unwrap_or(false)
5829}
5830
5831/// Device id scoped by runtime so Windows and WSL can share one xbp config home
5832/// without looking like the same upload source.
5833fn worktree_device_hardware_id(base_hardware_id: &str) -> String {
5834    format!("{base_hardware_id}@{}", worktree_runtime_key())
5835}
5836
5837fn ensure_mutation_event_identity(mut event: WorktreeMutationEvent) -> WorktreeMutationEvent {
5838    if event.fingerprint.as_ref().is_none_or(|value| value.is_empty()) {
5839        let fingerprint = mutation_event_fingerprint(&event);
5840        event.fingerprint = Some(fingerprint.clone());
5841        if event.id.is_empty() || !event.id.starts_with("wtmut_") {
5842            event.id = stable_event_id(&fingerprint);
5843        }
5844    } else if event.id.is_empty() {
5845        event.id = stable_event_id(event.fingerprint.as_deref().unwrap_or_default());
5846    }
5847    event
5848}
5849
5850fn ensure_commit_event_identity(mut commit: WorktreeCommitEvent) -> WorktreeCommitEvent {
5851    if commit.fingerprint.as_ref().is_none_or(|value| value.is_empty()) {
5852        let fingerprint = commit_event_fingerprint(&commit);
5853        commit.fingerprint = Some(fingerprint.clone());
5854        if commit.id.is_empty() || !commit.id.starts_with("wtmut_") {
5855            commit.id = stable_event_id(&fingerprint);
5856        }
5857    } else if commit.id.is_empty() {
5858        commit.id = stable_event_id(commit.fingerprint.as_deref().unwrap_or_default());
5859    }
5860    commit
5861}
5862
5863fn path_string_has_ignored_component(path: &str) -> bool {
5864    // No project root available; apply global + built-in rules only.
5865    watch_ignore_rules().ignores_relative(path)
5866}
5867
5868fn is_skipped_discovery_dir(path: &Path) -> bool {
5869    watch_ignore_rules().skips_discovery_dir(path)
5870}
5871
5872/// Relative path from `root` to `path` with `/` separators.
5873///
5874/// Uses `Path::strip_prefix` when possible, then falls back to slash-normalized
5875/// string prefix matching so Windows-style absolute paths still work when the
5876/// binary runs on Unix (WSL tests / mixed path sources from notify).
5877fn relative_watch_path(root: &Path, path: &Path) -> Option<String> {
5878    if let Ok(relative) = path.strip_prefix(root) {
5879        return Some(normalize_relative_watch_path(&relative.to_string_lossy()));
5880    }
5881
5882    let root_norm = normalize_absolute_watch_path(root);
5883    let path_norm = normalize_absolute_watch_path(path);
5884    if path_norm == root_norm {
5885        return Some(String::new());
5886    }
5887    let prefix = format!("{root_norm}/");
5888    path_norm
5889        .strip_prefix(&prefix)
5890        .map(|relative| normalize_relative_watch_path(relative))
5891}
5892
5893fn normalize_absolute_watch_path(path: &Path) -> String {
5894    let mut value = path.to_string_lossy().replace('\\', "/");
5895    // Drop Windows verbatim prefix if present after slash normalization.
5896    if let Some(stripped) = value.strip_prefix("//?/") {
5897        value = stripped.to_string();
5898    }
5899    // Case-fold drive letter for stable prefix matching (C: vs c:).
5900    if value.len() >= 2 {
5901        let bytes = value.as_bytes();
5902        if bytes[1] == b':' && bytes[0].is_ascii_alphabetic() {
5903            let mut chars = value.chars();
5904            let drive = chars.next().unwrap().to_ascii_lowercase();
5905            value = format!("{drive}{}", chars.as_str());
5906        }
5907    }
5908    value.trim_end_matches('/').to_string()
5909}
5910
5911fn relative_slash_path(root: &Path, path: &Path) -> Option<String> {
5912    relative_watch_path(root, path).or_else(|| {
5913        Some(normalize_relative_watch_path(
5914            &path.to_string_lossy().replace('\\', "/"),
5915        ))
5916    })
5917}
5918
5919fn is_ignored_path(root: &Path, path: &Path) -> bool {
5920    watch_ignore_rules_for_root(root).ignores_path(root, path)
5921}
5922
5923fn current_hostname() -> Option<String> {
5924    for key in ["COMPUTERNAME", "HOSTNAME", "NAME"] {
5925        if let Ok(value) = std::env::var(key) {
5926            let trimmed = value.trim();
5927            if !trimmed.is_empty() {
5928                return Some(trimmed.to_string());
5929            }
5930        }
5931    }
5932
5933    if let Ok(content) = fs::read_to_string("/etc/hostname") {
5934        let trimmed = content.trim();
5935        if !trimmed.is_empty() {
5936            return Some(trimmed.to_string());
5937        }
5938    }
5939
5940    // WSL often has a Linux hostname distinct from the Windows host; prefer it when set.
5941    if let Ok(output) = Command::new("hostname").output() {
5942        if output.status.success() {
5943            let value = String::from_utf8_lossy(&output.stdout).trim().to_string();
5944            if !value.is_empty() {
5945                return Some(value);
5946            }
5947        }
5948    }
5949
5950    None
5951}
5952
5953fn is_background_child() -> bool {
5954    std::env::var(BACKGROUND_CHILD_ENV)
5955        .ok()
5956        .map(|value| value == "1")
5957        .unwrap_or(false)
5958}
5959
5960#[allow(dead_code)]
5961fn content_sha256(path: &Path) -> Option<String> {
5962    let bytes = fs::read(path).ok()?;
5963    let mut hasher = Sha256::new();
5964    hasher.update(bytes);
5965    Some(format!("{:x}", hasher.finalize()))
5966}
5967
5968#[cfg(test)]
5969mod tests {
5970    use super::*;
5971
5972    #[test]
5973    fn parses_https_and_ssh_remote_urls() {
5974        assert_eq!(
5975            parse_remote_owner_repo("https://github.com/xylex-group/xbp.git"),
5976            Some(("xylex-group".to_string(), "xbp".to_string()))
5977        );
5978        assert_eq!(
5979            parse_remote_owner_repo("git@github.com:xylex-group/xbp.git"),
5980            Some(("xylex-group".to_string(), "xbp".to_string()))
5981        );
5982    }
5983
5984    #[test]
5985    fn sanitizes_branch_for_storage_path() {
5986        assert_eq!(
5987            sanitize_path_component("feature/worktree watcher"),
5988            "feature-worktree-watcher"
5989        );
5990    }
5991
5992    #[test]
5993    fn normalizes_windows_verbatim_repo_roots() {
5994        assert_eq!(
5995            normalize_windows_verbatim_path(PathBuf::from(
5996                r"\\?\C:\Users\floris\Documents\GitHub\mollie-api-rust"
5997            )),
5998            PathBuf::from(r"C:\Users\floris\Documents\GitHub\mollie-api-rust")
5999        );
6000    }
6001
6002    #[test]
6003    fn parent_path_matching_uses_repo_boundaries() {
6004        let repo = Path::new(r"C:\Users\floris\Documents\GitHub\xbp");
6005        assert!(path_belongs_to_repo(
6006            Path::new(r"C:\Users\floris\Documents\GitHub\xbp\src\main.rs"),
6007            repo
6008        ));
6009        assert!(!path_belongs_to_repo(
6010            Path::new(r"C:\Users\floris\Documents\GitHub\xbp-other\src\main.rs"),
6011            repo
6012        ));
6013    }
6014
6015    #[test]
6016    fn skips_heavy_discovery_directories() {
6017        assert!(is_skipped_discovery_dir(Path::new("node_modules")));
6018        assert!(is_skipped_discovery_dir(Path::new("target")));
6019        assert!(is_skipped_discovery_dir(Path::new("dist")));
6020        assert!(is_skipped_discovery_dir(Path::new(".next")));
6021        assert!(is_skipped_discovery_dir(Path::new(".nx")));
6022        assert!(is_skipped_discovery_dir(Path::new(".xbp")));
6023        assert!(is_skipped_discovery_dir(Path::new(".cargo")));
6024        assert!(is_skipped_discovery_dir(Path::new(".codex")));
6025        assert!(is_skipped_discovery_dir(Path::new(".cursor")));
6026        assert!(is_skipped_discovery_dir(Path::new(".agents")));
6027        assert!(is_skipped_discovery_dir(Path::new(".idea")));
6028        assert!(is_skipped_discovery_dir(Path::new(".github")));
6029        assert!(is_skipped_discovery_dir(Path::new(".postman")));
6030        assert!(is_skipped_discovery_dir(Path::new("mcps")));
6031        assert!(is_skipped_discovery_dir(Path::new("agent-tools")));
6032        assert!(is_skipped_discovery_dir(Path::new("target-publish")));
6033        assert!(is_skipped_discovery_dir(Path::new("__pycache__")));
6034        assert!(is_skipped_discovery_dir(Path::new("terminals")));
6035        assert!(is_skipped_discovery_dir(Path::new(".open-next")));
6036        assert!(!is_skipped_discovery_dir(Path::new("mollie-api-rust")));
6037    }
6038
6039    #[test]
6040    fn ignores_nested_generated_mutation_paths() {
6041        // Windows-style absolute paths must work even when tests run on Unix/WSL
6042        // (Path::strip_prefix does not treat `\` as a separator on Unix).
6043        let root = Path::new(r"C:\Users\floris\Documents\GitHub\athena");
6044        assert!(is_ignored_path(
6045            root,
6046            Path::new(
6047                r"C:\Users\floris\Documents\GitHub\athena\apps\docs\node_modules\@aws-sdk\client-lambda\dist-types\schemas"
6048            )
6049        ));
6050        assert!(is_ignored_path(
6051            root,
6052            Path::new(r"C:\Users\floris\Documents\GitHub\athena\target\debug\build")
6053        ));
6054        // Nx cache and OpenNext build output anywhere under the repo.
6055        assert!(is_ignored_path(
6056            root,
6057            Path::new(r"C:\Users\floris\Documents\GitHub\athena\.nx\cache\run.json")
6058        ));
6059        assert!(is_ignored_path(
6060            root,
6061            Path::new(
6062                r"C:\Users\floris\Documents\GitHub\athena\apps\docs\.open-next\server-functions\default\handler.mjs"
6063            )
6064        ));
6065        assert!(!is_ignored_path(
6066            root,
6067            Path::new(r"C:\Users\floris\Documents\GitHub\athena\apps\web\src\main.ts")
6068        ));
6069
6070        // Forward-slash form (notify / mixed hosts) and relative component checks.
6071        let root_slash = Path::new("C:/Users/floris/Documents/GitHub/athena");
6072        assert!(is_ignored_path(
6073            root_slash,
6074            Path::new(
6075                "C:/Users/floris/Documents/GitHub/athena/apps/docs/node_modules/@aws-sdk/client-lambda/dist-types/schemas"
6076            )
6077        ));
6078        assert!(path_string_has_ignored_component(
6079            "apps/docs/node_modules/@aws-sdk/client-lambda/dist-types/schemas"
6080        ));
6081    }
6082
6083    #[test]
6084    fn global_config_forbidden_paths_folders_and_banned_words() {
6085        let rules = WatchIgnoreRules::from_config(&WorktreeWatchConfig {
6086            forbidden_paths: vec!["secrets/".to_string(), "apps/web/.env.local".to_string()],
6087            forbidden_folders: vec![".cache".to_string(), "coverage".to_string()],
6088            banned_words: vec!["private-key".to_string(), "CREDENTIAL".to_string()],
6089        });
6090
6091        // Nested under a watched tree still blocked by prefix.
6092        assert!(rules.ignores_relative("secrets/prod/api.key"));
6093        assert!(rules.ignores_relative("apps/web/.env.local"));
6094        // Exact file ban is not a string-prefix of other filenames.
6095        assert!(!rules.ignores_relative("apps/web/.env.local.bak"));
6096
6097        // Folder name anywhere
6098        assert!(rules.ignores_relative("apps/web/.cache/tmp"));
6099        assert!(rules.ignores_relative("packages/foo/coverage/index.html"));
6100
6101        // Banned word (case-insensitive substring)
6102        assert!(rules.ignores_relative("docs/private-key-notes.md"));
6103        assert!(rules.ignores_relative("config/my-credential-store.json"));
6104
6105        // Still allow normal source
6106        assert!(!rules.ignores_relative("apps/web/src/main.ts"));
6107        assert!(!rules.skips_discovery_dir(Path::new("mollie-api-rust")));
6108        assert!(rules.skips_discovery_dir(Path::new(".cache")));
6109        assert!(rules.skips_discovery_dir(Path::new("node_modules"))); // built-in
6110    }
6111
6112    #[test]
6113    fn global_worktree_ignore_file_honors_mcps_patterns() {
6114        let root = std::env::temp_dir().join(format!(
6115            "xbp-global-watch-ignore-{}-{}",
6116            std::process::id(),
6117            std::time::SystemTime::now()
6118                .duration_since(std::time::UNIX_EPOCH)
6119                .expect("time")
6120                .as_nanos()
6121        ));
6122        let _ = fs::remove_dir_all(&root);
6123        fs::create_dir_all(&root).expect("global root");
6124
6125        let ignore_path = root.join(crate::utils::GLOBAL_WORKTREE_IGNORE_FILENAME);
6126        fs::write(&ignore_path, "mcps/\n/custom-global/\n").expect("write global ignore");
6127
6128        let rules = WatchIgnoreRules {
6129            forbidden_path_prefixes: Vec::new(),
6130            forbidden_folders: BTreeSet::new(),
6131            banned_words: Vec::new(),
6132            global_ignore: crate::utils::load_global_worktree_ignore_from_path(&ignore_path),
6133            project_ignore: crate::utils::XbpIgnoreSet::empty(),
6134        };
6135
6136        assert!(rules.ignores_relative("mcps/github/catalog.json"));
6137        assert!(rules.ignores_relative("apps/api/mcps/local"));
6138        assert!(rules.ignores_relative("custom-global/notes.md"));
6139        assert!(!rules.ignores_relative("apps/web/src/main.ts"));
6140        assert!(rules.skips_discovery_dir(Path::new("mcps")));
6141
6142        let _ = fs::remove_dir_all(root);
6143    }
6144
6145    #[test]
6146    fn project_xbpignore_is_honored_by_worktree_watch_rules() {
6147        let root = std::env::temp_dir().join(format!(
6148            "xbp-watch-ignore-{}-{}",
6149            std::process::id(),
6150            std::time::SystemTime::now()
6151                .duration_since(std::time::UNIX_EPOCH)
6152                .expect("time")
6153                .as_nanos()
6154        ));
6155        let _ = fs::remove_dir_all(&root);
6156        fs::create_dir_all(root.join(".xbp")).expect("xbp dir");
6157        fs::write(
6158            root.join(".xbp").join(".xbpignore"),
6159            "packages/icons/\n*.generated.ts\n",
6160        )
6161        .expect("write ignore");
6162        fs::write(
6163            root.join(".xbp").join("xbp.yaml"),
6164            "project_name: demo\nversion: 0.1.0\nport: 3000\nbuild_dir: ./\nignore_paths:\n  - e2e/\nwatch_ignore_paths:\n  - tmp-watch/\n",
6165        )
6166        .expect("write config");
6167
6168        let rules = WatchIgnoreRules::from_config_and_project(
6169            &WorktreeWatchConfig::default(),
6170            Some(&root),
6171        );
6172        // `.xbpignore` still applies to worktree-watch
6173        assert!(rules.ignores_relative("packages/icons/src/index.ts"));
6174        assert!(rules.ignores_relative("apps/web/foo.generated.ts"));
6175        // Discovery-only `ignore_paths` must NOT silence worktree-watch
6176        assert!(!rules.ignores_relative("e2e/login.spec.ts"));
6177        // Explicit `watch_ignore_paths` do apply
6178        assert!(rules.ignores_relative("tmp-watch/file.ts"));
6179        assert!(!rules.ignores_relative("apps/web/src/main.ts"));
6180
6181        let _ = fs::remove_dir_all(root);
6182    }
6183
6184    #[test]
6185    fn builds_coding_stats_by_file_and_filetype() {
6186        let identity = RepoIdentity {
6187            owner: "xylex-group".to_string(),
6188            name: "xbp".to_string(),
6189            branch: "main".to_string(),
6190            root: PathBuf::from(r"C:\Users\floris\Documents\GitHub\xbp"),
6191            head_sha: None,
6192        };
6193        let layout = SpoolLayout {
6194            root: PathBuf::from(r"C:\Users\floris\.xbp\mutations\xylex-group\xbp\main"),
6195            events_file: PathBuf::from("events.jsonl"),
6196            commits_file: PathBuf::from("commits.jsonl"),
6197            stats_file: PathBuf::from("stats.json"),
6198            watcher_state_file: PathBuf::from("watcher-state.json"),
6199            sync_log_file: PathBuf::from("sync-log.jsonl"),
6200            event_dedupe_file: PathBuf::from("event-dedupe.jsonl"),
6201            event_dedupe_dir: PathBuf::from("event-dedupe"),
6202            commit_dedupe_dir: PathBuf::from("commit-dedupe"),
6203        };
6204        let candidates = vec![SyncCandidate {
6205            path: PathBuf::from("events.jsonl"),
6206            records: SyncRecords::Events(vec![
6207                stats_test_event_record("2026-07-08T10:00:00Z", "src/main.rs", 4, 1),
6208                stats_test_event_record("2026-07-08T10:05:00Z", "src/main.rs", 2, 0),
6209                stats_test_event_record("2026-07-08T11:00:00Z", "README.md", 1, 3),
6210            ]),
6211        }];
6212
6213        let summary = build_stats_summary(&identity, &layout, &candidates, 15).unwrap();
6214
6215        assert_eq!(summary.total_events, 3);
6216        assert_eq!(summary.estimated_coding_seconds, 60 + 300 + 900);
6217        assert_eq!(summary.session_count, 2);
6218        assert_eq!(summary.observed_span_seconds, 60 * 60);
6219        assert_eq!(summary.added_lines, 7);
6220        assert_eq!(summary.removed_lines, 4);
6221        assert_eq!(summary.by_file[0].path, "README.md");
6222        assert_eq!(summary.by_file[0].estimated_coding_seconds, 900);
6223        assert_eq!(summary.by_filetype[0].name, "md");
6224        assert_eq!(summary.by_filetype[0].estimated_coding_seconds, 900);
6225        assert_eq!(summary.by_filetype[1].name, "rs");
6226        assert_eq!(summary.by_filetype[1].estimated_coding_seconds, 360);
6227    }
6228
6229    #[test]
6230    fn builds_repo_activity_across_branch_spools() {
6231        let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
6232        let main_root = temp_root.join("main");
6233        let feature_root = temp_root.join("feature-login");
6234        fs::create_dir_all(&main_root).unwrap();
6235        fs::create_dir_all(&feature_root).unwrap();
6236
6237        append_json_line(
6238            &main_root.join("events-main.jsonl"),
6239            &stats_test_event_record("2026-07-08T10:00:00Z", "src/main.rs", 4, 1),
6240        )
6241        .unwrap();
6242        append_json_line(
6243            &main_root.join("events-main.jsonl"),
6244            &stats_test_event_record("2026-07-08T10:10:00Z", "src/lib.rs", 2, 0),
6245        )
6246        .unwrap();
6247        append_json_line(
6248            &feature_root.join("events-feature.jsonl"),
6249            &stats_test_event_record("2026-07-08T11:00:00Z", "src/auth.rs", 10, 2),
6250        )
6251        .unwrap();
6252        append_json_line(
6253            &feature_root.join("events-feature.jsonl"),
6254            &stats_test_event_record("2026-07-08T11:30:00Z", "src/auth.rs", 3, 1),
6255        )
6256        .unwrap();
6257        append_json_line(
6258            &feature_root.join("commits-feature.jsonl"),
6259            &worktree_commit_test_event("previous", "current"),
6260        )
6261        .unwrap();
6262
6263        let identity = RepoIdentity {
6264            owner: "xylex-group".to_string(),
6265            name: "xbp".to_string(),
6266            branch: "main".to_string(),
6267            root: PathBuf::from(r"C:\Users\floris\Documents\GitHub\xbp"),
6268            head_sha: None,
6269        };
6270
6271        let summary = build_repo_activity_summary(&identity, &temp_root, 15).unwrap();
6272
6273        assert_eq!(summary.branch_count, 2);
6274        assert_eq!(summary.total_events, 4);
6275        assert_eq!(summary.total_commits, 1);
6276        assert_eq!(summary.session_count, 3);
6277        assert_eq!(summary.estimated_coding_seconds, 60 + 600 + 60 + 900);
6278        assert_eq!(summary.observed_span_seconds, 90 * 60);
6279        assert_eq!(summary.branches[0].branch_name, "feature-login");
6280        assert_eq!(summary.branches[0].session_count, 2);
6281        assert_eq!(summary.branches[0].observed_span_seconds, 30 * 60);
6282        assert_eq!(summary.branches[1].branch_name, "main");
6283
6284        let _ = fs::remove_dir_all(temp_root);
6285    }
6286
6287    #[test]
6288    fn mutation_fingerprint_ignores_random_event_id_but_keeps_time_bucket() {
6289        let first: WorktreeMutationEvent = serde_json::from_value(stats_test_event(
6290            "2026-07-08T10:00:00Z",
6291            "src/main.rs",
6292            4,
6293            1,
6294        ))
6295        .unwrap();
6296        let same_bucket: WorktreeMutationEvent = serde_json::from_value(stats_test_event(
6297            "2026-07-08T10:00:00Z",
6298            "src/main.rs",
6299            4,
6300            1,
6301        ))
6302        .unwrap();
6303        let next_bucket: WorktreeMutationEvent = serde_json::from_value(stats_test_event(
6304            "2026-07-08T10:00:06Z",
6305            "src/main.rs",
6306            4,
6307            1,
6308        ))
6309        .unwrap();
6310
6311        assert_eq!(
6312            mutation_event_fingerprint(&first),
6313            mutation_event_fingerprint(&same_bucket)
6314        );
6315        assert_ne!(
6316            mutation_event_fingerprint(&first),
6317            mutation_event_fingerprint(&next_bucket)
6318        );
6319
6320        // Fingerprint is path-root independent: absolute Windows/WSL spellings must not diverge.
6321        let mut windows_root = first.clone();
6322        windows_root.repo_root = r"C:\Users\floris\Documents\GitHub\xbp".to_string();
6323        let mut wsl_root = first.clone();
6324        wsl_root.repo_root = "/mnt/c/Users/floris/Documents/GitHub/xbp".to_string();
6325        assert_eq!(
6326            mutation_event_fingerprint(&windows_root),
6327            mutation_event_fingerprint(&wsl_root)
6328        );
6329
6330        let fingerprint = mutation_event_fingerprint(&first);
6331        assert_eq!(stable_event_id(&fingerprint), format!("wtmut_{fingerprint}"));
6332        assert_eq!(
6333            ensure_mutation_event_identity(first.clone()).id,
6334            stable_event_id(&fingerprint)
6335        );
6336    }
6337
6338    #[test]
6339    fn runtime_key_and_device_id_are_stable_and_scoped() {
6340        let key = worktree_runtime_key();
6341        assert!(!key.is_empty());
6342        assert_eq!(key, worktree_runtime_key());
6343        // On this host: windows | linux | macos | wsl-*
6344        assert!(
6345            key == "windows"
6346                || key == "linux"
6347                || key == "macos"
6348                || key.starts_with("wsl-")
6349                || key == "wsl",
6350            "unexpected runtime key {key}"
6351        );
6352        let device = worktree_device_hardware_id("xbp_hw_test");
6353        assert_eq!(device, format!("xbp_hw_test@{key}"));
6354    }
6355
6356    #[test]
6357    fn should_poll_only_for_wsl_mnt_paths() {
6358        // On non-WSL CI/dev hosts this is always false; the path heuristic itself is cheap.
6359        let mnt = Path::new("/mnt/c/Users/floris/Documents/GitHub/xbp");
6360        let home = Path::new("/home/floris/src/xbp");
6361        if is_wsl_runtime() {
6362            assert!(should_use_poll_watcher(mnt));
6363            assert!(!should_use_poll_watcher(home));
6364        } else {
6365            assert!(!should_use_poll_watcher(mnt));
6366            assert!(!should_use_poll_watcher(home));
6367        }
6368    }
6369
6370    #[test]
6371    fn repository_spool_root_uses_global_xbp_home_mutations() {
6372        let identity = RepoIdentity {
6373            owner: "xylex-group".to_string(),
6374            name: "xbp".to_string(),
6375            branch: "main".to_string(),
6376            root: PathBuf::from("/tmp/xbp"),
6377            head_sha: None,
6378        };
6379        let root = repository_spool_root(&identity).expect("global xbp home");
6380        let rendered = root.to_string_lossy().replace('\\', "/");
6381        assert!(
6382            rendered.contains("/mutations/"),
6383            "expected home-level mutations path, got {rendered}"
6384        );
6385        assert!(
6386            !rendered.contains("/mutations/runtimes/"),
6387            "spool must not be runtime-split when sharing one established home: {rendered}"
6388        );
6389        assert!(
6390            rendered.ends_with("xylex-group/xbp") || rendered.contains("xylex-group/xbp"),
6391            "expected owner/name suffix, got {rendered}"
6392        );
6393        // Must live under the established global root, never the repo path.
6394        assert!(!rendered.contains("/tmp/xbp/"));
6395    }
6396
6397    #[test]
6398    fn cross_platform_path_identity_reconciles_wsl_and_windows() {
6399        let windows = Path::new(r"C:\Users\szymon\.xbp\mutations\SuitsFinance\speedrun-formations\main");
6400        let wsl = Path::new(
6401            "/mnt/c/Users/szymon/.xbp/mutations/SuitsFinance/speedrun-formations/main",
6402        );
6403        assert_eq!(
6404            cross_platform_path_identity_key(windows),
6405            cross_platform_path_identity_key(wsl)
6406        );
6407    }
6408
6409    #[test]
6410    fn command_line_mentions_path_is_case_and_slash_insensitive() {
6411        let cmdline = r#""C:\xbp.exe" worktree-watch start --parent C:\Users\floris\Documents\GitHub"#;
6412        assert!(command_line_mentions_path(
6413            cmdline,
6414            "/mnt/c/users/floris/documents/github"
6415        ));
6416        assert!(command_line_mentions_path(
6417            cmdline,
6418            r"C:\Users\floris\Documents\GitHub"
6419        ));
6420        assert!(command_line_mentions_path(cmdline, "GitHub"));
6421    }
6422
6423    #[test]
6424    fn append_unique_mutation_event_is_idempotent_across_dedupe_instances() {
6425        let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
6426        fs::create_dir_all(&temp_root).unwrap();
6427        let layout = SpoolLayout {
6428            root: temp_root.clone(),
6429            events_file: temp_root.join("events.jsonl"),
6430            commits_file: temp_root.join("commits.jsonl"),
6431            stats_file: temp_root.join("stats.json"),
6432            watcher_state_file: temp_root.join("watcher-state.json"),
6433            sync_log_file: temp_root.join("sync-log.jsonl"),
6434            event_dedupe_file: temp_root.join("event-dedupe.jsonl"),
6435            event_dedupe_dir: temp_root.join("event-dedupe"),
6436            commit_dedupe_dir: temp_root.join("commit-dedupe"),
6437        };
6438        let mutation: WorktreeMutationEvent = serde_json::from_value(stats_test_event(
6439            "2026-07-08T10:00:00Z",
6440            "src/main.rs",
6441            4,
6442            1,
6443        ))
6444        .unwrap();
6445
6446        let first =
6447            append_unique_mutation_event(&layout, &mutation, &mut RecentEventDedupe::default())
6448                .unwrap();
6449        let second =
6450            append_unique_mutation_event(&layout, &mutation, &mut RecentEventDedupe::default())
6451                .unwrap();
6452
6453        assert!(first);
6454        assert!(!second);
6455        let events = read_jsonl_values(&layout.events_file).unwrap();
6456        assert_eq!(events.len(), 1);
6457        assert_eq!(
6458            events[0].get("fingerprint").and_then(Value::as_str),
6459            Some(mutation_event_fingerprint(&mutation).as_str())
6460        );
6461        let _ = fs::remove_dir_all(temp_root);
6462    }
6463
6464    #[test]
6465    fn reads_spool_records_as_typed_events() {
6466        let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
6467        fs::create_dir_all(&temp_root).unwrap();
6468        let events_file = temp_root.join("events.jsonl");
6469        append_json_line(
6470            &events_file,
6471            &stats_test_event_record("2026-07-08T10:00:00Z", "src/main.rs", 4, 1),
6472        )
6473        .unwrap();
6474
6475        let records = read_spool_records(&events_file, SyncFileKind::Events).unwrap();
6476
6477        match records {
6478            SyncRecords::Events(events) => {
6479                assert_eq!(events.len(), 1);
6480                assert_eq!(events[0].primary_path.as_deref(), Some("src/main.rs"));
6481            }
6482            SyncRecords::Commits(_) => panic!("expected typed event records"),
6483        }
6484        let _ = fs::remove_dir_all(temp_root);
6485    }
6486
6487    #[test]
6488    fn read_spool_records_skips_nul_only_padding() {
6489        let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
6490        fs::create_dir_all(&temp_root).unwrap();
6491        let events_file = temp_root.join("events.jsonl");
6492        fs::write(&events_file, vec![0; 4096]).unwrap();
6493
6494        let records = read_spool_records(&events_file, SyncFileKind::Events).unwrap();
6495
6496        match records {
6497            SyncRecords::Events(events) => assert!(events.is_empty()),
6498            SyncRecords::Commits(_) => panic!("expected event records"),
6499        }
6500        let _ = fs::remove_dir_all(temp_root);
6501    }
6502
6503    #[test]
6504    fn read_jsonl_strips_leading_nuls_before_json() {
6505        let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
6506        fs::create_dir_all(&temp_root).unwrap();
6507        let path = temp_root.join("event-dedupe.jsonl");
6508        let mut bytes = vec![0u8; 64];
6509        bytes.extend_from_slice(
6510            br#"{"fingerprint":"abc","firstSeenAt":"2026-07-09T00:00:00Z","eventKind":"modify","primaryPath":".xbp"}"#,
6511        );
6512        bytes.push(b'\n');
6513        fs::write(&path, bytes).unwrap();
6514
6515        let values = read_jsonl_values(&path).unwrap();
6516        assert_eq!(values.len(), 1);
6517        assert_eq!(
6518            values[0].get("fingerprint").and_then(Value::as_str),
6519            Some("abc")
6520        );
6521        let _ = fs::remove_dir_all(temp_root);
6522    }
6523
6524    #[test]
6525    fn read_jsonl_skips_garbage_lines() {
6526        let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
6527        fs::create_dir_all(&temp_root).unwrap();
6528        let path = temp_root.join("mixed.jsonl");
6529        fs::write(
6530            &path,
6531            "not-json\n{\"fingerprint\":\"ok\"}\n\n{\"broken\":\n",
6532        )
6533        .unwrap();
6534
6535        let values = read_jsonl_values(&path).unwrap();
6536        assert_eq!(values.len(), 1);
6537        assert_eq!(
6538            values[0].get("fingerprint").and_then(Value::as_str),
6539            Some("ok")
6540        );
6541        let _ = fs::remove_dir_all(temp_root);
6542    }
6543
6544    #[test]
6545    fn collect_spool_candidates_rewrites_legacy_generated_events() {
6546        let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
6547        fs::create_dir_all(&temp_root).unwrap();
6548        let events_file = temp_root.join("events-legacy.jsonl");
6549        append_json_line(
6550            &events_file,
6551            &stats_test_event_record(
6552                "2026-07-08T10:00:00Z",
6553                "apps/docs/node_modules/@aws-sdk/client-lambda/dist-types/schemas",
6554                0,
6555                0,
6556            ),
6557        )
6558        .unwrap();
6559        append_json_line(
6560            &events_file,
6561            &stats_test_event_record("2026-07-08T10:01:00Z", "apps/web/src/main.ts", 4, 1),
6562        )
6563        .unwrap();
6564        let layout = SpoolLayout {
6565            root: temp_root.clone(),
6566            events_file: temp_root.join("events.jsonl"),
6567            commits_file: temp_root.join("commits.jsonl"),
6568            stats_file: temp_root.join("stats.json"),
6569            watcher_state_file: temp_root.join("watcher-state.json"),
6570            sync_log_file: temp_root.join("sync-log.jsonl"),
6571            event_dedupe_file: temp_root.join("event-dedupe.jsonl"),
6572            event_dedupe_dir: temp_root.join("event-dedupe"),
6573            commit_dedupe_dir: temp_root.join("commit-dedupe"),
6574        };
6575
6576        let candidates = collect_spool_candidates(&layout, false).unwrap();
6577
6578        assert_eq!(candidates.len(), 1);
6579        match &candidates[0].records {
6580            SyncRecords::Events(events) => {
6581                assert_eq!(events.len(), 1);
6582                assert_eq!(
6583                    events[0].primary_path.as_deref(),
6584                    Some("apps/web/src/main.ts")
6585                );
6586            }
6587            SyncRecords::Commits(_) => panic!("expected event spool candidate"),
6588        }
6589        let rewritten = read_jsonl_records::<WorktreeMutationEvent>(&events_file).unwrap();
6590        assert_eq!(rewritten.len(), 1);
6591        assert_eq!(
6592            rewritten[0].primary_path.as_deref(),
6593            Some("apps/web/src/main.ts")
6594        );
6595        let _ = fs::remove_dir_all(temp_root);
6596    }
6597
6598    #[test]
6599    fn collect_spool_candidates_deletes_fully_ignored_event_files() {
6600        let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
6601        fs::create_dir_all(&temp_root).unwrap();
6602        let events_file = temp_root.join("events-legacy.jsonl");
6603        append_json_line(
6604            &events_file,
6605            &stats_test_event_record(
6606                "2026-07-08T10:00:00Z",
6607                "apps/docs/node_modules/@aws-sdk/client-lambda/dist-types/schemas",
6608                0,
6609                0,
6610            ),
6611        )
6612        .unwrap();
6613        let layout = SpoolLayout {
6614            root: temp_root.clone(),
6615            events_file: temp_root.join("events.jsonl"),
6616            commits_file: temp_root.join("commits.jsonl"),
6617            stats_file: temp_root.join("stats.json"),
6618            watcher_state_file: temp_root.join("watcher-state.json"),
6619            sync_log_file: temp_root.join("sync-log.jsonl"),
6620            event_dedupe_file: temp_root.join("event-dedupe.jsonl"),
6621            event_dedupe_dir: temp_root.join("event-dedupe"),
6622            commit_dedupe_dir: temp_root.join("commit-dedupe"),
6623        };
6624
6625        let candidates = collect_spool_candidates(&layout, false).unwrap();
6626
6627        assert!(candidates.is_empty());
6628        assert!(!events_file.exists());
6629        let _ = fs::remove_dir_all(temp_root);
6630    }
6631
6632    #[test]
6633    fn append_unique_commit_event_is_idempotent_across_watcher_instances() {
6634        let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
6635        fs::create_dir_all(&temp_root).unwrap();
6636        let layout = SpoolLayout {
6637            root: temp_root.clone(),
6638            events_file: temp_root.join("events.jsonl"),
6639            commits_file: temp_root.join("commits.jsonl"),
6640            stats_file: temp_root.join("stats.json"),
6641            watcher_state_file: temp_root.join("watcher-state.json"),
6642            sync_log_file: temp_root.join("sync-log.jsonl"),
6643            event_dedupe_file: temp_root.join("event-dedupe.jsonl"),
6644            event_dedupe_dir: temp_root.join("event-dedupe"),
6645            commit_dedupe_dir: temp_root.join("commit-dedupe"),
6646        };
6647        let commit = WorktreeCommitEvent {
6648            id: Uuid::new_v4().to_string(),
6649            fingerprint: None,
6650            repo_owner: "xylex-group".to_string(),
6651            repo_name: "xbp".to_string(),
6652            branch_name: "main".to_string(),
6653            repo_root: r"C:\Users\floris\Documents\GitHub\xbp".to_string(),
6654            previous_head_sha: Some("previous".to_string()),
6655            head_sha: "current".to_string(),
6656            subject: Some("test commit".to_string()),
6657            author_name: Some("Floris".to_string()),
6658            author_email: Some("floris@xylex.group".to_string()),
6659            committed_at: Some("2026-07-08T10:00:00Z".to_string()),
6660            occurred_at: Utc::now(),
6661        };
6662
6663        assert!(append_unique_commit_event(&layout, &commit).unwrap());
6664        assert!(!append_unique_commit_event(&layout, &commit).unwrap());
6665        let commits = read_jsonl_values(&layout.commits_file).unwrap();
6666        assert_eq!(commits.len(), 1);
6667        assert_eq!(
6668            commits[0].get("fingerprint").and_then(Value::as_str),
6669            Some(commit_event_fingerprint(&commit).as_str())
6670        );
6671        let _ = fs::remove_dir_all(temp_root);
6672    }
6673
6674    #[test]
6675    fn splits_worktree_mutation_uploads_into_bounded_batches() {
6676        let events = (0..(WORKTREE_MUTATION_SYNC_BATCH_RECORD_LIMIT + 1))
6677            .map(|index| {
6678                serde_json::from_value::<WorktreeMutationEvent>(stats_test_event(
6679                    "2026-07-08T10:00:00Z",
6680                    &format!("src/file-{index}.rs"),
6681                    1,
6682                    0,
6683                ))
6684                .unwrap()
6685            })
6686            .collect::<Vec<_>>();
6687        let commits = vec![
6688            worktree_commit_test_event("previous-a", "current-a"),
6689            worktree_commit_test_event("previous-b", "current-b"),
6690        ];
6691
6692        let batches = split_worktree_mutation_upload_batches(events, commits);
6693
6694        assert_eq!(batches.len(), 2);
6695        assert_eq!(
6696            batches[0].events.len() + batches[0].commits.len(),
6697            WORKTREE_MUTATION_SYNC_BATCH_RECORD_LIMIT
6698        );
6699        assert_eq!(batches[1].events.len(), 1);
6700        assert_eq!(batches[1].commits.len(), 2);
6701    }
6702
6703    #[test]
6704    fn splits_worktree_mutation_uploads_by_estimated_json_bytes() {
6705        let long_path = format!("src/{}.rs", "x".repeat(8_000));
6706        let events = (0..200)
6707            .map(|index| {
6708                serde_json::from_value::<WorktreeMutationEvent>(stats_test_event(
6709                    "2026-07-08T10:00:00Z",
6710                    &format!("{long_path}-{index}"),
6711                    1,
6712                    0,
6713                ))
6714                .unwrap()
6715            })
6716            .collect::<Vec<_>>();
6717
6718        let batches = split_worktree_mutation_upload_batches(events, Vec::new());
6719
6720        assert!(batches.len() > 1);
6721        for batch in batches {
6722            assert!(
6723                batch.estimated_json_bytes <= WORKTREE_MUTATION_SYNC_BATCH_JSON_BYTES,
6724                "batch estimated JSON bytes exceeded limit: {}",
6725                batch.estimated_json_bytes
6726            );
6727        }
6728    }
6729
6730    #[test]
6731    fn classifies_create_remove_and_rename_events() {
6732        assert_eq!(
6733            classify_event_kind(&EventKind::Create(CreateKind::File)),
6734            "file_create"
6735        );
6736        assert_eq!(
6737            classify_event_kind(&EventKind::Remove(RemoveKind::Folder)),
6738            "folder_remove"
6739        );
6740        assert_eq!(
6741            classify_event_kind(&EventKind::Modify(ModifyKind::Name(RenameMode::Both))),
6742            "rename_or_move"
6743        );
6744    }
6745
6746    #[test]
6747    fn retries_only_retryable_worktree_sync_statuses() {
6748        assert!(should_retry_worktree_mutation_sync_status(
6749            StatusCode::INTERNAL_SERVER_ERROR
6750        ));
6751        assert!(should_retry_worktree_mutation_sync_status(
6752            StatusCode::REQUEST_TIMEOUT
6753        ));
6754        assert!(should_retry_worktree_mutation_sync_status(
6755            StatusCode::TOO_MANY_REQUESTS
6756        ));
6757        assert!(!should_retry_worktree_mutation_sync_status(
6758            StatusCode::BAD_REQUEST
6759        ));
6760        assert!(!should_retry_worktree_mutation_sync_status(
6761            StatusCode::UNAUTHORIZED
6762        ));
6763        assert!(!should_retry_worktree_mutation_sync_status(
6764            StatusCode::PAYLOAD_TOO_LARGE
6765        ));
6766    }
6767
6768    #[test]
6769    fn match_github_repo_inventory_resolves_short_name_and_owner_repo() {
6770        let temp = std::env::temp_dir().join(format!("xbp-repo-lookup-{}", Uuid::new_v4()));
6771        let athena = temp.join("athena");
6772        let other = temp.join("other-athena");
6773        fs::create_dir_all(&athena).unwrap();
6774        fs::create_dir_all(&other).unwrap();
6775
6776        let repos = vec![
6777            GithubRepoRecord {
6778                owner: "xylex-group".to_string(),
6779                repo: "athena".to_string(),
6780                full_name: "xylex-group/athena".to_string(),
6781                path: athena.to_string_lossy().to_string(),
6782                remote_url: "https://github.com/xylex-group/athena.git".to_string(),
6783            },
6784            GithubRepoRecord {
6785                owner: "someone".to_string(),
6786                repo: "other".to_string(),
6787                full_name: "someone/other".to_string(),
6788                path: other.to_string_lossy().to_string(),
6789                remote_url: "https://github.com/someone/other.git".to_string(),
6790            },
6791        ];
6792
6793        let by_short = match_github_repo_inventory(&repos, "athena").unwrap().unwrap();
6794        assert_eq!(
6795            path_identity_key(&by_short),
6796            path_identity_key(athena.as_path())
6797        );
6798
6799        let by_full = match_github_repo_inventory(&repos, "xylex-group/athena")
6800            .unwrap()
6801            .unwrap();
6802        assert_eq!(
6803            path_identity_key(&by_full),
6804            path_identity_key(athena.as_path())
6805        );
6806
6807        assert!(match_github_repo_inventory(&repos, "missing")
6808            .unwrap()
6809            .is_none());
6810
6811        let _ = fs::remove_dir_all(temp);
6812    }
6813
6814    #[test]
6815    fn match_github_repo_inventory_errors_on_ambiguous_short_name() {
6816        let temp = std::env::temp_dir().join(format!("xbp-repo-lookup-ambig-{}", Uuid::new_v4()));
6817        let a = temp.join("a");
6818        let b = temp.join("b");
6819        fs::create_dir_all(&a).unwrap();
6820        fs::create_dir_all(&b).unwrap();
6821
6822        let repos = vec![
6823            GithubRepoRecord {
6824                owner: "one".to_string(),
6825                repo: "dup".to_string(),
6826                full_name: "one/dup".to_string(),
6827                path: a.to_string_lossy().to_string(),
6828                remote_url: "https://github.com/one/dup.git".to_string(),
6829            },
6830            GithubRepoRecord {
6831                owner: "two".to_string(),
6832                repo: "dup".to_string(),
6833                full_name: "two/dup".to_string(),
6834                path: b.to_string_lossy().to_string(),
6835                remote_url: "https://github.com/two/dup.git".to_string(),
6836            },
6837        ];
6838
6839        let err = match_github_repo_inventory(&repos, "dup").unwrap_err();
6840        assert!(err.contains("Ambiguous"));
6841        assert!(err.contains("one/dup") || err.contains(a.to_string_lossy().as_ref()));
6842
6843        let unique = match_github_repo_inventory(&repos, "two/dup").unwrap().unwrap();
6844        assert_eq!(path_identity_key(&unique), path_identity_key(b.as_path()));
6845
6846        let _ = fs::remove_dir_all(temp);
6847    }
6848
6849    fn stats_test_event(
6850        occurred_at: &str,
6851        path: &str,
6852        added_lines: u64,
6853        removed_lines: u64,
6854    ) -> Value {
6855        json!({
6856            "id": Uuid::new_v4().to_string(),
6857            "repoOwner": "xylex-group",
6858            "repoName": "xbp",
6859            "branchName": "main",
6860            "repoRoot": r"C:\Users\floris\Documents\GitHub\xbp",
6861            "headSha": null,
6862            "eventKind": "file_modify",
6863            "paths": [path],
6864            "primaryPath": path,
6865            "oldPath": null,
6866            "newPath": null,
6867            "addedLines": added_lines,
6868            "removedLines": removed_lines,
6869            "totalLines": added_lines + 42,
6870            "fileCreated": false,
6871            "fileRemoved": false,
6872            "folderCreated": false,
6873            "folderRemoved": false,
6874            "renamedOrMoved": false,
6875            "rawKind": "Modify(Data(Content))",
6876            "occurredAt": occurred_at,
6877        })
6878    }
6879
6880    fn stats_test_event_record(
6881        occurred_at: &str,
6882        path: &str,
6883        added_lines: u64,
6884        removed_lines: u64,
6885    ) -> WorktreeMutationEvent {
6886        serde_json::from_value(stats_test_event(
6887            occurred_at,
6888            path,
6889            added_lines,
6890            removed_lines,
6891        ))
6892        .unwrap()
6893    }
6894
6895    fn worktree_commit_test_event(previous_head_sha: &str, head_sha: &str) -> WorktreeCommitEvent {
6896        WorktreeCommitEvent {
6897            id: Uuid::new_v4().to_string(),
6898            fingerprint: None,
6899            repo_owner: "xylex-group".to_string(),
6900            repo_name: "xbp".to_string(),
6901            branch_name: "main".to_string(),
6902            repo_root: r"C:\Users\floris\Documents\GitHub\xbp".to_string(),
6903            previous_head_sha: Some(previous_head_sha.to_string()),
6904            head_sha: head_sha.to_string(),
6905            subject: Some("test commit".to_string()),
6906            author_name: Some("Floris".to_string()),
6907            author_email: Some("floris@xylex.group".to_string()),
6908            committed_at: Some("2026-07-08T10:00:00Z".to_string()),
6909            occurred_at: Utc::now(),
6910        }
6911    }
6912}