diff --git a/docs/book/src/cli/watch.md b/docs/book/src/cli/watch.md index f58f6b7..ddccfd5 100644 --- a/docs/book/src/cli/watch.md +++ b/docs/book/src/cli/watch.md @@ -46,9 +46,37 @@ The process responds to `Ctrl+C` and `SIGTERM` for clean shutdown. | `--repo ` | | Repository name for multi-repo indexing | | `--source ` | same as PATH | Source directory to watch (if different from config dir) | | `--debounce-ms ` | `500` | Debounce interval in milliseconds | +| `--reindex-interval-secs ` | `900` | Periodic full-tree reindex backstop (see below). `0` disables | | `--pid-file ` | | Write PID to this file for daemon management | | `--generate-systemd` | | Print a systemd service unit to stdout and exit | +## Reindex backstop + +File watchers are the fast path, but they are not a complete freshness +guarantee: a process restart, a high-churn burst, or a dropped inotify event +can leave content silently stale until the next manual reindex. + +To close that gap, `watch` runs a periodic **reindex backstop** (on by default, +every 15 minutes). Each sweep reconciles the whole source tree against the +index — re-embedding any file whose content hash drifted and pruning rows for +files that have disappeared from disk. Sweeps are incremental: unchanged files +are skipped by hash, so a sweep where the watcher kept up does almost no work. + +Tune the cadence with `--reindex-interval-secs`, or set it to `0` to disable the +backstop and rely on watcher events alone: + +```bash +# Reconcile every 5 minutes +bobbin watch --reindex-interval-secs 300 + +# Disable the backstop +bobbin watch --reindex-interval-secs 0 +``` + +`bobbin status` reports a **Freshness** signal — `stale` when the current git +HEAD is newer than the last index run — so drift is observable without waiting +for a search to miss. + ## Examples Watch the current directory: diff --git a/src/cli/index.rs b/src/cli/index.rs index 3d9e561..38ea935 100644 --- a/src/cli/index.rs +++ b/src/cli/index.rs @@ -1377,7 +1377,7 @@ fn read_indexable_content(path: &Path, config: &Config) -> Result { } /// Collect all files to index based on configuration patterns -fn collect_files(repo_root: &Path, config: &Config) -> Result> { +pub(crate) fn collect_files(repo_root: &Path, config: &Config) -> Result> { let mut files = Vec::new(); let mut include_globs: Vec = config.index.include.clone(); diff --git a/src/cli/status.rs b/src/cli/status.rs index c12d1ee..ebd54ea 100644 --- a/src/cli/status.rs +++ b/src/cli/status.rs @@ -40,6 +40,35 @@ struct StatusOutput { calibration: Option, #[serde(skip_serializing_if = "Option::is_none")] git: Option, + #[serde(skip_serializing_if = "Option::is_none")] + freshness: Option, +} + +/// Index freshness relative to git HEAD (bobbin #44). +#[derive(Serialize)] +struct Freshness { + /// True when HEAD's commit is newer than the last index run (or the repo + /// has commits but has never been indexed). + stale: bool, + /// Committer timestamp (unix seconds) of the current HEAD. + head_commit_time: i64, + /// Timestamp (unix seconds) of the most recent index run, if any. + #[serde(skip_serializing_if = "Option::is_none")] + last_indexed: Option, +} + +impl Freshness { + fn compute(head_commit_time: i64, last_indexed: Option) -> Self { + let stale = match last_indexed { + Some(ts) => head_commit_time > ts, + None => true, + }; + Freshness { + stale, + head_commit_time, + last_indexed, + } + } } #[derive(Serialize)] @@ -85,6 +114,7 @@ pub async fn run(args: StatusArgs, output: OutputConfig) -> Result<()> { stats: None, calibration: None, git: None, + freshness: None, }; println!("{}", serde_json::to_string_pretty(&json_output)?); } else if !output.quiet { @@ -111,10 +141,16 @@ pub async fn run(args: StatusArgs, output: OutputConfig) -> Result<()> { let calibration_result = calibrate::load_calibration(&repo_root); let cal_status = build_calibration_status(&calibration_result, stats.total_chunks); - // Git info - let git_status = GitAnalyzer::new(&repo_root) - .ok() - .map(|git| build_git_status(&git)); + // Git info + index freshness (bobbin #44): compare HEAD's commit time + // against the last index run. If HEAD is newer, committed changes have not + // been picked up — a watcher-drift signal that doesn't false-positive on + // quiet repos (no new commits => HEAD not newer => not stale). + let git = GitAnalyzer::new(&repo_root).ok(); + let git_status = git.as_ref().map(build_git_status); + let freshness = git + .as_ref() + .and_then(|g| g.get_head_commit_time()) + .map(|head_ts| Freshness::compute(head_ts, stats.last_indexed)); if output.json { let json_output = StatusOutput { @@ -124,6 +160,7 @@ pub async fn run(args: StatusArgs, output: OutputConfig) -> Result<()> { stats: Some(stats), calibration: Some(cal_status), git: git_status, + freshness, }; println!("{}", serde_json::to_string_pretty(&json_output)?); } else if !output.quiet { @@ -168,6 +205,23 @@ pub async fn run(args: StatusArgs, output: OutputConfig) -> Result<()> { println!(" Last indexed: {}", dt); } + // Freshness signal (bobbin #44): warn when HEAD outpaces the index. + if let Some(ref f) = freshness { + if f.stale { + let behind = f + .last_indexed + .map(|ts| format!(" (HEAD is {} newer)", format_duration(f.head_commit_time - ts))) + .unwrap_or_else(|| " (never indexed)".to_string()); + println!( + " Freshness: {}{}", + "stale — run `bobbin index`".yellow(), + behind.dimmed() + ); + } else { + println!(" Freshness: {}", "up to date".green()); + } + } + // Show dependency stats if let Ok((total_deps, resolved_deps)) = vector_store.get_dependency_stats().await { if total_deps > 0 { @@ -423,6 +477,20 @@ fn format_relative_time(ts: i64) -> String { } } +/// Format a positive duration in seconds as a compact human string. +fn format_duration(secs: i64) -> String { + let secs = secs.max(0); + if secs < 60 { + format!("{}s", secs) + } else if secs < 3600 { + format!("{}m", secs / 60) + } else if secs < 86400 { + format!("{}h", secs / 3600) + } else { + format!("{}d", secs / 86400) + } +} + fn format_age_days(days: u32) -> String { if days < 30 { format!("{} days", days) @@ -449,6 +517,7 @@ async fn run_remote(args: StatusArgs, output: OutputConfig, server_url: &str) -> repos: vec![], calibration: None, // Not available via remote git: None, // Not available via remote + freshness: None, // Not available via remote stats: Some(IndexStats { total_files: resp.index.total_files, total_chunks: resp.index.total_chunks, @@ -507,3 +576,47 @@ async fn run_remote(args: StatusArgs, output: OutputConfig, server_url: &str) -> Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_freshness_head_newer_is_stale() { + // HEAD committed after the last index run => stale. + let f = Freshness::compute(2_000, Some(1_000)); + assert!(f.stale); + assert_eq!(f.head_commit_time, 2_000); + assert_eq!(f.last_indexed, Some(1_000)); + } + + #[test] + fn test_freshness_head_older_is_fresh() { + // No commits since the last index (quiet repo) => not stale. This is + // the false-positive guard: age alone must not flag an idle repo. + let f = Freshness::compute(1_000, Some(2_000)); + assert!(!f.stale); + } + + #[test] + fn test_freshness_equal_is_fresh() { + let f = Freshness::compute(1_000, Some(1_000)); + assert!(!f.stale); + } + + #[test] + fn test_freshness_never_indexed_is_stale() { + let f = Freshness::compute(1_000, None); + assert!(f.stale); + assert_eq!(f.last_indexed, None); + } + + #[test] + fn test_format_duration_units() { + assert_eq!(format_duration(30), "30s"); + assert_eq!(format_duration(120), "2m"); + assert_eq!(format_duration(7_200), "2h"); + assert_eq!(format_duration(172_800), "2d"); + assert_eq!(format_duration(-5), "0s"); + } +} diff --git a/src/cli/watch.rs b/src/cli/watch.rs index dff9bed..5459e24 100644 --- a/src/cli/watch.rs +++ b/src/cli/watch.rs @@ -31,6 +31,13 @@ pub struct WatchArgs { #[arg(long, default_value_t = 500)] debounce_ms: u64, + /// Periodic full-tree reindex backstop, in seconds. Catches changes the + /// file watcher may have dropped (missed events, high-churn bursts). Each + /// sweep is incremental — unchanged files are skipped by hash — so it is + /// cheap when the watcher kept up. Set to 0 to disable. (bobbin #44) + #[arg(long, default_value_t = 900)] + reindex_interval_secs: u64, + /// Write PID to this file for daemon management #[arg(long)] pid_file: Option, @@ -168,6 +175,23 @@ pub async fn run(args: WatchArgs, output: OutputConfig) -> Result<()> { let mut compact_counter: usize = 0; let mut last_compact = Instant::now(); + // Periodic reindex backstop (bobbin #44). Fires one interval out (not + // immediately — the watcher was just started fresh) and reconciles the + // whole tree against the index, catching any events the watcher dropped. + let reindex_enabled = args.reindex_interval_secs > 0; + let reindex_period = Duration::from_secs(args.reindex_interval_secs.max(1)); + let mut reindex_interval = tokio::time::interval_at( + tokio::time::Instant::now() + reindex_period, + reindex_period, + ); + reindex_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + if !output.quiet && reindex_enabled { + println!( + " Reindex backstop: every {}s", + args.reindex_interval_secs + ); + } + let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?; @@ -329,6 +353,39 @@ pub async fn run(args: WatchArgs, output: OutputConfig) -> Result<()> { } } } + + _ = reindex_interval.tick(), if reindex_enabled => { + match run_reindex_backstop( + &source_root, + &config, + repo_name, + &mut vector_store, + &metadata_store, + &embed, + &mut parser, + &output, + ) + .await + { + Ok(stats) if stats.files_indexed > 0 || stats.files_removed > 0 => { + compact_counter += stats.files_indexed + stats.files_removed; + if !output.quiet { + println!( + " {} Backstop reconciled: {} reindexed, {} removed", + "~".cyan(), + stats.files_indexed, + stats.files_removed, + ); + } + } + Ok(_) => {} + Err(e) => { + if !output.quiet { + eprintln!(" {} Backstop error: {}", "!".yellow(), e); + } + } + } + } } } @@ -482,6 +539,74 @@ async fn reindex_files( Ok(stats) } +struct BackstopStats { + files_indexed: usize, + files_removed: usize, +} + +/// Periodic full-tree reconciliation (bobbin #44). +/// +/// Walks the whole source tree with the same collector `bobbin index` uses, +/// re-indexes any file whose content hash drifted from the stored one, and +/// prunes index rows for files that have disappeared from disk. Both halves +/// reuse the incremental paths (`reindex_files` hash-skips unchanged files), +/// so a sweep where the watcher kept up does almost no work. This is the +/// safety net for events the watcher missed (restarts, dropped events, +/// high-churn bursts). +#[allow(clippy::too_many_arguments)] +async fn run_reindex_backstop( + source_root: &Path, + config: &Config, + repo_name: &str, + vector_store: &mut VectorStore, + metadata_store: &MetadataStore, + embed: &Embedder, + parser: &mut Parser, + output: &OutputConfig, +) -> Result { + let files = super::index::collect_files(source_root, config)?; + + // Prune index rows for files that no longer exist on disk. collect_files + // only returns extant files, so anything indexed but absent was deleted + // while the watcher was down (or its Remove event was dropped). + let current: HashSet = files + .iter() + .map(|p| { + p.strip_prefix(source_root) + .unwrap_or(p) + .to_string_lossy() + .to_string() + }) + .collect(); + let indexed = metadata_store.get_all_indexed_files()?; + let removed: Vec = indexed.difference(¤t).cloned().collect(); + let mut files_removed = 0; + if !removed.is_empty() { + vector_store.delete_by_file(&removed).await?; + metadata_store.delete_file_hashes(&removed)?; + files_removed = removed.len(); + } + + // Reindex drifted files (reindex_files skips those whose hash is unchanged). + let stats = reindex_files( + &files, + source_root, + config, + repo_name, + vector_store, + metadata_store, + embed, + parser, + output, + ) + .await?; + + Ok(BackstopStats { + files_indexed: stats.files_indexed, + files_removed, + }) +} + /// Detect the git repo name for a directory by walking up to find `.git`. /// Returns the directory name containing `.git` (e.g. "moonraker" for /// /home/user/workspace/moonraker/src/main.rs). diff --git a/src/index/git.rs b/src/index/git.rs index cc12557..ef41b25 100644 --- a/src/index/git.rs +++ b/src/index/git.rs @@ -332,6 +332,23 @@ impl GitAnalyzer { Ok(String::from_utf8_lossy(&output.stdout).trim().to_string()) } + /// Get the committer timestamp (unix seconds) of the current HEAD commit. + /// + /// Used as a freshness reference: if HEAD is newer than the last index run, + /// committed changes have not yet been picked up (bobbin #44). Returns None + /// when there is no HEAD (empty repo) or git is unavailable. + pub fn get_head_commit_time(&self) -> Option { + let output = Command::new("git") + .args(["log", "-1", "--format=%ct"]) + .current_dir(&self.repo_root) + .output() + .ok()?; + if !output.status.success() { + return None; + } + String::from_utf8_lossy(&output.stdout).trim().parse::().ok() + } + /// Get the last commit timestamp for every file in the repo in one pass. /// /// Returns a map of relative file path -> unix timestamp of the most recent