use crate::commands::cli_session::{cli_request_client, resolve_cli_access_token};
use crate::config::{resolve_device_identity, ApiConfig};
use chrono::{DateTime, Utc};
use notify::event::{CreateKind, ModifyKind, RemoveKind, RenameMode};
use notify::{Config, Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
use reqwest::StatusCode;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, BTreeSet};
use std::fs::{self, File, OpenOptions};
use std::io::{BufRead, BufReader, Write};
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::sync::mpsc;
use std::time::{Duration, Instant};
use sysinfo::{Pid, System};
use uuid::Uuid;
const BACKGROUND_CHILD_ENV: &str = "XBP_WORKTREE_WATCH_BACKGROUND_CHILD";
const DEFAULT_REMOTE: &str = "origin";
const EVENT_DEDUPE_WINDOW_SECONDS: i64 = 5;
#[cfg(windows)]
const CREATE_NO_WINDOW: u32 = 0x08000000;
#[derive(Debug, Clone)]
pub struct WorktreeWatchTargetOptions {
pub repo: Option<PathBuf>,
pub parent: Option<PathBuf>,
}
#[derive(Debug, Clone)]
pub struct WorktreeWatchStartOptions {
pub target: WorktreeWatchTargetOptions,
pub detach: bool,
pub sync_interval_seconds: u64,
pub once: bool,
}
#[derive(Debug, Clone)]
pub struct WorktreeWatchSyncOptions {
pub target: WorktreeWatchTargetOptions,
pub dry_run: bool,
pub resync: bool,
}
#[derive(Debug, Clone)]
pub struct WorktreeWatchStatusOptions {
pub target: WorktreeWatchTargetOptions,
pub json: bool,
pub records: bool,
pub record_limit: usize,
pub stats: 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,
}
#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
struct WorktreeMutationEvent {
id: 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>,
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,
repo_owner: String,
repo_name: String,
branch_name: String,
repo_root: String,
previous_head_sha: Option<String>,
head_sha: String,
subject: Option<String>,
author_name: Option<String>,
author_email: Option<String>,
committed_at: Option<String>,
occurred_at: DateTime<Utc>,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct WorktreeMutationIngestPayload {
device: WorktreeMutationDevicePayload,
repository: WorktreeMutationRepositoryPayload,
events: Vec<WorktreeMutationEvent>,
commits: Vec<WorktreeCommitEvent>,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct WorktreeMutationDevicePayload {
hardware_id: String,
device_name: Option<String>,
hostname: Option<String>,
platform: String,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct WorktreeMutationRepositoryPayload {
owner: String,
name: String,
branch_name: String,
repo_root: String,
}
#[derive(Debug)]
struct SyncCandidate {
path: PathBuf,
kind: SyncFileKind,
records: Vec<Value>,
}
#[derive(Debug, Clone, Copy)]
enum SyncFileKind {
Events,
Commits,
}
#[derive(Debug, Serialize, Clone)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatchStatsSummary {
generated_at: DateTime<Utc>,
repository_owner: String,
repository_name: String,
branch_name: String,
repo_root: String,
spool_root: String,
session_gap_minutes: u64,
total_events: u64,
total_commits: u64,
first_event_at: Option<DateTime<Utc>>,
last_event_at: Option<DateTime<Utc>>,
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, Default)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatchFileStats {
path: String,
filetype: String,
event_count: u64,
estimated_coding_seconds: u64,
added_lines: u64,
removed_lines: u64,
}
#[derive(Debug, Serialize, Clone, Default)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatchBucketStats {
name: String,
event_count: u64,
estimated_coding_seconds: u64,
added_lines: u64,
removed_lines: u64,
}
#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatcherState {
pid: u32,
repo_root: String,
repository_owner: String,
repository_name: String,
branch_name: String,
executable: String,
started_at: DateTime<Utc>,
}
#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
struct 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, 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.target.parent.is_some() {
let identities = resolve_target_identities(&options.target)?;
if identities.is_empty() {
println!("No git repositories found under parent folder.");
return Ok(());
}
if options.detach {
for identity in &identities {
let path = spawn_detached_worktree_watch_for_identity(identity)?;
println!("Started background worktree watcher for {}", path.display());
}
println!(
"Started {} background worktree watcher(s).",
identities.len()
);
return Ok(());
}
if options.once {
for identity in &identities {
let layout = prepare_spool_layout(identity)?;
record_commit_snapshot(identity, &layout, None)?;
println!("Recorded current git state for {}", identity.root.display());
}
return Ok(());
}
return Err(
"Parent-folder watching starts one watcher per repository. Pass `--detach` to run them in the background, or `--once` to snapshot each repo."
.to_string(),
);
}
if options.detach {
return spawn_detached_worktree_watch(options.target.repo.as_deref()).map(|path| {
println!("Started background worktree watcher for {}", path.display());
});
}
let identity = resolve_repo_identity(options.target.repo.as_deref())?;
let layout = prepare_spool_layout(&identity)?;
println!(
"Watching {} and spooling mutations to {}",
identity.root.display(),
layout.root.display()
);
if options.once {
record_commit_snapshot(&identity, &layout, None)?;
return Ok(());
}
watch_foreground(identity, layout, options.sync_interval_seconds).await
}
pub async fn run_worktree_watch_sync(options: WorktreeWatchSyncOptions) -> Result<(), String> {
let identities = resolve_target_identities(&options.target)?;
let repo_count = identities.len();
let mut plans = Vec::new();
let mut total_files = 0usize;
let mut total_records = 0usize;
for identity in identities {
let layout = prepare_spool_layout(&identity)?;
let candidates = collect_spool_candidates(&layout, options.resync)?;
total_files += candidates.len();
total_records += candidates
.iter()
.map(|candidate| candidate.records.len())
.sum::<usize>();
if !candidates.is_empty() {
plans.push((identity, candidates));
}
}
if options.dry_run {
println!(
"Would sync {} record(s) from {} spool file(s) across {} repo(s).",
total_records, total_files, repo_count
);
if options.resync {
println!("Resync mode includes files already marked `.synced.*.jsonl`.");
}
return Ok(());
}
if plans.is_empty() {
print_no_sync_candidates_message(&options.target, options.resync)?;
return Ok(());
}
let upload_repo_count = plans.len();
for (identity, candidates) in plans {
sync_candidates(&identity, candidates, options.resync).await?;
}
println!(
"Synced {} worktree mutation record(s) across {} repo(s).",
total_records, upload_repo_count
);
Ok(())
}
pub async fn run_worktree_watch_status(options: WorktreeWatchStatusOptions) -> Result<(), String> {
let identities = resolve_target_identities(&options.target)?;
let mut payloads = Vec::new();
let mut total_files = 0usize;
let mut total_records = 0usize;
for identity in identities {
let status =
worktree_watch_status_payload(&identity, options.records, options.record_limit)?;
let stats = if options.stats {
Some(generate_and_store_stats(
&identity,
options.stats_gap_minutes,
)?)
} else {
None
};
let status = attach_optional_stats(status, stats)?;
total_files += status
.get("unsyncedFiles")
.and_then(Value::as_u64)
.unwrap_or(0) as usize;
total_records += status
.get("unsyncedRecords")
.and_then(Value::as_u64)
.unwrap_or(0) as usize;
payloads.push(status);
}
if options.json {
let payload = if options.target.parent.is_some() {
json!({
"repositories": payloads,
"repositoryCount": payloads.len(),
"unsyncedFiles": total_files,
"unsyncedRecords": total_records,
})
} else {
payloads.into_iter().next().unwrap_or_else(|| json!({}))
};
println!(
"{}",
serde_json::to_string_pretty(&payload)
.map_err(|error| format!("Failed to render status JSON: {error}"))?
);
} else {
for (index, payload) in payloads.iter().enumerate() {
if index > 0 {
println!();
}
println!(
"repo: {}/{}",
payload
.get("repositoryOwner")
.and_then(Value::as_str)
.unwrap_or("unknown"),
payload
.get("repositoryName")
.and_then(Value::as_str)
.unwrap_or("repository")
);
println!(
"branch: {}",
payload
.get("branchName")
.and_then(Value::as_str)
.unwrap_or("unknown")
);
println!(
"root: {}",
payload
.get("repoRoot")
.and_then(Value::as_str)
.unwrap_or("")
);
println!(
"spool: {}",
payload
.get("spoolRoot")
.and_then(Value::as_str)
.unwrap_or("")
);
println!(
"unsynced files: {}",
payload
.get("unsyncedFiles")
.and_then(Value::as_u64)
.unwrap_or(0)
);
println!(
"unsynced records: {}",
payload
.get("unsyncedRecords")
.and_then(Value::as_u64)
.unwrap_or(0)
);
println!(
"synced files: {}",
payload
.get("syncedFiles")
.and_then(Value::as_u64)
.unwrap_or(0)
);
println!(
"synced records: {}",
payload
.get("syncedRecords")
.and_then(Value::as_u64)
.unwrap_or(0)
);
print_last_sync(payload);
if options.records {
print_status_records(payload);
}
if options.stats {
print_status_stats(payload);
}
}
if options.target.parent.is_some() {
println!();
println!(
"total: {} repo(s), {} unsynced file(s), {} unsynced record(s)",
payloads.len(),
total_files,
total_records
);
}
}
Ok(())
}
fn attach_optional_stats(
mut payload: Value,
stats: Option<WorktreeWatchStatsSummary>,
) -> Result<Value, String> {
if let Some(stats) = stats {
if let Some(object) = payload.as_object_mut() {
object.insert(
"stats".to_string(),
serde_json::to_value(stats)
.map_err(|error| format!("Failed to render stats payload: {error}"))?,
);
}
}
Ok(payload)
}
pub fn spawn_detached_worktree_watch_for_current_repo() -> Result<Option<PathBuf>, String> {
if is_background_child() {
return Ok(None);
}
let identity = match resolve_repo_identity(None) {
Ok(identity) => identity,
Err(_) => return Ok(None),
};
spawn_detached_worktree_watch_for_identity(&identity).map(Some)
}
fn resolve_target_identities(
target: &WorktreeWatchTargetOptions,
) -> Result<Vec<RepoIdentity>, String> {
if target.repo.is_some() && target.parent.is_some() {
return Err("Pass either `--repo` or `--parent`, not both.".to_string());
}
if let Some(parent) = target.parent.as_deref() {
return discover_repo_identities_under(parent);
}
resolve_repo_identity(target.repo.as_deref()).map(|identity| vec![identity])
}
fn worktree_watch_status_payload(
identity: &RepoIdentity,
include_records: bool,
record_limit: usize,
) -> Result<Value, String> {
let layout = prepare_spool_layout(identity)?;
let candidates = collect_sync_candidates(&layout)?;
let all_candidates = collect_spool_candidates(&layout, true)?;
let record_count: usize = candidates
.iter()
.map(|candidate| candidate.records.len())
.sum();
let all_record_count: usize = all_candidates
.iter()
.map(|candidate| candidate.records.len())
.sum();
let synced_file_count = all_candidates.len().saturating_sub(candidates.len());
let synced_record_count = all_record_count.saturating_sub(record_count);
let mut payload = json!({
"repoRoot": identity.root.display().to_string(),
"repositoryOwner": identity.owner.clone(),
"repositoryName": identity.name.clone(),
"branchName": identity.branch.clone(),
"spoolRoot": layout.root.display().to_string(),
"unsyncedFiles": candidates.len(),
"unsyncedRecords": record_count,
"syncedFiles": synced_file_count,
"syncedRecords": synced_record_count,
"localSpoolFiles": all_candidates.len(),
"localSpoolRecords": all_record_count,
});
if let Some(last_sync) = read_last_sync_log_entry(&layout.sync_log_file)? {
if let Some(object) = payload.as_object_mut() {
object.insert(
"lastSync".to_string(),
serde_json::to_value(last_sync)
.map_err(|error| format!("Failed to render sync log payload: {error}"))?,
);
}
}
if include_records {
if let Some(object) = payload.as_object_mut() {
object.insert(
"records".to_string(),
collect_status_records(&candidates, record_limit)?,
);
}
}
Ok(payload)
}
fn collect_status_records(candidates: &[SyncCandidate], limit: usize) -> Result<Value, String> {
let available: usize = candidates
.iter()
.map(|candidate| candidate.records.len())
.sum();
let mut shown = 0usize;
let mut events = Vec::new();
let mut commits = Vec::new();
'candidates: for candidate in candidates {
for record in &candidate.records {
if shown >= limit {
break 'candidates;
}
match candidate.kind {
SyncFileKind::Events => {
let event = serde_json::from_value::<WorktreeMutationEvent>(record.clone())
.map_err(|error| {
format!(
"Failed to decode event from {}: {error}",
candidate.path.display()
)
})?;
events.push(json!({
"sourceFile": candidate.path.display().to_string(),
"occurredAt": event.occurred_at,
"eventKind": event.event_kind,
"primaryPath": event.primary_path,
"oldPath": event.old_path,
"newPath": event.new_path,
"paths": event.paths,
"addedLines": event.added_lines,
"removedLines": event.removed_lines,
"headSha": event.head_sha,
"rawKind": event.raw_kind,
}));
}
SyncFileKind::Commits => {
let commit = serde_json::from_value::<WorktreeCommitEvent>(record.clone())
.map_err(|error| {
format!(
"Failed to decode commit from {}: {error}",
candidate.path.display()
)
})?;
commits.push(json!({
"sourceFile": candidate.path.display().to_string(),
"occurredAt": commit.occurred_at,
"headSha": commit.head_sha,
"previousHeadSha": commit.previous_head_sha,
"subject": commit.subject,
"authorName": commit.author_name,
"authorEmail": commit.author_email,
"committedAt": commit.committed_at,
}));
}
}
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!(
" 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_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);
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 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.kind {
SyncFileKind::Events => {
for record in &candidate.records {
events.push(
serde_json::from_value::<WorktreeMutationEvent>(record.clone()).map_err(
|error| {
format!(
"Failed to decode event from {}: {error}",
candidate.path.display()
)
},
)?,
);
}
}
SyncFileKind::Commits => {
total_commits += candidate.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 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 {
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),
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 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")
}
}
fn watch_foreground(
identity: RepoIdentity,
layout: SpoolLayout,
sync_interval_seconds: u64,
) -> impl std::future::Future<Output = Result<(), String>> {
async move {
let (tx, rx) = mpsc::channel();
let mut watcher = RecommendedWatcher::new(
move |result| {
let _ = tx.send(result);
},
Config::default(),
)
.map_err(|error| format!("Failed to create filesystem watcher: {error}"))?;
watcher
.watch(&identity.root, RecursiveMode::Recursive)
.map_err(|error| {
format!(
"Failed to watch repository root {}: {error}",
identity.root.display()
)
})?;
let mut last_head = identity.head_sha.clone();
let mut last_commit_check = Instant::now();
let mut last_sync = Instant::now();
let mut event_dedupe = RecentEventDedupe {
fingerprints: load_event_dedupe_fingerprints(&layout.event_dedupe_file)?,
};
loop {
match rx.recv_timeout(Duration::from_millis(750)) {
Ok(Ok(event)) => {
if let Some(mutation) = mutation_event_from_notify(&identity, event) {
append_unique_mutation_event(&layout, &mutation, &mut event_dedupe)?;
}
}
Ok(Err(error)) => {
eprintln!("worktree watcher error: {error}");
}
Err(mpsc::RecvTimeoutError::Timeout) => {}
Err(mpsc::RecvTimeoutError::Disconnected) => {
return Err("Filesystem watcher disconnected.".to_string());
}
}
if last_commit_check.elapsed() >= Duration::from_secs(3) {
last_head = record_commit_snapshot(&identity, &layout, last_head.as_deref())?;
last_commit_check = Instant::now();
}
if sync_interval_seconds > 0
&& last_sync.elapsed() >= Duration::from_secs(sync_interval_seconds)
{
if let Err(error) = sync_spool_for_identity(&identity, &layout).await {
eprintln!("worktree mutation sync failed: {error}");
}
last_sync = Instant::now();
}
}
}
}
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))
.collect();
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 event_kind = classify_event_kind(&event.kind);
let is_dir = event
.paths
.first()
.and_then(|path| fs::metadata(path).ok())
.map(|metadata| metadata.is_dir())
.unwrap_or(false);
Some(WorktreeMutationEvent {
id: Uuid::new_v4().to_string(),
repo_owner: identity.owner.clone(),
repo_name: identity.name.clone(),
branch_name: identity.branch.clone(),
repo_root: identity.root.display().to_string(),
head_sha: current_head(&identity.root).ok().flatten(),
event_kind: event_kind.to_string(),
paths,
primary_path,
old_path,
new_path,
added_lines: line_counts.map(|counts| counts.0),
removed_lines: line_counts.map(|counts| counts.1),
file_created: matches!(
event.kind,
EventKind::Create(CreateKind::File | CreateKind::Any)
) && !is_dir,
file_removed: matches!(
event.kind,
EventKind::Remove(RemoveKind::File | RemoveKind::Any)
) && !is_dir,
folder_created: matches!(
event.kind,
EventKind::Create(CreateKind::Folder | CreateKind::Any)
) && is_dir,
folder_removed: matches!(
event.kind,
EventKind::Remove(RemoveKind::Folder | RemoveKind::Any)
) && is_dir,
renamed_or_moved: is_rename_or_move(&event.kind),
raw_kind: format!("{:?}", event.kind),
occurred_at: Utc::now(),
})
}
fn append_unique_mutation_event(
layout: &SpoolLayout,
mutation: &WorktreeMutationEvent,
dedupe: &mut RecentEventDedupe,
) -> Result<bool, String> {
let fingerprint = mutation_event_fingerprint(mutation);
if dedupe.fingerprints.contains(&fingerprint)
|| event_dedupe_file_contains(&layout.event_dedupe_file, &fingerprint)?
{
dedupe.fingerprints.insert(fingerprint);
return Ok(false);
}
append_json_line(&layout.events_file, mutation)?;
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 mutation_event_fingerprint(mutation: &WorktreeMutationEvent) -> String {
let bucket = mutation.occurred_at.timestamp() / EVENT_DEDUPE_WINDOW_SECONDS;
let payload = json!({
"repoOwner": mutation.repo_owner,
"repoName": mutation.repo_name,
"branchName": mutation.branch_name,
"repoRoot": mutation.repo_root,
"headSha": mutation.head_sha,
"eventKind": mutation.event_kind,
"paths": mutation.paths,
"primaryPath": mutation.primary_path,
"oldPath": mutation.old_path,
"newPath": mutation.new_path,
"addedLines": mutation.added_lines,
"removedLines": mutation.removed_lines,
"fileCreated": mutation.file_created,
"fileRemoved": mutation.file_removed,
"folderCreated": mutation.folder_created,
"folderRemoved": mutation.folder_removed,
"renamedOrMoved": mutation.renamed_or_moved,
"timeBucket": bucket,
});
let encoded = serde_json::to_vec(&payload).unwrap_or_default();
let mut hasher = Sha256::new();
hasher.update(encoded);
format!("{:x}", hasher.finalize())
}
fn load_event_dedupe_fingerprints(path: &Path) -> Result<BTreeSet<String>, String> {
let mut fingerprints = BTreeSet::new();
if !path.exists() {
return Ok(fingerprints);
}
for value in read_jsonl_values(path)? {
if let Some(fingerprint) = value.get("fingerprint").and_then(Value::as_str) {
fingerprints.insert(fingerprint.to_string());
}
}
Ok(fingerprints)
}
fn event_dedupe_file_contains(path: &Path, fingerprint: &str) -> Result<bool, String> {
if !path.exists() {
return Ok(false);
}
for value in read_jsonl_values(path)? {
if value
.get("fingerprint")
.and_then(Value::as_str)
.map(|value| value == fingerprint)
.unwrap_or(false)
{
return Ok(true);
}
}
Ok(false)
}
fn classify_event_kind(kind: &EventKind) -> &'static str {
match kind {
EventKind::Create(CreateKind::File) => "file_create",
EventKind::Create(CreateKind::Folder) => "folder_create",
EventKind::Create(_) => "create",
EventKind::Remove(RemoveKind::File) => "file_remove",
EventKind::Remove(RemoveKind::Folder) => "folder_remove",
EventKind::Remove(_) => "remove",
EventKind::Modify(ModifyKind::Name(
RenameMode::From | RenameMode::To | RenameMode::Both,
)) => "rename_or_move",
EventKind::Modify(ModifyKind::Data(_)) => "file_modify",
EventKind::Modify(ModifyKind::Metadata(_)) => "metadata_modify",
EventKind::Modify(_) => "modify",
_ => "other",
}
}
fn is_rename_or_move(kind: &EventKind) -> bool {
matches!(
kind,
EventKind::Modify(ModifyKind::Name(
RenameMode::From | RenameMode::To | RenameMode::Both
))
)
}
fn rename_paths(kind: &EventKind, paths: &[String]) -> (Option<String>, Option<String>) {
if !is_rename_or_move(kind) {
return (None, None);
}
match paths {
[old_path, new_path, ..] => (Some(old_path.clone()), Some(new_path.clone())),
[path] => match kind {
EventKind::Modify(ModifyKind::Name(RenameMode::From)) => (Some(path.clone()), None),
EventKind::Modify(ModifyKind::Name(RenameMode::To)) => (None, Some(path.clone())),
_ => (None, Some(path.clone())),
},
_ => (None, None),
}
}
fn record_commit_snapshot(
identity: &RepoIdentity,
layout: &SpoolLayout,
previous_head: Option<&str>,
) -> Result<Option<String>, String> {
let head = current_head(&identity.root)?;
let Some(head_sha) = head else {
return Ok(None);
};
if previous_head == Some(head_sha.as_str()) {
return Ok(Some(head_sha));
}
let commit = WorktreeCommitEvent {
id: Uuid::new_v4().to_string(),
repo_owner: identity.owner.clone(),
repo_name: identity.name.clone(),
branch_name: identity.branch.clone(),
repo_root: identity.root.display().to_string(),
previous_head_sha: previous_head.map(str::to_string),
head_sha: head_sha.clone(),
subject: git_output(&identity.root, &["log", "-1", "--pretty=%s"]).ok(),
author_name: git_output(&identity.root, &["log", "-1", "--pretty=%an"]).ok(),
author_email: git_output(&identity.root, &["log", "-1", "--pretty=%ae"]).ok(),
committed_at: git_output(&identity.root, &["log", "-1", "--pretty=%cI"]).ok(),
occurred_at: Utc::now(),
};
append_json_line(&layout.commits_file, &commit)?;
Ok(Some(head_sha))
}
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.kind {
SyncFileKind::Events => {
for record in &candidate.records {
events.push(
serde_json::from_value::<WorktreeMutationEvent>(record.clone()).map_err(
|error| {
format!(
"Failed to decode event from {}: {error}",
candidate.path.display()
)
},
)?,
);
}
}
SyncFileKind::Commits => {
for record in &candidate.records {
commits.push(
serde_json::from_value::<WorktreeCommitEvent>(record.clone()).map_err(
|error| {
format!(
"Failed to decode commit from {}: {error}",
candidate.path.display()
)
},
)?,
);
}
}
}
}
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 payload = WorktreeMutationIngestPayload {
device: WorktreeMutationDevicePayload {
hardware_id: device.hardware_id,
device_name: current_hostname(),
hostname: current_hostname(),
platform: std::env::consts::OS.to_string(),
},
repository: WorktreeMutationRepositoryPayload {
owner: identity.owner.clone(),
name: identity.name.clone(),
branch_name: identity.branch.clone(),
repo_root: identity.root.display().to_string(),
},
events,
commits,
};
let response = client
.post(endpoint.clone())
.bearer_auth(token)
.json(&payload)
.send()
.await
.map_err(|error| format!("Failed to upload worktree mutations: {error}"))?;
if response.status() == StatusCode::UNAUTHORIZED {
return Err(
"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(format!(
"Worktree mutation upload failed with {status}: {body}"
));
}
let status_code = response.status().as_u16();
let layout = prepare_spool_layout(identity)?;
append_sync_log(
identity,
&layout,
&endpoint,
status_code,
event_count,
commit_count,
&candidates,
resync,
)?;
for candidate in candidates {
if !is_synced_spool_path(&candidate.path) {
mark_synced(&candidate.path)?;
}
}
Ok(())
}
fn collect_sync_candidates(layout: &SpoolLayout) -> Result<Vec<SyncCandidate>, String> {
collect_spool_candidates(layout, false)
}
fn print_no_sync_candidates_message(
target: &WorktreeWatchTargetOptions,
resync: bool,
) -> Result<(), String> {
if resync {
println!("No local worktree mutation JSONL files found.");
return Ok(());
}
let identities = resolve_target_identities(target)?;
let mut synced_files = 0usize;
let mut synced_records = 0usize;
for identity in identities {
let layout = prepare_spool_layout(&identity)?;
let all_candidates = collect_spool_candidates(&layout, true)?;
let unsynced_candidates = collect_sync_candidates(&layout)?;
synced_files += all_candidates
.len()
.saturating_sub(unsynced_candidates.len());
let all_records = all_candidates
.iter()
.map(|candidate| candidate.records.len())
.sum::<usize>();
let unsynced_records = unsynced_candidates
.iter()
.map(|candidate| candidate.records.len())
.sum::<usize>();
synced_records += all_records.saturating_sub(unsynced_records);
}
if synced_files > 0 {
println!(
"No unsynced worktree mutation files found. Found {synced_files} already-synced local spool file(s) with {synced_records} record(s). Use `xbp worktree-watch sync --resync` to upload them again."
);
} else {
println!("No unsynced worktree mutation files found.");
}
Ok(())
}
fn collect_spool_candidates(
layout: &SpoolLayout,
include_synced: bool,
) -> Result<Vec<SyncCandidate>, String> {
let mut candidates = Vec::new();
if !layout.root.exists() {
return Ok(candidates);
}
for entry in fs::read_dir(&layout.root).map_err(|error| {
format!(
"Failed to read spool directory {}: {error}",
layout.root.display()
)
})? {
let path = entry
.map_err(|error| format!("Failed to read spool entry: {error}"))?
.path();
let Some(file_name) = path.file_name().and_then(|value| value.to_str()) else {
continue;
};
if !file_name.ends_with(".jsonl") || (!include_synced && file_name.contains(".synced.")) {
continue;
}
let kind = if file_name.starts_with("events-") {
SyncFileKind::Events
} else if file_name.starts_with("commits-") {
SyncFileKind::Commits
} else {
continue;
};
let records = read_jsonl_values(&path)?;
if records.is_empty() {
continue;
}
candidates.push(SyncCandidate {
path,
kind,
records,
});
}
Ok(candidates)
}
fn append_sync_log(
identity: &RepoIdentity,
layout: &SpoolLayout,
endpoint: &str,
status_code: u16,
event_count: usize,
commit_count: usize,
candidates: &[SyncCandidate],
resync: bool,
) -> Result<(), String> {
let entry = WorktreeWatchSyncLogEntry {
id: Uuid::new_v4().to_string(),
synced_at: Utc::now(),
endpoint: endpoint.to_string(),
repository_owner: identity.owner.clone(),
repository_name: identity.name.clone(),
branch_name: identity.branch.clone(),
repo_root: identity.root.display().to_string(),
resync,
status_code,
event_count,
commit_count,
spool_file_count: candidates.len(),
spool_files: candidates
.iter()
.map(|candidate| candidate.path.display().to_string())
.collect(),
};
append_json_line(&layout.sync_log_file, &entry)
}
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_jsonl_values(path: &Path) -> Result<Vec<Value>, 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 values = Vec::new();
for line in reader.lines() {
let line =
line.map_err(|error| format!("Failed to read spool file {}: {error}", path.display()))?;
if line.trim().is_empty() {
continue;
}
values.push(serde_json::from_str(&line).map_err(|error| {
format!(
"Failed to parse JSONL record in {}: {error}",
path.display()
)
})?);
}
Ok(values)
}
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_worktree_watch_for_identity(identity: &RepoIdentity) -> Result<PathBuf, String> {
let layout = prepare_spool_layout(identity)?;
replace_existing_watcher(identity, &layout)?;
let executable = std::env::current_exe()
.map_err(|error| format!("Failed to resolve current XBP executable: {error}"))?;
let mut command = Command::new(&executable);
command
.arg("worktree-watch")
.arg("start")
.arg("--repo")
.arg(&identity.root)
.env(BACKGROUND_CHILD_ENV, "1")
.current_dir(&identity.root)
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null());
#[cfg(windows)]
{
use std::os::windows::process::CommandExt;
command.creation_flags(CREATE_NO_WINDOW);
}
let child = command
.spawn()
.map_err(|error| format!("Failed to spawn background worktree watcher: {error}"))?;
write_watcher_state(identity, &layout, child.id(), &executable)?;
Ok(identity.root.clone())
}
fn replace_existing_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)?;
println!(
"Replaced existing background worktree watcher {} for {}",
state.pid,
identity.root.display()
);
}
let _ = fs::remove_file(&layout.watcher_state_file);
Ok(())
}
fn read_watcher_state(path: &Path) -> Result<Option<WorktreeWatcherState>, String> {
if !path.exists() {
return Ok(None);
}
let raw = fs::read_to_string(path)
.map_err(|error| format!("Failed to read watcher state {}: {error}", path.display()))?;
serde_json::from_str(&raw)
.map(Some)
.map_err(|error| format!("Failed to parse watcher state {}: {error}", path.display()))
}
fn write_watcher_state(
identity: &RepoIdentity,
layout: &SpoolLayout,
pid: u32,
executable: &Path,
) -> Result<(), String> {
let state = WorktreeWatcherState {
pid,
repo_root: identity.root.display().to_string(),
repository_owner: identity.owner.clone(),
repository_name: identity.name.clone(),
branch_name: identity.branch.clone(),
executable: executable.display().to_string(),
started_at: Utc::now(),
};
let rendered = serde_json::to_string_pretty(&state)
.map_err(|error| format!("Failed to serialize watcher state: {error}"))?;
fs::write(&layout.watcher_state_file, format!("{rendered}\n")).map_err(|error| {
format!(
"Failed to write watcher state {}: {error}",
layout.watcher_state_file.display()
)
})
}
fn same_repo_watcher_state(identity: &RepoIdentity, state: &WorktreeWatcherState) -> bool {
path_identity_key(&identity.root) == path_identity_key(Path::new(&state.repo_root))
&& identity.owner == state.repository_owner
&& identity.name == state.repository_name
&& identity.branch == state.branch_name
}
fn is_matching_watcher_process(state: &WorktreeWatcherState) -> bool {
if state.pid == std::process::id() {
return false;
}
let system = System::new_all();
let Some(process) = system.process(Pid::from_u32(state.pid)) else {
return false;
};
let command_line = process
.cmd()
.iter()
.map(|part| part.to_string_lossy())
.collect::<Vec<_>>()
.join(" ");
command_line.contains("worktree-watch")
&& command_line.contains("start")
&& command_line.contains(&state.repo_root)
}
fn stop_process(state: &WorktreeWatcherState) -> Result<(), String> {
let system = System::new_all();
let Some(process) = system.process(Pid::from_u32(state.pid)) else {
return Ok(());
};
if !is_matching_watcher_process(state) {
return Ok(());
}
if process.kill() {
Ok(())
} else {
Err(format!(
"Failed to stop existing watcher process {}",
state.pid
))
}
}
fn discover_repo_identities_under(parent: &Path) -> Result<Vec<RepoIdentity>, String> {
let parent = normalize_windows_verbatim_path(
fs::canonicalize(parent).unwrap_or_else(|_| parent.to_path_buf()),
);
if !parent.is_dir() {
return Err(format!(
"Parent folder {} does not exist or is not a directory.",
parent.display()
));
}
let mut identities = Vec::new();
let mut seen = BTreeSet::new();
discover_repo_identities_in_dir(&parent, &parent, &mut seen, &mut identities)?;
identities.sort_by(|left, right| left.root.cmp(&right.root));
Ok(identities)
}
fn discover_repo_identities_in_dir(
parent: &Path,
dir: &Path,
seen: &mut BTreeSet<String>,
identities: &mut Vec<RepoIdentity>,
) -> Result<(), String> {
if dir != parent && is_skipped_discovery_dir(dir) {
return Ok(());
}
if dir.join(".git").exists() {
if let Ok(identity) = resolve_repo_identity(Some(dir)) {
let key = path_identity_key(&identity.root);
if seen.insert(key) {
identities.push(identity);
}
return Ok(());
}
}
let entries = fs::read_dir(dir)
.map_err(|error| format!("Failed to read folder {}: {error}", dir.display()))?;
for entry in entries {
let entry = entry.map_err(|error| {
format!(
"Failed to read folder entry under {}: {error}",
dir.display()
)
})?;
let file_type = entry
.file_type()
.map_err(|error| format!("Failed to inspect {}: {error}", entry.path().display()))?;
if file_type.is_dir() {
discover_repo_identities_in_dir(parent, &entry.path(), seen, identities)?;
}
}
Ok(())
}
fn resolve_repo_identity(repo: Option<&Path>) -> Result<RepoIdentity, String> {
let start = match repo {
Some(path) => path.to_path_buf(),
None => std::env::current_dir()
.map_err(|error| format!("Failed to resolve current directory: {error}"))?,
};
let root_raw = git_output(&start, &["rev-parse", "--show-toplevel"])?;
let root = normalize_windows_verbatim_path(
fs::canonicalize(root_raw.trim()).unwrap_or_else(|_| PathBuf::from(root_raw.trim())),
);
let branch = git_output(&root, &["rev-parse", "--abbrev-ref", "HEAD"])
.unwrap_or_else(|_| "unknown".to_string());
let remote = git_output(&root, &["remote", "get-url", DEFAULT_REMOTE]).unwrap_or_default();
let (owner, name) = parse_remote_owner_repo(&remote).unwrap_or_else(|| {
let name = root
.file_name()
.and_then(|value| value.to_str())
.unwrap_or("repository")
.to_string();
("unknown".to_string(), name)
});
let head_sha = current_head(&root)?;
Ok(RepoIdentity {
owner,
name,
branch: sanitize_path_component(branch.trim()),
root,
head_sha,
})
}
fn prepare_spool_layout(identity: &RepoIdentity) -> Result<SpoolLayout, String> {
let home = dirs::home_dir().ok_or_else(|| "Failed to resolve home directory.".to_string())?;
let run_id = Uuid::new_v4().to_string();
let root = home
.join(".xbp")
.join("mutations")
.join(sanitize_path_component(&identity.owner))
.join(sanitize_path_component(&identity.name))
.join(sanitize_path_component(&identity.branch));
fs::create_dir_all(&root).map_err(|error| {
format!(
"Failed to create worktree mutation spool {}: {error}",
root.display()
)
})?;
Ok(SpoolLayout {
events_file: root.join(format!("events-{run_id}.jsonl")),
commits_file: root.join(format!("commits-{run_id}.jsonl")),
stats_file: root.join("stats.json"),
watcher_state_file: root.join("watcher-state.json"),
sync_log_file: root.join("sync-log.jsonl"),
event_dedupe_file: root.join("event-dedupe.jsonl"),
root,
})
}
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 parse_numstat_count(value: Option<&str>) -> u64 {
value.and_then(|raw| raw.parse::<u64>().ok()).unwrap_or(0)
}
fn current_head(repo_root: &Path) -> Result<Option<String>, String> {
match git_output(repo_root, &["rev-parse", "HEAD"]) {
Ok(value) => Ok(Some(value)),
Err(error)
if error.contains("unknown revision") || error.contains("ambiguous argument") =>
{
Ok(None)
}
Err(error) => Err(error),
}
}
fn git_output(repo_root: &Path, args: &[&str]) -> Result<String, String> {
let output = Command::new("git")
.args(args)
.current_dir(repo_root)
.output()
.map_err(|error| format!("Failed to run git {}: {error}", args.join(" ")))?;
if !output.status.success() {
return Err(String::from_utf8_lossy(&output.stderr).trim().to_string());
}
Ok(String::from_utf8_lossy(&output.stdout).trim().to_string())
}
fn parse_remote_owner_repo(remote: &str) -> Option<(String, String)> {
let trimmed = remote.trim().trim_end_matches(".git");
if trimmed.is_empty() {
return None;
}
let path_part = if !trimmed.contains("://") {
if let Some((_, path)) = trimmed.rsplit_once(':') {
path
} else {
trimmed
}
} else {
trimmed
.trim_start_matches("https://")
.trim_start_matches("http://")
.trim_start_matches("ssh://")
.split_once('/')
.map(|(_, path)| path)
.unwrap_or(trimmed)
};
let mut parts = path_part.rsplitn(2, '/');
let name = parts.next()?.trim();
let owner = parts.next()?.trim();
if owner.is_empty() || name.is_empty() {
return None;
}
Some((
sanitize_path_component(owner),
sanitize_path_component(name.trim_end_matches(".git")),
))
}
fn sanitize_path_component(value: &str) -> String {
let sanitized: String = value
.chars()
.map(|ch| {
if ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.') {
ch
} else {
'-'
}
})
.collect();
sanitized
.trim_matches('-')
.chars()
.take(160)
.collect::<String>()
}
fn normalize_windows_verbatim_path(path: PathBuf) -> PathBuf {
PathBuf::from(strip_windows_verbatim_prefix(&path.to_string_lossy()))
}
fn strip_windows_verbatim_prefix(input: &str) -> &str {
input.strip_prefix(r"\\?\").unwrap_or(input)
}
fn path_identity_key(path: &Path) -> String {
let value = path.to_string_lossy().to_string();
if cfg!(windows) {
value.to_ascii_lowercase()
} else {
value
}
}
fn is_skipped_discovery_dir(path: &Path) -> bool {
let Some(name) = path.file_name().and_then(|value| value.to_str()) else {
return false;
};
matches!(
name,
".git"
| ".hg"
| ".svn"
| ".next"
| ".turbo"
| ".vercel"
| "node_modules"
| "target"
| "dist"
| "build"
)
}
fn relative_slash_path(root: &Path, path: &Path) -> Option<String> {
let relative = path.strip_prefix(root).ok().unwrap_or(path);
Some(relative.to_string_lossy().replace('\\', "/"))
}
fn is_ignored_path(root: &Path, path: &Path) -> bool {
let Some(relative) = relative_slash_path(root, path) else {
return false;
};
relative == ".git"
|| relative.starts_with(".git/")
|| relative == "target"
|| relative.starts_with("target/")
}
fn current_hostname() -> Option<String> {
std::env::var("COMPUTERNAME")
.or_else(|_| std::env::var("HOSTNAME"))
.ok()
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
}
fn is_background_child() -> bool {
std::env::var(BACKGROUND_CHILD_ENV)
.ok()
.map(|value| value == "1")
.unwrap_or(false)
}
#[allow(dead_code)]
fn content_sha256(path: &Path) -> Option<String> {
let bytes = fs::read(path).ok()?;
let mut hasher = Sha256::new();
hasher.update(bytes);
Some(format!("{:x}", hasher.finalize()))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn parses_https_and_ssh_remote_urls() {
assert_eq!(
parse_remote_owner_repo("https://github.com/xylex-group/xbp.git"),
Some(("xylex-group".to_string(), "xbp".to_string()))
);
assert_eq!(
parse_remote_owner_repo("git@github.com:xylex-group/xbp.git"),
Some(("xylex-group".to_string(), "xbp".to_string()))
);
}
#[test]
fn sanitizes_branch_for_storage_path() {
assert_eq!(
sanitize_path_component("feature/worktree watcher"),
"feature-worktree-watcher"
);
}
#[test]
fn normalizes_windows_verbatim_repo_roots() {
assert_eq!(
normalize_windows_verbatim_path(PathBuf::from(
r"\\?\C:\Users\floris\Documents\GitHub\mollie-api-rust"
)),
PathBuf::from(r"C:\Users\floris\Documents\GitHub\mollie-api-rust")
);
}
#[test]
fn skips_heavy_discovery_directories() {
assert!(is_skipped_discovery_dir(Path::new("node_modules")));
assert!(is_skipped_discovery_dir(Path::new("target")));
assert!(is_skipped_discovery_dir(Path::new(".next")));
assert!(!is_skipped_discovery_dir(Path::new("mollie-api-rust")));
}
#[test]
fn 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"),
};
let candidates = vec![SyncCandidate {
path: PathBuf::from("events.jsonl"),
kind: SyncFileKind::Events,
records: vec![
stats_test_event("2026-07-08T10:00:00Z", "src/main.rs", 4, 1),
stats_test_event("2026-07-08T10:05:00Z", "src/main.rs", 2, 0),
stats_test_event("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.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 mutation_fingerprint_ignores_random_event_id_but_keeps_time_bucket() {
let first: WorktreeMutationEvent = serde_json::from_value(stats_test_event(
"2026-07-08T10:00:00Z",
"src/main.rs",
4,
1,
))
.unwrap();
let same_bucket: WorktreeMutationEvent = serde_json::from_value(stats_test_event(
"2026-07-08T10:00:00Z",
"src/main.rs",
4,
1,
))
.unwrap();
let next_bucket: WorktreeMutationEvent = serde_json::from_value(stats_test_event(
"2026-07-08T10:00:06Z",
"src/main.rs",
4,
1,
))
.unwrap();
assert_eq!(
mutation_event_fingerprint(&first),
mutation_event_fingerprint(&same_bucket)
);
assert_ne!(
mutation_event_fingerprint(&first),
mutation_event_fingerprint(&next_bucket)
);
}
#[test]
fn 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"
);
}
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,
"fileCreated": false,
"fileRemoved": false,
"folderCreated": false,
"folderRemoved": false,
"renamedOrMoved": false,
"rawKind": "Modify(Data(Content))",
"occurredAt": occurred_at,
})
}
}