agentmux_srv/backend/
files_watcher.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Directory watcher for the Files pane (`fs.watch` / `fs.unwatch`).
5//!
6//! Sibling of `media_file_watcher.rs` on the same shared `FsWatchPool`, with
7//! a different contract: a Files pane re-lists the whole folder on any
8//! change, so this publishes a coalesced "this folder changed" signal per
9//! directory rather than per file, for creates, modifications and removals
10//! alike.
11//!
12//! Spec: docs/specs/SPEC_FILE_BROWSER_PANE_2026_10_01.md §6.3, §10.
13
14use std::collections::{HashMap, HashSet, VecDeque};
15use std::path::{Path, PathBuf};
16use std::sync::{Arc, Mutex};
17use std::time::Duration;
18
19use tokio::sync::broadcast;
20
21use super::fs_ops::platform::display_path;
22use super::fs_watch::{FsWatchPool, Subscription};
23use super::mps::Broker;
24
25/// MPS event telling a Files pane to re-list a folder. Scoped to
26/// `block:<id>`; payload `{ dir }`, the folder's display path exactly as
27/// `fs.list` returns it.
28pub const EVENT_FILES_CHANGED: &str = "files:changed";
29
30/// Changes to one folder within this window are folded into one event
31/// (spec §10: one patch per ~100 ms window, applied within 200 ms).
32const COALESCE: Duration = Duration::from_millis(150);
33
34/// Watches one block may hold. Past this the oldest is dropped (spec §10:
35/// "hard cap, LRU"). Also bounds what a pane that never unwatches (a
36/// window reload skips its dispose) can leak.
37pub const MAX_WATCHES_PER_BLOCK: usize = 64;
38
39struct Watch {
40    block_id: String,
41    dir: PathBuf,
42    sub: Subscription,
43}
44
45#[derive(Default)]
46struct Inner {
47    watches: HashMap<String, Watch>,
48    /// Each block's watch ids, oldest first.
49    by_block: HashMap<String, VecDeque<String>>,
50    /// Canonical directory -> (display path, watch ids on it).
51    by_dir: HashMap<PathBuf, (String, HashSet<String>)>,
52    /// Directories with a publish already scheduled.
53    pending: HashSet<PathBuf>,
54}
55
56pub struct FilesWatcher {
57    pool: Arc<FsWatchPool>,
58    broker: Arc<Broker>,
59    inner: Mutex<Inner>,
60}
61
62impl FilesWatcher {
63    /// Construct and start the watcher on `pool`'s shared event stream.
64    pub fn new(pool: Arc<FsWatchPool>, broker: Arc<Broker>) -> Arc<Self> {
65        let this = Arc::new(Self { pool: pool.clone(), broker, inner: Mutex::new(Inner::default()) });
66        let mut events = pool.events();
67        let worker = this.clone();
68        tokio::spawn(async move {
69            loop {
70                match events.recv().await {
71                    Ok(ev) => worker.on_fs_event(&ev.path),
72                    Err(broadcast::error::RecvError::Closed) => break,
73                    // Events were dropped, so any folder may have changed
74                    // unseen: re-list them all. The frontend re-lists on each
75                    // event, so this is a full rescan.
76                    Err(broadcast::error::RecvError::Lagged(skipped)) => {
77                        tracing::warn!(skipped, "files watcher lagged behind the fs_watch stream; rescanning every watched folder");
78                        worker.on_lagged();
79                    }
80                }
81            }
82        });
83        this
84    }
85
86    fn lock(&self) -> std::sync::MutexGuard<'_, Inner> {
87        self.inner.lock().unwrap_or_else(|e| e.into_inner())
88    }
89
90    /// Watch the canonical directory `dir` for `block_id`; returns the id to
91    /// unwatch with. Each call is its own registration (the pool
92    /// reference-counts the OS watch), so two panes on one folder, or one
93    /// pane watching it twice, never cancel each other.
94    pub fn watch(&self, dir: &Path, block_id: &str) -> String {
95        let watch_id = uuid::Uuid::new_v4().to_string();
96        let sub = self.pool.subscribe_dir(dir);
97        let mut inner = self.lock();
98        inner.watches.insert(watch_id.clone(), Watch { block_id: block_id.to_string(), dir: dir.to_path_buf(), sub });
99        inner
100            .by_dir
101            .entry(dir.to_path_buf())
102            .or_insert_with(|| (display_path(dir), HashSet::new()))
103            .1
104            .insert(watch_id.clone());
105        let ids = inner.by_block.entry(block_id.to_string()).or_default();
106        ids.push_back(watch_id.clone());
107        let evicted: Vec<String> = if ids.len() > MAX_WATCHES_PER_BLOCK {
108            ids.drain(..ids.len() - MAX_WATCHES_PER_BLOCK).collect()
109        } else {
110            Vec::new()
111        };
112        for id in &evicted {
113            self.remove_locked(&mut inner, id);
114        }
115        if !evicted.is_empty() {
116            tracing::debug!(block_id, dropped = evicted.len(), "fs.watch: per-block cap reached, dropped the oldest");
117        }
118        tracing::debug!(dir = %dir.display(), block_id, watch_id, "fs.watch: started");
119        watch_id
120    }
121
122    /// Stop a watch. An unknown id (already dropped by the cap, or never
123    /// issued) is not an error.
124    pub fn unwatch(&self, watch_id: &str) {
125        let mut inner = self.lock();
126        let Some(block_id) = inner.watches.get(watch_id).map(|w| w.block_id.clone()) else {
127            return;
128        };
129        if let Some(ids) = inner.by_block.get_mut(&block_id) {
130            ids.retain(|id| id != watch_id);
131            if ids.is_empty() {
132                inner.by_block.remove(&block_id);
133            }
134        }
135        self.remove_locked(&mut inner, watch_id);
136    }
137
138    /// Drop `watch_id` from `watches` and `by_dir` and release its pool
139    /// subscription. The caller has already removed it from `by_block`.
140    fn remove_locked(&self, inner: &mut Inner, watch_id: &str) {
141        let Some(w) = inner.watches.remove(watch_id) else { return };
142        if let Some((_, ids)) = inner.by_dir.get_mut(&w.dir) {
143            ids.remove(watch_id);
144            if ids.is_empty() {
145                inner.by_dir.remove(&w.dir);
146            }
147        }
148        self.pool.unsubscribe(w.sub);
149    }
150
151    /// A raw pool event. A change to an entry means its parent folder's
152    /// listing changed; a change to a watched folder itself (removed,
153    /// renamed) is reported for that folder too, so its pane re-lists and
154    /// finds it gone.
155    fn on_fs_event(self: &Arc<Self>, path: &Path) {
156        let parent = path.parent();
157        let (parent_hit, self_hit) = {
158            let inner = self.lock();
159            // The pool is shared with every other watcher; most events are
160            // not for us.
161            if inner.by_dir.is_empty() {
162                return;
163            }
164            (parent.filter(|p| inner.by_dir.contains_key(*p)).map(Path::to_path_buf), inner.by_dir.contains_key(path))
165        };
166        // `notify` usually reports paths under the exact (canonical) path
167        // that was watched, but not always (symlinked ancestors, `\\?\`
168        // prefixing on Windows), so fall back to the canonical form, as
169        // `media_file_watcher` does. Outside the lock: it is a syscall.
170        let parent_hit = parent_hit.or_else(|| {
171            let canonical = parent?.canonicalize().ok()?;
172            self.lock().by_dir.contains_key(&canonical).then_some(canonical)
173        });
174        if let Some(dir) = parent_hit {
175            self.schedule(dir);
176        }
177        if self_hit {
178            self.schedule(path.to_path_buf());
179        }
180    }
181
182    fn on_lagged(self: &Arc<Self>) {
183        let dirs: Vec<PathBuf> = self.lock().by_dir.keys().cloned().collect();
184        for dir in dirs {
185            self.schedule(dir);
186        }
187    }
188
189    /// Publish for `dir` once `COALESCE` from now, unless a publish is
190    /// already scheduled, in which case this change rides along with it.
191    fn schedule(self: &Arc<Self>, dir: PathBuf) {
192        if !self.lock().pending.insert(dir.clone()) {
193            return;
194        }
195        let this = self.clone();
196        tokio::spawn(async move {
197            tokio::time::sleep(COALESCE).await;
198            let target = {
199                let mut inner = this.lock();
200                inner.pending.remove(&dir);
201                // Re-read the watchers now: one may have unwatched in the
202                // meantime.
203                inner.by_dir.get(&dir).map(|(display, ids)| {
204                    let blocks: HashSet<String> =
205                        ids.iter().filter_map(|id| inner.watches.get(id)).map(|w| w.block_id.clone()).collect();
206                    (display.clone(), blocks)
207                })
208            };
209            if let Some((display, blocks)) = target {
210                publish_files_changed(&this.broker, &display, blocks.iter());
211            }
212        });
213    }
214
215    #[cfg(test)]
216    fn watch_count(&self, block_id: &str) -> usize {
217        self.lock().by_block.get(block_id).map_or(0, VecDeque::len)
218    }
219}
220
221/// Publish `EVENT_FILES_CHANGED` for `display_dir`, scoped to each block.
222/// Never a global broadcast.
223fn publish_files_changed<'a>(broker: &Broker, display_dir: &str, block_ids: impl Iterator<Item = &'a String>) {
224    let scopes: Vec<String> = block_ids.map(|id| format!("block:{id}")).collect();
225    if scopes.is_empty() {
226        return;
227    }
228    broker.publish_event::<crate::backend::mps_events::FilesChanged>(
229        &crate::backend::mps_events::FilesChangedEvent { dir: display_dir.to_string() },
230        scopes,
231        0,
232    );
233}
234
235#[cfg(test)]
236mod tests {
237    use super::*;
238    use crate::backend::mps::MuxEvent;
239    use std::sync::Mutex as StdMutex;
240
241    struct TestClient {
242        events: StdMutex<Vec<(String, MuxEvent)>>,
243    }
244
245    impl super::super::mps::WpsClient for Arc<TestClient> {
246        fn send_event(&self, route_id: &str, event: MuxEvent) {
247            self.events.lock().unwrap().push((route_id.to_string(), event));
248        }
249    }
250
251    fn subscribed_broker(route: &str, block: &str) -> (Arc<Broker>, Arc<TestClient>) {
252        let broker = Arc::new(Broker::new());
253        let client = Arc::new(TestClient { events: StdMutex::new(Vec::new()) });
254        broker.set_client(Box::new(client.clone()));
255        broker.subscribe(
256            route,
257            super::super::mps::SubscriptionRequest {
258                event: EVENT_FILES_CHANGED.to_string(),
259                scopes: vec![format!("block:{block}")],
260                allscopes: false,
261            },
262        );
263        (broker, client)
264    }
265
266    #[tokio::test]
267    async fn each_block_keeps_at_most_the_cap_dropping_the_oldest() {
268        let watcher = FilesWatcher::new(FsWatchPool::new(), Arc::new(Broker::new()));
269        let dir = tempfile::tempdir().unwrap();
270        let canonical = dir.path().canonicalize().unwrap();
271        let first = watcher.watch(&canonical, "b1");
272        for _ in 0..MAX_WATCHES_PER_BLOCK {
273            watcher.watch(&canonical, "b1");
274        }
275        assert_eq!(watcher.watch_count("b1"), MAX_WATCHES_PER_BLOCK);
276        assert!(!watcher.lock().watches.contains_key(&first), "the oldest was dropped");
277
278        // Another block is unaffected by b1's cap.
279        watcher.watch(&canonical, "b2");
280        assert_eq!(watcher.watch_count("b2"), 1);
281    }
282
283    #[tokio::test]
284    async fn unwatch_releases_and_unknown_ids_are_fine() {
285        let watcher = FilesWatcher::new(FsWatchPool::new(), Arc::new(Broker::new()));
286        let dir = tempfile::tempdir().unwrap();
287        let canonical = dir.path().canonicalize().unwrap();
288        let a = watcher.watch(&canonical, "b1");
289        let b = watcher.watch(&canonical, "b2");
290        watcher.unwatch(&a);
291        assert!(watcher.lock().by_dir.contains_key(&canonical), "b2 still watches it");
292        watcher.unwatch(&b);
293        assert!(watcher.lock().by_dir.is_empty());
294        assert!(watcher.lock().by_block.is_empty());
295        watcher.unwatch(&b);
296        watcher.unwatch("never-issued");
297    }
298
299    #[test]
300    fn publish_is_scoped_to_the_watching_blocks() {
301        let (broker, client) = subscribed_broker("route-1", "abc");
302        publish_files_changed(&broker, "/tmp/x", ["abc".to_string(), "xyz".to_string()].iter());
303        let events = client.events.lock().unwrap();
304        assert_eq!(events.len(), 1);
305        assert_eq!(events[0].1.event, EVENT_FILES_CHANGED);
306        assert_eq!(events[0].1.data.as_ref().unwrap()["dir"], "/tmp/x");
307        assert_eq!(events[0].1.persist, 0);
308    }
309
310    #[tokio::test]
311    async fn a_real_change_publishes_once_for_its_folder() {
312        let (broker, client) = subscribed_broker("route-e2e", "e2e");
313        let watcher = FilesWatcher::new(FsWatchPool::new(), broker);
314        let dir = tempfile::tempdir().unwrap();
315        let canonical = dir.path().canonicalize().unwrap();
316        watcher.watch(&canonical, "e2e");
317        tokio::time::sleep(Duration::from_millis(200)).await;
318
319        // A burst of changes inside one coalescing window.
320        for i in 0..5 {
321            std::fs::write(canonical.join(format!("f{i}.txt")), "x").unwrap();
322        }
323        let saw = tokio::time::timeout(Duration::from_secs(5), async {
324            loop {
325                if !client.events.lock().unwrap().is_empty() {
326                    return true;
327                }
328                tokio::time::sleep(Duration::from_millis(25)).await;
329            }
330        })
331        .await
332        .unwrap_or(false);
333        assert!(saw, "a write in a watched folder must publish files:changed");
334
335        tokio::time::sleep(Duration::from_millis(400)).await;
336        let events = client.events.lock().unwrap();
337        assert!(events.len() <= 2, "a burst is coalesced, got {} events", events.len());
338        assert_eq!(events[0].1.data.as_ref().unwrap()["dir"], display_path(&canonical));
339    }
340
341    #[tokio::test]
342    async fn lag_rescans_every_watched_folder() {
343        let (broker, client) = subscribed_broker("route-lag", "lag");
344        let watcher = FilesWatcher::new(FsWatchPool::new(), broker);
345        let (d1, d2) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
346        watcher.watch(&d1.path().canonicalize().unwrap(), "lag");
347        watcher.watch(&d2.path().canonicalize().unwrap(), "lag");
348        watcher.on_lagged();
349        tokio::time::sleep(COALESCE + Duration::from_millis(200)).await;
350        let dirs: HashSet<String> = client
351            .events
352            .lock()
353            .unwrap()
354            .iter()
355            .map(|(_, e)| e.data.as_ref().unwrap()["dir"].as_str().unwrap().to_string())
356            .collect();
357        assert_eq!(dirs.len(), 2, "one event per watched folder");
358    }
359}
agentmux_srv/backend/files_watcher.rs

agentmux_srv/backend/
files_watcher.rs
