rpi_cli/
fs_watch.rs

1//! Lightweight file watching for live reload.
2//!
3//! Port of the upstream `packages/coding-agent/src/utils/fs-watch.ts`.
4//! uses Node's `fs.watch` with a retry-on-error wrapper to (a) keep the git
5//! footer's branch/status fresh and (b) hot-reload the active theme file.
6//!
7//! rpi watches by polling modification times. That avoids a native watcher
8//! dependency and behaves predictably on every platform (network drives, WSL
9//! mounts, editors that replace files by rename). The retry delay matches
10//! native [`FS_WATCH_RETRY_DELAY_MS`].
11
12use std::collections::HashMap;
13use std::path::{Path, PathBuf};
14use std::sync::atomic::{AtomicBool, Ordering};
15use std::sync::Arc;
16use std::time::{Duration, SystemTime};
17
18/// Native `FS_WATCH_RETRY_DELAY_MS`: how long to wait after a watch error
19/// before retrying.
20pub const FS_WATCH_RETRY_DELAY_MS: u64 = 5_000;
21
22/// Default poll interval. Short enough to feel immediate, long enough that a
23/// few watched files cost nothing on the main thread.
24pub const DEFAULT_POLL_INTERVAL_MS: u64 = 1_000;
25
26fn mtime(path: &Path) -> Option<SystemTime> {
27    std::fs::metadata(path).ok().and_then(|m| m.modified().ok())
28}
29
30fn snapshot(paths: &[PathBuf]) -> HashMap<PathBuf, Option<SystemTime>> {
31    paths.iter().map(|p| (p.clone(), mtime(p))).collect()
32}
33
34/// A running watch. Dropping it does **not** stop the watch; call
35/// [`WatchHandle::close`] to stop it deterministically (native `closeWatcher`).
36pub struct WatchHandle {
37    stop: Arc<AtomicBool>,
38    join: Option<tokio::task::JoinHandle<()>>,
39}
40
41impl WatchHandle {
42    /// Signal the watch to stop and wait for its task to finish.
43    pub fn close(mut self) {
44        self.stop.store(true, Ordering::SeqCst);
45        if let Some(join) = self.join.take() {
46            join.abort();
47        }
48    }
49
50    /// Request stop without joining (useful from a drop path).
51    pub fn request_stop(&self) {
52        self.stop.store(true, Ordering::SeqCst);
53    }
54}
55
56impl Drop for WatchHandle {
57    fn drop(&mut self) {
58        self.stop.store(true, Ordering::SeqCst);
59    }
60}
61
62/// Watch `paths` for mtime changes, invoking `on_change` whenever any watched
63/// path changes (content or existence). Missing paths are tracked too: creating
64/// or deleting a watched path fires `on_change`.
65///
66/// Poll failures are tolerated: the watcher waits [`FS_WATCH_RETRY_DELAY_MS`] and
67/// retries, never panicking (native `watchWithErrorHandler`).
68pub fn watch_paths<F>(paths: Vec<PathBuf>, poll_interval: Duration, on_change: F) -> WatchHandle
69where
70    F: Fn() + Send + Sync + 'static,
71{
72    let stop = Arc::new(AtomicBool::new(false));
73    let stop_task = stop.clone();
74    let join = tokio::spawn(async move {
75        let mut prev = snapshot(&paths);
76        loop {
77            tokio::time::sleep(poll_interval).await;
78            if stop_task.load(Ordering::SeqCst) {
79                break;
80            }
81            let current = match std::panic::catch_unwind(|| snapshot(&paths)) {
82                Ok(snap) => snap,
83                Err(_) => {
84                    // A metadata read blew up; wait the retry delay and try
85                    // again rather than tearing down the watch.
86                    tokio::time::sleep(Duration::from_millis(FS_WATCH_RETRY_DELAY_MS)).await;
87                    continue;
88                }
89            };
90            if current != prev {
91                prev = current;
92                on_change();
93            }
94        }
95    });
96    WatchHandle {
97        stop,
98        join: Some(join),
99    }
100}
101
102/// Convenience for a single path.
103pub fn watch_path<F>(path: PathBuf, on_change: F) -> WatchHandle
104where
105    F: Fn() + Send + Sync + 'static,
106{
107    watch_paths(
108        vec![path],
109        Duration::from_millis(DEFAULT_POLL_INTERVAL_MS),
110        on_change,
111    )
112}
113
114#[cfg(test)]
115mod tests {
116    use super::*;
117    use std::sync::atomic::AtomicUsize;
118
119    #[test]
120    fn detects_content_change() {
121        let dir = tempfile::tempdir().unwrap();
122        let file = dir.path().join("theme.json");
123        std::fs::write(&file, "a").unwrap();
124
125        let hits = Arc::new(AtomicUsize::new(0));
126        let hits_task = hits.clone();
127        let rt = tokio::runtime::Builder::new_current_thread()
128            .enable_time()
129            .build()
130            .unwrap();
131        rt.block_on(async {
132            let handle = watch_paths(vec![file.clone()], Duration::from_millis(20), move || {
133                hits_task.fetch_add(1, Ordering::SeqCst);
134            });
135            tokio::time::sleep(Duration::from_millis(60)).await;
136            std::fs::write(&file, "bb").unwrap();
137            // mtime resolution: ensure a visible change.
138            tokio::time::sleep(Duration::from_millis(120)).await;
139            handle.close();
140        });
141        assert!(
142            hits.load(Ordering::SeqCst) >= 1,
143            "expected at least one change"
144        );
145    }
146
147    #[test]
148    fn close_stops_watch() {
149        let dir = tempfile::tempdir().unwrap();
150        let file = dir.path().join("f");
151        std::fs::write(&file, "x").unwrap();
152        let rt = tokio::runtime::Builder::new_current_thread()
153            .enable_time()
154            .build()
155            .unwrap();
156        rt.block_on(async {
157            let handle = watch_path(file, || {});
158            handle.close();
159        });
160    }
161}
rpi_cli/fs_watch.rs

rpi_cli/
fs_watch.rs
