agentmux_srv/backend/
media_file_watcher.rs

1// Copyright 2025-2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Filesystem watcher for directories a Media pane is pointed at — detects
5//! new/changed media files (e.g. a freshly-downloaded ComfyUI render
6//! landing in a project's `clips/` folder) and pushes a per-block "a
7//! matching file changed" wake signal so panes update without a manual
8//! reload.
9//!
10//! Migrated onto the shared `fs_watch::FsWatchPool`
11//! (SPEC_SHARED_FS_WATCHER_FRAMEWORK_2026_08_07.md), alongside
12//! `editor_file_watcher.rs` in the same PR. Directory-mode sibling of that
13//! file: that one watches individually-opened files (one entry per open
14//! tab); this one watches whole directories directly, filtered by a
15//! per-subscriber extension set, since a Media pane's job is "show me the
16//! latest render in this folder," not "watch this one exact file."
17//!
18//! Spec: docs/specs/SPEC_MEDIA_PANE_2026_07_26.md
19
20use std::collections::{HashMap, HashSet};
21use std::path::{Path, PathBuf};
22use std::sync::atomic::{AtomicU64, Ordering};
23use std::sync::{Arc, Mutex};
24use std::time::Duration;
25
26use tokio::sync::broadcast;
27
28use super::fs_watch::{FsWatchEventKind, FsWatchPool, Subscription};
29use super::mps::Broker;
30
31/// MPS event fired when a file matching a Media pane's extension filter is
32/// created/modified inside a directory that pane is watching. Scoped
33/// per-block (`block:<id>`) via `MuxEvent::scopes`, matching
34/// `EVENT_EDITOR_FILE_CHANGED`'s pattern. Payload is just the changed file's
35/// path — a wake signal, not content; the frontend re-fetches via
36/// `GET /agentmux/stream-local-file`.
37pub const EVENT_MEDIA_FILE_CHANGED: &str = "media:file_changed";
38
39const DEBOUNCE: Duration = Duration::from_millis(300);
40
41struct Inner {
42    /// Canonicalized directory -> (block id -> lowercase extensions that
43    /// block cares about, no leading dot). A directory has a live pool
44    /// subscription exactly while this map has at least one entry for it.
45    watched_dirs: HashMap<PathBuf, HashMap<String, HashSet<String>>>,
46    /// One pool subscription per watched directory — see
47    /// `editor_file_watcher.rs::Inner::pool_subs`'s doc comment for why this
48    /// bookkeeping is separate from the pool's own OS-level refcounting.
49    pool_subs: HashMap<PathBuf, Subscription>,
50}
51
52/// Per-file debounce generation counters, same collapsing-burst-writes
53/// purpose as `editor_file_watcher.rs`'s, keyed by the full changed-file
54/// path (not the directory) so unrelated files in the same watched
55/// directory debounce independently.
56type DebounceGens = Mutex<HashMap<PathBuf, Arc<AtomicU64>>>;
57
58pub struct MediaFileWatcher {
59    pool: Arc<FsWatchPool>,
60    inner: Mutex<Inner>,
61    debounce_gens: DebounceGens,
62    broker: Arc<Broker>,
63}
64
65impl MediaFileWatcher {
66    /// Construct and start the watcher, subscribing to `pool`'s shared
67    /// broadcast stream. Always succeeds — see `EditorFileWatcher::new`'s
68    /// doc comment for why there's no `Option` to check anymore.
69    pub fn new(pool: Arc<FsWatchPool>, broker: Arc<Broker>) -> Arc<Self> {
70        let this = Arc::new(Self {
71            pool: pool.clone(),
72            inner: Mutex::new(Inner { watched_dirs: HashMap::new(), pool_subs: HashMap::new() }),
73            debounce_gens: Mutex::new(HashMap::new()),
74            broker,
75        });
76
77        let mut events = pool.events();
78        let worker = this.clone();
79        tokio::spawn(async move {
80            loop {
81                match events.recv().await {
82                    // Match the pre-migration filter exactly: only Create/Modify
83                    // trigger a reload publish — a Removed event otherwise makes
84                    // a pane try to load a file that's no longer there (reagent
85                    // P1 on PR #2458).
86                    Ok(ev) if matches!(ev.kind, FsWatchEventKind::Created | FsWatchEventKind::Modified) => {
87                        worker.handle_fs_event(ev.path)
88                    }
89                    Ok(_) => continue,
90                    Err(broadcast::error::RecvError::Closed) => break,
91                    Err(broadcast::error::RecvError::Lagged(skipped)) => {
92                        tracing::warn!(skipped, "media file watcher lagged behind the fs_watch broadcast stream");
93                    }
94                }
95            }
96        });
97
98        this
99    }
100
101    /// Start watching `dir` on behalf of `block_id`, notifying only for
102    /// files whose extension (lowercase, no dot — e.g. "webm") appears in
103    /// `extensions`. Idempotent for the same (dir, block_id) — a second call
104    /// replaces that block's extension set rather than adding a duplicate
105    /// subscription. Called when a Media pane points at a directory.
106    pub fn watch_directory(&self, dir: &Path, block_id: &str, extensions: &[String]) {
107        let canonical = dir.canonicalize().unwrap_or_else(|_| dir.to_path_buf());
108        let ext_set: HashSet<String> = extensions.iter().map(|e| e.to_lowercase()).collect();
109
110        let mut inner = self.inner.lock().unwrap();
111        let subscribers = inner.watched_dirs.entry(canonical.clone()).or_default();
112        let is_new_dir = subscribers.is_empty();
113        subscribers.insert(block_id.to_string(), ext_set);
114
115        if is_new_dir {
116            let sub = self.pool.subscribe_dir(&canonical);
117            inner.pool_subs.insert(canonical.clone(), sub);
118        }
119        tracing::debug!(dir = %canonical.display(), block_id, "media dir watch: started/updated");
120    }
121
122    /// Stop watching `dir` on behalf of `block_id`. No-op if that pairing
123    /// wasn't watched. Called on path change / pane dispose.
124    pub fn unwatch_directory(&self, dir: &Path, block_id: &str) {
125        let canonical = dir.canonicalize().unwrap_or_else(|_| dir.to_path_buf());
126        let mut inner = self.inner.lock().unwrap();
127        let Some(subscribers) = inner.watched_dirs.get_mut(&canonical) else {
128            return;
129        };
130        subscribers.remove(block_id);
131        if !subscribers.is_empty() {
132            return;
133        }
134        inner.watched_dirs.remove(&canonical);
135        if let Some(sub) = inner.pool_subs.remove(&canonical) {
136            self.pool.unsubscribe(sub);
137        }
138        drop(inner);
139
140        // Prune debounce entries for files under this now-unwatched dir —
141        // same rationale as EditorFileWatcher::unwatch_path: don't let
142        // debounce_gens grow unbounded across a long process lifetime.
143        let mut gens = self.debounce_gens.lock().unwrap();
144        gens.retain(|path, _| path.parent() != Some(canonical.as_path()));
145
146        tracing::debug!(dir = %canonical.display(), block_id, "media dir watch: stopped");
147    }
148
149    fn handle_fs_event(self: &Arc<Self>, changed_path: PathBuf) {
150        let Some(dir) = changed_path.parent().map(Path::to_path_buf) else {
151            return;
152        };
153        let canonical_dir = dir.canonicalize().unwrap_or_else(|_| dir.clone());
154
155        let matched_block_ids: Vec<String> = {
156            let inner = self.inner.lock().unwrap();
157            // `notify` events aren't always pre-canonicalized (symlinked
158            // ancestors, `\\?\` UNC prefixing on Windows) — try both,
159            // mirroring EditorFileWatcher::handle_fs_event.
160            let subscribers = inner
161                .watched_dirs
162                .get(&canonical_dir)
163                .or_else(|| inner.watched_dirs.get(&dir));
164            let Some(subscribers) = subscribers else { return };
165
166            let ext = changed_path
167                .extension()
168                .and_then(|e| e.to_str())
169                .map(|e| e.to_lowercase())
170                .unwrap_or_default();
171            if ext.is_empty() {
172                return;
173            }
174            subscribers
175                .iter()
176                .filter(|(_, exts)| exts.contains(&ext))
177                .map(|(block_id, _)| block_id.clone())
178                .collect()
179        };
180        if matched_block_ids.is_empty() {
181            return;
182        }
183
184        let gen_counter = {
185            let mut gens = self.debounce_gens.lock().unwrap();
186            gens.entry(changed_path.clone())
187                .or_insert_with(|| Arc::new(AtomicU64::new(0)))
188                .clone()
189        };
190        let my_gen = gen_counter.fetch_add(1, Ordering::SeqCst) + 1;
191
192        let this = self.clone();
193        tokio::spawn(async move {
194            tokio::time::sleep(DEBOUNCE).await;
195            if gen_counter.load(Ordering::SeqCst) != my_gen {
196                return; // superseded by a later event for this file
197            }
198            // Belt-and-suspenders on top of the `Created | Modified` filter
199            // in `new()`'s event loop — see EditorFileWatcher::handle_fs_event's
200            // matching comment: on macOS, FSEvents' directory-granularity
201            // reporting can get a plain file removal reclassified as
202            // `Modify` by `notify`'s own heuristic (confirmed failing on
203            // macOS CI, test_removed_event_does_not_trigger_publish; passes
204            // on Windows/Linux). Re-check real existence at publish time
205            // instead of trusting the EventKind alone.
206            if !changed_path.exists() {
207                return;
208            }
209            publish_media_file_changed(&this.broker, &changed_path, &matched_block_ids);
210        });
211    }
212}
213
214/// Publish `EVENT_MEDIA_FILE_CHANGED`, scoped to every block watching the
215/// changed file's directory with a matching extension. Mirrors
216/// `publish_editor_file_changed`'s per-block scoping — never a global
217/// broadcast.
218fn publish_media_file_changed(broker: &Broker, path: &Path, block_ids: &[String]) {
219    broker.publish_event::<crate::backend::mps_events::MediaFileChanged>(
220        &crate::backend::mps_events::PathChangedEvent { path: path.to_string_lossy().into_owned() },
221        block_ids.iter().map(|id| format!("block:{id}")).collect(),
222        0,
223    );
224}
225
226#[cfg(test)]
227mod tests {
228    use super::*;
229    use crate::backend::mps::MuxEvent;
230    use std::sync::Mutex as StdMutex;
231
232    struct TestClient {
233        events: StdMutex<Vec<(String, MuxEvent)>>,
234    }
235
236    impl super::super::mps::WpsClient for Arc<TestClient> {
237        fn send_event(&self, route_id: &str, event: MuxEvent) {
238            self.events.lock().unwrap().push((route_id.to_string(), event));
239        }
240    }
241
242    #[tokio::test]
243    async fn test_watch_unwatch_refcounting_does_not_panic() {
244        let broker = Arc::new(Broker::new());
245        let pool = FsWatchPool::new();
246        let watcher = MediaFileWatcher::new(pool, broker);
247
248        let tmp_dir = std::env::temp_dir().join("agentmux_media_watch_test");
249        std::fs::create_dir_all(&tmp_dir).unwrap();
250        let exts = vec!["webm".to_string(), "png".to_string()];
251
252        watcher.watch_directory(&tmp_dir, "block-1", &exts);
253        watcher.watch_directory(&tmp_dir, "block-2", &exts);
254        {
255            let inner = watcher.inner.lock().unwrap();
256            let canonical = tmp_dir.canonicalize().unwrap();
257            assert_eq!(inner.watched_dirs.get(&canonical).map(|s| s.len()), Some(2));
258        }
259
260        watcher.unwatch_directory(&tmp_dir, "block-1");
261        {
262            let inner = watcher.inner.lock().unwrap();
263            let canonical = tmp_dir.canonicalize().unwrap();
264            assert_eq!(inner.watched_dirs.get(&canonical).map(|s| s.len()), Some(1));
265        }
266
267        watcher.unwatch_directory(&tmp_dir, "block-2");
268        {
269            let inner = watcher.inner.lock().unwrap();
270            let canonical = tmp_dir.canonicalize().unwrap();
271            assert!(!inner.watched_dirs.contains_key(&canonical));
272        }
273
274        let _ = std::fs::remove_dir_all(&tmp_dir);
275    }
276
277    #[test]
278    fn test_publish_scopes_to_every_subscribed_block() {
279        let broker = Arc::new(Broker::new());
280        let client = Arc::new(TestClient { events: StdMutex::new(Vec::new()) });
281        broker.set_client(Box::new(client.clone()));
282
283        broker.subscribe(
284            "route-1",
285            super::super::mps::SubscriptionRequest {
286                event: EVENT_MEDIA_FILE_CHANGED.to_string(),
287                scopes: vec!["block:abc".to_string()],
288                allscopes: false,
289            },
290        );
291
292        publish_media_file_changed(&broker, Path::new("/tmp/clips/shot.webm"), &["abc".to_string(), "xyz".to_string()]);
293
294        let events = client.events.lock().unwrap();
295        assert_eq!(events.len(), 1);
296        assert_eq!(events[0].0, "route-1");
297        assert_eq!(events[0].1.event, EVENT_MEDIA_FILE_CHANGED);
298    }
299
300    #[test]
301    fn test_extension_filter_excludes_non_matching_block() {
302        // A block watching only "png" should not be scoped in when a
303        // ".webm" file changes in the same directory.
304        let broker = Arc::new(Broker::new());
305        let client = Arc::new(TestClient { events: StdMutex::new(Vec::new()) });
306        broker.set_client(Box::new(client.clone()));
307
308        broker.subscribe(
309            "route-png-only",
310            super::super::mps::SubscriptionRequest {
311                event: EVENT_MEDIA_FILE_CHANGED.to_string(),
312                scopes: vec!["block:png-block".to_string()],
313                allscopes: false,
314            },
315        );
316
317        // Simulate what handle_fs_event's filtering step would decide: only
318        // "webm-block" matched, "png-block" did not, so only it is published.
319        publish_media_file_changed(&broker, Path::new("/tmp/clips/shot.webm"), &["webm-block".to_string()]);
320
321        let events = client.events.lock().unwrap();
322        assert_eq!(events.len(), 0, "png-only block must not receive a .webm change event");
323    }
324
325    #[tokio::test]
326    async fn test_end_to_end_real_fs_write_triggers_publish_for_matching_extension() {
327        // Regression for the migration itself: a real notify event flowing
328        // through the shared pool must still reach this watcher's own
329        // extension-filter + debounce + publish path.
330        let broker = Arc::new(Broker::new());
331        let client = Arc::new(TestClient { events: StdMutex::new(Vec::new()) });
332        broker.set_client(Box::new(client.clone()));
333        broker.subscribe(
334            "route-e2e",
335            super::super::mps::SubscriptionRequest {
336                event: EVENT_MEDIA_FILE_CHANGED.to_string(),
337                scopes: vec!["block:e2e".to_string()],
338                allscopes: false,
339            },
340        );
341
342        let pool = FsWatchPool::new();
343        let watcher = MediaFileWatcher::new(pool, broker);
344
345        let dir = std::env::temp_dir().join("agentmux_media_watch_e2e_test");
346        let _ = std::fs::remove_dir_all(&dir);
347        std::fs::create_dir_all(&dir).unwrap();
348
349        watcher.watch_directory(&dir, "e2e", &["webm".to_string()]);
350        tokio::time::sleep(Duration::from_millis(200)).await;
351        std::fs::write(dir.join("render.webm"), b"fake video bytes").unwrap();
352
353        let saw_it = tokio::time::timeout(Duration::from_secs(5), async {
354            loop {
355                if !client.events.lock().unwrap().is_empty() {
356                    return true;
357                }
358                tokio::time::sleep(Duration::from_millis(50)).await;
359            }
360        })
361        .await
362        .unwrap_or(false);
363
364        assert!(saw_it, "expected a real .webm write to reach the publish path via the shared pool");
365
366        let _ = std::fs::remove_dir_all(&dir);
367    }
368
369    #[tokio::test]
370    async fn test_removed_event_does_not_trigger_publish() {
371        // reagent P1 on PR #2458: the migration to FsWatchPool initially
372        // dropped the pre-migration Create/Modify-only filter, so deleting a
373        // file in a watched media dir would still publish
374        // EVENT_MEDIA_FILE_CHANGED, sending the pane to reload a file that's
375        // no longer there. Confirm the filter is back.
376        let broker = Arc::new(Broker::new());
377        let client = Arc::new(TestClient { events: StdMutex::new(Vec::new()) });
378        broker.set_client(Box::new(client.clone()));
379        broker.subscribe(
380            "route-removed",
381            super::super::mps::SubscriptionRequest {
382                event: EVENT_MEDIA_FILE_CHANGED.to_string(),
383                scopes: vec!["block:removed".to_string()],
384                allscopes: false,
385            },
386        );
387
388        let pool = FsWatchPool::new();
389        let watcher = MediaFileWatcher::new(pool, broker);
390
391        let dir = std::env::temp_dir().join("agentmux_media_watch_removed_test");
392        let _ = std::fs::remove_dir_all(&dir);
393        std::fs::create_dir_all(&dir).unwrap();
394        let file = dir.join("render.webm");
395        std::fs::write(&file, b"fake video bytes").unwrap();
396
397        watcher.watch_directory(&dir, "removed", &["webm".to_string()]);
398        tokio::time::sleep(Duration::from_millis(200)).await;
399        std::fs::remove_file(&file).unwrap();
400
401        // Give the pool's fs backend time to deliver (and this watcher time
402        // to wrongly act on) the Removed event before asserting silence.
403        tokio::time::sleep(Duration::from_millis(500)).await;
404        assert!(
405            client.events.lock().unwrap().is_empty(),
406            "a Removed event must not trigger EVENT_MEDIA_FILE_CHANGED"
407        );
408
409        let _ = std::fs::remove_dir_all(&dir);
410    }
411}
agentmux_srv/backend/media_file_watcher.rs

agentmux_srv/backend/
media_file_watcher.rs
