agent_procs/daemon/
watcher.rs

1//! File watcher with debounce for watch-mode process restarts.
2
3use globset::{Glob, GlobSet, GlobSetBuilder};
4use notify::{Event, RecommendedWatcher, RecursiveMode, Watcher};
5use std::path::{Path, PathBuf};
6use std::time::Duration;
7use tokio::sync::mpsc;
8
9/// Default patterns always ignored by watchers.
10const DEFAULT_IGNORE: &[&str] = &[
11    "**/.git/**",
12    "**/node_modules/**",
13    "**/target/**",
14    "**/__pycache__/**",
15    "**/*.pyc",
16    "**/.DS_Store",
17];
18
19const DEBOUNCE_MS: u64 = 500;
20
21/// Handle to a running file watcher. Drop to stop watching.
22pub struct WatchHandle {
23    _watcher: RecommendedWatcher,
24    _debounce_handle: tokio::task::JoinHandle<()>,
25}
26
27/// Create a file watcher that sends the process name to `restart_tx`
28/// when watched files change (after debounce).
29pub fn create_watcher(
30    paths: &[String],
31    ignore: Option<&[String]>,
32    base_dir: &Path,
33    process_name: String,
34    restart_tx: mpsc::Sender<String>,
35) -> Result<WatchHandle, String> {
36    let ignore_set = build_ignore_set(ignore)?;
37    let watch_set = build_watch_set(paths)?;
38    let base_for_filter = base_dir.to_path_buf();
39
40    let (event_tx, mut event_rx) = mpsc::channel::<PathBuf>(256);
41
42    let watcher = RecommendedWatcher::new(
43        move |res: Result<Event, notify::Error>| {
44            if let Ok(event) = res {
45                for path in event.paths {
46                    // Filter: path must match watch globs and not match ignore globs
47                    let relative = path.strip_prefix(&base_for_filter).unwrap_or(&path);
48                    if !watch_set.is_match(relative) && !watch_set.is_match(&path) {
49                        continue;
50                    }
51                    if ignore_set.is_match(relative) || ignore_set.is_match(&path) {
52                        continue;
53                    }
54                    let _ = event_tx.blocking_send(path);
55                }
56            }
57        },
58        notify::Config::default(),
59    )
60    .map_err(|e| format!("failed to create watcher: {}", e))?;
61
62    // Watch the base directory recursively
63    let dirs = resolve_watch_dirs(paths, base_dir);
64    let mut watcher = watcher;
65    for dir in &dirs {
66        let _ = watcher.watch(dir, RecursiveMode::Recursive);
67    }
68
69    // Debounce task
70    let debounce_handle = tokio::spawn(async move {
71        loop {
72            // Wait for first event
73            let Some(_path) = event_rx.recv().await else {
74                break;
75            };
76
77            // Drain events during debounce window
78            let deadline = tokio::time::Instant::now() + Duration::from_millis(DEBOUNCE_MS);
79            loop {
80                match tokio::time::timeout_at(deadline, event_rx.recv()).await {
81                    Ok(Some(_)) => {}   // more events, keep waiting
82                    Ok(None) => return, // channel closed
83                    Err(_) => break,    // timeout — debounce complete
84                }
85            }
86
87            // Send restart signal
88            if restart_tx.send(process_name.clone()).await.is_err() {
89                break; // receiver dropped
90            }
91        }
92    });
93
94    Ok(WatchHandle {
95        _watcher: watcher,
96        _debounce_handle: debounce_handle,
97    })
98}
99
100fn build_ignore_set(user_ignore: Option<&[String]>) -> Result<GlobSet, String> {
101    let mut builder = GlobSetBuilder::new();
102    for pat in DEFAULT_IGNORE {
103        builder.add(Glob::new(pat).map_err(|e| e.to_string())?);
104    }
105    if let Some(patterns) = user_ignore {
106        for pat in patterns {
107            builder.add(Glob::new(pat).map_err(|e| e.to_string())?);
108        }
109    }
110    builder.build().map_err(|e| e.to_string())
111}
112
113fn build_watch_set(paths: &[String]) -> Result<GlobSet, String> {
114    let mut builder = GlobSetBuilder::new();
115    for pat in paths {
116        builder.add(Glob::new(pat).map_err(|e| e.to_string())?);
117    }
118    builder.build().map_err(|e| e.to_string())
119}
120
121fn resolve_watch_dirs(_patterns: &[String], base: &Path) -> Vec<PathBuf> {
122    // Watch base directory recursively; glob filtering happens in the event callback
123    vec![base.to_path_buf()]
124}
agent_procs/daemon/watcher.rs

agent_procs/daemon/
watcher.rs
