use crate::codetime::GithubRepoRecord;
use crate::commands::cli_session::{cli_request_client, resolve_cli_access_token};
use crate::commands::terminal_table::{render_table, TableStyle};
use crate::config::{
ensure_global_xbp_paths, map_windows_path_to_wsl_mnt, map_wsl_mnt_path_to_windows,
resolve_device_identity, resolve_global_xbp_root_dir, resolve_worktree_watch_config, ApiConfig,
SshConfig, WorktreeWatchConfig,
};
use chrono::{DateTime, Utc};
use colored::Colorize;
use notify::event::{CreateKind, ModifyKind, RemoveKind, RenameMode};
use notify::{Config, Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
use reqwest::StatusCode;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, BTreeSet};
use std::fs::{self, File, OpenOptions};
use std::io::{BufRead, BufReader, Write};
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::sync::{mpsc, OnceLock};
use std::time::{Duration, Instant};
use sysinfo::{Pid, ProcessRefreshKind, ProcessesToUpdate, System, UpdateKind};
use uuid::Uuid;
const BACKGROUND_CHILD_ENV: &str = "XBP_WORKTREE_WATCH_BACKGROUND_CHILD";
const DEFAULT_REMOTE: &str = "origin";
const EVENT_DEDUPE_WINDOW_SECONDS: i64 = 5;
const WORKTREE_MUTATION_SYNC_BATCH_RECORD_LIMIT: usize = 250;
const WORKTREE_MUTATION_SYNC_BATCH_JSON_BYTES: usize = 768 * 1024;
const WORKTREE_MUTATION_SYNC_MAX_ATTEMPTS: usize = 3;
const WORKTREE_MUTATION_SYNC_RETRY_DELAY_SECONDS: u64 = 2;
#[derive(Debug, Clone, Default)]
struct WatchIgnoreRules {
forbidden_path_prefixes: Vec<String>,
forbidden_folders: BTreeSet<String>,
banned_words: Vec<String>,
global_ignore: crate::utils::XbpIgnoreSet,
project_ignore: crate::utils::XbpIgnoreSet,
}
impl WatchIgnoreRules {
fn load() -> Self {
Self::from_config(&resolve_worktree_watch_config())
}
fn from_config(config: &WorktreeWatchConfig) -> Self {
Self::from_config_and_project(config, None)
}
fn from_config_and_project(config: &WorktreeWatchConfig, project_root: Option<&Path>) -> Self {
let mut forbidden_folders = built_in_forbidden_folders();
for folder in &config.forbidden_folders {
let normalized = normalize_folder_token(folder);
if !normalized.is_empty() {
forbidden_folders.insert(normalized);
}
}
let forbidden_path_prefixes = config
.forbidden_paths
.iter()
.map(|path| normalize_path_prefix(path))
.filter(|path| !path.is_empty())
.collect::<Vec<_>>();
let banned_words = config
.banned_words
.iter()
.map(|word| word.trim().to_ascii_lowercase())
.filter(|word| !word.is_empty())
.collect::<Vec<_>>();
let project_ignore = project_root
.map(|root| load_project_watch_ignore(root))
.unwrap_or_default();
Self {
forbidden_path_prefixes,
forbidden_folders,
banned_words,
global_ignore: crate::config::load_global_worktree_ignore(),
project_ignore,
}
}
fn with_project_root(mut self, project_root: &Path) -> Self {
self.project_ignore = load_project_watch_ignore(project_root);
self
}
fn ignores_relative(&self, relative: &str) -> bool {
let normalized = normalize_relative_watch_path(relative);
if normalized.is_empty() {
return false;
}
if self.global_ignore.is_ignored_relative(&normalized, false)
|| self.global_ignore.is_ignored_relative(&normalized, true)
{
return true;
}
if self.project_ignore.is_ignored_relative(&normalized, false)
|| self.project_ignore.is_ignored_relative(&normalized, true)
{
return true;
}
let lower = normalized.to_ascii_lowercase();
let components = lower
.split('/')
.filter(|component| !component.is_empty())
.collect::<Vec<_>>();
if components
.iter()
.any(|component| self.forbidden_folders.contains(*component))
{
return true;
}
for prefix in &self.forbidden_path_prefixes {
if lower == *prefix || lower.starts_with(&format!("{prefix}/")) {
return true;
}
if !prefix.contains('/') && components.iter().any(|component| *component == prefix) {
return true;
}
}
for word in &self.banned_words {
if lower.contains(word) {
return true;
}
}
false
}
fn ignores_path(&self, root: &Path, path: &Path) -> bool {
if let Some(relative) = relative_watch_path(root, path) {
return self.ignores_relative(&relative);
}
false
}
fn skips_discovery_dir(&self, path: &Path) -> bool {
let Some(name) = path.file_name().and_then(|value| value.to_str()) else {
return false;
};
let trimmed = name.trim();
if self
.forbidden_folders
.contains(&trimmed.to_ascii_lowercase())
{
return true;
}
if self.global_ignore.skips_dir_name(trimmed) {
return true;
}
self.project_ignore.skips_dir_name(trimmed)
}
}
fn load_project_watch_ignore(project_root: &Path) -> crate::utils::XbpIgnoreSet {
use crate::strategies::XbpConfig;
use crate::utils::{load_project_xbp_ignore, parse_config_with_auto_heal};
let mut extra = Vec::new();
if let Some(found) = crate::utils::find_xbp_config_upwards(project_root) {
if let Ok(content) = fs::read_to_string(&found.config_path) {
if let Ok((config, _)) = parse_config_with_auto_heal::<XbpConfig>(&content, found.kind)
{
extra = config.watch_ignore_paths;
}
}
return load_project_xbp_ignore(&found.project_root, &extra);
}
load_project_xbp_ignore(project_root, &extra)
}
fn built_in_forbidden_folders() -> BTreeSet<String> {
use crate::utils::DEFAULT_NOISE_DIR_NAMES;
DEFAULT_NOISE_DIR_NAMES
.iter()
.map(|name| name.to_ascii_lowercase())
.collect()
}
fn normalize_folder_token(raw: &str) -> String {
raw.trim()
.trim_matches(|ch| ch == '/' || ch == '\\')
.to_ascii_lowercase()
}
fn normalize_path_prefix(raw: &str) -> String {
let trimmed = raw.trim().replace('\\', "/");
let without_dot = trimmed
.trim_start_matches("./")
.trim_matches(|ch| ch == '/' || ch == '\\');
without_dot.to_ascii_lowercase()
}
fn normalize_relative_watch_path(raw: &str) -> String {
raw.trim()
.replace('\\', "/")
.trim_start_matches("./")
.trim_matches('/')
.to_string()
}
fn watch_ignore_rules() -> &'static WatchIgnoreRules {
static RULES: OnceLock<WatchIgnoreRules> = OnceLock::new();
RULES.get_or_init(WatchIgnoreRules::load)
}
fn watch_ignore_rules_for_root(root: &Path) -> WatchIgnoreRules {
watch_ignore_rules().clone().with_project_root(root)
}
#[cfg(windows)]
const CREATE_NO_WINDOW: u32 = 0x08000000;
#[derive(Debug, Clone, Default)]
pub struct WorktreeWatchTargetOptions {
pub repo: Option<PathBuf>,
pub parent: Option<PathBuf>,
pub repos: Vec<PathBuf>,
}
pub fn resolve_default_worktree_watch_status_target() -> Result<WorktreeWatchTargetOptions, String> {
if let Some(parent) = prefer_running_parent_root() {
return Ok(WorktreeWatchTargetOptions {
repo: None,
parent: Some(parent),
repos: Vec::new(),
});
}
if let Some(parent) = prefer_any_parent_root() {
return Ok(WorktreeWatchTargetOptions {
repo: None,
parent: Some(parent),
repos: Vec::new(),
});
}
normalize_worktree_watch_target(WorktreeWatchTargetOptions {
repo: None,
parent: None,
repos: Vec::new(),
})
}
pub fn normalize_worktree_watch_target(
target: WorktreeWatchTargetOptions,
) -> Result<WorktreeWatchTargetOptions, String> {
if target.parent.is_some() && target.repo.is_some() {
return Err("Pass either `--repo` or `--parent`, not both.".to_string());
}
if target.parent.is_some() && !target.repos.is_empty() {
return Err("Pass either `--repos` or `--parent`, not both.".to_string());
}
if target.repo.is_some() && !target.repos.is_empty() {
return Err("Pass either `--repo` or `--repos`, not both.".to_string());
}
if let Some(parent) = target.parent.as_ref() {
let resolved = if parent.is_dir() {
normalize_windows_verbatim_path(
fs::canonicalize(parent).unwrap_or_else(|_| parent.clone()),
)
} else {
if let Some(found) = resolve_parent_folder_hint(parent)? {
found
} else {
return Err(format!(
"Parent folder not found: {}. Pass an absolute path or a folder under Documents/GitHub.",
parent.display()
));
}
};
return Ok(WorktreeWatchTargetOptions {
repo: None,
parent: Some(resolved),
repos: Vec::new(),
});
}
if !target.repos.is_empty() {
let mut repos = Vec::new();
let mut seen = BTreeSet::new();
for repo in &target.repos {
let path = resolve_repo_path_input(Some(repo.as_path()))?;
let absolute = normalize_windows_verbatim_path(
fs::canonicalize(&path).unwrap_or(path),
);
let key = path_identity_key(&absolute);
if seen.insert(key) {
repos.push(absolute);
}
}
return Ok(WorktreeWatchTargetOptions {
repo: None,
parent: None,
repos,
});
}
let identity = resolve_repo_identity(target.repo.as_deref())?;
Ok(WorktreeWatchTargetOptions {
repo: Some(identity.root),
parent: None,
repos: Vec::new(),
})
}
fn resolve_parent_folder_hint(parent: &Path) -> Result<Option<PathBuf>, String> {
if parent.is_absolute() {
return Ok(None);
}
let name = parent
.file_name()
.and_then(|value| value.to_str())
.unwrap_or_default();
if name.is_empty() {
return Ok(None);
}
let mut candidates: Vec<PathBuf> = Vec::new();
if let Some(home) = dirs::home_dir() {
candidates.push(home.join("Documents").join("GitHub").join(name));
candidates.push(home.join("Documents").join("github").join(name));
candidates.push(home.join("Documents").join(name));
candidates.push(home.join("src").join(name));
candidates.push(home.join(name));
}
if name.eq_ignore_ascii_case("github") {
if let Some(home) = dirs::home_dir() {
candidates.insert(0, home.join("Documents").join("GitHub"));
candidates.insert(1, home.join("Documents").join("github"));
}
}
for candidate in candidates {
if candidate.is_dir() {
return Ok(Some(normalize_windows_verbatim_path(
fs::canonicalize(&candidate).unwrap_or(candidate),
)));
}
}
Ok(None)
}
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct WorktreeWatchTraySnapshot {
pub target_label: String,
pub mode: String,
pub parent: Option<String>,
pub repository_count: usize,
pub running_watchers: usize,
pub any_running: bool,
pub all_running: bool,
pub parent_watcher_running: bool,
pub parent_watcher_pid: Option<u32>,
pub total_unsynced_files: u64,
pub total_unsynced_records: u64,
pub repositories: Vec<WorktreeWatchTrayRepoStatus>,
pub stats_line: Option<String>,
}
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct WorktreeWatchTrayRepoStatus {
pub root: String,
pub owner: String,
pub name: String,
pub branch: String,
pub running: bool,
pub pid: Option<u32>,
pub unsynced_files: u64,
pub unsynced_records: u64,
}
#[derive(Debug, Clone)]
pub struct WorktreeWatchStartOptions {
pub target: WorktreeWatchTargetOptions,
pub detach: bool,
pub sync_interval_seconds: u64,
pub once: bool,
}
#[derive(Debug, Clone)]
pub struct WorktreeWatchSyncOptions {
pub target: WorktreeWatchTargetOptions,
pub dry_run: bool,
pub resync: bool,
}
#[derive(Debug, Clone)]
pub struct WorktreeWatchStopOptions {
pub target: WorktreeWatchTargetOptions,
pub force: bool,
}
#[derive(Debug, Clone)]
pub struct WorktreeWatchStatusOptions {
pub target: WorktreeWatchTargetOptions,
pub json: bool,
pub records: bool,
pub record_limit: usize,
pub stats: bool,
pub repo_activity: bool,
pub stats_gap_minutes: u64,
}
#[derive(Debug, Clone)]
struct RepoIdentity {
owner: String,
name: String,
branch: String,
root: PathBuf,
head_sha: Option<String>,
}
#[derive(Debug, Clone)]
struct SpoolLayout {
root: PathBuf,
events_file: PathBuf,
commits_file: PathBuf,
stats_file: PathBuf,
watcher_state_file: PathBuf,
sync_log_file: PathBuf,
event_dedupe_file: PathBuf,
event_dedupe_dir: PathBuf,
commit_dedupe_dir: PathBuf,
}
#[derive(Debug, Clone)]
struct ParentSpoolLayout {
watcher_state_file: PathBuf,
}
#[derive(Debug)]
struct WatchedRepo {
identity: RepoIdentity,
layout: SpoolLayout,
last_head: Option<String>,
event_dedupe: RecentEventDedupe,
}
#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
struct WorktreeMutationEvent {
id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
fingerprint: Option<String>,
repo_owner: String,
repo_name: String,
branch_name: String,
repo_root: String,
head_sha: Option<String>,
event_kind: String,
paths: Vec<String>,
primary_path: Option<String>,
old_path: Option<String>,
new_path: Option<String>,
added_lines: Option<u64>,
removed_lines: Option<u64>,
total_lines: Option<u64>,
file_created: bool,
file_removed: bool,
folder_created: bool,
folder_removed: bool,
renamed_or_moved: bool,
raw_kind: String,
occurred_at: DateTime<Utc>,
}
#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
struct WorktreeCommitEvent {
id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
fingerprint: Option<String>,
repo_owner: String,
repo_name: String,
branch_name: String,
repo_root: String,
previous_head_sha: Option<String>,
head_sha: String,
subject: Option<String>,
author_name: Option<String>,
author_email: Option<String>,
committed_at: Option<String>,
occurred_at: DateTime<Utc>,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct WorktreeMutationIngestPayload {
device: WorktreeMutationDevicePayload,
repository: WorktreeMutationRepositoryPayload,
events: Vec<WorktreeMutationEvent>,
commits: Vec<WorktreeCommitEvent>,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct WorktreeMutationDevicePayload {
hardware_id: String,
device_name: Option<String>,
hostname: Option<String>,
platform: String,
#[serde(skip_serializing_if = "Option::is_none")]
runtime_key: Option<String>,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct WorktreeMutationRepositoryPayload {
owner: String,
name: String,
branch_name: String,
repo_root: String,
}
#[derive(Debug)]
struct SyncCandidate {
path: PathBuf,
records: SyncRecords,
}
#[derive(Debug)]
struct WorktreeMutationUploadBatch {
events: Vec<WorktreeMutationEvent>,
commits: Vec<WorktreeCommitEvent>,
estimated_json_bytes: usize,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum WorktreeMutationSyncErrorKind {
Retryable,
NonRetryable,
}
#[derive(Debug)]
struct WorktreeMutationSyncError {
kind: WorktreeMutationSyncErrorKind,
message: String,
}
struct SyncLogAppend<'a> {
identity: &'a RepoIdentity,
layout: &'a SpoolLayout,
endpoint: &'a str,
status_code: u16,
event_count: usize,
commit_count: usize,
candidates: &'a [SyncCandidate],
resync: bool,
}
#[derive(Debug, Clone, Copy)]
enum SyncFileKind {
Events,
Commits,
}
#[derive(Debug)]
enum SyncRecords {
Events(Vec<WorktreeMutationEvent>),
Commits(Vec<WorktreeCommitEvent>),
}
impl SyncRecords {
fn len(&self) -> usize {
match self {
Self::Events(records) => records.len(),
Self::Commits(records) => records.len(),
}
}
fn is_empty(&self) -> bool {
self.len() == 0
}
}
#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatchStatsSummary {
generated_at: DateTime<Utc>,
repository_owner: String,
repository_name: String,
branch_name: String,
repo_root: String,
spool_root: String,
session_gap_minutes: u64,
total_events: u64,
total_commits: u64,
first_event_at: Option<DateTime<Utc>>,
last_event_at: Option<DateTime<Utc>>,
session_count: u64,
observed_span_seconds: u64,
estimated_coding_seconds: u64,
added_lines: u64,
removed_lines: u64,
by_file: Vec<WorktreeWatchFileStats>,
by_filetype: Vec<WorktreeWatchBucketStats>,
by_event_kind: Vec<WorktreeWatchBucketStats>,
}
#[derive(Debug, Serialize, Clone)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatchRepoActivitySummary {
generated_at: DateTime<Utc>,
repository_owner: String,
repository_name: String,
repo_root: String,
repository_spool_root: String,
stats_file: String,
session_gap_minutes: u64,
branch_count: u64,
total_events: u64,
total_commits: u64,
first_event_at: Option<DateTime<Utc>>,
last_event_at: Option<DateTime<Utc>>,
session_count: u64,
observed_span_seconds: u64,
estimated_coding_seconds: u64,
added_lines: u64,
removed_lines: u64,
branches: Vec<WorktreeWatchBranchActivityStats>,
}
#[derive(Debug, Serialize, Clone)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatchBranchActivityStats {
branch_name: String,
spool_root: String,
total_events: u64,
total_commits: u64,
first_event_at: Option<DateTime<Utc>>,
last_event_at: Option<DateTime<Utc>>,
session_count: u64,
observed_span_seconds: u64,
estimated_coding_seconds: u64,
added_lines: u64,
removed_lines: u64,
}
#[derive(Debug, Serialize, Deserialize, Clone, Default)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatchFileStats {
path: String,
filetype: String,
event_count: u64,
estimated_coding_seconds: u64,
added_lines: u64,
removed_lines: u64,
}
#[derive(Debug, Serialize, Deserialize, Clone, Default)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatchBucketStats {
name: String,
event_count: u64,
estimated_coding_seconds: u64,
added_lines: u64,
removed_lines: u64,
}
#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatcherState {
pid: u32,
repo_root: String,
repository_owner: String,
repository_name: String,
branch_name: String,
executable: String,
started_at: DateTime<Utc>,
#[serde(default, skip_serializing_if = "Option::is_none")]
runtime_key: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
platform: Option<String>,
}
#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
struct ParentWorktreeWatcherState {
pid: u32,
parent_root: String,
executable: String,
started_at: DateTime<Utc>,
#[serde(default, skip_serializing_if = "Option::is_none")]
runtime_key: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
platform: Option<String>,
}
#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatchSyncLogEntry {
id: String,
synced_at: DateTime<Utc>,
endpoint: String,
repository_owner: String,
repository_name: String,
branch_name: String,
repo_root: String,
resync: bool,
status_code: u16,
event_count: usize,
commit_count: usize,
spool_file_count: usize,
spool_files: Vec<String>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum StopWatcherOutcome {
Stopped(u32),
RemovedStaleState,
NoState,
}
#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatchEventDedupeRecord {
fingerprint: String,
first_seen_at: DateTime<Utc>,
event_kind: String,
primary_path: Option<String>,
}
#[derive(Debug, Default)]
struct RecentEventDedupe {
fingerprints: BTreeSet<String>,
}
pub async fn run_worktree_watch_start(options: WorktreeWatchStartOptions) -> Result<(), String> {
if options.detach {
return run_worktree_watch_start_detached(options);
}
if let Some(parent) = options.target.parent.as_deref() {
let identities = resolve_target_identities(&options.target)?;
if identities.is_empty() {
println!("No git repositories found under parent folder.");
return Ok(());
}
if options.once {
for identity in &identities {
let layout = prepare_spool_layout(identity)?;
record_commit_snapshot(identity, &layout, None)?;
println!("Recorded current git state for {}", identity.root.display());
}
return Ok(());
}
let parent = canonical_parent_path(parent)?;
println!(
"Watching {} recursively and routing mutations for {} repo(s).",
parent.display(),
identities.len()
);
return watch_parent_foreground(parent, identities, options.sync_interval_seconds).await;
}
let identity = resolve_repo_identity(options.target.repo.as_deref())?;
let layout = prepare_spool_layout(&identity)?;
println!(
"Watching {} and spooling mutations to {} [runtime={}]",
identity.root.display(),
layout.root.display(),
worktree_runtime_key()
);
if options.once {
record_commit_snapshot(&identity, &layout, None)?;
return Ok(());
}
watch_foreground(identity, layout, options.sync_interval_seconds).await
}
pub fn run_worktree_watch_start_detached(
options: WorktreeWatchStartOptions,
) -> Result<(), String> {
if let Some(parent) = options.target.parent.as_deref() {
let parent_path = canonical_parent_path(parent)?;
let repo_count = count_git_roots_under(&parent_path)?;
if repo_count == 0 {
println!("No git repositories found under parent folder.");
return Ok(());
}
let path = spawn_detached_parent_worktree_watch(&parent_path)?;
println!(
"Started one background worktree watcher for {} covering {} repo(s).",
path.display(),
repo_count
);
if let Some(port) =
crate::commands::worktree_watch_api::resolve_worktree_watch_api_port(None)
{
println!(
"Local API (when free): http://127.0.0.1:{port} — dashboard: apps/worktree-watch-app/"
);
}
return Ok(());
}
if !options.target.repos.is_empty() {
for repo in &options.target.repos {
let path = spawn_detached_worktree_watch(Some(repo.as_path()))?;
println!("Started background worktree watcher for {}", path.display());
}
if let Some(port) =
crate::commands::worktree_watch_api::resolve_worktree_watch_api_port(None)
{
println!(
"Local API (when free): http://127.0.0.1:{port} — dashboard: apps/worktree-watch-app/"
);
}
return Ok(());
}
spawn_detached_worktree_watch(options.target.repo.as_deref()).map(|path| {
println!("Started background worktree watcher for {}", path.display());
if let Some(port) =
crate::commands::worktree_watch_api::resolve_worktree_watch_api_port(None)
{
println!(
"Local API (when free): http://127.0.0.1:{port} — dashboard: apps/worktree-watch-app/"
);
}
})
}
pub async fn run_worktree_watch_sync(options: WorktreeWatchSyncOptions) -> Result<(), String> {
let identities = resolve_target_identities(&options.target)?;
let repo_count = identities.len();
let mut plans = Vec::new();
let mut total_files = 0usize;
let mut total_records = 0usize;
for identity in identities {
let layout = prepare_spool_layout(&identity)?;
let candidates = collect_spool_candidates(&layout, options.resync)?;
total_files += candidates.len();
total_records += candidates
.iter()
.map(|candidate| candidate.records.len())
.sum::<usize>();
if !candidates.is_empty() {
plans.push((identity, candidates));
}
}
if options.dry_run {
println!(
"Would sync {} record(s) from {} spool file(s) across {} repo(s).",
total_records, total_files, repo_count
);
if options.resync {
println!("Resync mode includes files already marked `.synced.*.jsonl`.");
}
return Ok(());
}
if plans.is_empty() {
print_no_sync_candidates_message(&options.target, options.resync)?;
return Ok(());
}
let upload_repo_count = plans.len();
for (identity, candidates) in plans {
sync_candidates(&identity, candidates, options.resync).await?;
}
println!(
"Synced {} worktree mutation record(s) across {} repo(s).",
total_records, upload_repo_count
);
Ok(())
}
pub fn run_worktree_watch_stop(options: WorktreeWatchStopOptions) -> Result<(), String> {
if let Some(parent) = options.target.parent.as_deref() {
let parent = canonical_parent_path(parent)?;
match stop_existing_parent_watcher(&parent, options.force)? {
StopWatcherOutcome::Stopped(pid) => {
println!(
"Stopped background parent worktree watcher {} for {}",
pid,
parent.display()
);
}
StopWatcherOutcome::RemovedStaleState => {
println!(
"Removed stale parent worktree watcher state for {}",
parent.display()
);
}
StopWatcherOutcome::NoState => {
println!("No parent worktree watcher state for {}", parent.display());
}
}
}
let identities = resolve_target_identities(&options.target)?;
if identities.is_empty() {
println!("No git repositories found for worktree watcher stop.");
return Ok(());
}
let mut stopped = 0usize;
let mut stale = 0usize;
let mut missing = 0usize;
for identity in identities {
let layout = prepare_spool_layout(&identity)?;
match stop_existing_watcher(&identity, &layout, options.force)? {
StopWatcherOutcome::Stopped(pid) => {
stopped += 1;
println!(
"Stopped background worktree watcher {} for {}",
pid,
identity.root.display()
);
}
StopWatcherOutcome::RemovedStaleState => {
stale += 1;
println!(
"Removed stale worktree watcher state for {}",
identity.root.display()
);
}
StopWatcherOutcome::NoState => {
missing += 1;
println!(
"No background worktree watcher state for {}",
identity.root.display()
);
}
}
}
println!(
"Worktree watcher stop complete: {stopped} stopped, {stale} stale state file(s) removed, {missing} without state."
);
Ok(())
}
pub async fn run_worktree_watch_status(options: WorktreeWatchStatusOptions) -> Result<(), String> {
let need_heavy = options.records || options.stats || options.repo_activity;
let snapshot = collect_worktree_watch_tray_snapshot(&options.target)?;
let mut payloads = Vec::new();
let mut total_files = 0usize;
let mut total_records = 0usize;
for tray_repo in &snapshot.repositories {
let identity = identity_from_tray_repo(tray_repo);
let mut status = if need_heavy {
worktree_watch_status_payload(&identity, options.records, options.record_limit)?
} else {
worktree_watch_status_from_tray_repo(tray_repo)?
};
if let Some(object) = status.as_object_mut() {
object.insert("running".to_string(), json!(tray_repo.running));
object.insert("pid".to_string(), json!(tray_repo.pid));
if let Some(last_sync) = object.get("lastSync") {
if let Some(synced_at) = last_sync.get("syncedAt").and_then(Value::as_str) {
if let Ok(dt) = DateTime::parse_from_rfc3339(synced_at) {
let age = (Utc::now() - dt.with_timezone(&Utc))
.num_seconds()
.max(0) as u64;
object.insert("lastSyncSecondsAgo".to_string(), json!(age));
}
}
}
}
let stats = if options.stats {
Some(generate_and_store_stats(
&identity,
options.stats_gap_minutes,
)?)
} else {
None
};
let repo_activity = if options.repo_activity {
Some(generate_and_store_repo_activity(
&identity,
options.stats_gap_minutes,
)?)
} else {
None
};
if options.stats || options.repo_activity {
if let Err(err) = crate::commands::todos::effort::recompute_effort(
&identity.root,
options.stats_gap_minutes,
true,
) {
eprintln!("note: todo effort recompute: {err}");
}
}
let status = attach_optional_stats(status, stats, repo_activity)?;
total_files += status
.get("unsyncedFiles")
.and_then(Value::as_u64)
.unwrap_or(0) as usize;
total_records += status
.get("unsyncedRecords")
.and_then(Value::as_u64)
.unwrap_or(0) as usize;
payloads.push(status);
}
if options.json {
let payload = if options.target.parent.is_some() || payloads.len() > 1 {
json!({
"targetLabel": snapshot.target_label,
"mode": snapshot.mode,
"parent": snapshot.parent,
"parentWatcherRunning": snapshot.parent_watcher_running,
"parentWatcherPid": snapshot.parent_watcher_pid,
"anyRunning": snapshot.any_running,
"allRunning": snapshot.all_running,
"runningWatchers": snapshot.running_watchers,
"repositories": payloads,
"repositoryCount": payloads.len(),
"unsyncedFiles": total_files,
"unsyncedRecords": total_records,
})
} else {
payloads.into_iter().next().unwrap_or_else(|| json!({}))
};
println!(
"{}",
serde_json::to_string_pretty(&payload)
.map_err(|error| format!("Failed to render status JSON: {error}"))?
);
} else {
print_status_overview(&snapshot, total_files, total_records);
print_status_matrix(&payloads, &snapshot);
if need_heavy {
for (index, payload) in payloads.iter().enumerate() {
if index > 0 {
println!();
}
if options.records {
println!(
"{} {}/{}",
"records".bright_black(),
payload
.get("repositoryOwner")
.and_then(Value::as_str)
.unwrap_or("unknown"),
payload
.get("repositoryName")
.and_then(Value::as_str)
.unwrap_or("repository")
);
print_status_records(payload);
}
if options.stats {
print_status_stats(payload);
}
if options.repo_activity {
print_repo_activity(payload);
}
}
}
}
Ok(())
}
fn print_status_overview(
snapshot: &WorktreeWatchTraySnapshot,
total_files: usize,
total_records: usize,
) {
let running_label = if snapshot.any_running {
"RUNNING".green().bold()
} else {
"STOPPED".red().bold()
};
println!(
"{} {} {}",
"worktree-watch".bright_magenta().bold(),
running_label,
snapshot.target_label.bright_white()
);
if snapshot.mode == "parent" {
let parent_label = if snapshot.parent_watcher_running {
format!(
"parent watcher {}",
snapshot
.parent_watcher_pid
.map(|pid| format!("pid {pid}"))
.unwrap_or_else(|| "active".to_string())
)
.green()
.to_string()
} else {
"parent watcher not detected".red().to_string()
};
println!(
" {} {} {} repo(s) covered",
parent_label,
"·".bright_black(),
snapshot.repository_count
);
} else {
println!(
" {} watching · {} repo(s)",
snapshot.running_watchers, snapshot.repository_count
);
}
println!(
" {} unsynced record(s) in {} file(s)",
total_records.to_string().yellow(),
total_files
);
if let Some(line) = snapshot.stats_line.as_ref() {
println!(" {}", line.bright_black());
}
println!();
}
fn print_status_matrix(payloads: &[Value], snapshot: &WorktreeWatchTraySnapshot) {
if payloads.is_empty() {
println!("{}", "No repositories found for this target.".yellow());
return;
}
let headers = [
"REPO",
"BRANCH",
"STATE",
"PID",
"UNSYNCED",
"LAST SYNC",
"AGE",
];
let mut rows = Vec::with_capacity(payloads.len());
for payload in payloads {
let owner = payload
.get("repositoryOwner")
.or_else(|| payload.get("owner"))
.and_then(Value::as_str)
.unwrap_or("unknown");
let name = payload
.get("repositoryName")
.or_else(|| payload.get("name"))
.and_then(Value::as_str)
.unwrap_or("repository");
let branch = payload
.get("branchName")
.or_else(|| payload.get("branch"))
.and_then(Value::as_str)
.unwrap_or("unknown");
let running = payload
.get("running")
.and_then(Value::as_bool)
.unwrap_or(false);
let pid = payload
.get("pid")
.and_then(Value::as_u64)
.map(|value| value.to_string())
.unwrap_or_else(|| {
if running {
snapshot
.parent_watcher_pid
.map(|value| value.to_string())
.unwrap_or_else(|| "—".to_string())
} else {
"—".to_string()
}
});
let unsynced = payload
.get("unsyncedRecords")
.and_then(Value::as_u64)
.unwrap_or(0);
let (last_sync_at, age) = last_sync_display(payload);
let state = if running {
"RUN".green().bold().to_string()
} else {
"OFF".red().to_string()
};
let unsynced_cell = if unsynced > 0 {
unsynced.to_string().yellow().to_string()
} else {
"0".bright_black().to_string()
};
let age_cell = if age == "never" {
age.bright_black().to_string()
} else {
age.cyan().to_string()
};
rows.push(vec![
format!("{owner}/{name}"),
branch.to_string(),
state,
pid,
unsynced_cell,
last_sync_at,
age_cell,
]);
}
print!(
"{}",
render_table(&headers, &rows, TableStyle::Box, "")
);
let running = snapshot.running_watchers;
let stopped = snapshot
.repository_count
.saturating_sub(snapshot.running_watchers);
println!();
println!(
"{} {} running · {} stopped · {} total",
"summary".bright_black(),
running.to_string().green().bold(),
stopped.to_string().red(),
snapshot.repository_count
);
}
fn last_sync_display(payload: &Value) -> (String, String) {
let Some(last_sync) = payload.get("lastSync") else {
return ("—".to_string(), "never".to_string());
};
let synced_at = last_sync
.get("syncedAt")
.and_then(Value::as_str)
.unwrap_or("unknown");
let age = if let Some(seconds) = payload
.get("lastSyncSecondsAgo")
.and_then(Value::as_u64)
{
format_age_seconds(seconds)
} else if let Ok(dt) = DateTime::parse_from_rfc3339(synced_at) {
let seconds = (Utc::now() - dt.with_timezone(&Utc)).num_seconds().max(0) as u64;
format_age_seconds(seconds)
} else {
"unknown".to_string()
};
let compact = if synced_at.len() >= 19 {
synced_at[11..19].to_string()
} else {
synced_at.to_string()
};
(compact, age)
}
fn format_age_seconds(seconds: u64) -> String {
if seconds < 60 {
format!("{seconds}s ago")
} else if seconds < 3600 {
format!("{}m ago", seconds / 60)
} else if seconds < 86400 {
format!("{}h ago", seconds / 3600)
} else {
format!("{}d ago", seconds / 86400)
}
}
fn attach_optional_stats(
mut payload: Value,
stats: Option<WorktreeWatchStatsSummary>,
repo_activity: Option<WorktreeWatchRepoActivitySummary>,
) -> Result<Value, String> {
if let Some(object) = payload.as_object_mut() {
if let Some(stats) = stats {
object.insert(
"stats".to_string(),
serde_json::to_value(stats)
.map_err(|error| format!("Failed to render stats payload: {error}"))?,
);
}
if let Some(repo_activity) = repo_activity {
object.insert(
"repositoryActivity".to_string(),
serde_json::to_value(repo_activity)
.map_err(|error| format!("Failed to render repo activity payload: {error}"))?,
);
}
}
Ok(payload)
}
pub fn spawn_detached_worktree_watch_for_current_repo() -> Result<Option<PathBuf>, String> {
if is_background_child() {
return Ok(None);
}
let identity = match resolve_repo_identity(None) {
Ok(identity) => identity,
Err(_) => return Ok(None),
};
spawn_detached_worktree_watch_for_identity(&identity).map(Some)
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct WorktreeWatchDiagnostic {
pub global_xbp_root: String,
pub mutations_root: String,
pub runtime_key: String,
pub platform: String,
pub is_wsl: bool,
pub repo_owner: Option<String>,
pub repo_name: Option<String>,
pub branch: Option<String>,
pub branch_spool: Option<String>,
pub branch_spool_exists: bool,
pub watcher_state_path: Option<String>,
pub watcher_state_present: bool,
pub watcher_running: bool,
pub event_jsonl_files: usize,
pub commit_jsonl_files: usize,
pub alternate_spool_roots: Vec<String>,
pub notes: Vec<String>,
}
#[derive(Debug, Clone)]
pub struct MutationActivityEvent {
pub paths: Vec<String>,
pub primary_path: Option<String>,
pub occurred_at: DateTime<Utc>,
pub added_lines: Option<u64>,
pub removed_lines: Option<u64>,
pub head_sha: Option<String>,
pub branch_name: String,
}
#[derive(Debug, Clone)]
pub struct CommitActivityEvent {
pub occurred_at: DateTime<Utc>,
pub head_sha: String,
pub subject: Option<String>,
pub branch_name: String,
pub repo_root: String,
}
pub fn load_repo_mutation_activity(
owner: &str,
name: &str,
) -> Result<Vec<MutationActivityEvent>, String> {
let identity = RepoIdentity {
owner: owner.to_string(),
name: name.to_string(),
branch: "main".to_string(),
root: PathBuf::from("."),
head_sha: None,
};
let mut events = Vec::new();
for repo_root in repository_spool_search_roots(&identity)? {
if !repo_root.is_dir() {
continue;
}
for entry in fs::read_dir(&repo_root).map_err(|e| {
format!(
"Failed to read mutation spool {}: {e}",
repo_root.display()
)
})? {
let entry = entry.map_err(|e| format!("Failed to read spool entry: {e}"))?;
if !entry.file_type().map(|t| t.is_dir()).unwrap_or(false) {
continue;
}
let branch_layout = spool_layout_for_existing_root(&entry.path());
let candidates = collect_spool_candidates(&branch_layout, true)?;
for candidate in candidates {
if let SyncRecords::Events(records) = candidate.records {
for record in records {
events.push(MutationActivityEvent {
paths: record.paths,
primary_path: record.primary_path,
occurred_at: record.occurred_at,
added_lines: record.added_lines,
removed_lines: record.removed_lines,
head_sha: record.head_sha,
branch_name: record.branch_name,
});
}
}
}
}
}
events.sort_by_key(|e| e.occurred_at);
Ok(events)
}
pub fn load_repo_commit_activity(
owner: &str,
name: &str,
) -> Result<Vec<CommitActivityEvent>, String> {
let identity = RepoIdentity {
owner: owner.to_string(),
name: name.to_string(),
branch: "main".to_string(),
root: PathBuf::from("."),
head_sha: None,
};
let mut commits = Vec::new();
for repo_root in repository_spool_search_roots(&identity)? {
if !repo_root.is_dir() {
continue;
}
for entry in fs::read_dir(&repo_root).map_err(|e| {
format!(
"Failed to read mutation spool {}: {e}",
repo_root.display()
)
})? {
let entry = entry.map_err(|e| format!("Failed to read spool entry: {e}"))?;
if !entry.file_type().map(|t| t.is_dir()).unwrap_or(false) {
continue;
}
let branch_layout = spool_layout_for_existing_root(&entry.path());
let candidates = collect_spool_candidates(&branch_layout, true)?;
for candidate in candidates {
if let SyncRecords::Commits(records) = candidate.records {
for record in records {
commits.push(CommitActivityEvent {
occurred_at: record.occurred_at,
head_sha: record.head_sha,
subject: record.subject,
branch_name: record.branch_name,
repo_root: record.repo_root,
});
}
}
}
}
}
commits.sort_by_key(|c| c.occurred_at);
Ok(commits)
}
pub fn collect_worktree_watch_diagnostic() -> WorktreeWatchDiagnostic {
let runtime_key = worktree_runtime_key();
let platform = std::env::consts::OS.to_string();
let is_wsl = is_wsl_runtime();
let mut notes = Vec::new();
let global_xbp_root = match ensure_global_xbp_paths() {
Ok(paths) => paths.root_dir,
Err(error) => {
notes.push(format!("Failed to resolve global XBP home: {error}"));
resolve_global_xbp_root_dir()
}
};
let mutations_root = global_xbp_root.join("mutations");
let identity = resolve_repo_identity(None).ok();
let (
repo_owner,
repo_name,
branch,
branch_spool,
branch_spool_exists,
watcher_state_path,
watcher_state_present,
watcher_running,
event_jsonl_files,
commit_jsonl_files,
alternate_spool_roots,
) = if let Some(identity) = identity.as_ref() {
let layout = prepare_spool_layout(identity).ok();
let branch_spool = layout
.as_ref()
.map(|value| value.root.display().to_string());
let branch_spool_exists = layout
.as_ref()
.map(|value| value.root.exists())
.unwrap_or(false);
let watcher_state_path = layout
.as_ref()
.map(|value| value.watcher_state_file.display().to_string());
let watcher_state_present = layout
.as_ref()
.map(|value| {
watcher_state_read_candidates(&value.watcher_state_file)
.iter()
.any(|candidate| candidate.exists())
})
.unwrap_or(false);
let watcher_running = layout
.as_ref()
.and_then(|value| read_watcher_state(&value.watcher_state_file).ok().flatten())
.map(|state| is_matching_watcher_process(&state))
.unwrap_or(false);
let mut event_jsonl_files = 0usize;
let mut commit_jsonl_files = 0usize;
if let Some(layout) = layout.as_ref() {
if let Ok(entries) = fs::read_dir(&layout.root) {
for entry in entries.flatten() {
let name = entry.file_name().to_string_lossy().to_string();
if name.starts_with("events-") && name.ends_with(".jsonl") {
event_jsonl_files += 1;
} else if name.starts_with("commits-") && name.ends_with(".jsonl") {
commit_jsonl_files += 1;
}
}
}
}
let alternate_spool_roots = repository_spool_search_roots(identity)
.unwrap_or_default()
.into_iter()
.map(|path| {
path.join(sanitize_path_component(&identity.branch))
.display()
.to_string()
})
.filter(|path| {
layout
.as_ref()
.map(|value| path_identity_key(Path::new(path)) != path_identity_key(&value.root))
.unwrap_or(true)
})
.filter(|path| Path::new(path).exists())
.collect::<Vec<_>>();
let root_display = global_xbp_root
.display()
.to_string()
.replace('\\', "/")
.to_ascii_lowercase();
if is_wsl && !root_display.starts_with("/mnt/") {
notes.push(
"WSL global XBP home is not on a Windows drive mount. If Windows has mutations under C:\\Users\\…\\.xbp, set XBP_HOME=/mnt/c/Users/<you>/.xbp so status/sync share that spool."
.to_string(),
);
}
if is_wsl && root_display.starts_with("/mnt/") {
notes.push(
"Using a Windows-profile-backed XBP home from WSL; spoils are reconcilable with native Windows."
.to_string(),
);
}
if !alternate_spool_roots.is_empty() {
notes.push(format!(
"Found {} alternate spool location(s) with historical data (legacy homes / runtime layouts).",
alternate_spool_roots.len()
));
}
(
Some(identity.owner.clone()),
Some(identity.name.clone()),
Some(identity.branch.clone()),
branch_spool,
branch_spool_exists,
watcher_state_path,
watcher_state_present,
watcher_running,
event_jsonl_files,
commit_jsonl_files,
alternate_spool_roots,
)
} else {
notes.push(
"Not inside a git repository with a resolvable remote; showing global XBP home only."
.to_string(),
);
(
None,
None,
None,
None,
false,
None,
false,
false,
0,
0,
Vec::new(),
)
};
WorktreeWatchDiagnostic {
global_xbp_root: global_xbp_root.display().to_string(),
mutations_root: mutations_root.display().to_string(),
runtime_key,
platform,
is_wsl,
repo_owner,
repo_name,
branch,
branch_spool,
branch_spool_exists,
watcher_state_path,
watcher_state_present,
watcher_running,
event_jsonl_files,
commit_jsonl_files,
alternate_spool_roots,
notes,
}
}
fn resolve_target_identities(
target: &WorktreeWatchTargetOptions,
) -> Result<Vec<RepoIdentity>, String> {
if target.repo.is_some() && target.parent.is_some() {
return Err("Pass either `--repo` or `--parent`, not both.".to_string());
}
if !target.repos.is_empty() && target.parent.is_some() {
return Err("Pass either `--repos` or `--parent`, not both.".to_string());
}
if target.repo.is_some() && !target.repos.is_empty() {
return Err("Pass either `--repo` or `--repos`, not both.".to_string());
}
if let Some(parent) = target.parent.as_deref() {
return discover_repo_identities_under(parent);
}
if !target.repos.is_empty() {
let mut identities = Vec::new();
let mut seen = BTreeSet::new();
for repo in &target.repos {
let identity = resolve_repo_identity(Some(repo.as_path()))?;
let key = path_identity_key(&identity.root);
if seen.insert(key) {
identities.push(identity);
}
}
return Ok(identities);
}
resolve_repo_identity(target.repo.as_deref()).map(|identity| vec![identity])
}
pub fn collect_worktree_watch_tray_snapshot(
target: &WorktreeWatchTargetOptions,
) -> Result<WorktreeWatchTraySnapshot, String> {
let identities = resolve_target_identities(target)?;
let mode = if target.parent.is_some() {
"parent"
} else if !target.repos.is_empty() {
"repos"
} else {
"repo"
}
.to_string();
let parent_label = target
.parent
.as_ref()
.map(|path| path.display().to_string());
let target_label = if let Some(parent) = parent_label.as_ref() {
format!("parent:{parent}")
} else if !target.repos.is_empty() {
format!("{} repo(s)", target.repos.len())
} else if let Some(repo) = target.repo.as_ref() {
repo.display().to_string()
} else {
identities
.first()
.map(|identity| identity.root.display().to_string())
.unwrap_or_else(|| "current repo".to_string())
};
let mut parent_state: Option<ParentWorktreeWatcherState> = None;
if let Some(parent) = target.parent.as_deref() {
if let Ok(parent) = canonical_parent_path(parent) {
if let Ok(layout) = prepare_parent_spool_layout(&parent) {
parent_state = read_parent_watcher_state(&layout.watcher_state_file)?;
}
}
}
let mut repo_states: Vec<(usize, WorktreeWatcherState)> = Vec::new();
let mut layouts: Vec<SpoolLayout> = Vec::with_capacity(identities.len());
for (index, identity) in identities.iter().enumerate() {
let layout = prepare_spool_layout(identity)?;
if let Some(state) = read_watcher_state(&layout.watcher_state_file)? {
if same_repo_watcher_state(identity, &state) {
repo_states.push((index, state));
}
}
layouts.push(layout);
}
let mut pids: Vec<u32> = repo_states.iter().map(|(_, state)| state.pid).collect();
if let Some(state) = parent_state.as_ref() {
pids.push(state.pid);
}
let process_system = process_system_for_pids(&pids);
let mut parent_watcher_running = false;
let mut parent_watcher_pid = None;
if let Some(state) = parent_state.as_ref() {
if is_live_parent_watcher_process_with_system(&process_system, state) {
parent_watcher_running = true;
parent_watcher_pid = Some(state.pid);
}
}
let mut repositories = Vec::new();
let mut running_watchers = 0usize;
let mut total_unsynced_files = 0u64;
let mut total_unsynced_records = 0u64;
for (index, identity) in identities.iter().enumerate() {
let layout = &layouts[index];
let counts = count_spool_summary(layout)?;
let unsynced_files = counts.unsynced_files;
let unsynced_records = counts.unsynced_records;
total_unsynced_files += unsynced_files;
total_unsynced_records += unsynced_records;
let (running, pid) = match repo_states
.iter()
.find(|(repo_index, _)| *repo_index == index)
.map(|(_, state)| state)
{
Some(state) if is_live_watcher_process_with_system(&process_system, state) => {
(true, Some(state.pid))
}
_ => (false, None),
};
let running = running || parent_watcher_running;
let pid = pid.or(if running {
parent_watcher_pid
} else {
None
});
if running {
running_watchers += 1;
}
repositories.push(WorktreeWatchTrayRepoStatus {
root: normalize_repo_root_for_payload(&identity.root),
owner: identity.owner.clone(),
name: identity.name.clone(),
branch: identity.branch.clone(),
running,
pid,
unsynced_files,
unsynced_records,
});
}
let repository_count = repositories.len();
let any_running = parent_watcher_running || running_watchers > 0;
let all_running = repository_count > 0 && running_watchers >= repository_count;
let stats_line = Some(format!(
"{repository_count} repo(s) · {running_watchers} watching · {total_unsynced_records} unsynced · {total_unsynced_files} file(s)"
));
Ok(WorktreeWatchTraySnapshot {
target_label,
mode,
parent: parent_label,
repository_count,
running_watchers,
any_running,
all_running,
parent_watcher_running,
parent_watcher_pid,
total_unsynced_files,
total_unsynced_records,
repositories,
stats_line,
})
}
pub const WORKTREE_WATCH_API_DEFAULT_STATS_GAP_MINUTES: u64 = 30;
pub fn collect_worktree_watch_api_status(
target: &WorktreeWatchTargetOptions,
last_error: Option<&str>,
) -> Result<Value, String> {
let snapshot = collect_worktree_watch_tray_snapshot(target)?;
let mut repositories = Vec::new();
for tray_repo in &snapshot.repositories {
let mut status = worktree_watch_status_from_tray_repo(tray_repo)?;
if let Some(object) = status.as_object_mut() {
object.insert("running".to_string(), json!(tray_repo.running));
object.insert("pid".to_string(), json!(tray_repo.pid));
object.insert("owner".to_string(), json!(tray_repo.owner));
object.insert("name".to_string(), json!(tray_repo.name));
object.insert("branch".to_string(), json!(tray_repo.branch));
object.insert("root".to_string(), json!(tray_repo.root));
if let Some(spool) = object.get("spoolRoot").cloned() {
object.insert("spoolPath".to_string(), spool);
}
object.insert("lastError".to_string(), Value::Null);
if let Some(last_sync) = object.get("lastSync") {
if let Some(synced_at) = last_sync.get("syncedAt").and_then(Value::as_str) {
if let Ok(dt) = DateTime::parse_from_rfc3339(synced_at) {
let age = (Utc::now() - dt.with_timezone(&Utc))
.num_seconds()
.max(0) as u64;
object.insert("lastSyncSecondsAgo".to_string(), json!(age));
}
}
}
}
repositories.push(status);
}
Ok(json!({
"targetLabel": snapshot.target_label,
"mode": snapshot.mode,
"parent": snapshot.parent,
"repositoryCount": snapshot.repository_count,
"runningWatchers": snapshot.running_watchers,
"anyRunning": snapshot.any_running,
"allRunning": snapshot.all_running,
"parentWatcherRunning": snapshot.parent_watcher_running,
"parentWatcherPid": snapshot.parent_watcher_pid,
"totalUnsyncedFiles": snapshot.total_unsynced_files,
"totalUnsyncedRecords": snapshot.total_unsynced_records,
"unsyncedFiles": snapshot.total_unsynced_files,
"unsyncedRecords": snapshot.total_unsynced_records,
"statsLine": snapshot.stats_line,
"lastError": last_error,
"repositories": repositories,
}))
}
fn identity_from_tray_repo(repo: &WorktreeWatchTrayRepoStatus) -> RepoIdentity {
RepoIdentity {
owner: repo.owner.clone(),
name: repo.name.clone(),
branch: repo.branch.clone(),
root: parent_root_as_local_path(&repo.root),
head_sha: None,
}
}
fn worktree_watch_status_from_tray_repo(
repo: &WorktreeWatchTrayRepoStatus,
) -> Result<Value, String> {
let identity = identity_from_tray_repo(repo);
let layout = prepare_spool_layout(&identity)?;
let mut payload = json!({
"repoRoot": repo.root,
"repositoryOwner": repo.owner,
"repositoryName": repo.name,
"branchName": repo.branch,
"platform": std::env::consts::OS,
"runtimeKey": worktree_runtime_key(),
"spoolRoot": layout.root.display().to_string(),
"unsyncedFiles": repo.unsynced_files,
"unsyncedRecords": repo.unsynced_records,
"syncedFiles": 0,
"syncedRecords": 0,
"localSpoolFiles": repo.unsynced_files,
"localSpoolRecords": repo.unsynced_records,
"running": repo.running,
"pid": repo.pid,
});
if let Some(last_sync) = read_last_sync_log_entry(&layout.sync_log_file)? {
if let Some(object) = payload.as_object_mut() {
object.insert(
"lastSync".to_string(),
serde_json::to_value(last_sync)
.map_err(|error| format!("Failed to render sync log payload: {error}"))?,
);
}
}
Ok(payload)
}
pub fn collect_worktree_watch_api_repositories(
target: &WorktreeWatchTargetOptions,
) -> Result<Value, String> {
let status = collect_worktree_watch_api_status(target, None)?;
let repositories = status
.get("repositories")
.cloned()
.unwrap_or_else(|| json!([]));
Ok(json!({ "repositories": repositories }))
}
pub fn collect_worktree_watch_api_stats(
target: &WorktreeWatchTargetOptions,
session_gap_minutes: u64,
) -> Result<Value, String> {
let identities = resolve_target_identities(target)?;
if identities.is_empty() {
return Ok(json!({
"generatedAt": null,
"message": "no stats yet",
"repositories": [],
}));
}
let gap = if session_gap_minutes == 0 {
WORKTREE_WATCH_API_DEFAULT_STATS_GAP_MINUTES
} else {
session_gap_minutes
};
let mut repositories = Vec::new();
for identity in &identities {
let stats = read_or_generate_stats(identity, gap, Duration::from_secs(120))?;
repositories.push(json!({
"owner": identity.owner,
"name": identity.name,
"branch": identity.branch,
"repositoryOwner": identity.owner,
"repositoryName": identity.name,
"branchName": identity.branch,
"repoRoot": normalize_repo_root_for_payload(&identity.root),
"root": normalize_repo_root_for_payload(&identity.root),
"stats": stats,
}));
}
if repositories.len() == 1 {
let first = repositories.remove(0);
let stats = first
.get("stats")
.cloned()
.unwrap_or(Value::Null);
if let Some(mut object) = stats.as_object().cloned() {
object.insert(
"repositories".to_string(),
json!([{
"owner": first.get("owner"),
"name": first.get("name"),
"branch": first.get("branch"),
"stats": first.get("stats"),
}]),
);
return Ok(Value::Object(object));
}
return Ok(first);
}
Ok(json!({
"repositories": repositories,
"repositoryCount": repositories.len(),
}))
}
fn worktree_watch_status_payload_light(identity: &RepoIdentity) -> Result<Value, String> {
let layout = prepare_spool_layout(identity)?;
let counts = count_spool_summary(&layout)?;
let mut payload = json!({
"repoRoot": normalize_repo_root_for_payload(&identity.root),
"repositoryOwner": identity.owner.clone(),
"repositoryName": identity.name.clone(),
"branchName": identity.branch.clone(),
"platform": std::env::consts::OS,
"runtimeKey": worktree_runtime_key(),
"spoolRoot": layout.root.display().to_string(),
"unsyncedFiles": counts.unsynced_files,
"unsyncedRecords": counts.unsynced_records,
"syncedFiles": counts.synced_files,
"syncedRecords": counts.synced_records,
"localSpoolFiles": counts.unsynced_files + counts.synced_files,
"localSpoolRecords": counts.unsynced_records + counts.synced_records,
});
if let Some(last_sync) = read_last_sync_log_entry(&layout.sync_log_file)? {
if let Some(object) = payload.as_object_mut() {
object.insert(
"lastSync".to_string(),
serde_json::to_value(last_sync)
.map_err(|error| format!("Failed to render sync log payload: {error}"))?,
);
}
}
Ok(payload)
}
fn worktree_watch_status_payload(
identity: &RepoIdentity,
include_records: bool,
record_limit: usize,
) -> Result<Value, String> {
if !include_records {
return worktree_watch_status_payload_light(identity);
}
let layout = prepare_spool_layout(identity)?;
let candidates = collect_sync_candidates(&layout)?;
let all_candidates = collect_spool_candidates(&layout, true)?;
let record_count: usize = candidates
.iter()
.map(|candidate| candidate.records.len())
.sum();
let all_record_count: usize = all_candidates
.iter()
.map(|candidate| candidate.records.len())
.sum();
let synced_file_count = all_candidates.len().saturating_sub(candidates.len());
let synced_record_count = all_record_count.saturating_sub(record_count);
let mut payload = json!({
"repoRoot": normalize_repo_root_for_payload(&identity.root),
"repositoryOwner": identity.owner.clone(),
"repositoryName": identity.name.clone(),
"branchName": identity.branch.clone(),
"platform": std::env::consts::OS,
"runtimeKey": worktree_runtime_key(),
"spoolRoot": layout.root.display().to_string(),
"unsyncedFiles": candidates.len(),
"unsyncedRecords": record_count,
"syncedFiles": synced_file_count,
"syncedRecords": synced_record_count,
"localSpoolFiles": all_candidates.len(),
"localSpoolRecords": all_record_count,
});
if let Some(last_sync) = read_last_sync_log_entry(&layout.sync_log_file)? {
if let Some(object) = payload.as_object_mut() {
object.insert(
"lastSync".to_string(),
serde_json::to_value(last_sync)
.map_err(|error| format!("Failed to render sync log payload: {error}"))?,
);
}
}
if let Some(object) = payload.as_object_mut() {
object.insert(
"records".to_string(),
collect_status_records(&candidates, record_limit)?,
);
}
Ok(payload)
}
#[derive(Debug, Default, Clone, Copy)]
struct SpoolCountSummary {
unsynced_files: u64,
unsynced_records: u64,
synced_files: u64,
synced_records: u64,
}
fn count_spool_summary(layout: &SpoolLayout) -> Result<SpoolCountSummary, String> {
let mut roots = vec![layout.root.clone()];
if let Some(branch) = layout.root.file_name().and_then(|name| name.to_str()) {
if let Some(repo_root) = layout.root.parent() {
if let (Some(name), Some(owner)) = (
repo_root.file_name().and_then(|n| n.to_str()),
repo_root
.parent()
.and_then(|p| p.file_name())
.and_then(|n| n.to_str()),
) {
let identity = RepoIdentity {
owner: owner.to_string(),
name: name.to_string(),
branch: branch.to_string(),
root: PathBuf::new(),
head_sha: None,
};
if let Ok(search_roots) = repository_spool_search_roots(&identity) {
for search_root in search_roots {
let branch_root = search_root.join(branch);
if !roots.iter().any(|existing| {
path_identity_key(existing) == path_identity_key(&branch_root)
}) {
roots.push(branch_root);
}
}
}
}
}
}
let mut summary = SpoolCountSummary::default();
let mut seen_files = BTreeSet::new();
for root in roots {
if !root.exists() {
continue;
}
for entry in fs::read_dir(&root).map_err(|error| {
format!(
"Failed to read spool directory {}: {error}",
root.display()
)
})? {
let path = entry
.map_err(|error| format!("Failed to read spool entry: {error}"))?
.path();
let Some(file_name) = path.file_name().and_then(|value| value.to_str()) else {
continue;
};
if !file_name.ends_with(".jsonl") {
continue;
}
if !(file_name.starts_with("events-") || file_name.starts_with("commits-")) {
continue;
}
let identity_key = path_identity_key(&path);
if !seen_files.insert(identity_key) {
continue;
}
let lines = count_nonempty_lines(&path)?;
if file_name.contains(".synced.") {
summary.synced_files += 1;
summary.synced_records += lines;
} else {
summary.unsynced_files += 1;
summary.unsynced_records += lines;
}
}
}
Ok(summary)
}
fn count_nonempty_lines(path: &Path) -> Result<u64, String> {
let file = File::open(path)
.map_err(|error| format!("Failed to open spool file {}: {error}", path.display()))?;
let reader = BufReader::new(file);
let mut count = 0u64;
for line in reader.lines() {
let line =
line.map_err(|error| format!("Failed to read spool file {}: {error}", path.display()))?;
if !line.trim().is_empty() {
count += 1;
}
}
Ok(count)
}
fn read_or_generate_stats(
identity: &RepoIdentity,
session_gap_minutes: u64,
max_age: Duration,
) -> Result<WorktreeWatchStatsSummary, String> {
let layout = prepare_spool_layout(identity)?;
if layout.stats_file.exists() {
if let Ok(raw) = fs::read_to_string(&layout.stats_file) {
if let Ok(stats) = serde_json::from_str::<WorktreeWatchStatsSummary>(&raw) {
let age = Utc::now()
.signed_duration_since(stats.generated_at)
.to_std()
.unwrap_or(Duration::from_secs(u64::MAX));
if age <= max_age {
return Ok(stats);
}
}
}
}
generate_and_store_stats(identity, session_gap_minutes)
}
fn collect_status_records(candidates: &[SyncCandidate], limit: usize) -> Result<Value, String> {
let available: usize = candidates
.iter()
.map(|candidate| candidate.records.len())
.sum();
let mut shown = 0usize;
let mut events = Vec::new();
let mut commits = Vec::new();
'candidates: for candidate in candidates {
match &candidate.records {
SyncRecords::Events(records) => {
for event in records {
if shown >= limit {
break 'candidates;
}
events.push(json!({
"sourceFile": candidate.path.display().to_string(),
"occurredAt": event.occurred_at,
"eventKind": event.event_kind.clone(),
"primaryPath": event.primary_path.clone(),
"oldPath": event.old_path.clone(),
"newPath": event.new_path.clone(),
"paths": event.paths.clone(),
"addedLines": event.added_lines,
"removedLines": event.removed_lines,
"totalLines": event.total_lines,
"headSha": event.head_sha.clone(),
"rawKind": event.raw_kind.clone(),
}));
shown += 1;
}
}
SyncRecords::Commits(records) => {
for commit in records {
if shown >= limit {
break 'candidates;
}
commits.push(json!({
"sourceFile": candidate.path.display().to_string(),
"occurredAt": commit.occurred_at,
"headSha": commit.head_sha.clone(),
"previousHeadSha": commit.previous_head_sha.clone(),
"subject": commit.subject.clone(),
"authorName": commit.author_name.clone(),
"authorEmail": commit.author_email.clone(),
"committedAt": commit.committed_at.clone(),
}));
shown += 1;
}
}
}
}
Ok(json!({
"available": available,
"shown": shown,
"truncated": shown < available,
"events": events,
"commits": commits,
}))
}
fn print_status_records(payload: &Value) {
let Some(records) = payload.get("records") else {
return;
};
let shown = records.get("shown").and_then(Value::as_u64).unwrap_or(0);
if shown == 0 {
println!("records: none");
return;
}
println!("records:");
if let Some(events) = records.get("events").and_then(Value::as_array) {
if !events.is_empty() {
println!(" mutations:");
for event in events {
print_status_event_record(event);
}
}
}
if let Some(commits) = records.get("commits").and_then(Value::as_array) {
if !commits.is_empty() {
println!(" commits:");
for commit in commits {
print_status_commit_record(commit);
}
}
}
if records
.get("truncated")
.and_then(Value::as_bool)
.unwrap_or(false)
{
let available = records
.get("available")
.and_then(Value::as_u64)
.unwrap_or(shown);
println!(" showing {shown} of {available} record(s)");
}
}
fn print_status_stats(payload: &Value) {
let Some(stats) = payload.get("stats") else {
return;
};
println!("stats:");
println!(
" coding time: {}",
format_duration(
stats
.get("estimatedCodingSeconds")
.and_then(Value::as_u64)
.unwrap_or(0)
)
);
println!(
" sessions: {}",
stats
.get("sessionCount")
.and_then(Value::as_u64)
.unwrap_or(0)
);
println!(
" observed span: {}",
format_duration(
stats
.get("observedSpanSeconds")
.and_then(Value::as_u64)
.unwrap_or(0)
)
);
println!(
" events: {}",
stats
.get("totalEvents")
.and_then(Value::as_u64)
.unwrap_or(0)
);
println!(
" commits: {}",
stats
.get("totalCommits")
.and_then(Value::as_u64)
.unwrap_or(0)
);
println!(
" lines: +{} -{}",
stats.get("addedLines").and_then(Value::as_u64).unwrap_or(0),
stats
.get("removedLines")
.and_then(Value::as_u64)
.unwrap_or(0)
);
if let Some(spool_root) = value_str(stats, "spoolRoot") {
println!(
" stored: {}",
Path::new(spool_root).join("stats.json").display()
);
}
print_stats_bucket_section(stats, "byFiletype", " by filetype:", 10);
print_stats_bucket_section(stats, "byEventKind", " by event kind:", 10);
print_stats_file_section(stats, 10);
}
fn print_repo_activity(payload: &Value) {
let Some(activity) = payload.get("repositoryActivity") else {
return;
};
println!("repo activity:");
println!(
" coding time: {}",
format_duration(
activity
.get("estimatedCodingSeconds")
.and_then(Value::as_u64)
.unwrap_or(0)
)
);
println!(
" observed span: {}",
format_duration(
activity
.get("observedSpanSeconds")
.and_then(Value::as_u64)
.unwrap_or(0)
)
);
println!(
" branches: {}",
activity
.get("branchCount")
.and_then(Value::as_u64)
.unwrap_or(0)
);
println!(
" sessions: {}, events: {}, commits: {}, lines: +{} -{}",
activity
.get("sessionCount")
.and_then(Value::as_u64)
.unwrap_or(0),
activity
.get("totalEvents")
.and_then(Value::as_u64)
.unwrap_or(0),
activity
.get("totalCommits")
.and_then(Value::as_u64)
.unwrap_or(0),
activity
.get("addedLines")
.and_then(Value::as_u64)
.unwrap_or(0),
activity
.get("removedLines")
.and_then(Value::as_u64)
.unwrap_or(0)
);
if let Some(stats_file) = value_str(activity, "statsFile") {
println!(" stored: {stats_file}");
}
let Some(branches) = activity.get("branches").and_then(Value::as_array) else {
return;
};
if branches.is_empty() {
return;
}
println!(" by branch:");
for branch in branches.iter().take(20) {
let name = value_str(branch, "branchName").unwrap_or("unknown");
let coding_seconds = branch
.get("estimatedCodingSeconds")
.and_then(Value::as_u64)
.unwrap_or(0);
let observed_seconds = branch
.get("observedSpanSeconds")
.and_then(Value::as_u64)
.unwrap_or(0);
let sessions = branch
.get("sessionCount")
.and_then(Value::as_u64)
.unwrap_or(0);
let events = branch
.get("totalEvents")
.and_then(Value::as_u64)
.unwrap_or(0);
let commits = branch
.get("totalCommits")
.and_then(Value::as_u64)
.unwrap_or(0);
println!(
" {name}: {} coding, {} span, {} session(s), {} event(s), {} commit(s)",
format_duration(coding_seconds),
format_duration(observed_seconds),
sessions,
events,
commits
);
}
}
#[allow(dead_code)]
fn print_last_sync(payload: &Value) {
let Some(last_sync) = payload.get("lastSync") else {
return;
};
let synced_at = value_str(last_sync, "syncedAt").unwrap_or("unknown-time");
let status = last_sync
.get("statusCode")
.and_then(Value::as_u64)
.unwrap_or(0);
let events = last_sync
.get("eventCount")
.and_then(Value::as_u64)
.unwrap_or(0);
let commits = last_sync
.get("commitCount")
.and_then(Value::as_u64)
.unwrap_or(0);
let files = last_sync
.get("spoolFileCount")
.and_then(Value::as_u64)
.unwrap_or(0);
let resync = last_sync
.get("resync")
.and_then(Value::as_bool)
.unwrap_or(false);
println!(
"last sync: {synced_at}, HTTP {status}, {events} event(s), {commits} commit(s), {files} file(s), resync={resync}"
);
}
fn print_stats_bucket_section(stats: &Value, key: &str, title: &str, limit: usize) {
let Some(items) = stats.get(key).and_then(Value::as_array) else {
return;
};
if items.is_empty() {
return;
}
println!("{title}");
for item in items.iter().take(limit) {
let name = value_str(item, "name").unwrap_or("unknown");
let events = item.get("eventCount").and_then(Value::as_u64).unwrap_or(0);
let seconds = item
.get("estimatedCodingSeconds")
.and_then(Value::as_u64)
.unwrap_or(0);
let added = item.get("addedLines").and_then(Value::as_u64).unwrap_or(0);
let removed = item
.get("removedLines")
.and_then(Value::as_u64)
.unwrap_or(0);
println!(
" {name}: {}, {} event(s), +{} -{}",
format_duration(seconds),
events,
added,
removed
);
}
}
fn print_stats_file_section(stats: &Value, limit: usize) {
let Some(items) = stats.get("byFile").and_then(Value::as_array) else {
return;
};
if items.is_empty() {
return;
}
println!(" by file:");
for item in items.iter().take(limit) {
let path = value_str(item, "path").unwrap_or("(no path)");
let filetype = value_str(item, "filetype").unwrap_or("(none)");
let events = item.get("eventCount").and_then(Value::as_u64).unwrap_or(0);
let seconds = item
.get("estimatedCodingSeconds")
.and_then(Value::as_u64)
.unwrap_or(0);
let added = item.get("addedLines").and_then(Value::as_u64).unwrap_or(0);
let removed = item
.get("removedLines")
.and_then(Value::as_u64)
.unwrap_or(0);
println!(
" {path} [{filetype}]: {}, {} event(s), +{} -{}",
format_duration(seconds),
events,
added,
removed
);
}
}
fn print_status_event_record(event: &Value) {
let occurred_at = value_str(event, "occurredAt").unwrap_or("unknown-time");
let kind = value_str(event, "eventKind").unwrap_or("event");
let primary_path = event_display_path(event);
let added = event.get("addedLines").and_then(Value::as_u64).unwrap_or(0);
let removed = event
.get("removedLines")
.and_then(Value::as_u64)
.unwrap_or(0);
let total_lines = event.get("totalLines").and_then(Value::as_u64);
if let Some(total_lines) = total_lines {
println!(
" {occurred_at} {kind} {primary_path} (+{added} -{removed}, {total_lines} lines)"
);
} else {
println!(" {occurred_at} {kind} {primary_path} (+{added} -{removed})");
}
if let Some(head_sha) = value_str(event, "headSha") {
println!(" head: {}", short_sha(head_sha));
}
if let Some(paths) = event.get("paths").and_then(Value::as_array) {
if paths.len() > 1 {
let rendered = paths
.iter()
.filter_map(Value::as_str)
.collect::<Vec<_>>()
.join(", ");
println!(" paths: {rendered}");
}
}
}
fn print_status_commit_record(commit: &Value) {
let occurred_at = value_str(commit, "occurredAt").unwrap_or("unknown-time");
let head_sha = value_str(commit, "headSha")
.map(short_sha)
.unwrap_or_else(|| "unknown".to_string());
let subject = value_str(commit, "subject").unwrap_or("(no subject)");
println!(" {occurred_at} {head_sha} {subject}");
if let Some(author) = value_str(commit, "authorName") {
println!(" author: {author}");
}
}
fn event_display_path(event: &Value) -> String {
let old_path = value_str(event, "oldPath");
let new_path = value_str(event, "newPath");
if let (Some(old_path), Some(new_path)) = (old_path, new_path) {
return format!("{old_path} -> {new_path}");
}
value_str(event, "primaryPath")
.map(str::to_string)
.or_else(|| {
event
.get("paths")
.and_then(Value::as_array)
.and_then(|paths| paths.first())
.and_then(Value::as_str)
.map(str::to_string)
})
.unwrap_or_else(|| "(no path)".to_string())
}
fn value_str<'a>(value: &'a Value, key: &str) -> Option<&'a str> {
value.get(key).and_then(Value::as_str)
}
fn short_sha(value: &str) -> String {
value.chars().take(12).collect()
}
fn generate_and_store_stats(
identity: &RepoIdentity,
session_gap_minutes: u64,
) -> Result<WorktreeWatchStatsSummary, String> {
let layout = prepare_spool_layout(identity)?;
let candidates = collect_spool_candidates(&layout, true)?;
let summary = build_stats_summary(identity, &layout, &candidates, session_gap_minutes)?;
let rendered = serde_json::to_string_pretty(&summary)
.map_err(|error| format!("Failed to serialize stats: {error}"))?;
fs::write(&layout.stats_file, format!("{rendered}\n")).map_err(|error| {
format!(
"Failed to write stats file {}: {error}",
layout.stats_file.display()
)
})?;
Ok(summary)
}
fn generate_and_store_repo_activity(
identity: &RepoIdentity,
session_gap_minutes: u64,
) -> Result<WorktreeWatchRepoActivitySummary, String> {
let repository_spool_root = repository_spool_root(identity)?;
let summary =
build_repo_activity_summary(identity, &repository_spool_root, session_gap_minutes)?;
let rendered = serde_json::to_string_pretty(&summary)
.map_err(|error| format!("Failed to serialize repo activity: {error}"))?;
fs::create_dir_all(&repository_spool_root).map_err(|error| {
format!(
"Failed to create repository spool root {}: {error}",
repository_spool_root.display()
)
})?;
let stats_file = repository_spool_root.join("repo-activity.json");
fs::write(&stats_file, format!("{rendered}\n")).map_err(|error| {
format!(
"Failed to write repo activity file {}: {error}",
stats_file.display()
)
})?;
Ok(summary)
}
fn build_repo_activity_summary(
identity: &RepoIdentity,
repository_spool_root: &Path,
session_gap_minutes: u64,
) -> Result<WorktreeWatchRepoActivitySummary, String> {
let mut branches = Vec::new();
if repository_spool_root.exists() {
for entry in fs::read_dir(repository_spool_root).map_err(|error| {
format!(
"Failed to read repository spool root {}: {error}",
repository_spool_root.display()
)
})? {
let entry =
entry.map_err(|error| format!("Failed to read repository spool entry: {error}"))?;
let file_type = entry.file_type().map_err(|error| {
format!("Failed to inspect {}: {error}", entry.path().display())
})?;
if !file_type.is_dir() {
continue;
}
let branch_name = entry.file_name().to_string_lossy().to_string();
let branch_root = entry.path();
let branch_layout = spool_layout_for_existing_root(&branch_root);
let branch_identity = RepoIdentity {
branch: branch_name.clone(),
..identity.clone()
};
let candidates = collect_spool_candidates(&branch_layout, true)?;
let stats = build_stats_summary(
&branch_identity,
&branch_layout,
&candidates,
session_gap_minutes,
)?;
if stats.total_events == 0 && stats.total_commits == 0 {
continue;
}
branches.push(WorktreeWatchBranchActivityStats {
branch_name,
spool_root: branch_root.display().to_string(),
total_events: stats.total_events,
total_commits: stats.total_commits,
first_event_at: stats.first_event_at,
last_event_at: stats.last_event_at,
session_count: stats.session_count,
observed_span_seconds: stats.observed_span_seconds,
estimated_coding_seconds: stats.estimated_coding_seconds,
added_lines: stats.added_lines,
removed_lines: stats.removed_lines,
});
}
}
branches.sort_by(|left, right| {
right
.estimated_coding_seconds
.cmp(&left.estimated_coding_seconds)
.then_with(|| right.total_events.cmp(&left.total_events))
.then_with(|| left.branch_name.cmp(&right.branch_name))
});
let first_event_at = branches
.iter()
.filter_map(|branch| branch.first_event_at)
.min();
let last_event_at = branches
.iter()
.filter_map(|branch| branch.last_event_at)
.max();
let observed_span_seconds = observed_span_seconds(first_event_at, last_event_at);
let stats_file = repository_spool_root.join("repo-activity.json");
Ok(WorktreeWatchRepoActivitySummary {
generated_at: Utc::now(),
repository_owner: identity.owner.clone(),
repository_name: identity.name.clone(),
repo_root: identity.root.display().to_string(),
repository_spool_root: repository_spool_root.display().to_string(),
stats_file: stats_file.display().to_string(),
session_gap_minutes,
branch_count: branches.len() as u64,
total_events: branches.iter().map(|branch| branch.total_events).sum(),
total_commits: branches.iter().map(|branch| branch.total_commits).sum(),
first_event_at,
last_event_at,
session_count: branches.iter().map(|branch| branch.session_count).sum(),
observed_span_seconds,
estimated_coding_seconds: branches
.iter()
.map(|branch| branch.estimated_coding_seconds)
.sum(),
added_lines: branches.iter().map(|branch| branch.added_lines).sum(),
removed_lines: branches.iter().map(|branch| branch.removed_lines).sum(),
branches,
})
}
fn build_stats_summary(
identity: &RepoIdentity,
layout: &SpoolLayout,
candidates: &[SyncCandidate],
session_gap_minutes: u64,
) -> Result<WorktreeWatchStatsSummary, String> {
let mut events = Vec::new();
let mut total_commits = 0u64;
for candidate in candidates {
match &candidate.records {
SyncRecords::Events(records) => events.extend(records.iter().cloned()),
SyncRecords::Commits(records) => total_commits += records.len() as u64,
}
}
events.sort_by_key(|event| event.occurred_at);
let gap_seconds = session_gap_minutes.saturating_mul(60).max(60);
let mut previous_at: Option<DateTime<Utc>> = None;
let mut session_count = 0u64;
let mut total_seconds = 0u64;
let mut total_added = 0u64;
let mut total_removed = 0u64;
let mut by_file: BTreeMap<String, WorktreeWatchFileStats> = BTreeMap::new();
let mut by_filetype: BTreeMap<String, WorktreeWatchBucketStats> = BTreeMap::new();
let mut by_event_kind: BTreeMap<String, WorktreeWatchBucketStats> = BTreeMap::new();
for event in &events {
if starts_new_coding_session(previous_at, event.occurred_at, gap_seconds) {
session_count += 1;
}
let seconds = coding_seconds_for_event(previous_at, event.occurred_at, gap_seconds);
previous_at = Some(event.occurred_at);
let added = event.added_lines.unwrap_or(0);
let removed = event.removed_lines.unwrap_or(0);
let path = stats_event_path(event);
let filetype = filetype_for_path(&path);
total_seconds += seconds;
total_added += added;
total_removed += removed;
update_file_stats(&mut by_file, &path, &filetype, seconds, added, removed);
update_bucket_stats(&mut by_filetype, &filetype, seconds, added, removed);
update_bucket_stats(
&mut by_event_kind,
&event.event_kind,
seconds,
added,
removed,
);
}
Ok(WorktreeWatchStatsSummary {
generated_at: Utc::now(),
repository_owner: identity.owner.clone(),
repository_name: identity.name.clone(),
branch_name: identity.branch.clone(),
repo_root: identity.root.display().to_string(),
spool_root: layout.root.display().to_string(),
session_gap_minutes,
total_events: events.len() as u64,
total_commits,
first_event_at: events.first().map(|event| event.occurred_at),
last_event_at: events.last().map(|event| event.occurred_at),
session_count,
observed_span_seconds: observed_span_seconds(
events.first().map(|event| event.occurred_at),
events.last().map(|event| event.occurred_at),
),
estimated_coding_seconds: total_seconds,
added_lines: total_added,
removed_lines: total_removed,
by_file: sorted_file_stats(by_file),
by_filetype: sorted_bucket_stats(by_filetype),
by_event_kind: sorted_bucket_stats(by_event_kind),
})
}
fn coding_seconds_for_event(
previous_at: Option<DateTime<Utc>>,
occurred_at: DateTime<Utc>,
gap_seconds: u64,
) -> u64 {
let Some(previous_at) = previous_at else {
return 60;
};
let delta = occurred_at.signed_duration_since(previous_at).num_seconds();
if delta <= 0 {
0
} else {
(delta as u64).min(gap_seconds)
}
}
fn starts_new_coding_session(
previous_at: Option<DateTime<Utc>>,
occurred_at: DateTime<Utc>,
gap_seconds: u64,
) -> bool {
let Some(previous_at) = previous_at else {
return true;
};
let delta = occurred_at.signed_duration_since(previous_at).num_seconds();
delta > gap_seconds as i64
}
fn observed_span_seconds(
first_event_at: Option<DateTime<Utc>>,
last_event_at: Option<DateTime<Utc>>,
) -> u64 {
let (Some(first_event_at), Some(last_event_at)) = (first_event_at, last_event_at) else {
return 0;
};
last_event_at
.signed_duration_since(first_event_at)
.num_seconds()
.max(0) as u64
}
fn update_file_stats(
by_file: &mut BTreeMap<String, WorktreeWatchFileStats>,
path: &str,
filetype: &str,
seconds: u64,
added: u64,
removed: u64,
) {
let stats = by_file
.entry(path.to_string())
.or_insert_with(|| WorktreeWatchFileStats {
path: path.to_string(),
filetype: filetype.to_string(),
..WorktreeWatchFileStats::default()
});
stats.event_count += 1;
stats.estimated_coding_seconds += seconds;
stats.added_lines += added;
stats.removed_lines += removed;
}
fn update_bucket_stats(
buckets: &mut BTreeMap<String, WorktreeWatchBucketStats>,
name: &str,
seconds: u64,
added: u64,
removed: u64,
) {
let stats = buckets
.entry(name.to_string())
.or_insert_with(|| WorktreeWatchBucketStats {
name: name.to_string(),
..WorktreeWatchBucketStats::default()
});
stats.event_count += 1;
stats.estimated_coding_seconds += seconds;
stats.added_lines += added;
stats.removed_lines += removed;
}
fn sorted_file_stats(
by_file: BTreeMap<String, WorktreeWatchFileStats>,
) -> Vec<WorktreeWatchFileStats> {
let mut stats = by_file.into_values().collect::<Vec<_>>();
stats.sort_by(|left, right| {
right
.estimated_coding_seconds
.cmp(&left.estimated_coding_seconds)
.then_with(|| right.event_count.cmp(&left.event_count))
.then_with(|| left.path.cmp(&right.path))
});
stats
}
fn sorted_bucket_stats(
buckets: BTreeMap<String, WorktreeWatchBucketStats>,
) -> Vec<WorktreeWatchBucketStats> {
let mut stats = buckets.into_values().collect::<Vec<_>>();
stats.sort_by(|left, right| {
right
.estimated_coding_seconds
.cmp(&left.estimated_coding_seconds)
.then_with(|| right.event_count.cmp(&left.event_count))
.then_with(|| left.name.cmp(&right.name))
});
stats
}
fn stats_event_path(event: &WorktreeMutationEvent) -> String {
event
.new_path
.as_deref()
.or(event.primary_path.as_deref())
.or_else(|| event.paths.first().map(String::as_str))
.unwrap_or("(no path)")
.to_string()
}
fn filetype_for_path(path: &str) -> String {
Path::new(path)
.extension()
.and_then(|value| value.to_str())
.map(|value| value.to_ascii_lowercase())
.filter(|value| !value.is_empty())
.unwrap_or_else(|| "(none)".to_string())
}
fn format_duration(seconds: u64) -> String {
let hours = seconds / 3600;
let minutes = (seconds % 3600) / 60;
let remaining_seconds = seconds % 60;
if hours > 0 {
format!("{hours}h {minutes}m")
} else if minutes > 0 {
format!("{minutes}m {remaining_seconds}s")
} else {
format!("{remaining_seconds}s")
}
}
async fn watch_foreground(
identity: RepoIdentity,
layout: SpoolLayout,
sync_interval_seconds: u64,
) -> Result<(), String> {
if let Ok(executable) = std::env::current_exe() {
let _ = write_watcher_state(&identity, &layout, std::process::id(), &executable);
}
let api_target = WorktreeWatchTargetOptions {
repo: Some(identity.root.clone()),
parent: None,
repos: Vec::new(),
};
if let Some(port) = crate::commands::worktree_watch_api::spawn_worktree_watch_api_in_background(
api_target,
sync_interval_seconds,
None,
) {
if !is_background_child() {
crate::commands::worktree_watch_api::print_api_listen_hint(port);
} else {
tracing::info!(
"worktree-watch local API listening on http://127.0.0.1:{port}"
);
}
}
let (tx, rx) = mpsc::channel();
let mut watcher = create_path_watcher(&identity.root, tx)?;
watcher
.watch(&identity.root, RecursiveMode::Recursive)
.map_err(|error| {
format!(
"Failed to watch repository root {}: {error}",
identity.root.display()
)
})?;
log_watcher_backend(&identity.root, "repository");
let mut last_head = identity.head_sha.clone();
let mut last_commit_check = Instant::now();
let mut last_sync = Instant::now();
let mut event_dedupe = RecentEventDedupe {
fingerprints: load_event_dedupe_fingerprints(&layout.event_dedupe_file)?,
};
loop {
match rx.recv_timeout(Duration::from_millis(750)) {
Ok(Ok(event)) => {
if let Some(mutation) = mutation_event_from_notify(&identity, event) {
append_unique_mutation_event(&layout, &mutation, &mut event_dedupe)?;
}
}
Ok(Err(error)) => {
eprintln!("worktree watcher error: {error}");
}
Err(mpsc::RecvTimeoutError::Timeout) => {}
Err(mpsc::RecvTimeoutError::Disconnected) => {
return Err("Filesystem watcher disconnected.".to_string());
}
}
if last_commit_check.elapsed() >= Duration::from_secs(3) {
last_head = record_commit_snapshot(&identity, &layout, last_head.as_deref())?;
last_commit_check = Instant::now();
}
if sync_interval_seconds > 0
&& last_sync.elapsed() >= Duration::from_secs(sync_interval_seconds)
{
if let Err(error) = sync_spool_for_identity(&identity, &layout).await {
eprintln!("worktree mutation sync failed: {error}");
}
last_sync = Instant::now();
}
}
}
async fn watch_parent_foreground(
parent_root: PathBuf,
identities: Vec<RepoIdentity>,
sync_interval_seconds: u64,
) -> Result<(), String> {
if let Ok(layout) = prepare_parent_spool_layout(&parent_root) {
if let Ok(executable) = std::env::current_exe() {
let _ = write_parent_watcher_state(
&parent_root,
&layout,
std::process::id(),
&executable,
);
}
}
let api_target = WorktreeWatchTargetOptions {
repo: None,
parent: Some(parent_root.clone()),
repos: Vec::new(),
};
if let Some(port) = crate::commands::worktree_watch_api::spawn_worktree_watch_api_in_background(
api_target,
sync_interval_seconds,
None,
) {
if !is_background_child() {
crate::commands::worktree_watch_api::print_api_listen_hint(port);
} else {
tracing::info!(
"worktree-watch local API listening on http://127.0.0.1:{port}"
);
}
}
let (tx, rx) = mpsc::channel();
let mut watcher = create_path_watcher(&parent_root, tx)?;
watcher
.watch(&parent_root, RecursiveMode::Recursive)
.map_err(|error| {
format!(
"Failed to watch parent folder {}: {error}",
parent_root.display()
)
})?;
log_watcher_backend(&parent_root, "parent");
let mut repos = prepare_watched_repos(identities)?;
let mut last_commit_check = Instant::now();
let mut last_sync = Instant::now();
loop {
match rx.recv_timeout(Duration::from_millis(750)) {
Ok(Ok(event)) => {
for (repo_index, routed_event) in route_parent_event(&repos, &event) {
let repo = &mut repos[repo_index];
if let Some(mutation) = mutation_event_from_notify(&repo.identity, routed_event)
{
append_unique_mutation_event(
&repo.layout,
&mutation,
&mut repo.event_dedupe,
)?;
}
}
}
Ok(Err(error)) => {
eprintln!("parent worktree watcher error: {error}");
}
Err(mpsc::RecvTimeoutError::Timeout) => {}
Err(mpsc::RecvTimeoutError::Disconnected) => {
return Err("Parent filesystem watcher disconnected.".to_string());
}
}
if last_commit_check.elapsed() >= Duration::from_secs(3) {
for repo in &mut repos {
repo.last_head = record_commit_snapshot(
&repo.identity,
&repo.layout,
repo.last_head.as_deref(),
)?;
}
last_commit_check = Instant::now();
}
if sync_interval_seconds > 0
&& last_sync.elapsed() >= Duration::from_secs(sync_interval_seconds)
{
for repo in &repos {
if let Err(error) = sync_spool_for_identity(&repo.identity, &repo.layout).await {
eprintln!(
"worktree mutation sync failed for {}: {error}",
repo.identity.root.display()
);
}
}
last_sync = Instant::now();
}
}
}
enum PathWatcher {
Recommended(RecommendedWatcher),
Poll(notify::PollWatcher),
}
impl PathWatcher {
fn watch(&mut self, path: &Path, mode: RecursiveMode) -> notify::Result<()> {
match self {
Self::Recommended(watcher) => watcher.watch(path, mode),
Self::Poll(watcher) => watcher.watch(path, mode),
}
}
}
fn create_path_watcher(
path: &Path,
tx: mpsc::Sender<notify::Result<Event>>,
) -> Result<PathWatcher, String> {
let config = Config::default();
if should_use_poll_watcher(path) {
let poll_config = config.with_poll_interval(Duration::from_secs(2));
notify::PollWatcher::new(
move |result| {
let _ = tx.send(result);
},
poll_config,
)
.map(PathWatcher::Poll)
.map_err(|error| format!("Failed to create poll filesystem watcher: {error}"))
} else {
RecommendedWatcher::new(
move |result| {
let _ = tx.send(result);
},
config,
)
.map(PathWatcher::Recommended)
.map_err(|error| format!("Failed to create filesystem watcher: {error}"))
}
}
fn should_use_poll_watcher(path: &Path) -> bool {
if !is_wsl_runtime() {
return false;
}
let key = path_identity_key(path).replace('\\', "/");
key.starts_with("/mnt/") || key.contains("/mnt/")
}
fn log_watcher_backend(path: &Path, label: &str) {
if should_use_poll_watcher(path) {
eprintln!(
"worktree-watch: using poll watcher for {label} {} (WSL mount; inotify is unreliable)",
path.display()
);
}
}
fn prepare_watched_repos(identities: Vec<RepoIdentity>) -> Result<Vec<WatchedRepo>, String> {
let mut repos = Vec::with_capacity(identities.len());
for identity in identities {
let layout = prepare_spool_layout(&identity)?;
let event_dedupe = RecentEventDedupe {
fingerprints: load_event_dedupe_fingerprints(&layout.event_dedupe_file)?,
};
repos.push(WatchedRepo {
last_head: identity.head_sha.clone(),
identity,
layout,
event_dedupe,
});
}
repos.sort_by_key(|repo| std::cmp::Reverse(repo.identity.root.components().count()));
Ok(repos)
}
fn route_parent_event(repos: &[WatchedRepo], event: &Event) -> Vec<(usize, Event)> {
let mut paths_by_repo: BTreeMap<usize, Vec<PathBuf>> = BTreeMap::new();
for path in &event.paths {
for (index, repo) in repos.iter().enumerate() {
if path_belongs_to_repo(path, &repo.identity.root) {
let paths = paths_by_repo.entry(index).or_default();
if !paths.iter().any(|existing| existing == path) {
paths.push(path.clone());
}
break;
}
}
}
let mut routed = Vec::new();
for (index, paths) in paths_by_repo {
let mut repo_event = event.clone();
repo_event.paths = paths;
routed.push((index, repo_event));
}
routed
}
fn path_belongs_to_repo(path: &Path, repo_root: &Path) -> bool {
let path_key = path_identity_key(path).replace('\\', "/");
let root_key = path_identity_key(repo_root).replace('\\', "/");
path_key == root_key || path_key.starts_with(&format!("{root_key}/"))
}
fn mutation_event_from_notify(
identity: &RepoIdentity,
event: Event,
) -> Option<WorktreeMutationEvent> {
if event.kind.is_access() || event.kind.is_other() {
return None;
}
let paths: Vec<String> = event
.paths
.iter()
.filter(|path| !is_ignored_path(&identity.root, path))
.filter_map(|path| relative_slash_path(&identity.root, path))
.fold(Vec::new(), |mut paths, path| {
if !paths.iter().any(|existing| existing == &path) {
paths.push(path);
}
paths
});
if paths.is_empty() {
return None;
}
let primary_path = paths.first().cloned();
let (old_path, new_path) = rename_paths(&event.kind, &paths);
let line_counts = primary_path
.as_deref()
.and_then(|path| git_numstat_for_path(&identity.root, path).ok());
let total_lines = primary_path
.as_deref()
.and_then(|path| file_line_count_for_path(&identity.root, path).ok());
let event_kind = classify_event_kind(&event.kind);
let is_dir = event
.paths
.first()
.and_then(|path| fs::metadata(path).ok())
.map(|metadata| metadata.is_dir())
.unwrap_or(false);
let occurred_at = Utc::now();
let mut mutation = WorktreeMutationEvent {
id: String::new(),
fingerprint: None,
repo_owner: identity.owner.clone(),
repo_name: identity.name.clone(),
branch_name: identity.branch.clone(),
repo_root: normalize_repo_root_for_payload(&identity.root),
head_sha: current_head(&identity.root).ok().flatten(),
event_kind: event_kind.to_string(),
paths,
primary_path,
old_path,
new_path,
added_lines: line_counts.map(|counts| counts.0),
removed_lines: line_counts.map(|counts| counts.1),
total_lines,
file_created: matches!(
event.kind,
EventKind::Create(CreateKind::File | CreateKind::Any)
) && !is_dir,
file_removed: matches!(
event.kind,
EventKind::Remove(RemoveKind::File | RemoveKind::Any)
) && !is_dir,
folder_created: matches!(
event.kind,
EventKind::Create(CreateKind::Folder | CreateKind::Any)
) && is_dir,
folder_removed: matches!(
event.kind,
EventKind::Remove(RemoveKind::Folder | RemoveKind::Any)
) && is_dir,
renamed_or_moved: is_rename_or_move(&event.kind),
raw_kind: format!("{:?}", event.kind),
occurred_at,
};
let fingerprint = mutation_event_fingerprint(&mutation);
mutation.fingerprint = Some(fingerprint.clone());
mutation.id = stable_event_id(&fingerprint);
Some(mutation)
}
fn append_unique_mutation_event(
layout: &SpoolLayout,
mutation: &WorktreeMutationEvent,
dedupe: &mut RecentEventDedupe,
) -> Result<bool, String> {
let fingerprint = mutation
.fingerprint
.clone()
.unwrap_or_else(|| mutation_event_fingerprint(mutation));
if dedupe.fingerprints.contains(&fingerprint)
|| event_dedupe_file_contains(&layout.event_dedupe_file, &fingerprint)?
{
dedupe.fingerprints.insert(fingerprint);
return Ok(false);
}
if !claim_event_fingerprint(layout, &fingerprint)? {
dedupe.fingerprints.insert(fingerprint);
return Ok(false);
}
let mut mutation = mutation.clone();
mutation.fingerprint = Some(fingerprint.clone());
if mutation.id.is_empty() {
mutation.id = stable_event_id(&fingerprint);
}
if let Err(error) = append_json_line(&layout.events_file, &mutation) {
release_event_fingerprint_claim(layout, &fingerprint);
return Err(error);
}
append_json_line(
&layout.event_dedupe_file,
&WorktreeWatchEventDedupeRecord {
fingerprint: fingerprint.clone(),
first_seen_at: Utc::now(),
event_kind: mutation.event_kind.clone(),
primary_path: mutation.primary_path.clone(),
},
)?;
dedupe.fingerprints.insert(fingerprint);
Ok(true)
}
fn claim_event_fingerprint(layout: &SpoolLayout, fingerprint: &str) -> Result<bool, String> {
fs::create_dir_all(&layout.event_dedupe_dir).map_err(|error| {
format!(
"Failed to create event dedupe directory {}: {error}",
layout.event_dedupe_dir.display()
)
})?;
let marker = event_dedupe_marker_path(layout, fingerprint);
match OpenOptions::new()
.write(true)
.create_new(true)
.open(&marker)
{
Ok(mut file) => {
writeln!(file, "{}", Utc::now().to_rfc3339()).map_err(|error| {
format!(
"Failed to write event dedupe marker {}: {error}",
marker.display()
)
})?;
Ok(true)
}
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => Ok(false),
Err(error) => Err(format!(
"Failed to create event dedupe marker {}: {error}",
marker.display()
)),
}
}
fn release_event_fingerprint_claim(layout: &SpoolLayout, fingerprint: &str) {
let _ = fs::remove_file(event_dedupe_marker_path(layout, fingerprint));
}
fn event_dedupe_marker_path(layout: &SpoolLayout, fingerprint: &str) -> PathBuf {
layout.event_dedupe_dir.join(format!("{fingerprint}.seen"))
}
fn mutation_event_fingerprint(mutation: &WorktreeMutationEvent) -> String {
let bucket = mutation.occurred_at.timestamp() / EVENT_DEDUPE_WINDOW_SECONDS;
let payload = json!({
"schema": 3,
"repoOwner": mutation.repo_owner,
"repoName": mutation.repo_name,
"branchName": mutation.branch_name,
"headSha": mutation.head_sha,
"eventKind": mutation.event_kind,
"paths": normalize_path_list_for_fingerprint(&mutation.paths),
"primaryPath": mutation.primary_path.as_deref().map(normalize_relative_watch_path),
"oldPath": mutation.old_path.as_deref().map(normalize_relative_watch_path),
"newPath": mutation.new_path.as_deref().map(normalize_relative_watch_path),
"addedLines": mutation.added_lines,
"removedLines": mutation.removed_lines,
"fileCreated": mutation.file_created,
"fileRemoved": mutation.file_removed,
"folderCreated": mutation.folder_created,
"folderRemoved": mutation.folder_removed,
"renamedOrMoved": mutation.renamed_or_moved,
"timeBucket": bucket,
});
let encoded = serde_json::to_vec(&payload).unwrap_or_default();
let mut hasher = Sha256::new();
hasher.update(encoded);
format!("{:x}", hasher.finalize())
}
fn normalize_path_list_for_fingerprint(paths: &[String]) -> Vec<String> {
let mut normalized = paths
.iter()
.map(|path| normalize_relative_watch_path(path))
.filter(|path| !path.is_empty())
.collect::<Vec<_>>();
normalized.sort();
normalized.dedup();
normalized
}
fn stable_event_id(fingerprint: &str) -> String {
format!("wtmut_{fingerprint}")
}
fn normalize_repo_root_for_payload(root: &Path) -> String {
path_identity_key(root).replace('\\', "/")
}
fn load_event_dedupe_fingerprints(path: &Path) -> Result<BTreeSet<String>, String> {
let mut fingerprints = BTreeSet::new();
if !path.exists() {
return Ok(fingerprints);
}
for value in read_jsonl_values(path)? {
if let Some(fingerprint) = value.get("fingerprint").and_then(Value::as_str) {
fingerprints.insert(fingerprint.to_string());
}
}
Ok(fingerprints)
}
fn event_dedupe_file_contains(path: &Path, fingerprint: &str) -> Result<bool, String> {
if !path.exists() {
return Ok(false);
}
for value in read_jsonl_values(path)? {
if value
.get("fingerprint")
.and_then(Value::as_str)
.map(|value| value == fingerprint)
.unwrap_or(false)
{
return Ok(true);
}
}
Ok(false)
}
fn classify_event_kind(kind: &EventKind) -> &'static str {
match kind {
EventKind::Create(CreateKind::File) => "file_create",
EventKind::Create(CreateKind::Folder) => "folder_create",
EventKind::Create(_) => "create",
EventKind::Remove(RemoveKind::File) => "file_remove",
EventKind::Remove(RemoveKind::Folder) => "folder_remove",
EventKind::Remove(_) => "remove",
EventKind::Modify(ModifyKind::Name(
RenameMode::From | RenameMode::To | RenameMode::Both,
)) => "rename_or_move",
EventKind::Modify(ModifyKind::Data(_)) => "file_modify",
EventKind::Modify(ModifyKind::Metadata(_)) => "metadata_modify",
EventKind::Modify(_) => "modify",
_ => "other",
}
}
fn is_rename_or_move(kind: &EventKind) -> bool {
matches!(
kind,
EventKind::Modify(ModifyKind::Name(
RenameMode::From | RenameMode::To | RenameMode::Both
))
)
}
fn rename_paths(kind: &EventKind, paths: &[String]) -> (Option<String>, Option<String>) {
if !is_rename_or_move(kind) {
return (None, None);
}
match paths {
[old_path, new_path, ..] => (Some(old_path.clone()), Some(new_path.clone())),
[path] => match kind {
EventKind::Modify(ModifyKind::Name(RenameMode::From)) => (Some(path.clone()), None),
EventKind::Modify(ModifyKind::Name(RenameMode::To)) => (None, Some(path.clone())),
_ => (None, Some(path.clone())),
},
_ => (None, None),
}
}
fn record_commit_snapshot(
identity: &RepoIdentity,
layout: &SpoolLayout,
previous_head: Option<&str>,
) -> Result<Option<String>, String> {
let head = current_head(&identity.root)?;
let Some(head_sha) = head else {
return Ok(None);
};
if previous_head == Some(head_sha.as_str()) {
return Ok(Some(head_sha));
}
let mut commit = WorktreeCommitEvent {
id: String::new(),
fingerprint: None,
repo_owner: identity.owner.clone(),
repo_name: identity.name.clone(),
branch_name: identity.branch.clone(),
repo_root: normalize_repo_root_for_payload(&identity.root),
previous_head_sha: previous_head.map(str::to_string),
head_sha: head_sha.clone(),
subject: git_output(&identity.root, &["log", "-1", "--pretty=%s"]).ok(),
author_name: git_output(&identity.root, &["log", "-1", "--pretty=%an"]).ok(),
author_email: git_output(&identity.root, &["log", "-1", "--pretty=%ae"]).ok(),
committed_at: git_output(&identity.root, &["log", "-1", "--pretty=%cI"]).ok(),
occurred_at: Utc::now(),
};
let fingerprint = commit_event_fingerprint(&commit);
commit.fingerprint = Some(fingerprint.clone());
commit.id = stable_event_id(&fingerprint);
append_unique_commit_event(layout, &commit)?;
Ok(Some(head_sha))
}
fn append_unique_commit_event(
layout: &SpoolLayout,
commit: &WorktreeCommitEvent,
) -> Result<bool, String> {
let fingerprint = commit
.fingerprint
.clone()
.unwrap_or_else(|| commit_event_fingerprint(commit));
if !claim_commit_fingerprint(layout, &fingerprint)? {
return Ok(false);
}
let mut commit = commit.clone();
commit.fingerprint = Some(fingerprint.clone());
if commit.id.is_empty() {
commit.id = stable_event_id(&fingerprint);
}
if let Err(error) = append_json_line(&layout.commits_file, &commit) {
release_commit_fingerprint_claim(layout, &fingerprint);
return Err(error);
}
Ok(true)
}
fn commit_event_fingerprint(commit: &WorktreeCommitEvent) -> String {
let payload = json!({
"schema": 3,
"repoOwner": commit.repo_owner,
"repoName": commit.repo_name,
"branchName": commit.branch_name,
"previousHeadSha": commit.previous_head_sha,
"headSha": commit.head_sha,
});
let encoded = serde_json::to_vec(&payload).unwrap_or_default();
let mut hasher = Sha256::new();
hasher.update(encoded);
format!("{:x}", hasher.finalize())
}
fn claim_commit_fingerprint(layout: &SpoolLayout, fingerprint: &str) -> Result<bool, String> {
fs::create_dir_all(&layout.commit_dedupe_dir).map_err(|error| {
format!(
"Failed to create commit dedupe directory {}: {error}",
layout.commit_dedupe_dir.display()
)
})?;
let marker = commit_dedupe_marker_path(layout, fingerprint);
match OpenOptions::new()
.write(true)
.create_new(true)
.open(&marker)
{
Ok(mut file) => {
writeln!(file, "{}", Utc::now().to_rfc3339()).map_err(|error| {
format!(
"Failed to write commit dedupe marker {}: {error}",
marker.display()
)
})?;
Ok(true)
}
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => Ok(false),
Err(error) => Err(format!(
"Failed to create commit dedupe marker {}: {error}",
marker.display()
)),
}
}
fn release_commit_fingerprint_claim(layout: &SpoolLayout, fingerprint: &str) {
let _ = fs::remove_file(commit_dedupe_marker_path(layout, fingerprint));
}
fn commit_dedupe_marker_path(layout: &SpoolLayout, fingerprint: &str) -> PathBuf {
layout.commit_dedupe_dir.join(format!("{fingerprint}.seen"))
}
async fn sync_spool_for_identity(
identity: &RepoIdentity,
layout: &SpoolLayout,
) -> Result<(), String> {
let candidates = collect_sync_candidates(layout)?;
if candidates.is_empty() {
return Ok(());
}
sync_candidates(identity, candidates, false).await
}
async fn sync_candidates(
identity: &RepoIdentity,
candidates: Vec<SyncCandidate>,
resync: bool,
) -> Result<(), String> {
let mut events = Vec::new();
let mut commits = Vec::new();
for candidate in &candidates {
match &candidate.records {
SyncRecords::Events(records) => events.extend(records.iter().cloned()),
SyncRecords::Commits(records) => commits.extend(records.iter().cloned()),
}
}
let token = resolve_cli_access_token()?;
let device = resolve_device_identity()?;
let client = cli_request_client()?;
let api = ApiConfig::from_env();
let endpoint = api.cli_worktree_mutations_endpoint();
let event_count = events.len();
let commit_count = commits.len();
let hardware_id = worktree_device_hardware_id(&device.hardware_id);
let hostname = current_hostname();
let platform = std::env::consts::OS.to_string();
let runtime_key = worktree_runtime_key();
let repo_root = normalize_repo_root_for_payload(&identity.root);
let events = events
.into_iter()
.map(ensure_mutation_event_identity)
.collect::<Vec<_>>();
let commits = commits
.into_iter()
.map(ensure_commit_event_identity)
.collect::<Vec<_>>();
let batches = split_worktree_mutation_upload_batches(events.clone(), commits.clone());
let batch_count = batches.len();
if batch_count > 1 {
println!(
"Uploading {} worktree mutation record(s) in {} batch(es) for {}/{} on {} [{}].",
event_count + commit_count,
batch_count,
identity.owner,
identity.name,
identity.branch,
runtime_key
);
}
let mut status_code = 202;
for attempt in 1..=WORKTREE_MUTATION_SYNC_MAX_ATTEMPTS {
let attempt_batches =
split_worktree_mutation_upload_batches(events.clone(), commits.clone());
match upload_worktree_mutation_batches(
&client,
&endpoint,
&token,
&hardware_id,
hostname.as_deref(),
&platform,
&runtime_key,
identity,
&repo_root,
attempt_batches,
)
.await
{
Ok(code) => {
status_code = code;
break;
}
Err(error)
if error.kind == WorktreeMutationSyncErrorKind::Retryable
&& attempt < WORKTREE_MUTATION_SYNC_MAX_ATTEMPTS =>
{
eprintln!(
"Worktree mutation sync attempt {attempt}/{} failed for {}/{} on {}: {} Restarting from the first batch in {} second(s).",
WORKTREE_MUTATION_SYNC_MAX_ATTEMPTS,
identity.owner,
identity.name,
identity.branch,
error.message,
WORKTREE_MUTATION_SYNC_RETRY_DELAY_SECONDS
);
tokio::time::sleep(Duration::from_secs(
WORKTREE_MUTATION_SYNC_RETRY_DELAY_SECONDS,
))
.await;
}
Err(error) => return Err(error.message),
}
}
let layout = prepare_spool_layout(identity)?;
append_sync_log(SyncLogAppend {
identity,
layout: &layout,
endpoint: &endpoint,
status_code,
event_count,
commit_count,
candidates: &candidates,
resync,
})?;
for candidate in candidates {
if !is_synced_spool_path(&candidate.path) {
mark_synced(&candidate.path)?;
}
}
Ok(())
}
async fn upload_worktree_mutation_batches(
client: &reqwest::Client,
endpoint: &str,
token: &str,
hardware_id: &str,
hostname: Option<&str>,
platform: &str,
runtime_key: &str,
identity: &RepoIdentity,
repo_root: &str,
batches: Vec<WorktreeMutationUploadBatch>,
) -> Result<u16, WorktreeMutationSyncError> {
let batch_count = batches.len();
let mut status_code = 202;
for (batch_index, batch) in batches.into_iter().enumerate() {
let batch_event_count = batch.events.len();
let batch_commit_count = batch.commits.len();
let payload = WorktreeMutationIngestPayload {
device: WorktreeMutationDevicePayload {
hardware_id: hardware_id.to_string(),
device_name: hostname.map(str::to_string),
hostname: hostname.map(str::to_string),
platform: platform.to_string(),
runtime_key: Some(runtime_key.to_string()),
},
repository: WorktreeMutationRepositoryPayload {
owner: identity.owner.clone(),
name: identity.name.clone(),
branch_name: identity.branch.clone(),
repo_root: repo_root.to_string(),
},
events: batch.events,
commits: batch.commits,
};
let response = client
.post(endpoint)
.bearer_auth(token)
.json(&payload)
.send()
.await
.map_err(|error| WorktreeMutationSyncError {
kind: WorktreeMutationSyncErrorKind::Retryable,
message: format!(
"Failed to upload worktree mutation batch {}/{}: {error}",
batch_index + 1,
batch_count
),
})?;
if response.status() == StatusCode::UNAUTHORIZED {
return Err(WorktreeMutationSyncError {
kind: WorktreeMutationSyncErrorKind::NonRetryable,
message: "Your stored CLI session is no longer valid. Run `xbp login` again."
.to_string(),
});
}
if !response.status().is_success() {
let status = response.status();
let body = response.text().await.unwrap_or_default();
return Err(WorktreeMutationSyncError {
kind: if should_retry_worktree_mutation_sync_status(status) {
WorktreeMutationSyncErrorKind::Retryable
} else {
WorktreeMutationSyncErrorKind::NonRetryable
},
message: format!(
"Worktree mutation upload batch {}/{} failed with {status}: {body}",
batch_index + 1,
batch_count
),
});
}
status_code = response.status().as_u16();
if batch_count > 1 {
println!(
"Uploaded worktree mutation batch {}/{} ({} event(s), {} commit(s)).",
batch_index + 1,
batch_count,
batch_event_count,
batch_commit_count
);
}
}
Ok(status_code)
}
fn should_retry_worktree_mutation_sync_status(status: StatusCode) -> bool {
status.is_server_error()
|| status == StatusCode::REQUEST_TIMEOUT
|| status == StatusCode::TOO_MANY_REQUESTS
}
fn split_worktree_mutation_upload_batches(
events: Vec<WorktreeMutationEvent>,
commits: Vec<WorktreeCommitEvent>,
) -> Vec<WorktreeMutationUploadBatch> {
let mut batches = Vec::new();
let mut current = empty_worktree_mutation_upload_batch();
for event in events {
let estimated_json_bytes = serialized_json_len(&event);
if should_start_next_worktree_mutation_upload_batch(¤t, estimated_json_bytes) {
batches.push(current);
current = empty_worktree_mutation_upload_batch();
}
current.estimated_json_bytes += estimated_json_bytes;
current.events.push(event);
}
for commit in commits {
let estimated_json_bytes = serialized_json_len(&commit);
if should_start_next_worktree_mutation_upload_batch(¤t, estimated_json_bytes) {
batches.push(current);
current = empty_worktree_mutation_upload_batch();
}
current.estimated_json_bytes += estimated_json_bytes;
current.commits.push(commit);
}
if current_record_count(¤t) > 0 {
batches.push(current);
}
batches
}
fn empty_worktree_mutation_upload_batch() -> WorktreeMutationUploadBatch {
WorktreeMutationUploadBatch {
events: Vec::new(),
commits: Vec::new(),
estimated_json_bytes: 0,
}
}
fn should_start_next_worktree_mutation_upload_batch(
batch: &WorktreeMutationUploadBatch,
next_record_json_bytes: usize,
) -> bool {
if current_record_count(batch) == 0 {
return false;
}
current_record_count(batch) >= WORKTREE_MUTATION_SYNC_BATCH_RECORD_LIMIT
|| batch.estimated_json_bytes + next_record_json_bytes
> WORKTREE_MUTATION_SYNC_BATCH_JSON_BYTES
}
fn current_record_count(batch: &WorktreeMutationUploadBatch) -> usize {
batch.events.len() + batch.commits.len()
}
fn serialized_json_len<T: Serialize>(value: &T) -> usize {
serde_json::to_vec(value)
.map(|encoded| encoded.len())
.unwrap_or(0)
}
fn collect_sync_candidates(layout: &SpoolLayout) -> Result<Vec<SyncCandidate>, String> {
collect_spool_candidates(layout, false)
}
fn print_no_sync_candidates_message(
target: &WorktreeWatchTargetOptions,
resync: bool,
) -> Result<(), String> {
if resync {
println!("No local worktree mutation JSONL files found.");
return Ok(());
}
let identities = resolve_target_identities(target)?;
let mut synced_files = 0usize;
let mut synced_records = 0usize;
for identity in identities {
let layout = prepare_spool_layout(&identity)?;
let all_candidates = collect_spool_candidates(&layout, true)?;
let unsynced_candidates = collect_sync_candidates(&layout)?;
synced_files += all_candidates
.len()
.saturating_sub(unsynced_candidates.len());
let all_records = all_candidates
.iter()
.map(|candidate| candidate.records.len())
.sum::<usize>();
let unsynced_records = unsynced_candidates
.iter()
.map(|candidate| candidate.records.len())
.sum::<usize>();
synced_records += all_records.saturating_sub(unsynced_records);
}
if synced_files > 0 {
println!(
"No unsynced worktree mutation files found. Found {synced_files} already-synced local spool file(s) with {synced_records} record(s). Use `xbp worktree-watch sync --resync` to upload them again."
);
} else {
println!("No unsynced worktree mutation files found.");
}
Ok(())
}
fn collect_spool_candidates(
layout: &SpoolLayout,
include_synced: bool,
) -> Result<Vec<SyncCandidate>, String> {
let mut roots = vec![layout.root.clone()];
if let Some(branch) = layout.root.file_name().and_then(|name| name.to_str()) {
if let Some(repo_root) = layout.root.parent() {
if let (Some(name), Some(owner)) = (
repo_root.file_name().and_then(|n| n.to_str()),
repo_root.parent().and_then(|p| p.file_name()).and_then(|n| n.to_str()),
) {
let identity = RepoIdentity {
owner: owner.to_string(),
name: name.to_string(),
branch: branch.to_string(),
root: PathBuf::new(),
head_sha: None,
};
if let Ok(search_roots) = repository_spool_search_roots(&identity) {
for search_root in search_roots {
let branch_root = search_root.join(branch);
if !roots.iter().any(|existing| {
path_identity_key(existing) == path_identity_key(&branch_root)
}) {
roots.push(branch_root);
}
}
}
}
}
}
let mut candidates = Vec::new();
let mut seen_files = BTreeSet::new();
for root in roots {
collect_spool_candidates_from_root(&root, include_synced, &mut candidates, &mut seen_files)?;
}
Ok(candidates)
}
fn collect_spool_candidates_from_root(
root: &Path,
include_synced: bool,
candidates: &mut Vec<SyncCandidate>,
seen_files: &mut BTreeSet<String>,
) -> Result<(), String> {
if !root.exists() {
return Ok(());
}
for entry in fs::read_dir(root).map_err(|error| {
format!(
"Failed to read spool directory {}: {error}",
root.display()
)
})? {
let path = entry
.map_err(|error| format!("Failed to read spool entry: {error}"))?
.path();
let Some(file_name) = path.file_name().and_then(|value| value.to_str()) else {
continue;
};
if !file_name.ends_with(".jsonl") || (!include_synced && file_name.contains(".synced.")) {
continue;
}
let kind = if file_name.starts_with("events-") {
SyncFileKind::Events
} else if file_name.starts_with("commits-") {
SyncFileKind::Commits
} else {
continue;
};
let identity_key = path_identity_key(&path);
if !seen_files.insert(identity_key) {
continue;
}
let records = sanitize_spool_records(&path, kind, read_spool_records(&path, kind)?)?;
if records.is_empty() {
continue;
}
candidates.push(SyncCandidate { path, records });
}
Ok(())
}
fn append_sync_log(request: SyncLogAppend<'_>) -> Result<(), String> {
let entry = WorktreeWatchSyncLogEntry {
id: Uuid::new_v4().to_string(),
synced_at: Utc::now(),
endpoint: request.endpoint.to_string(),
repository_owner: request.identity.owner.clone(),
repository_name: request.identity.name.clone(),
branch_name: request.identity.branch.clone(),
repo_root: request.identity.root.display().to_string(),
resync: request.resync,
status_code: request.status_code,
event_count: request.event_count,
commit_count: request.commit_count,
spool_file_count: request.candidates.len(),
spool_files: request
.candidates
.iter()
.map(|candidate| candidate.path.display().to_string())
.collect(),
};
append_json_line(&request.layout.sync_log_file, &entry)
}
fn sanitize_spool_records(
path: &Path,
kind: SyncFileKind,
records: SyncRecords,
) -> Result<SyncRecords, String> {
match (kind, records) {
(SyncFileKind::Events, SyncRecords::Events(records)) => {
let mut sanitized = Vec::with_capacity(records.len());
let mut changed = false;
for record in records {
let original = serde_json::to_value(&record).ok();
let Some(record) = sanitize_spooled_event(record) else {
changed = true;
continue;
};
if !changed {
changed = serde_json::to_value(&record).ok() != original;
}
sanitized.push(record);
}
if changed {
rewrite_jsonl_records(path, &sanitized)?;
}
Ok(SyncRecords::Events(sanitized))
}
(_, records) => Ok(records),
}
}
fn sanitize_spooled_event(mut event: WorktreeMutationEvent) -> Option<WorktreeMutationEvent> {
event.paths = dedupe_filtered_spool_paths(event.paths);
event.primary_path = sanitize_optional_spool_path(event.primary_path);
event.old_path = sanitize_optional_spool_path(event.old_path);
event.new_path = sanitize_optional_spool_path(event.new_path);
if event.paths.is_empty() {
if let Some(path) = event.primary_path.clone() {
event.paths.push(path);
}
if let Some(path) = event.old_path.clone() {
if !event.paths.iter().any(|existing| existing == &path) {
event.paths.push(path);
}
}
if let Some(path) = event.new_path.clone() {
if !event.paths.iter().any(|existing| existing == &path) {
event.paths.push(path);
}
}
}
if event.primary_path.is_none() {
event.primary_path = event.paths.first().cloned();
}
if event.paths.is_empty() {
return None;
}
Some(event)
}
fn sanitize_optional_spool_path(path: Option<String>) -> Option<String> {
path.and_then(sanitize_spool_path)
}
fn sanitize_spool_path(path: String) -> Option<String> {
let trimmed = path.trim();
if trimmed.is_empty() || watch_ignore_rules().ignores_relative(trimmed) {
return None;
}
Some(trimmed.replace('\\', "/"))
}
fn dedupe_filtered_spool_paths(paths: Vec<String>) -> Vec<String> {
let mut sanitized = Vec::with_capacity(paths.len());
for path in paths {
let Some(path) = sanitize_spool_path(path) else {
continue;
};
if !sanitized.iter().any(|existing| existing == &path) {
sanitized.push(path);
}
}
sanitized
}
fn rewrite_jsonl_records<T: Serialize>(path: &Path, records: &[T]) -> Result<(), String> {
if records.is_empty() {
if path.exists() {
fs::remove_file(path).map_err(|error| {
format!(
"Failed to remove emptied spool file {}: {error}",
path.display()
)
})?;
}
return Ok(());
}
let temp_path = path.with_extension(format!(
"{}.{}.tmp",
path.extension()
.and_then(|value| value.to_str())
.unwrap_or("jsonl"),
Uuid::new_v4()
));
{
let mut file = File::create(&temp_path).map_err(|error| {
format!(
"Failed to create temporary spool file {}: {error}",
temp_path.display()
)
})?;
for record in records {
let line = serde_json::to_string(record)
.map_err(|error| format!("Failed to serialize worktree event: {error}"))?;
writeln!(file, "{line}").map_err(|error| {
format!(
"Failed to write temporary spool file {}: {error}",
temp_path.display()
)
})?;
}
}
if path.exists() {
fs::remove_file(path).map_err(|error| {
format!(
"Failed to replace rewritten spool file {}: {error}",
path.display()
)
})?;
}
fs::rename(&temp_path, path)
.map_err(|error| format!("Failed to rewrite spool file {}: {error}", path.display()))
}
fn read_last_sync_log_entry(path: &Path) -> Result<Option<WorktreeWatchSyncLogEntry>, String> {
if !path.exists() {
return Ok(None);
}
let values = read_jsonl_values(path)?;
let Some(value) = values.into_iter().last() else {
return Ok(None);
};
serde_json::from_value(value)
.map(Some)
.map_err(|error| format!("Failed to decode sync log {}: {error}", path.display()))
}
fn is_synced_spool_path(path: &Path) -> bool {
path.file_name()
.and_then(|value| value.to_str())
.map(|value| value.contains(".synced."))
.unwrap_or(false)
}
fn mark_synced(path: &Path) -> Result<(), String> {
let Some(file_name) = path.file_name().and_then(|value| value.to_str()) else {
return Ok(());
};
let synced_name = file_name.replace(
".jsonl",
&format!(".synced.{}.jsonl", Utc::now().timestamp()),
);
let synced_path = path.with_file_name(synced_name);
fs::rename(path, &synced_path).map_err(|error| {
format!(
"Failed to mark spool file {} as synced: {error}",
path.display()
)
})
}
fn read_spool_records(path: &Path, kind: SyncFileKind) -> Result<SyncRecords, String> {
match kind {
SyncFileKind::Events => read_jsonl_records::<WorktreeMutationEvent>(path)
.map(SyncRecords::Events)
.map_err(|error| format!("Failed to decode event from {}: {error}", path.display())),
SyncFileKind::Commits => read_jsonl_records::<WorktreeCommitEvent>(path)
.map(SyncRecords::Commits)
.map_err(|error| format!("Failed to decode commit from {}: {error}", path.display())),
}
}
fn sanitize_jsonl_line(line: &str) -> Option<&str> {
let trimmed = line.trim().trim_matches('\0').trim();
if trimmed.is_empty() {
None
} else {
Some(trimmed)
}
}
fn read_jsonl_records<T>(path: &Path) -> Result<Vec<T>, String>
where
T: for<'de> Deserialize<'de>,
{
let file = File::open(path)
.map_err(|error| format!("Failed to open spool file {}: {error}", path.display()))?;
let reader = BufReader::new(file);
let mut records = Vec::new();
for (line_no, line) in reader.lines().enumerate() {
let line =
line.map_err(|error| format!("Failed to read spool file {}: {error}", path.display()))?;
let Some(payload) = sanitize_jsonl_line(&line) else {
continue;
};
match serde_json::from_str(payload) {
Ok(record) => records.push(record),
Err(error) => {
eprintln!(
"warning: skipping unparseable JSONL record in {} at line {}: {error}",
path.display(),
line_no + 1
);
}
}
}
Ok(records)
}
fn read_jsonl_values(path: &Path) -> Result<Vec<Value>, String> {
read_jsonl_records(path)
}
fn spawn_detached_worktree_watch(repo: Option<&Path>) -> Result<PathBuf, String> {
let identity = resolve_repo_identity(repo)?;
spawn_detached_worktree_watch_for_identity(&identity)
}
fn spawn_detached_parent_worktree_watch(parent: &Path) -> Result<PathBuf, String> {
let parent = canonical_parent_path(parent)?;
let layout = prepare_parent_spool_layout(&parent)?;
replace_existing_parent_watcher(&parent, &layout)?;
let executable = std::env::current_exe()
.map_err(|error| format!("Failed to resolve current XBP executable: {error}"))?;
let mut command = Command::new(&executable);
command
.arg("worktree-watch")
.arg("start")
.arg("--parent")
.arg(&parent)
.env(BACKGROUND_CHILD_ENV, "1")
.current_dir(&parent)
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null());
configure_detached_child_process(&mut command);
let child = command
.spawn()
.map_err(|error| format!("Failed to spawn background parent worktree watcher: {error}"))?;
write_parent_watcher_state(&parent, &layout, child.id(), &executable)?;
Ok(parent)
}
fn spawn_detached_worktree_watch_for_identity(identity: &RepoIdentity) -> Result<PathBuf, String> {
let layout = prepare_spool_layout(identity)?;
replace_existing_watcher(identity, &layout)?;
let executable = std::env::current_exe()
.map_err(|error| format!("Failed to resolve current XBP executable: {error}"))?;
let mut command = Command::new(&executable);
command
.arg("worktree-watch")
.arg("start")
.arg("--repo")
.arg(&identity.root)
.env(BACKGROUND_CHILD_ENV, "1")
.current_dir(&identity.root)
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null());
configure_detached_child_process(&mut command);
let child: std::process::Child = command
.spawn()
.map_err(|error| format!("Failed to spawn background worktree watcher: {error}"))?;
write_watcher_state(identity, &layout, child.id(), &executable)?;
Ok(identity.root.clone())
}
fn configure_detached_child_process(command: &mut Command) {
#[cfg(windows)]
{
use std::os::windows::process::CommandExt;
command.creation_flags(CREATE_NO_WINDOW);
}
#[cfg(unix)]
{
use std::os::unix::process::CommandExt;
unsafe {
command.pre_exec(|| {
extern "C" {
fn setsid() -> i32;
}
let _ = setsid();
Ok(())
});
}
}
}
fn replace_existing_parent_watcher(
parent: &Path,
layout: &ParentSpoolLayout,
) -> Result<(), String> {
let Some(state) = read_parent_watcher_state(&layout.watcher_state_file)? else {
return Ok(());
};
if path_identity_key(parent) != path_identity_key(Path::new(&state.parent_root)) {
return Ok(());
}
if is_matching_parent_watcher_process(&state) {
stop_parent_process(&state, false)?;
println!(
"Replaced existing background parent worktree watcher {} for {}",
state.pid,
parent.display()
);
let _ = fs::remove_file(&layout.watcher_state_file);
return Ok(());
}
if parent_watcher_state_pid_exists_with_same_executable(&state) {
return Err(format!(
"Existing parent watcher state for {} points to running XBP process {}, but its command line could not be verified. Run `xbp worktree-watch stop --parent \"{}\" --force` before starting another detached parent watcher.",
parent.display(),
state.pid,
parent.display()
));
}
let _ = fs::remove_file(&layout.watcher_state_file);
Ok(())
}
fn replace_existing_watcher(identity: &RepoIdentity, layout: &SpoolLayout) -> Result<(), String> {
let Some(state) = read_watcher_state(&layout.watcher_state_file)? else {
return Ok(());
};
if !same_repo_watcher_state(identity, &state) {
return Ok(());
}
if is_matching_watcher_process(&state) {
stop_process(&state, false)?;
println!(
"Replaced existing background worktree watcher {} for {}",
state.pid,
identity.root.display()
);
let _ = fs::remove_file(&layout.watcher_state_file);
return Ok(());
}
if watcher_state_pid_exists_with_same_executable(&state) {
return Err(format!(
"Existing watcher state for {} points to running XBP process {}, but its command line could not be verified. Run `xbp worktree-watch stop --repo \"{}\" --force` before starting another detached watcher.",
identity.root.display(),
state.pid,
identity.root.display()
));
}
let _ = fs::remove_file(&layout.watcher_state_file);
Ok(())
}
fn stop_existing_watcher(
identity: &RepoIdentity,
layout: &SpoolLayout,
force: bool,
) -> Result<StopWatcherOutcome, String> {
let Some(state) = read_watcher_state(&layout.watcher_state_file)? else {
return Ok(StopWatcherOutcome::NoState);
};
if !same_repo_watcher_state(identity, &state) {
return Ok(StopWatcherOutcome::NoState);
}
if is_matching_watcher_process(&state) || force {
let pid = state.pid;
stop_process(&state, force)?;
let _ = fs::remove_file(&layout.watcher_state_file);
return Ok(StopWatcherOutcome::Stopped(pid));
}
if watcher_state_pid_exists_with_same_executable(&state) {
return Err(format!(
"Watcher state for {} points to running XBP process {}, but its command line could not be verified. Re-run with `--force` to stop that PID.",
identity.root.display(),
state.pid
));
}
let _ = fs::remove_file(&layout.watcher_state_file);
Ok(StopWatcherOutcome::RemovedStaleState)
}
fn stop_existing_parent_watcher(parent: &Path, force: bool) -> Result<StopWatcherOutcome, String> {
let layout = prepare_parent_spool_layout(parent)?;
let Some(state) = read_parent_watcher_state(&layout.watcher_state_file)? else {
return Ok(StopWatcherOutcome::NoState);
};
if path_identity_key(parent) != path_identity_key(Path::new(&state.parent_root)) {
return Ok(StopWatcherOutcome::NoState);
}
if is_matching_parent_watcher_process(&state) || force {
let pid = state.pid;
stop_parent_process(&state, force)?;
let _ = fs::remove_file(&layout.watcher_state_file);
return Ok(StopWatcherOutcome::Stopped(pid));
}
if parent_watcher_state_pid_exists_with_same_executable(&state) {
return Err(format!(
"Parent watcher state for {} points to running XBP process {}, but its command line could not be verified. Re-run with `--force` to stop that PID.",
parent.display(),
state.pid
));
}
let _ = fs::remove_file(&layout.watcher_state_file);
Ok(StopWatcherOutcome::RemovedStaleState)
}
fn read_watcher_state(path: &Path) -> Result<Option<WorktreeWatcherState>, String> {
for candidate in watcher_state_read_candidates(path) {
if !candidate.exists() {
continue;
}
let raw = fs::read_to_string(&candidate).map_err(|error| {
format!(
"Failed to read watcher state {}: {error}",
candidate.display()
)
})?;
return serde_json::from_str(&raw).map(Some).map_err(|error| {
format!(
"Failed to parse watcher state {}: {error}",
candidate.display()
)
});
}
Ok(None)
}
fn watcher_state_read_candidates(path: &Path) -> Vec<PathBuf> {
let mut candidates = vec![path.to_path_buf()];
if let Some(parent) = path.parent() {
let legacy = parent.join("watcher-state.json");
if legacy != path {
candidates.push(legacy);
}
}
candidates
}
fn read_parent_watcher_state(path: &Path) -> Result<Option<ParentWorktreeWatcherState>, String> {
if !path.exists() {
return Ok(None);
}
let raw = fs::read_to_string(path).map_err(|error| {
format!(
"Failed to read parent watcher state {}: {error}",
path.display()
)
})?;
serde_json::from_str(&raw).map(Some).map_err(|error| {
format!(
"Failed to parse parent watcher state {}: {error}",
path.display()
)
})
}
fn write_watcher_state(
identity: &RepoIdentity,
layout: &SpoolLayout,
pid: u32,
executable: &Path,
) -> Result<(), String> {
let state = WorktreeWatcherState {
pid,
repo_root: normalize_repo_root_for_payload(&identity.root),
repository_owner: identity.owner.clone(),
repository_name: identity.name.clone(),
branch_name: identity.branch.clone(),
executable: executable.display().to_string(),
started_at: Utc::now(),
runtime_key: Some(worktree_runtime_key()),
platform: Some(std::env::consts::OS.to_string()),
};
let rendered = serde_json::to_string_pretty(&state)
.map_err(|error| format!("Failed to serialize watcher state: {error}"))?;
fs::write(&layout.watcher_state_file, format!("{rendered}\n")).map_err(|error| {
format!(
"Failed to write watcher state {}: {error}",
layout.watcher_state_file.display()
)
})
}
fn write_parent_watcher_state(
parent: &Path,
layout: &ParentSpoolLayout,
pid: u32,
executable: &Path,
) -> Result<(), String> {
let state = ParentWorktreeWatcherState {
pid,
parent_root: normalize_repo_root_for_payload(parent),
executable: executable.display().to_string(),
started_at: Utc::now(),
runtime_key: Some(worktree_runtime_key()),
platform: Some(std::env::consts::OS.to_string()),
};
let rendered = serde_json::to_string_pretty(&state)
.map_err(|error| format!("Failed to serialize parent watcher state: {error}"))?;
fs::write(&layout.watcher_state_file, format!("{rendered}\n")).map_err(|error| {
format!(
"Failed to write parent watcher state {}: {error}",
layout.watcher_state_file.display()
)
})
}
fn same_repo_watcher_state(identity: &RepoIdentity, state: &WorktreeWatcherState) -> bool {
if !runtime_keys_compatible(state.runtime_key.as_deref()) {
return false;
}
path_identity_key(&identity.root) == path_identity_key(Path::new(&state.repo_root))
&& identity.owner == state.repository_owner
&& identity.name == state.repository_name
&& identity.branch == state.branch_name
}
fn runtime_keys_compatible(state_runtime: Option<&str>) -> bool {
let current = worktree_runtime_key();
match state_runtime {
Some(key) => key == current,
None => !is_wsl_runtime(),
}
}
fn process_system_for_pids(pids: &[u32]) -> System {
let mut system = System::new();
if pids.is_empty() {
return system;
}
let pid_list: Vec<Pid> = pids
.iter()
.copied()
.filter(|pid| *pid != 0 && *pid != std::process::id())
.map(Pid::from_u32)
.collect();
if pid_list.is_empty() {
return system;
}
system.refresh_processes_specifics(
ProcessesToUpdate::Some(&pid_list),
true,
ProcessRefreshKind::everything()
.with_cmd(UpdateKind::Always)
.with_exe(UpdateKind::Always),
);
system
}
fn is_matching_watcher_process(state: &WorktreeWatcherState) -> bool {
let system = process_system_for_pids(&[state.pid]);
is_matching_watcher_process_with_system(&system, state)
}
fn is_live_watcher_process_with_system(system: &System, state: &WorktreeWatcherState) -> bool {
if state.pid == std::process::id() {
return runtime_keys_compatible(state.runtime_key.as_deref());
}
is_matching_watcher_process_with_system(system, state)
}
fn is_matching_watcher_process_with_system(
system: &System,
state: &WorktreeWatcherState,
) -> bool {
if state.pid == std::process::id() {
return false;
}
let Some(process) = system.process(Pid::from_u32(state.pid)) else {
return false;
};
let command_line = process
.cmd()
.iter()
.map(|part| part.to_string_lossy())
.collect::<Vec<_>>()
.join(" ");
let command_matches = command_line_contains_ignore_case(&command_line, "worktree-watch")
&& command_line_contains_ignore_case(&command_line, "start")
&& command_line_mentions_path(&command_line, &state.repo_root);
command_matches || watcher_state_process_identity_matches(process, state)
}
fn is_matching_parent_watcher_process(state: &ParentWorktreeWatcherState) -> bool {
let system = process_system_for_pids(&[state.pid]);
is_matching_parent_watcher_process_with_system(&system, state)
}
fn is_live_parent_watcher_process_with_system(
system: &System,
state: &ParentWorktreeWatcherState,
) -> bool {
if state.pid == std::process::id() {
return runtime_keys_compatible(state.runtime_key.as_deref());
}
is_matching_parent_watcher_process_with_system(system, state)
}
fn is_matching_parent_watcher_process_with_system(
system: &System,
state: &ParentWorktreeWatcherState,
) -> bool {
if state.pid == std::process::id() {
return false;
}
if !runtime_keys_compatible(state.runtime_key.as_deref()) {
return false;
}
let Some(process) = system.process(Pid::from_u32(state.pid)) else {
return false;
};
let command_line = process
.cmd()
.iter()
.map(|part| part.to_string_lossy())
.collect::<Vec<_>>()
.join(" ");
let command_matches = command_line_contains_ignore_case(&command_line, "worktree-watch")
&& command_line_contains_ignore_case(&command_line, "start")
&& command_line_contains_ignore_case(&command_line, "--parent")
&& command_line_mentions_path(&command_line, &state.parent_root);
command_matches
|| parent_watcher_state_process_identity_matches(process, state)
|| (process_executable_matches_path(process, Path::new(&state.executable))
&& process_looks_like_worktree_watch(process))
}
fn process_looks_like_worktree_watch(process: &sysinfo::Process) -> bool {
let command_line = process
.cmd()
.iter()
.map(|part| part.to_string_lossy())
.collect::<Vec<_>>()
.join(" ");
if command_line_contains_ignore_case(&command_line, "worktree-watch") {
return true;
}
command_line.trim().is_empty()
}
fn command_line_contains_ignore_case(haystack: &str, needle: &str) -> bool {
haystack.to_ascii_lowercase().contains(&needle.to_ascii_lowercase())
}
fn command_line_mentions_path(command_line: &str, path: &str) -> bool {
if path.is_empty() {
return false;
}
if command_line.contains(path) {
return true;
}
let command_lower = command_line.to_ascii_lowercase().replace('\\', "/");
let path_lower = path.to_ascii_lowercase().replace('\\', "/");
if command_lower.contains(&path_lower) {
return true;
}
let path_key = path_identity_key(Path::new(path));
if !path_key.is_empty() && command_lower.contains(&path_key) {
return true;
}
for token in command_line.split_whitespace() {
let cleaned = token.trim_matches('"').trim_matches('\'');
if cleaned.is_empty() {
continue;
}
if path_identity_key(Path::new(cleaned)) == path_key {
return true;
}
}
Path::new(path)
.file_name()
.and_then(|name| name.to_str())
.map(|name| command_line_contains_ignore_case(command_line, name))
.unwrap_or(false)
}
fn watcher_state_pid_exists_with_same_executable(state: &WorktreeWatcherState) -> bool {
if state.pid == std::process::id() {
return false;
}
let system = process_system_for_pids(&[state.pid]);
let Some(process) = system.process(Pid::from_u32(state.pid)) else {
return false;
};
process_executable_matches_state(process, state)
}
fn parent_watcher_state_pid_exists_with_same_executable(
state: &ParentWorktreeWatcherState,
) -> bool {
if state.pid == std::process::id() {
return false;
}
let system = process_system_for_pids(&[state.pid]);
let Some(process) = system.process(Pid::from_u32(state.pid)) else {
return false;
};
process_executable_matches_path(process, Path::new(&state.executable))
}
fn watcher_state_process_identity_matches(
process: &sysinfo::Process,
state: &WorktreeWatcherState,
) -> bool {
process_executable_matches_state(process, state)
&& process_start_time_matches(process.start_time(), state.started_at)
}
fn parent_watcher_state_process_identity_matches(
process: &sysinfo::Process,
state: &ParentWorktreeWatcherState,
) -> bool {
process_executable_matches_path(process, Path::new(&state.executable))
&& process_start_time_matches(process.start_time(), state.started_at)
}
fn process_executable_matches_state(
process: &sysinfo::Process,
state: &WorktreeWatcherState,
) -> bool {
process_executable_matches_path(process, Path::new(&state.executable))
}
fn process_executable_matches_path(process: &sysinfo::Process, executable: &Path) -> bool {
let expected = path_identity_key(executable);
process
.exe()
.map(|path| path_identity_key(path) == expected)
.unwrap_or(false)
}
fn process_start_time_matches(process_started: u64, state_started: DateTime<Utc>) -> bool {
if process_started == 0 {
return true;
}
let process_started = process_started as i64;
let state_started = state_started.timestamp();
(process_started - state_started).abs() <= 120
}
fn stop_process(state: &WorktreeWatcherState, force: bool) -> Result<(), String> {
let system = process_system_for_pids(&[state.pid]);
let Some(process) = system.process(Pid::from_u32(state.pid)) else {
return Ok(());
};
if !force && !is_matching_watcher_process_with_system(&system, state) {
return Ok(());
}
if process.kill() {
Ok(())
} else {
Err(format!(
"Failed to stop existing watcher process {}",
state.pid
))
}
}
fn stop_parent_process(state: &ParentWorktreeWatcherState, force: bool) -> Result<(), String> {
let system = process_system_for_pids(&[state.pid]);
let Some(process) = system.process(Pid::from_u32(state.pid)) else {
return Ok(());
};
if !force && !is_matching_parent_watcher_process_with_system(&system, state) {
return Ok(());
}
if process.kill() {
Ok(())
} else {
Err(format!(
"Failed to stop existing parent watcher process {}",
state.pid
))
}
}
fn prefer_running_parent_root() -> Option<PathBuf> {
let mut candidates = list_parent_watcher_state_files().ok()?;
candidates.sort_by(|left, right| right.1.started_at.cmp(&left.1.started_at));
for (_path, state) in candidates {
let system = process_system_for_pids(&[state.pid]);
if !is_live_parent_watcher_process_with_system(&system, &state) {
continue;
}
let root = parent_root_as_local_path(&state.parent_root);
if root.is_dir() {
return Some(root);
}
return Some(root);
}
None
}
fn prefer_any_parent_root() -> Option<PathBuf> {
let mut candidates = list_parent_watcher_state_files().ok()?;
candidates.sort_by(|left, right| right.1.started_at.cmp(&left.1.started_at));
for (_path, state) in candidates {
let root = parent_root_as_local_path(&state.parent_root);
if root.is_dir() {
return Some(root);
}
}
None
}
fn count_git_roots_under(parent: &Path) -> Result<usize, String> {
let mut roots = Vec::new();
let mut seen = BTreeSet::new();
collect_git_roots_in_dir(parent, parent, &mut seen, &mut roots)?;
Ok(roots.len())
}
fn list_parent_watcher_state_files() -> Result<Vec<(PathBuf, ParentWorktreeWatcherState)>, String> {
let parents_root = global_mutations_root()?.join("parents");
if !parents_root.is_dir() {
return Ok(Vec::new());
}
let mut out = Vec::new();
for entry in fs::read_dir(&parents_root)
.map_err(|error| format!("Failed to read {}: {error}", parents_root.display()))?
{
let entry = entry.map_err(|error| format!("Failed to read parent spool entry: {error}"))?;
if !entry.file_type().map(|ft| ft.is_dir()).unwrap_or(false) {
continue;
}
for file in fs::read_dir(entry.path()).map_err(|error| {
format!(
"Failed to read parent spool {}: {error}",
entry.path().display()
)
})? {
let file = file.map_err(|error| format!("Failed to read parent state entry: {error}"))?;
let path = file.path();
let Some(name) = path.file_name().and_then(|value| value.to_str()) else {
continue;
};
if !(name.starts_with("watcher-state") && name.ends_with(".json")) {
continue;
}
if let Ok(Some(state)) = read_parent_watcher_state(&path) {
if let Some(runtime) = state.runtime_key.as_deref() {
if runtime != worktree_runtime_key() {
if !runtime_keys_compatible(Some(runtime)) {
continue;
}
}
} else if !runtime_keys_compatible(None) {
continue;
}
out.push((path, state));
}
}
}
Ok(out)
}
fn parent_root_as_local_path(parent_root: &str) -> PathBuf {
let trimmed = parent_root.trim();
if trimmed.is_empty() {
return PathBuf::new();
}
if let Some(windows) = map_wsl_mnt_path_to_windows(trimmed) {
return PathBuf::from(windows);
}
let as_path = PathBuf::from(trimmed);
if as_path.is_dir() {
return as_path;
}
if let Some(windows) = map_wsl_mnt_path_to_windows(&trimmed.replace('\\', "/")) {
return PathBuf::from(windows);
}
as_path
}
fn discover_repo_identities_under(parent: &Path) -> Result<Vec<RepoIdentity>, String> {
let parent = canonical_parent_path(parent)?;
let mut roots = Vec::new();
let mut seen = BTreeSet::new();
collect_git_roots_in_dir(&parent, &parent, &mut seen, &mut roots)?;
let identities = resolve_repo_identities_parallel(&roots);
let mut identities = identities.into_iter().flatten().collect::<Vec<_>>();
identities.sort_by(|left, right| left.root.cmp(&right.root));
Ok(identities)
}
fn collect_git_roots_in_dir(
parent: &Path,
dir: &Path,
seen: &mut BTreeSet<String>,
roots: &mut Vec<PathBuf>,
) -> Result<(), String> {
if dir != parent && is_skipped_discovery_dir(dir) {
return Ok(());
}
if dir.join(".git").exists() {
let key = path_identity_key(dir);
if seen.insert(key) {
roots.push(dir.to_path_buf());
}
return Ok(());
}
let entries = fs::read_dir(dir)
.map_err(|error| format!("Failed to read folder {}: {error}", dir.display()))?;
for entry in entries {
let entry = entry.map_err(|error| {
format!(
"Failed to read folder entry under {}: {error}",
dir.display()
)
})?;
let file_type = entry
.file_type()
.map_err(|error| format!("Failed to inspect {}: {error}", entry.path().display()))?;
if file_type.is_dir() {
collect_git_roots_in_dir(parent, &entry.path(), seen, roots)?;
}
}
Ok(())
}
fn resolve_repo_identities_parallel(roots: &[PathBuf]) -> Vec<Option<RepoIdentity>> {
if roots.is_empty() {
return Vec::new();
}
if roots.len() == 1 {
return vec![resolve_repo_identity_light(Some(&roots[0])).ok()];
}
let thread_count = roots.len().min(8).max(1);
let chunk_size = (roots.len() + thread_count - 1) / thread_count;
let mut handles = Vec::new();
for chunk in roots.chunks(chunk_size) {
let chunk = chunk.to_vec();
handles.push(std::thread::spawn(move || {
chunk
.iter()
.map(|root| resolve_repo_identity_light(Some(root)).ok())
.collect::<Vec<_>>()
}));
}
let mut out = Vec::with_capacity(roots.len());
for handle in handles {
match handle.join() {
Ok(part) => out.extend(part),
Err(_) => {
}
}
}
if out.len() < roots.len() {
for root in roots.iter().skip(out.len()) {
out.push(resolve_repo_identity_light(Some(root)).ok());
}
}
out
}
fn resolve_repo_identity_light(repo: Option<&Path>) -> Result<RepoIdentity, String> {
let start = resolve_repo_path_input(repo)?;
let root_raw = git_output(&start, &["rev-parse", "--show-toplevel"]).map_err(|error| {
if repo.is_some() {
format!(
"Failed to run git rev-parse --show-toplevel in {}: {error}",
start.display()
)
} else {
format!("Failed to run git rev-parse --show-toplevel: {error}")
}
})?;
let root = normalize_windows_verbatim_path(
fs::canonicalize(root_raw.trim()).unwrap_or_else(|_| PathBuf::from(root_raw.trim())),
);
let branch = git_output(&root, &["rev-parse", "--abbrev-ref", "HEAD"])
.unwrap_or_else(|_| "unknown".to_string());
let remote = git_output(&root, &["remote", "get-url", DEFAULT_REMOTE]).unwrap_or_default();
let (owner, name) = parse_remote_owner_repo(&remote).unwrap_or_else(|| {
let name = root
.file_name()
.and_then(|value| value.to_str())
.unwrap_or("repository")
.to_string();
("unknown".to_string(), name)
});
Ok(RepoIdentity {
owner,
name,
branch: sanitize_path_component(branch.trim()),
root,
head_sha: None,
})
}
fn resolve_repo_identity(repo: Option<&Path>) -> Result<RepoIdentity, String> {
let start = resolve_repo_path_input(repo)?;
let root_raw = git_output(&start, &["rev-parse", "--show-toplevel"]).map_err(|error| {
if repo.is_some() {
format!(
"Failed to run git rev-parse --show-toplevel in {}: {error}",
start.display()
)
} else {
format!("Failed to run git rev-parse --show-toplevel: {error}")
}
})?;
let root = normalize_windows_verbatim_path(
fs::canonicalize(root_raw.trim()).unwrap_or_else(|_| PathBuf::from(root_raw.trim())),
);
let branch = git_output(&root, &["rev-parse", "--abbrev-ref", "HEAD"])
.unwrap_or_else(|_| "unknown".to_string());
let remote = git_output(&root, &["remote", "get-url", DEFAULT_REMOTE]).unwrap_or_default();
let (owner, name) = parse_remote_owner_repo(&remote).unwrap_or_else(|| {
let name = root
.file_name()
.and_then(|value| value.to_str())
.unwrap_or("repository")
.to_string();
("unknown".to_string(), name)
});
let head_sha = current_head(&root)?;
Ok(RepoIdentity {
owner,
name,
branch: sanitize_path_component(branch.trim()),
root,
head_sha,
})
}
fn resolve_repo_path_input(repo: Option<&Path>) -> Result<PathBuf, String> {
let Some(input) = repo else {
return std::env::current_dir()
.map_err(|error| format!("Failed to resolve current directory: {error}"));
};
if path_looks_like_existing_dir(input) {
return Ok(input.to_path_buf());
}
let token = input.to_string_lossy();
let token = token.trim();
if token.is_empty() {
return Err("--repo value is empty".to_string());
}
if let Some(path) = lookup_repo_path_from_system_inventory(token)? {
return Ok(path);
}
if input.components().count() > 1
|| token.contains('/')
|| token.contains('\\')
|| token.contains(':')
{
return Ok(input.to_path_buf());
}
Err(format!(
"Could not resolve `--repo {token}` as a filesystem path or inventory name. \
Refresh with `xbp diag --refresh-system-inventory`, pass a full path, or use `--parent`."
))
}
fn path_looks_like_existing_dir(path: &Path) -> bool {
path.is_dir()
|| fs::metadata(path)
.map(|meta| meta.is_dir())
.unwrap_or(false)
}
fn lookup_repo_path_from_system_inventory(token: &str) -> Result<Option<PathBuf>, String> {
let config = match SshConfig::load() {
Ok(config) => config,
Err(_) => return Ok(None),
};
let Some(inventory) = config.system_inventory.as_ref() else {
return Ok(None);
};
if let Some(path) = match_github_repo_inventory(&inventory.github_repos, token)? {
return Ok(Some(path));
}
let needle = normalize_repo_lookup_token(token);
let mut project_hits: Vec<PathBuf> = Vec::new();
for project in &inventory.xbp_projects {
let name = normalize_repo_lookup_token(&project.name);
let root_name = Path::new(&project.root)
.file_name()
.and_then(|value| value.to_str())
.map(normalize_repo_lookup_token)
.unwrap_or_default();
if name == needle || root_name == needle {
let path = PathBuf::from(&project.root);
if path_looks_like_existing_dir(&path)
&& !project_hits.iter().any(|existing| {
path_identity_key(existing.as_path()) == path_identity_key(path.as_path())
})
{
project_hits.push(path);
}
}
}
match project_hits.len() {
0 => Ok(None),
1 => Ok(Some(project_hits.remove(0))),
_ => Err(format_ambiguous_repo_matches(token, &project_hits)),
}
}
fn match_github_repo_inventory(
repos: &[GithubRepoRecord],
token: &str,
) -> Result<Option<PathBuf>, String> {
let needle = normalize_repo_lookup_token(token);
if needle.is_empty() {
return Ok(None);
}
let mut exact_path: Vec<PathBuf> = Vec::new();
let mut full_name_hits: Vec<PathBuf> = Vec::new();
let mut short_name_hits: Vec<PathBuf> = Vec::new();
for record in repos {
let path = PathBuf::from(record.path.trim());
if record.path.trim().is_empty() || !path_looks_like_existing_dir(&path) {
continue;
}
let path_key = path_identity_key(&path);
let full_name = normalize_repo_lookup_token(&record.full_name);
let short = normalize_repo_lookup_token(&record.repo);
let owner_repo = normalize_repo_lookup_token(&format!("{}/{}", record.owner, record.repo));
let folder = path
.file_name()
.and_then(|value| value.to_str())
.map(normalize_repo_lookup_token)
.unwrap_or_default();
let push_unique = |bucket: &mut Vec<PathBuf>| {
if !bucket
.iter()
.any(|existing| path_identity_key(existing.as_path()) == path_key)
{
bucket.push(path.clone());
}
};
if path_identity_key(Path::new(record.path.trim())) == path_identity_key(Path::new(token))
|| normalize_repo_lookup_token(&record.path) == needle
{
push_unique(&mut exact_path);
}
if full_name == needle || owner_repo == needle {
push_unique(&mut full_name_hits);
}
if short == needle || folder == needle {
push_unique(&mut short_name_hits);
}
}
if let Some(path) = exact_path.into_iter().next() {
return Ok(Some(path));
}
match full_name_hits.len() {
0 => {}
1 => return Ok(Some(full_name_hits.remove(0))),
_ => return Err(format_ambiguous_repo_matches(token, &full_name_hits)),
}
match short_name_hits.len() {
0 => Ok(None),
1 => Ok(Some(short_name_hits.remove(0))),
_ => Err(format_ambiguous_repo_matches(token, &short_name_hits)),
}
}
fn normalize_repo_lookup_token(value: &str) -> String {
value
.trim()
.trim_end_matches('/')
.trim_end_matches('\\')
.trim_end_matches(".git")
.replace('\\', "/")
.to_ascii_lowercase()
}
fn format_ambiguous_repo_matches(token: &str, paths: &[PathBuf]) -> String {
let list = paths
.iter()
.map(|path| format!(" - {}", path.display()))
.collect::<Vec<_>>()
.join("\n");
format!(
"Ambiguous `--repo {token}` matches multiple inventory paths:\n{list}\n\
Use a full path or `owner/repo` (for example `xylex-group/{token}`)."
)
}
fn prepare_spool_layout(identity: &RepoIdentity) -> Result<SpoolLayout, String> {
let root = repository_spool_root(identity)?.join(sanitize_path_component(&identity.branch));
fs::create_dir_all(&root).map_err(|error| {
format!(
"Failed to create worktree mutation spool {}: {error}",
root.display()
)
})?;
let event_dedupe_dir = root.join("event-dedupe");
fs::create_dir_all(&event_dedupe_dir).map_err(|error| {
format!(
"Failed to create worktree mutation dedupe dir {}: {error}",
event_dedupe_dir.display()
)
})?;
let commit_dedupe_dir = root.join("commit-dedupe");
fs::create_dir_all(&commit_dedupe_dir).map_err(|error| {
format!(
"Failed to create worktree commit dedupe dir {}: {error}",
commit_dedupe_dir.display()
)
})?;
Ok(spool_layout_for_root(
root,
Some(Uuid::new_v4().to_string()),
))
}
fn global_mutations_root() -> Result<PathBuf, String> {
let paths = ensure_global_xbp_paths()?;
Ok(paths.root_dir.join("mutations"))
}
fn repository_spool_root(identity: &RepoIdentity) -> Result<PathBuf, String> {
Ok(global_mutations_root()?
.join(sanitize_path_component(&identity.owner))
.join(sanitize_path_component(&identity.name)))
}
fn repository_spool_search_roots(identity: &RepoIdentity) -> Result<Vec<PathBuf>, String> {
let owner = sanitize_path_component(&identity.owner);
let name = sanitize_path_component(&identity.name);
let mut roots: Vec<PathBuf> = Vec::new();
let mut push = |path: PathBuf| {
if roots.iter().any(|existing| {
path_identity_key(existing.as_path()) == path_identity_key(path.as_path())
}) {
return;
}
roots.push(path);
};
push(repository_spool_root(identity)?);
for base in legacy_mutations_base_candidates() {
push(base.join(&owner).join(&name));
let runtimes = base.join("runtimes");
if runtimes.is_dir() {
if let Ok(entries) = fs::read_dir(&runtimes) {
for entry in entries.flatten() {
if entry.file_type().map(|ft| ft.is_dir()).unwrap_or(false) {
push(entry.path().join(&owner).join(&name));
}
}
}
}
}
Ok(roots)
}
fn legacy_mutations_base_candidates() -> Vec<PathBuf> {
let mut bases: Vec<PathBuf> = Vec::new();
let mut push = |path: PathBuf| {
if bases
.iter()
.any(|existing| path_identity_key(existing.as_path()) == path_identity_key(path.as_path()))
{
return;
}
bases.push(path);
};
if let Ok(primary) = global_mutations_root() {
push(primary);
}
push(resolve_global_xbp_root_dir().join("mutations"));
if let Some(home) = dirs::home_dir() {
push(home.join(".xbp").join("mutations"));
push(home.join(".config").join("xbp").join("mutations"));
}
if let Ok(primary) = global_mutations_root() {
let rendered = primary.to_string_lossy().replace('\\', "/");
if let Some(mnt) = map_windows_path_to_wsl_mnt(&rendered) {
push(PathBuf::from(mnt));
}
if let Some(windows) = map_wsl_mnt_path_to_windows(&rendered) {
if let Some(mnt) = map_windows_path_to_wsl_mnt(&windows) {
push(PathBuf::from(mnt));
}
push(PathBuf::from(windows));
}
}
bases
}
fn spool_layout_for_existing_root(root: &Path) -> SpoolLayout {
spool_layout_for_root(root.to_path_buf(), None)
}
fn spool_layout_for_root(root: PathBuf, run_id: Option<String>) -> SpoolLayout {
let run_id = run_id.unwrap_or_else(|| Uuid::new_v4().to_string());
let event_dedupe_dir = root.join("event-dedupe");
let commit_dedupe_dir = root.join("commit-dedupe");
let watcher_state_file = root.join(format!(
"watcher-state-{}.json",
sanitize_path_component(&worktree_runtime_key())
));
SpoolLayout {
events_file: root.join(format!("events-{run_id}.jsonl")),
commits_file: root.join(format!("commits-{run_id}.jsonl")),
stats_file: root.join("stats.json"),
watcher_state_file,
sync_log_file: root.join("sync-log.jsonl"),
event_dedupe_file: root.join("event-dedupe.jsonl"),
event_dedupe_dir,
commit_dedupe_dir,
root,
}
}
fn prepare_parent_spool_layout(parent: &Path) -> Result<ParentSpoolLayout, String> {
let parent = canonical_parent_path(parent)?;
let parent_key = cross_platform_path_identity_key(&parent);
let mut hasher = Sha256::new();
hasher.update(parent_key.as_bytes());
let digest = format!("{:x}", hasher.finalize());
let label = parent
.file_name()
.and_then(|value| value.to_str())
.map(sanitize_path_component)
.filter(|value| !value.is_empty())
.unwrap_or_else(|| "parent".to_string());
let root = global_mutations_root()?
.join("parents")
.join(format!("{}-{}", label, &digest[..16]));
fs::create_dir_all(&root).map_err(|error| {
format!(
"Failed to create parent worktree watcher state dir {}: {error}",
root.display()
)
})?;
Ok(ParentSpoolLayout {
watcher_state_file: root.join(format!(
"watcher-state-{}.json",
sanitize_path_component(&worktree_runtime_key())
)),
})
}
fn canonical_parent_path(parent: &Path) -> Result<PathBuf, String> {
let normalized = normalize_windows_verbatim_path(
fs::canonicalize(parent).unwrap_or_else(|_| parent.to_path_buf()),
);
if !normalized.is_dir() {
return Err(format!(
"Parent folder {} does not exist or is not a directory.",
normalized.display()
));
}
Ok(normalized)
}
fn append_json_line<T: Serialize>(path: &Path, value: &T) -> Result<(), String> {
let mut file = OpenOptions::new()
.create(true)
.append(true)
.open(path)
.map_err(|error| format!("Failed to open spool file {}: {error}", path.display()))?;
let line = serde_json::to_string(value)
.map_err(|error| format!("Failed to serialize worktree event: {error}"))?;
writeln!(file, "{line}")
.map_err(|error| format!("Failed to append spool file {}: {error}", path.display()))
}
fn git_numstat_for_path(repo_root: &Path, relative_path: &str) -> Result<(u64, u64), String> {
let output = Command::new("git")
.args(["diff", "--numstat", "--"])
.arg(relative_path)
.current_dir(repo_root)
.output()
.map_err(|error| format!("Failed to run git diff --numstat: {error}"))?;
if !output.status.success() {
return Ok((0, 0));
}
let stdout = String::from_utf8_lossy(&output.stdout);
let mut added = 0;
let mut removed = 0;
for line in stdout.lines() {
let mut parts = line.split_whitespace();
added += parse_numstat_count(parts.next());
removed += parse_numstat_count(parts.next());
}
Ok((added, removed))
}
fn file_line_count_for_path(repo_root: &Path, relative_path: &str) -> Result<u64, String> {
let path = repo_root.join(relative_path);
let content = fs::read_to_string(&path)
.map_err(|error| format!("Failed to read file {}: {error}", path.display()))?;
Ok(content.lines().count() as u64)
}
fn parse_numstat_count(value: Option<&str>) -> u64 {
value.and_then(|raw| raw.parse::<u64>().ok()).unwrap_or(0)
}
fn current_head(repo_root: &Path) -> Result<Option<String>, String> {
match git_output(repo_root, &["rev-parse", "HEAD"]) {
Ok(value) => Ok(Some(value)),
Err(error)
if error.contains("unknown revision") || error.contains("ambiguous argument") =>
{
Ok(None)
}
Err(error) => Err(error),
}
}
fn git_output(repo_root: &Path, args: &[&str]) -> Result<String, String> {
let output = Command::new("git")
.args(args)
.current_dir(repo_root)
.output()
.map_err(|error| format!("Failed to run git {}: {error}", args.join(" ")))?;
if !output.status.success() {
return Err(String::from_utf8_lossy(&output.stderr).trim().to_string());
}
Ok(String::from_utf8_lossy(&output.stdout).trim().to_string())
}
fn parse_remote_owner_repo(remote: &str) -> Option<(String, String)> {
let trimmed = remote.trim().trim_end_matches(".git");
if trimmed.is_empty() {
return None;
}
let path_part = if !trimmed.contains("://") {
if let Some((_, path)) = trimmed.rsplit_once(':') {
path
} else {
trimmed
}
} else {
trimmed
.trim_start_matches("https://")
.trim_start_matches("http://")
.trim_start_matches("ssh://")
.split_once('/')
.map(|(_, path)| path)
.unwrap_or(trimmed)
};
let mut parts = path_part.rsplitn(2, '/');
let name = parts.next()?.trim();
let owner = parts.next()?.trim();
if owner.is_empty() || name.is_empty() {
return None;
}
Some((
sanitize_path_component(owner),
sanitize_path_component(name.trim_end_matches(".git")),
))
}
fn sanitize_path_component(value: &str) -> String {
let sanitized: String = value
.chars()
.map(|ch| {
if ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.') {
ch
} else {
'-'
}
})
.collect();
sanitized
.trim_matches('-')
.chars()
.take(160)
.collect::<String>()
}
fn normalize_windows_verbatim_path(path: PathBuf) -> PathBuf {
PathBuf::from(strip_windows_verbatim_prefix(&path.to_string_lossy()))
}
fn strip_windows_verbatim_prefix(input: &str) -> &str {
input.strip_prefix(r"\\?\").unwrap_or(input)
}
fn path_identity_key(path: &Path) -> String {
cross_platform_path_identity_key(path)
}
fn cross_platform_path_identity_key(path: &Path) -> String {
let mut value = path.to_string_lossy().replace('\\', "/");
if let Some(mnt) = map_windows_path_to_wsl_mnt(&value) {
value = mnt;
} else if let Some(windows) = map_wsl_mnt_path_to_windows(&value) {
if let Some(mnt) = map_windows_path_to_wsl_mnt(&windows) {
value = mnt;
} else {
value = windows.replace('\\', "/");
}
}
value.to_ascii_lowercase()
}
fn worktree_runtime_key() -> String {
static KEY: OnceLock<String> = OnceLock::new();
KEY.get_or_init(|| {
if is_wsl_runtime() {
let distro = std::env::var("WSL_DISTRO_NAME")
.ok()
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
.unwrap_or_else(|| "wsl".to_string());
format!("wsl-{}", sanitize_path_component(&distro).to_ascii_lowercase())
} else {
std::env::consts::OS.to_string()
}
})
.clone()
}
fn is_wsl_runtime() -> bool {
if !cfg!(target_os = "linux") {
return false;
}
if std::env::var_os("WSL_DISTRO_NAME").is_some() || std::env::var_os("WSL_INTEROP").is_some() {
return true;
}
fs::read_to_string("/proc/version")
.map(|content| {
let lower = content.to_ascii_lowercase();
lower.contains("microsoft") || lower.contains("wsl")
})
.unwrap_or(false)
}
fn worktree_device_hardware_id(base_hardware_id: &str) -> String {
format!("{base_hardware_id}@{}", worktree_runtime_key())
}
fn ensure_mutation_event_identity(mut event: WorktreeMutationEvent) -> WorktreeMutationEvent {
if event.fingerprint.as_ref().is_none_or(|value| value.is_empty()) {
let fingerprint = mutation_event_fingerprint(&event);
event.fingerprint = Some(fingerprint.clone());
if event.id.is_empty() || !event.id.starts_with("wtmut_") {
event.id = stable_event_id(&fingerprint);
}
} else if event.id.is_empty() {
event.id = stable_event_id(event.fingerprint.as_deref().unwrap_or_default());
}
event
}
fn ensure_commit_event_identity(mut commit: WorktreeCommitEvent) -> WorktreeCommitEvent {
if commit.fingerprint.as_ref().is_none_or(|value| value.is_empty()) {
let fingerprint = commit_event_fingerprint(&commit);
commit.fingerprint = Some(fingerprint.clone());
if commit.id.is_empty() || !commit.id.starts_with("wtmut_") {
commit.id = stable_event_id(&fingerprint);
}
} else if commit.id.is_empty() {
commit.id = stable_event_id(commit.fingerprint.as_deref().unwrap_or_default());
}
commit
}
fn path_string_has_ignored_component(path: &str) -> bool {
watch_ignore_rules().ignores_relative(path)
}
fn is_skipped_discovery_dir(path: &Path) -> bool {
watch_ignore_rules().skips_discovery_dir(path)
}
fn relative_watch_path(root: &Path, path: &Path) -> Option<String> {
if let Ok(relative) = path.strip_prefix(root) {
return Some(normalize_relative_watch_path(&relative.to_string_lossy()));
}
let root_norm = normalize_absolute_watch_path(root);
let path_norm = normalize_absolute_watch_path(path);
if path_norm == root_norm {
return Some(String::new());
}
let prefix = format!("{root_norm}/");
path_norm
.strip_prefix(&prefix)
.map(|relative| normalize_relative_watch_path(relative))
}
fn normalize_absolute_watch_path(path: &Path) -> String {
let mut value = path.to_string_lossy().replace('\\', "/");
if let Some(stripped) = value.strip_prefix("//?/") {
value = stripped.to_string();
}
if value.len() >= 2 {
let bytes = value.as_bytes();
if bytes[1] == b':' && bytes[0].is_ascii_alphabetic() {
let mut chars = value.chars();
let drive = chars.next().unwrap().to_ascii_lowercase();
value = format!("{drive}{}", chars.as_str());
}
}
value.trim_end_matches('/').to_string()
}
fn relative_slash_path(root: &Path, path: &Path) -> Option<String> {
relative_watch_path(root, path).or_else(|| {
Some(normalize_relative_watch_path(
&path.to_string_lossy().replace('\\', "/"),
))
})
}
fn is_ignored_path(root: &Path, path: &Path) -> bool {
watch_ignore_rules_for_root(root).ignores_path(root, path)
}
fn current_hostname() -> Option<String> {
for key in ["COMPUTERNAME", "HOSTNAME", "NAME"] {
if let Ok(value) = std::env::var(key) {
let trimmed = value.trim();
if !trimmed.is_empty() {
return Some(trimmed.to_string());
}
}
}
if let Ok(content) = fs::read_to_string("/etc/hostname") {
let trimmed = content.trim();
if !trimmed.is_empty() {
return Some(trimmed.to_string());
}
}
if let Ok(output) = Command::new("hostname").output() {
if output.status.success() {
let value = String::from_utf8_lossy(&output.stdout).trim().to_string();
if !value.is_empty() {
return Some(value);
}
}
}
None
}
fn is_background_child() -> bool {
std::env::var(BACKGROUND_CHILD_ENV)
.ok()
.map(|value| value == "1")
.unwrap_or(false)
}
#[allow(dead_code)]
fn content_sha256(path: &Path) -> Option<String> {
let bytes = fs::read(path).ok()?;
let mut hasher = Sha256::new();
hasher.update(bytes);
Some(format!("{:x}", hasher.finalize()))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn parses_https_and_ssh_remote_urls() {
assert_eq!(
parse_remote_owner_repo("https://github.com/xylex-group/xbp.git"),
Some(("xylex-group".to_string(), "xbp".to_string()))
);
assert_eq!(
parse_remote_owner_repo("git@github.com:xylex-group/xbp.git"),
Some(("xylex-group".to_string(), "xbp".to_string()))
);
}
#[test]
fn sanitizes_branch_for_storage_path() {
assert_eq!(
sanitize_path_component("feature/worktree watcher"),
"feature-worktree-watcher"
);
}
#[test]
fn normalizes_windows_verbatim_repo_roots() {
assert_eq!(
normalize_windows_verbatim_path(PathBuf::from(
r"\\?\C:\Users\floris\Documents\GitHub\mollie-api-rust"
)),
PathBuf::from(r"C:\Users\floris\Documents\GitHub\mollie-api-rust")
);
}
#[test]
fn parent_path_matching_uses_repo_boundaries() {
let repo = Path::new(r"C:\Users\floris\Documents\GitHub\xbp");
assert!(path_belongs_to_repo(
Path::new(r"C:\Users\floris\Documents\GitHub\xbp\src\main.rs"),
repo
));
assert!(!path_belongs_to_repo(
Path::new(r"C:\Users\floris\Documents\GitHub\xbp-other\src\main.rs"),
repo
));
}
#[test]
fn skips_heavy_discovery_directories() {
assert!(is_skipped_discovery_dir(Path::new("node_modules")));
assert!(is_skipped_discovery_dir(Path::new("target")));
assert!(is_skipped_discovery_dir(Path::new("dist")));
assert!(is_skipped_discovery_dir(Path::new(".next")));
assert!(is_skipped_discovery_dir(Path::new(".nx")));
assert!(is_skipped_discovery_dir(Path::new(".xbp")));
assert!(is_skipped_discovery_dir(Path::new(".cargo")));
assert!(is_skipped_discovery_dir(Path::new(".codex")));
assert!(is_skipped_discovery_dir(Path::new(".cursor")));
assert!(is_skipped_discovery_dir(Path::new(".agents")));
assert!(is_skipped_discovery_dir(Path::new(".idea")));
assert!(is_skipped_discovery_dir(Path::new(".github")));
assert!(is_skipped_discovery_dir(Path::new(".postman")));
assert!(is_skipped_discovery_dir(Path::new("mcps")));
assert!(is_skipped_discovery_dir(Path::new("agent-tools")));
assert!(is_skipped_discovery_dir(Path::new("target-publish")));
assert!(is_skipped_discovery_dir(Path::new("__pycache__")));
assert!(is_skipped_discovery_dir(Path::new("terminals")));
assert!(is_skipped_discovery_dir(Path::new(".open-next")));
assert!(!is_skipped_discovery_dir(Path::new("mollie-api-rust")));
}
#[test]
fn ignores_nested_generated_mutation_paths() {
let root = Path::new(r"C:\Users\floris\Documents\GitHub\athena");
assert!(is_ignored_path(
root,
Path::new(
r"C:\Users\floris\Documents\GitHub\athena\apps\docs\node_modules\@aws-sdk\client-lambda\dist-types\schemas"
)
));
assert!(is_ignored_path(
root,
Path::new(r"C:\Users\floris\Documents\GitHub\athena\target\debug\build")
));
assert!(is_ignored_path(
root,
Path::new(r"C:\Users\floris\Documents\GitHub\athena\.nx\cache\run.json")
));
assert!(is_ignored_path(
root,
Path::new(
r"C:\Users\floris\Documents\GitHub\athena\apps\docs\.open-next\server-functions\default\handler.mjs"
)
));
assert!(!is_ignored_path(
root,
Path::new(r"C:\Users\floris\Documents\GitHub\athena\apps\web\src\main.ts")
));
let root_slash = Path::new("C:/Users/floris/Documents/GitHub/athena");
assert!(is_ignored_path(
root_slash,
Path::new(
"C:/Users/floris/Documents/GitHub/athena/apps/docs/node_modules/@aws-sdk/client-lambda/dist-types/schemas"
)
));
assert!(path_string_has_ignored_component(
"apps/docs/node_modules/@aws-sdk/client-lambda/dist-types/schemas"
));
}
#[test]
fn global_config_forbidden_paths_folders_and_banned_words() {
let rules = WatchIgnoreRules::from_config(&WorktreeWatchConfig {
forbidden_paths: vec!["secrets/".to_string(), "apps/web/.env.local".to_string()],
forbidden_folders: vec![".cache".to_string(), "coverage".to_string()],
banned_words: vec!["private-key".to_string(), "CREDENTIAL".to_string()],
});
assert!(rules.ignores_relative("secrets/prod/api.key"));
assert!(rules.ignores_relative("apps/web/.env.local"));
assert!(!rules.ignores_relative("apps/web/.env.local.bak"));
assert!(rules.ignores_relative("apps/web/.cache/tmp"));
assert!(rules.ignores_relative("packages/foo/coverage/index.html"));
assert!(rules.ignores_relative("docs/private-key-notes.md"));
assert!(rules.ignores_relative("config/my-credential-store.json"));
assert!(!rules.ignores_relative("apps/web/src/main.ts"));
assert!(!rules.skips_discovery_dir(Path::new("mollie-api-rust")));
assert!(rules.skips_discovery_dir(Path::new(".cache")));
assert!(rules.skips_discovery_dir(Path::new("node_modules"))); }
#[test]
fn global_worktree_ignore_file_honors_mcps_patterns() {
let root = std::env::temp_dir().join(format!(
"xbp-global-watch-ignore-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.expect("time")
.as_nanos()
));
let _ = fs::remove_dir_all(&root);
fs::create_dir_all(&root).expect("global root");
let ignore_path = root.join(crate::utils::GLOBAL_WORKTREE_IGNORE_FILENAME);
fs::write(&ignore_path, "mcps/\n/custom-global/\n").expect("write global ignore");
let rules = WatchIgnoreRules {
forbidden_path_prefixes: Vec::new(),
forbidden_folders: BTreeSet::new(),
banned_words: Vec::new(),
global_ignore: crate::utils::load_global_worktree_ignore_from_path(&ignore_path),
project_ignore: crate::utils::XbpIgnoreSet::empty(),
};
assert!(rules.ignores_relative("mcps/github/catalog.json"));
assert!(rules.ignores_relative("apps/api/mcps/local"));
assert!(rules.ignores_relative("custom-global/notes.md"));
assert!(!rules.ignores_relative("apps/web/src/main.ts"));
assert!(rules.skips_discovery_dir(Path::new("mcps")));
let _ = fs::remove_dir_all(root);
}
#[test]
fn project_xbpignore_is_honored_by_worktree_watch_rules() {
let root = std::env::temp_dir().join(format!(
"xbp-watch-ignore-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.expect("time")
.as_nanos()
));
let _ = fs::remove_dir_all(&root);
fs::create_dir_all(root.join(".xbp")).expect("xbp dir");
fs::write(
root.join(".xbp").join(".xbpignore"),
"packages/icons/\n*.generated.ts\n",
)
.expect("write ignore");
fs::write(
root.join(".xbp").join("xbp.yaml"),
"project_name: demo\nversion: 0.1.0\nport: 3000\nbuild_dir: ./\nignore_paths:\n - e2e/\nwatch_ignore_paths:\n - tmp-watch/\n",
)
.expect("write config");
let rules = WatchIgnoreRules::from_config_and_project(
&WorktreeWatchConfig::default(),
Some(&root),
);
assert!(rules.ignores_relative("packages/icons/src/index.ts"));
assert!(rules.ignores_relative("apps/web/foo.generated.ts"));
assert!(!rules.ignores_relative("e2e/login.spec.ts"));
assert!(rules.ignores_relative("tmp-watch/file.ts"));
assert!(!rules.ignores_relative("apps/web/src/main.ts"));
let _ = fs::remove_dir_all(root);
}
#[test]
fn builds_coding_stats_by_file_and_filetype() {
let identity = RepoIdentity {
owner: "xylex-group".to_string(),
name: "xbp".to_string(),
branch: "main".to_string(),
root: PathBuf::from(r"C:\Users\floris\Documents\GitHub\xbp"),
head_sha: None,
};
let layout = SpoolLayout {
root: PathBuf::from(r"C:\Users\floris\.xbp\mutations\xylex-group\xbp\main"),
events_file: PathBuf::from("events.jsonl"),
commits_file: PathBuf::from("commits.jsonl"),
stats_file: PathBuf::from("stats.json"),
watcher_state_file: PathBuf::from("watcher-state.json"),
sync_log_file: PathBuf::from("sync-log.jsonl"),
event_dedupe_file: PathBuf::from("event-dedupe.jsonl"),
event_dedupe_dir: PathBuf::from("event-dedupe"),
commit_dedupe_dir: PathBuf::from("commit-dedupe"),
};
let candidates = vec![SyncCandidate {
path: PathBuf::from("events.jsonl"),
records: SyncRecords::Events(vec![
stats_test_event_record("2026-07-08T10:00:00Z", "src/main.rs", 4, 1),
stats_test_event_record("2026-07-08T10:05:00Z", "src/main.rs", 2, 0),
stats_test_event_record("2026-07-08T11:00:00Z", "README.md", 1, 3),
]),
}];
let summary = build_stats_summary(&identity, &layout, &candidates, 15).unwrap();
assert_eq!(summary.total_events, 3);
assert_eq!(summary.estimated_coding_seconds, 60 + 300 + 900);
assert_eq!(summary.session_count, 2);
assert_eq!(summary.observed_span_seconds, 60 * 60);
assert_eq!(summary.added_lines, 7);
assert_eq!(summary.removed_lines, 4);
assert_eq!(summary.by_file[0].path, "README.md");
assert_eq!(summary.by_file[0].estimated_coding_seconds, 900);
assert_eq!(summary.by_filetype[0].name, "md");
assert_eq!(summary.by_filetype[0].estimated_coding_seconds, 900);
assert_eq!(summary.by_filetype[1].name, "rs");
assert_eq!(summary.by_filetype[1].estimated_coding_seconds, 360);
}
#[test]
fn builds_repo_activity_across_branch_spools() {
let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
let main_root = temp_root.join("main");
let feature_root = temp_root.join("feature-login");
fs::create_dir_all(&main_root).unwrap();
fs::create_dir_all(&feature_root).unwrap();
append_json_line(
&main_root.join("events-main.jsonl"),
&stats_test_event_record("2026-07-08T10:00:00Z", "src/main.rs", 4, 1),
)
.unwrap();
append_json_line(
&main_root.join("events-main.jsonl"),
&stats_test_event_record("2026-07-08T10:10:00Z", "src/lib.rs", 2, 0),
)
.unwrap();
append_json_line(
&feature_root.join("events-feature.jsonl"),
&stats_test_event_record("2026-07-08T11:00:00Z", "src/auth.rs", 10, 2),
)
.unwrap();
append_json_line(
&feature_root.join("events-feature.jsonl"),
&stats_test_event_record("2026-07-08T11:30:00Z", "src/auth.rs", 3, 1),
)
.unwrap();
append_json_line(
&feature_root.join("commits-feature.jsonl"),
&worktree_commit_test_event("previous", "current"),
)
.unwrap();
let identity = RepoIdentity {
owner: "xylex-group".to_string(),
name: "xbp".to_string(),
branch: "main".to_string(),
root: PathBuf::from(r"C:\Users\floris\Documents\GitHub\xbp"),
head_sha: None,
};
let summary = build_repo_activity_summary(&identity, &temp_root, 15).unwrap();
assert_eq!(summary.branch_count, 2);
assert_eq!(summary.total_events, 4);
assert_eq!(summary.total_commits, 1);
assert_eq!(summary.session_count, 3);
assert_eq!(summary.estimated_coding_seconds, 60 + 600 + 60 + 900);
assert_eq!(summary.observed_span_seconds, 90 * 60);
assert_eq!(summary.branches[0].branch_name, "feature-login");
assert_eq!(summary.branches[0].session_count, 2);
assert_eq!(summary.branches[0].observed_span_seconds, 30 * 60);
assert_eq!(summary.branches[1].branch_name, "main");
let _ = fs::remove_dir_all(temp_root);
}
#[test]
fn mutation_fingerprint_ignores_random_event_id_but_keeps_time_bucket() {
let first: WorktreeMutationEvent = serde_json::from_value(stats_test_event(
"2026-07-08T10:00:00Z",
"src/main.rs",
4,
1,
))
.unwrap();
let same_bucket: WorktreeMutationEvent = serde_json::from_value(stats_test_event(
"2026-07-08T10:00:00Z",
"src/main.rs",
4,
1,
))
.unwrap();
let next_bucket: WorktreeMutationEvent = serde_json::from_value(stats_test_event(
"2026-07-08T10:00:06Z",
"src/main.rs",
4,
1,
))
.unwrap();
assert_eq!(
mutation_event_fingerprint(&first),
mutation_event_fingerprint(&same_bucket)
);
assert_ne!(
mutation_event_fingerprint(&first),
mutation_event_fingerprint(&next_bucket)
);
let mut windows_root = first.clone();
windows_root.repo_root = r"C:\Users\floris\Documents\GitHub\xbp".to_string();
let mut wsl_root = first.clone();
wsl_root.repo_root = "/mnt/c/Users/floris/Documents/GitHub/xbp".to_string();
assert_eq!(
mutation_event_fingerprint(&windows_root),
mutation_event_fingerprint(&wsl_root)
);
let fingerprint = mutation_event_fingerprint(&first);
assert_eq!(stable_event_id(&fingerprint), format!("wtmut_{fingerprint}"));
assert_eq!(
ensure_mutation_event_identity(first.clone()).id,
stable_event_id(&fingerprint)
);
}
#[test]
fn runtime_key_and_device_id_are_stable_and_scoped() {
let key = worktree_runtime_key();
assert!(!key.is_empty());
assert_eq!(key, worktree_runtime_key());
assert!(
key == "windows"
|| key == "linux"
|| key == "macos"
|| key.starts_with("wsl-")
|| key == "wsl",
"unexpected runtime key {key}"
);
let device = worktree_device_hardware_id("xbp_hw_test");
assert_eq!(device, format!("xbp_hw_test@{key}"));
}
#[test]
fn should_poll_only_for_wsl_mnt_paths() {
let mnt = Path::new("/mnt/c/Users/floris/Documents/GitHub/xbp");
let home = Path::new("/home/floris/src/xbp");
if is_wsl_runtime() {
assert!(should_use_poll_watcher(mnt));
assert!(!should_use_poll_watcher(home));
} else {
assert!(!should_use_poll_watcher(mnt));
assert!(!should_use_poll_watcher(home));
}
}
#[test]
fn repository_spool_root_uses_global_xbp_home_mutations() {
let identity = RepoIdentity {
owner: "xylex-group".to_string(),
name: "xbp".to_string(),
branch: "main".to_string(),
root: PathBuf::from("/tmp/xbp"),
head_sha: None,
};
let root = repository_spool_root(&identity).expect("global xbp home");
let rendered = root.to_string_lossy().replace('\\', "/");
assert!(
rendered.contains("/mutations/"),
"expected home-level mutations path, got {rendered}"
);
assert!(
!rendered.contains("/mutations/runtimes/"),
"spool must not be runtime-split when sharing one established home: {rendered}"
);
assert!(
rendered.ends_with("xylex-group/xbp") || rendered.contains("xylex-group/xbp"),
"expected owner/name suffix, got {rendered}"
);
assert!(!rendered.contains("/tmp/xbp/"));
}
#[test]
fn cross_platform_path_identity_reconciles_wsl_and_windows() {
let windows = Path::new(r"C:\Users\szymon\.xbp\mutations\SuitsFinance\speedrun-formations\main");
let wsl = Path::new(
"/mnt/c/Users/szymon/.xbp/mutations/SuitsFinance/speedrun-formations/main",
);
assert_eq!(
cross_platform_path_identity_key(windows),
cross_platform_path_identity_key(wsl)
);
}
#[test]
fn command_line_mentions_path_is_case_and_slash_insensitive() {
let cmdline = r#""C:\xbp.exe" worktree-watch start --parent C:\Users\floris\Documents\GitHub"#;
assert!(command_line_mentions_path(
cmdline,
"/mnt/c/users/floris/documents/github"
));
assert!(command_line_mentions_path(
cmdline,
r"C:\Users\floris\Documents\GitHub"
));
assert!(command_line_mentions_path(cmdline, "GitHub"));
}
#[test]
fn append_unique_mutation_event_is_idempotent_across_dedupe_instances() {
let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
fs::create_dir_all(&temp_root).unwrap();
let layout = SpoolLayout {
root: temp_root.clone(),
events_file: temp_root.join("events.jsonl"),
commits_file: temp_root.join("commits.jsonl"),
stats_file: temp_root.join("stats.json"),
watcher_state_file: temp_root.join("watcher-state.json"),
sync_log_file: temp_root.join("sync-log.jsonl"),
event_dedupe_file: temp_root.join("event-dedupe.jsonl"),
event_dedupe_dir: temp_root.join("event-dedupe"),
commit_dedupe_dir: temp_root.join("commit-dedupe"),
};
let mutation: WorktreeMutationEvent = serde_json::from_value(stats_test_event(
"2026-07-08T10:00:00Z",
"src/main.rs",
4,
1,
))
.unwrap();
let first =
append_unique_mutation_event(&layout, &mutation, &mut RecentEventDedupe::default())
.unwrap();
let second =
append_unique_mutation_event(&layout, &mutation, &mut RecentEventDedupe::default())
.unwrap();
assert!(first);
assert!(!second);
let events = read_jsonl_values(&layout.events_file).unwrap();
assert_eq!(events.len(), 1);
assert_eq!(
events[0].get("fingerprint").and_then(Value::as_str),
Some(mutation_event_fingerprint(&mutation).as_str())
);
let _ = fs::remove_dir_all(temp_root);
}
#[test]
fn reads_spool_records_as_typed_events() {
let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
fs::create_dir_all(&temp_root).unwrap();
let events_file = temp_root.join("events.jsonl");
append_json_line(
&events_file,
&stats_test_event_record("2026-07-08T10:00:00Z", "src/main.rs", 4, 1),
)
.unwrap();
let records = read_spool_records(&events_file, SyncFileKind::Events).unwrap();
match records {
SyncRecords::Events(events) => {
assert_eq!(events.len(), 1);
assert_eq!(events[0].primary_path.as_deref(), Some("src/main.rs"));
}
SyncRecords::Commits(_) => panic!("expected typed event records"),
}
let _ = fs::remove_dir_all(temp_root);
}
#[test]
fn read_spool_records_skips_nul_only_padding() {
let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
fs::create_dir_all(&temp_root).unwrap();
let events_file = temp_root.join("events.jsonl");
fs::write(&events_file, vec![0; 4096]).unwrap();
let records = read_spool_records(&events_file, SyncFileKind::Events).unwrap();
match records {
SyncRecords::Events(events) => assert!(events.is_empty()),
SyncRecords::Commits(_) => panic!("expected event records"),
}
let _ = fs::remove_dir_all(temp_root);
}
#[test]
fn read_jsonl_strips_leading_nuls_before_json() {
let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
fs::create_dir_all(&temp_root).unwrap();
let path = temp_root.join("event-dedupe.jsonl");
let mut bytes = vec![0u8; 64];
bytes.extend_from_slice(
br#"{"fingerprint":"abc","firstSeenAt":"2026-07-09T00:00:00Z","eventKind":"modify","primaryPath":".xbp"}"#,
);
bytes.push(b'\n');
fs::write(&path, bytes).unwrap();
let values = read_jsonl_values(&path).unwrap();
assert_eq!(values.len(), 1);
assert_eq!(
values[0].get("fingerprint").and_then(Value::as_str),
Some("abc")
);
let _ = fs::remove_dir_all(temp_root);
}
#[test]
fn read_jsonl_skips_garbage_lines() {
let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
fs::create_dir_all(&temp_root).unwrap();
let path = temp_root.join("mixed.jsonl");
fs::write(
&path,
"not-json\n{\"fingerprint\":\"ok\"}\n\n{\"broken\":\n",
)
.unwrap();
let values = read_jsonl_values(&path).unwrap();
assert_eq!(values.len(), 1);
assert_eq!(
values[0].get("fingerprint").and_then(Value::as_str),
Some("ok")
);
let _ = fs::remove_dir_all(temp_root);
}
#[test]
fn collect_spool_candidates_rewrites_legacy_generated_events() {
let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
fs::create_dir_all(&temp_root).unwrap();
let events_file = temp_root.join("events-legacy.jsonl");
append_json_line(
&events_file,
&stats_test_event_record(
"2026-07-08T10:00:00Z",
"apps/docs/node_modules/@aws-sdk/client-lambda/dist-types/schemas",
0,
0,
),
)
.unwrap();
append_json_line(
&events_file,
&stats_test_event_record("2026-07-08T10:01:00Z", "apps/web/src/main.ts", 4, 1),
)
.unwrap();
let layout = SpoolLayout {
root: temp_root.clone(),
events_file: temp_root.join("events.jsonl"),
commits_file: temp_root.join("commits.jsonl"),
stats_file: temp_root.join("stats.json"),
watcher_state_file: temp_root.join("watcher-state.json"),
sync_log_file: temp_root.join("sync-log.jsonl"),
event_dedupe_file: temp_root.join("event-dedupe.jsonl"),
event_dedupe_dir: temp_root.join("event-dedupe"),
commit_dedupe_dir: temp_root.join("commit-dedupe"),
};
let candidates = collect_spool_candidates(&layout, false).unwrap();
assert_eq!(candidates.len(), 1);
match &candidates[0].records {
SyncRecords::Events(events) => {
assert_eq!(events.len(), 1);
assert_eq!(
events[0].primary_path.as_deref(),
Some("apps/web/src/main.ts")
);
}
SyncRecords::Commits(_) => panic!("expected event spool candidate"),
}
let rewritten = read_jsonl_records::<WorktreeMutationEvent>(&events_file).unwrap();
assert_eq!(rewritten.len(), 1);
assert_eq!(
rewritten[0].primary_path.as_deref(),
Some("apps/web/src/main.ts")
);
let _ = fs::remove_dir_all(temp_root);
}
#[test]
fn collect_spool_candidates_deletes_fully_ignored_event_files() {
let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
fs::create_dir_all(&temp_root).unwrap();
let events_file = temp_root.join("events-legacy.jsonl");
append_json_line(
&events_file,
&stats_test_event_record(
"2026-07-08T10:00:00Z",
"apps/docs/node_modules/@aws-sdk/client-lambda/dist-types/schemas",
0,
0,
),
)
.unwrap();
let layout = SpoolLayout {
root: temp_root.clone(),
events_file: temp_root.join("events.jsonl"),
commits_file: temp_root.join("commits.jsonl"),
stats_file: temp_root.join("stats.json"),
watcher_state_file: temp_root.join("watcher-state.json"),
sync_log_file: temp_root.join("sync-log.jsonl"),
event_dedupe_file: temp_root.join("event-dedupe.jsonl"),
event_dedupe_dir: temp_root.join("event-dedupe"),
commit_dedupe_dir: temp_root.join("commit-dedupe"),
};
let candidates = collect_spool_candidates(&layout, false).unwrap();
assert!(candidates.is_empty());
assert!(!events_file.exists());
let _ = fs::remove_dir_all(temp_root);
}
#[test]
fn append_unique_commit_event_is_idempotent_across_watcher_instances() {
let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
fs::create_dir_all(&temp_root).unwrap();
let layout = SpoolLayout {
root: temp_root.clone(),
events_file: temp_root.join("events.jsonl"),
commits_file: temp_root.join("commits.jsonl"),
stats_file: temp_root.join("stats.json"),
watcher_state_file: temp_root.join("watcher-state.json"),
sync_log_file: temp_root.join("sync-log.jsonl"),
event_dedupe_file: temp_root.join("event-dedupe.jsonl"),
event_dedupe_dir: temp_root.join("event-dedupe"),
commit_dedupe_dir: temp_root.join("commit-dedupe"),
};
let commit = WorktreeCommitEvent {
id: Uuid::new_v4().to_string(),
fingerprint: None,
repo_owner: "xylex-group".to_string(),
repo_name: "xbp".to_string(),
branch_name: "main".to_string(),
repo_root: r"C:\Users\floris\Documents\GitHub\xbp".to_string(),
previous_head_sha: Some("previous".to_string()),
head_sha: "current".to_string(),
subject: Some("test commit".to_string()),
author_name: Some("Floris".to_string()),
author_email: Some("floris@xylex.group".to_string()),
committed_at: Some("2026-07-08T10:00:00Z".to_string()),
occurred_at: Utc::now(),
};
assert!(append_unique_commit_event(&layout, &commit).unwrap());
assert!(!append_unique_commit_event(&layout, &commit).unwrap());
let commits = read_jsonl_values(&layout.commits_file).unwrap();
assert_eq!(commits.len(), 1);
assert_eq!(
commits[0].get("fingerprint").and_then(Value::as_str),
Some(commit_event_fingerprint(&commit).as_str())
);
let _ = fs::remove_dir_all(temp_root);
}
#[test]
fn splits_worktree_mutation_uploads_into_bounded_batches() {
let events = (0..(WORKTREE_MUTATION_SYNC_BATCH_RECORD_LIMIT + 1))
.map(|index| {
serde_json::from_value::<WorktreeMutationEvent>(stats_test_event(
"2026-07-08T10:00:00Z",
&format!("src/file-{index}.rs"),
1,
0,
))
.unwrap()
})
.collect::<Vec<_>>();
let commits = vec![
worktree_commit_test_event("previous-a", "current-a"),
worktree_commit_test_event("previous-b", "current-b"),
];
let batches = split_worktree_mutation_upload_batches(events, commits);
assert_eq!(batches.len(), 2);
assert_eq!(
batches[0].events.len() + batches[0].commits.len(),
WORKTREE_MUTATION_SYNC_BATCH_RECORD_LIMIT
);
assert_eq!(batches[1].events.len(), 1);
assert_eq!(batches[1].commits.len(), 2);
}
#[test]
fn splits_worktree_mutation_uploads_by_estimated_json_bytes() {
let long_path = format!("src/{}.rs", "x".repeat(8_000));
let events = (0..200)
.map(|index| {
serde_json::from_value::<WorktreeMutationEvent>(stats_test_event(
"2026-07-08T10:00:00Z",
&format!("{long_path}-{index}"),
1,
0,
))
.unwrap()
})
.collect::<Vec<_>>();
let batches = split_worktree_mutation_upload_batches(events, Vec::new());
assert!(batches.len() > 1);
for batch in batches {
assert!(
batch.estimated_json_bytes <= WORKTREE_MUTATION_SYNC_BATCH_JSON_BYTES,
"batch estimated JSON bytes exceeded limit: {}",
batch.estimated_json_bytes
);
}
}
#[test]
fn classifies_create_remove_and_rename_events() {
assert_eq!(
classify_event_kind(&EventKind::Create(CreateKind::File)),
"file_create"
);
assert_eq!(
classify_event_kind(&EventKind::Remove(RemoveKind::Folder)),
"folder_remove"
);
assert_eq!(
classify_event_kind(&EventKind::Modify(ModifyKind::Name(RenameMode::Both))),
"rename_or_move"
);
}
#[test]
fn retries_only_retryable_worktree_sync_statuses() {
assert!(should_retry_worktree_mutation_sync_status(
StatusCode::INTERNAL_SERVER_ERROR
));
assert!(should_retry_worktree_mutation_sync_status(
StatusCode::REQUEST_TIMEOUT
));
assert!(should_retry_worktree_mutation_sync_status(
StatusCode::TOO_MANY_REQUESTS
));
assert!(!should_retry_worktree_mutation_sync_status(
StatusCode::BAD_REQUEST
));
assert!(!should_retry_worktree_mutation_sync_status(
StatusCode::UNAUTHORIZED
));
assert!(!should_retry_worktree_mutation_sync_status(
StatusCode::PAYLOAD_TOO_LARGE
));
}
#[test]
fn match_github_repo_inventory_resolves_short_name_and_owner_repo() {
let temp = std::env::temp_dir().join(format!("xbp-repo-lookup-{}", Uuid::new_v4()));
let athena = temp.join("athena");
let other = temp.join("other-athena");
fs::create_dir_all(&athena).unwrap();
fs::create_dir_all(&other).unwrap();
let repos = vec![
GithubRepoRecord {
owner: "xylex-group".to_string(),
repo: "athena".to_string(),
full_name: "xylex-group/athena".to_string(),
path: athena.to_string_lossy().to_string(),
remote_url: "https://github.com/xylex-group/athena.git".to_string(),
},
GithubRepoRecord {
owner: "someone".to_string(),
repo: "other".to_string(),
full_name: "someone/other".to_string(),
path: other.to_string_lossy().to_string(),
remote_url: "https://github.com/someone/other.git".to_string(),
},
];
let by_short = match_github_repo_inventory(&repos, "athena").unwrap().unwrap();
assert_eq!(
path_identity_key(&by_short),
path_identity_key(athena.as_path())
);
let by_full = match_github_repo_inventory(&repos, "xylex-group/athena")
.unwrap()
.unwrap();
assert_eq!(
path_identity_key(&by_full),
path_identity_key(athena.as_path())
);
assert!(match_github_repo_inventory(&repos, "missing")
.unwrap()
.is_none());
let _ = fs::remove_dir_all(temp);
}
#[test]
fn match_github_repo_inventory_errors_on_ambiguous_short_name() {
let temp = std::env::temp_dir().join(format!("xbp-repo-lookup-ambig-{}", Uuid::new_v4()));
let a = temp.join("a");
let b = temp.join("b");
fs::create_dir_all(&a).unwrap();
fs::create_dir_all(&b).unwrap();
let repos = vec![
GithubRepoRecord {
owner: "one".to_string(),
repo: "dup".to_string(),
full_name: "one/dup".to_string(),
path: a.to_string_lossy().to_string(),
remote_url: "https://github.com/one/dup.git".to_string(),
},
GithubRepoRecord {
owner: "two".to_string(),
repo: "dup".to_string(),
full_name: "two/dup".to_string(),
path: b.to_string_lossy().to_string(),
remote_url: "https://github.com/two/dup.git".to_string(),
},
];
let err = match_github_repo_inventory(&repos, "dup").unwrap_err();
assert!(err.contains("Ambiguous"));
assert!(err.contains("one/dup") || err.contains(a.to_string_lossy().as_ref()));
let unique = match_github_repo_inventory(&repos, "two/dup").unwrap().unwrap();
assert_eq!(path_identity_key(&unique), path_identity_key(b.as_path()));
let _ = fs::remove_dir_all(temp);
}
fn stats_test_event(
occurred_at: &str,
path: &str,
added_lines: u64,
removed_lines: u64,
) -> Value {
json!({
"id": Uuid::new_v4().to_string(),
"repoOwner": "xylex-group",
"repoName": "xbp",
"branchName": "main",
"repoRoot": r"C:\Users\floris\Documents\GitHub\xbp",
"headSha": null,
"eventKind": "file_modify",
"paths": [path],
"primaryPath": path,
"oldPath": null,
"newPath": null,
"addedLines": added_lines,
"removedLines": removed_lines,
"totalLines": added_lines + 42,
"fileCreated": false,
"fileRemoved": false,
"folderCreated": false,
"folderRemoved": false,
"renamedOrMoved": false,
"rawKind": "Modify(Data(Content))",
"occurredAt": occurred_at,
})
}
fn stats_test_event_record(
occurred_at: &str,
path: &str,
added_lines: u64,
removed_lines: u64,
) -> WorktreeMutationEvent {
serde_json::from_value(stats_test_event(
occurred_at,
path,
added_lines,
removed_lines,
))
.unwrap()
}
fn worktree_commit_test_event(previous_head_sha: &str, head_sha: &str) -> WorktreeCommitEvent {
WorktreeCommitEvent {
id: Uuid::new_v4().to_string(),
fingerprint: None,
repo_owner: "xylex-group".to_string(),
repo_name: "xbp".to_string(),
branch_name: "main".to_string(),
repo_root: r"C:\Users\floris\Documents\GitHub\xbp".to_string(),
previous_head_sha: Some(previous_head_sha.to_string()),
head_sha: head_sha.to_string(),
subject: Some("test commit".to_string()),
author_name: Some("Floris".to_string()),
author_email: Some("floris@xylex.group".to_string()),
committed_at: Some("2026-07-08T10:00:00Z".to_string()),
occurred_at: Utc::now(),
}
}
}