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#[derive(Debug, Clone, Default)]
42struct WatchIgnoreRules {
43 forbidden_path_prefixes: Vec<String>,
45 forbidden_folders: BTreeSet<String>,
47 banned_words: Vec<String>,
49 global_ignore: crate::utils::XbpIgnoreSet,
51 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 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 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
236fn 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 pub repos: Vec<PathBuf>,
250}
251
252pub 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 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
281pub 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 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 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 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#[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 #[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 #[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 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
826pub fn run_worktree_watch_start_detached(
831 options: WorktreeWatchStartOptions,
832) -> Result<(), String> {
833 if let Some(parent) = options.target.parent.as_deref() {
834 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 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 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 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 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#[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#[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#[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
1407pub 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
1457pub 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
1505pub 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
1703pub 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 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 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 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 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 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
1848pub const WORKTREE_WATCH_API_DEFAULT_STATS_GAP_MINUTES: u64 = 30;
1850
1851pub 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 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
1953pub 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
1965pub 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 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
2032fn 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 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
2135fn 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 if let Ok(executable) = std::env::current_exe() {
3039 let _ = write_watcher_state(&identity, &layout, std::process::id(), &executable);
3040 }
3041
3042 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 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
3218enum 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 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 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 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 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 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(¤t, 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(¤t, 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(¤t) > 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 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
4369fn 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 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 #[cfg(unix)]
4481 {
4482 use std::os::unix::process::CommandExt;
4483 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 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 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 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 is_matching_watcher_process_with_system(&system, state)
4777}
4778
4779fn 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
4814fn 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 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 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 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 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 if process_started == 0 {
4966 return true;
4967 }
4968 let process_started = process_started as i64;
4969 let state_started = state_started.timestamp();
4970 (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
5010fn prefer_running_parent_root() -> Option<PathBuf> {
5012 let mut candidates = list_parent_watcher_state_files().ok()?;
5013 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
5029fn 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 if let Some(runtime) = state.runtime_key.as_deref() {
5079 if runtime != worktree_runtime_key() {
5080 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 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 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 }
5195 }
5196 }
5197 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
5206fn 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
5280fn 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 if let Some(path) = lookup_repo_path_from_system_inventory(token)? {
5305 return Ok(path);
5306 }
5307
5308 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 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
5487fn 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
5500fn 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 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 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 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 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 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
5779fn 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 value.to_ascii_lowercase()
5793}
5794
5795fn 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 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
5831fn 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 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
5872fn 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 if let Some(stripped) = value.strip_prefix("//?/") {
5897 value = stripped.to_string();
5898 }
5899 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 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 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 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 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 assert!(rules.ignores_relative("secrets/prod/api.key"));
6093 assert!(rules.ignores_relative("apps/web/.env.local"));
6094 assert!(!rules.ignores_relative("apps/web/.env.local.bak"));
6096
6097 assert!(rules.ignores_relative("apps/web/.cache/tmp"));
6099 assert!(rules.ignores_relative("packages/foo/coverage/index.html"));
6100
6101 assert!(rules.ignores_relative("docs/private-key-notes.md"));
6103 assert!(rules.ignores_relative("config/my-credential-store.json"));
6104
6105 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"))); }
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 assert!(rules.ignores_relative("packages/icons/src/index.ts"));
6174 assert!(rules.ignores_relative("apps/web/foo.generated.ts"));
6175 assert!(!rules.ignores_relative("e2e/login.spec.ts"));
6177 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 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 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 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 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}