agentmux_srv/backend/
editor_file_watcher.rs

1// Copyright 2025-2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Filesystem watcher for files open in editor/preview panes — detects
5//! external changes (an agent's `Edit`/`Write` tool, another process, a
6//! second AgentMux window) and pushes a per-block "this file changed on
7//! disk" wake signal so panes can refresh instead of silently going stale.
8//!
9//! Migrated onto the shared `fs_watch::FsWatchPool`
10//! (SPEC_SHARED_FS_WATCHER_FRAMEWORK_2026_08_07.md) — this module keeps only
11//! its own domain-specific bookkeeping (which block ids have which paths
12//! open, per-path debounce, and the `EVENT_EDITOR_FILE_CHANGED` publish
13//! shape). The actual `notify` construction, OS-level watch/unwatch, and
14//! recovery-on-failure now live in the pool, shared with every other
15//! watcher that migrates onto it (`config_watcher_fs.rs` already has;
16//! `media_file_watcher.rs`, below in the same PR, is the directory-mode
17//! sibling migrating alongside this one).
18//!
19//! Spec: docs/specs/SPEC_EDITOR_LIVE_FILE_RELOAD_2026_07_18.md
20
21use std::collections::HashMap;
22use std::path::{Path, PathBuf};
23use std::sync::atomic::{AtomicU64, Ordering};
24use std::sync::{Arc, Mutex};
25use std::time::Duration;
26
27use tokio::sync::broadcast;
28
29use super::fs_watch::{FsWatchEventKind, FsWatchPool, Subscription};
30use super::mps::Broker;
31
32/// MPS event fired when a file open in at least one editor tab changes on
33/// disk. Scoped per-block (`block:<id>`) via `MuxEvent::scopes`, matching
34/// `EVENT_CONTROLLER_STATUS`/`EVENT_BLOCK_ACTIVITY`'s existing pattern —
35/// never a global broadcast, so panes on unrelated files aren't notified.
36/// Payload is deliberately just the path (a wake signal, not content); the
37/// frontend re-fetches via the existing `readeditorfile` RPC.
38pub const EVENT_EDITOR_FILE_CHANGED: &str = "editor:file_changed";
39
40const DEBOUNCE: Duration = Duration::from_millis(300);
41
42struct Inner {
43    /// Canonicalized file path -> block ids with a tab open on it.
44    watched_paths: HashMap<PathBuf, std::collections::HashSet<String>>,
45    /// One pool subscription per watched path — held so it can be handed
46    /// back to `pool.unsubscribe` once the last block id for that path
47    /// unsubscribes. The pool does its own OS-level refcounting internally;
48    /// this is a *different* refcount (by block id, which the pool has no
49    /// concept of), so both layers of bookkeeping are genuinely needed.
50    pool_subs: HashMap<PathBuf, Subscription>,
51}
52
53/// Per-path debounce generation counters. A new fs event for a path bumps
54/// its counter; the delayed publish task only fires if its captured
55/// generation is still current when its sleep completes, so a burst of
56/// writes (e.g. an editor's multi-syscall save) collapses into one event.
57type DebounceGens = Mutex<HashMap<PathBuf, Arc<AtomicU64>>>;
58
59pub struct EditorFileWatcher {
60    pool: Arc<FsWatchPool>,
61    inner: Mutex<Inner>,
62    debounce_gens: DebounceGens,
63    broker: Arc<Broker>,
64}
65
66impl EditorFileWatcher {
67    /// Construct and start the watcher, subscribing to `pool`'s shared
68    /// broadcast stream. Unlike the pre-migration version, this always
69    /// succeeds — `FsWatchPool` itself absorbs the "can the OS backend even
70    /// construct" failure mode (see its own `health()`), so there's no
71    /// `Option` for callers to check anymore.
72    pub fn new(pool: Arc<FsWatchPool>, broker: Arc<Broker>) -> Arc<Self> {
73        let this = Arc::new(Self {
74            pool: pool.clone(),
75            inner: Mutex::new(Inner { watched_paths: HashMap::new(), pool_subs: HashMap::new() }),
76            debounce_gens: Mutex::new(HashMap::new()),
77            broker,
78        });
79
80        let mut events = pool.events();
81        let worker = this.clone();
82        tokio::spawn(async move {
83            loop {
84                match events.recv().await {
85                    // Match the pre-migration filter exactly: only Create/Modify
86                    // trigger a reload wake — a Removed event otherwise makes a
87                    // pane try to reload a file that's no longer there (reagent
88                    // P1 on PR #2458).
89                    Ok(ev) if matches!(ev.kind, FsWatchEventKind::Created | FsWatchEventKind::Modified) => {
90                        worker.handle_fs_event(ev.path)
91                    }
92                    Ok(_) => continue,
93                    Err(broadcast::error::RecvError::Closed) => break,
94                    Err(broadcast::error::RecvError::Lagged(skipped)) => {
95                        tracing::warn!(skipped, "editor file watcher lagged behind the fs_watch broadcast stream");
96                    }
97                }
98            }
99        });
100
101        this
102    }
103
104    /// Start watching `path` on behalf of `block_id`. Idempotent — calling
105    /// again for the same (path, block_id) is a no-op. Called from the
106    /// `watcheditorfile` RPC handler when a tab finishes loading a file.
107    pub fn watch_path(&self, path: &Path, block_id: &str) {
108        let canonical = path.canonicalize().unwrap_or_else(|_| path.to_path_buf());
109        let mut inner = self.inner.lock().unwrap();
110        let block_ids = inner.watched_paths.entry(canonical.clone()).or_default();
111        let is_new_path = block_ids.is_empty();
112        let newly_inserted = block_ids.insert(block_id.to_string());
113        if !is_new_path {
114            if newly_inserted {
115                tracing::debug!(path = %canonical.display(), block_id, "editor file watch: added subscriber");
116            }
117            return;
118        }
119
120        let sub = self.pool.subscribe_file(&canonical);
121        inner.pool_subs.insert(canonical.clone(), sub);
122        tracing::debug!(path = %canonical.display(), block_id, "editor file watch: started");
123    }
124
125    /// Stop watching `path` on behalf of `block_id`. No-op if that pairing
126    /// wasn't watched. Called on tab close / pane dispose.
127    pub fn unwatch_path(&self, path: &Path, block_id: &str) {
128        let canonical = path.canonicalize().unwrap_or_else(|_| path.to_path_buf());
129        let mut inner = self.inner.lock().unwrap();
130        let Some(block_ids) = inner.watched_paths.get_mut(&canonical) else {
131            return;
132        };
133        block_ids.remove(block_id);
134        if !block_ids.is_empty() {
135            return;
136        }
137        inner.watched_paths.remove(&canonical);
138        if let Some(sub) = inner.pool_subs.remove(&canonical) {
139            self.pool.unsubscribe(sub);
140        }
141        drop(inner);
142
143        // Prune the debounce-generation entry now that nothing watches this
144        // path — otherwise every distinct path ever externally modified
145        // during the process's lifetime accumulates here forever, even
146        // after its last watcher unsubscribed. Any debounce task already in
147        // flight still holds its own Arc clone of the counter, so this is
148        // safe to drop from the map immediately; a future re-watch of the
149        // same path just mints a fresh counter in handle_fs_event.
150        self.debounce_gens.lock().unwrap().remove(&canonical);
151
152        tracing::debug!(path = %canonical.display(), block_id, "editor file watch: stopped");
153    }
154
155    fn handle_fs_event(self: &Arc<Self>, changed_path: PathBuf) {
156        let canonical = changed_path.canonicalize().unwrap_or_else(|_| changed_path.clone());
157        let matched = {
158            let inner = self.inner.lock().unwrap();
159            // `notify` events aren't always pre-canonicalized (symlinked
160            // ancestors, `\\?\` UNC prefixing on Windows) — fall back to a
161            // raw-path match so we don't silently miss a watched file.
162            if inner.watched_paths.contains_key(&canonical) {
163                Some(canonical.clone())
164            } else if inner.watched_paths.contains_key(&changed_path) {
165                Some(changed_path.clone())
166            } else {
167                None
168            }
169        };
170        let Some(path) = matched else { return };
171
172        let gen_counter = {
173            let mut gens = self.debounce_gens.lock().unwrap();
174            gens.entry(path.clone())
175                .or_insert_with(|| Arc::new(AtomicU64::new(0)))
176                .clone()
177        };
178        let my_gen = gen_counter.fetch_add(1, Ordering::SeqCst) + 1;
179
180        let this = self.clone();
181        tokio::spawn(async move {
182            tokio::time::sleep(DEBOUNCE).await;
183            if gen_counter.load(Ordering::SeqCst) != my_gen {
184                return; // superseded by a later event for this path
185            }
186            let block_ids: Vec<String> = {
187                let inner = this.inner.lock().unwrap();
188                inner
189                    .watched_paths
190                    .get(&path)
191                    .map(|s| s.iter().cloned().collect())
192                    .unwrap_or_default()
193            };
194            if block_ids.is_empty() {
195                return; // last watcher unsubscribed while we were debouncing
196            }
197            // Belt-and-suspenders on top of the `Created | Modified` filter
198            // in `new()`'s event loop: on macOS, FSEvents (without the
199            // per-file-events flag `notify`'s recommended_watcher doesn't
200            // request) reports changes at directory granularity, and the
201            // crate's own heuristic for turning that into an EventKind can
202            // misclassify a plain `remove_file` as `Modify` rather than
203            // `Remove` — confirmed failing on macOS CI
204            // (test_removed_event_does_not_trigger_publish) while passing on
205            // Windows/Linux, whose backends report Remove correctly for this
206            // exact case. Re-check real existence at publish time rather
207            // than trusting the EventKind alone — this also correctly keeps
208            // firing for the legitimate "temp-write-then-rename" atomic-save
209            // pattern, since the file exists again well before DEBOUNCE
210            // elapses.
211            if !path.exists() {
212                return;
213            }
214            publish_editor_file_changed(&this.broker, &path, &block_ids);
215        });
216    }
217}
218
219/// Publish `EVENT_EDITOR_FILE_CHANGED`, scoped to every block that has a tab
220/// open on `path`. Mirrors `publish_block_activity`'s per-block scoping
221/// (`crates/srv/src/backend/mps.rs`) — not a global broadcast.
222fn publish_editor_file_changed(broker: &Broker, path: &Path, block_ids: &[String]) {
223    broker.publish_event::<crate::backend::mps_events::EditorFileChanged>(
224        &crate::backend::mps_events::PathChangedEvent { path: path.to_string_lossy().into_owned() },
225        block_ids.iter().map(|id| format!("block:{id}")).collect(),
226        0,
227    );
228}
229
230#[cfg(test)]
231mod tests {
232    use super::*;
233    use crate::backend::mps::MuxEvent;
234    use std::sync::Mutex as StdMutex;
235
236    struct TestClient {
237        events: StdMutex<Vec<(String, MuxEvent)>>,
238    }
239
240    impl super::super::mps::WpsClient for Arc<TestClient> {
241        fn send_event(&self, route_id: &str, event: MuxEvent) {
242            self.events.lock().unwrap().push((route_id.to_string(), event));
243        }
244    }
245
246    #[tokio::test]
247    async fn test_watch_unwatch_refcounting_does_not_panic() {
248        let broker = Arc::new(Broker::new());
249        let pool = FsWatchPool::new();
250        let watcher = EditorFileWatcher::new(pool, broker);
251
252        let tmp = std::env::temp_dir().join("agentmux_editor_watch_test.txt");
253        std::fs::write(&tmp, "hello").unwrap();
254
255        watcher.watch_path(&tmp, "block-1");
256        watcher.watch_path(&tmp, "block-2");
257        {
258            let inner = watcher.inner.lock().unwrap();
259            let canonical = tmp.canonicalize().unwrap();
260            assert_eq!(inner.watched_paths.get(&canonical).map(|s| s.len()), Some(2));
261        }
262
263        watcher.unwatch_path(&tmp, "block-1");
264        {
265            let inner = watcher.inner.lock().unwrap();
266            let canonical = tmp.canonicalize().unwrap();
267            assert_eq!(inner.watched_paths.get(&canonical).map(|s| s.len()), Some(1));
268        }
269
270        watcher.unwatch_path(&tmp, "block-2");
271        {
272            let inner = watcher.inner.lock().unwrap();
273            let canonical = tmp.canonicalize().unwrap();
274            assert!(!inner.watched_paths.contains_key(&canonical));
275        }
276
277        let _ = std::fs::remove_file(&tmp);
278    }
279
280    #[tokio::test]
281    async fn test_unwatch_prunes_debounce_gen_entry() {
282        // Regression: debounce_gens used to only ever grow — an entry was
283        // minted on the first fs event for a path but never removed when
284        // the last watcher for that path unsubscribed, so the map grew
285        // unbounded over a long-running process's lifetime.
286        let broker = Arc::new(Broker::new());
287        let pool = FsWatchPool::new();
288        let watcher = EditorFileWatcher::new(pool, broker);
289
290        let tmp = std::env::temp_dir().join("agentmux_editor_watch_debounce_test.txt");
291        std::fs::write(&tmp, "hello").unwrap();
292        let canonical = tmp.canonicalize().unwrap();
293
294        watcher.watch_path(&tmp, "block-1");
295        // Simulate a debounce-generation entry the way handle_fs_event would
296        // create one, without depending on a real fs-event round trip.
297        watcher.debounce_gens.lock().unwrap().insert(canonical.clone(), Arc::new(AtomicU64::new(3)));
298        assert!(watcher.debounce_gens.lock().unwrap().contains_key(&canonical));
299
300        watcher.unwatch_path(&tmp, "block-1");
301        assert!(
302            !watcher.debounce_gens.lock().unwrap().contains_key(&canonical),
303            "debounce_gens entry must be pruned once the last watcher for a path unsubscribes"
304        );
305
306        let _ = std::fs::remove_file(&tmp);
307    }
308
309    #[test]
310    fn test_publish_scopes_to_every_subscribed_block() {
311        let broker = Arc::new(Broker::new());
312        let client = Arc::new(TestClient { events: StdMutex::new(Vec::new()) });
313        broker.set_client(Box::new(client.clone()));
314
315        broker.subscribe(
316            "route-1",
317            super::super::mps::SubscriptionRequest {
318                event: EVENT_EDITOR_FILE_CHANGED.to_string(),
319                scopes: vec!["block:abc".to_string()],
320                allscopes: false,
321            },
322        );
323
324        publish_editor_file_changed(&broker, Path::new("/tmp/foo.md"), &["abc".to_string(), "xyz".to_string()]);
325
326        let events = client.events.lock().unwrap();
327        assert_eq!(events.len(), 1);
328        assert_eq!(events[0].0, "route-1");
329        assert_eq!(events[0].1.event, EVENT_EDITOR_FILE_CHANGED);
330    }
331
332    #[tokio::test]
333    async fn test_end_to_end_real_fs_write_triggers_publish() {
334        // Regression for the migration itself: a real notify event flowing
335        // through the shared pool must still reach this watcher's own
336        // debounce + publish path, not just the bookkeeping unit tests above.
337        let broker = Arc::new(Broker::new());
338        let client = Arc::new(TestClient { events: StdMutex::new(Vec::new()) });
339        broker.set_client(Box::new(client.clone()));
340        broker.subscribe(
341            "route-e2e",
342            super::super::mps::SubscriptionRequest {
343                event: EVENT_EDITOR_FILE_CHANGED.to_string(),
344                scopes: vec!["block:e2e".to_string()],
345                allscopes: false,
346            },
347        );
348
349        let pool = FsWatchPool::new();
350        let watcher = EditorFileWatcher::new(pool, broker);
351
352        let dir = std::env::temp_dir().join("agentmux_editor_watch_e2e_test");
353        let _ = std::fs::remove_dir_all(&dir);
354        std::fs::create_dir_all(&dir).unwrap();
355        let file = dir.join("watched.md");
356        std::fs::write(&file, "v1").unwrap();
357
358        watcher.watch_path(&file, "e2e");
359        tokio::time::sleep(Duration::from_millis(200)).await;
360        std::fs::write(&file, "v2").unwrap();
361
362        let saw_it = tokio::time::timeout(Duration::from_secs(5), async {
363            loop {
364                if !client.events.lock().unwrap().is_empty() {
365                    return true;
366                }
367                tokio::time::sleep(Duration::from_millis(50)).await;
368            }
369        })
370        .await
371        .unwrap_or(false);
372
373        assert!(saw_it, "expected a real on-disk change to reach the publish path via the shared pool");
374
375        let _ = std::fs::remove_dir_all(&dir);
376    }
377
378    #[tokio::test]
379    async fn test_removed_event_does_not_trigger_publish() {
380        // reagent P1 on PR #2458: the migration to FsWatchPool initially
381        // dropped the pre-migration Create/Modify-only filter, so a Removed
382        // event (deleting the watched file) would wake the pane to reload a
383        // file that's no longer there. Confirm the filter is back.
384        let broker = Arc::new(Broker::new());
385        let client = Arc::new(TestClient { events: StdMutex::new(Vec::new()) });
386        broker.set_client(Box::new(client.clone()));
387        broker.subscribe(
388            "route-removed",
389            super::super::mps::SubscriptionRequest {
390                event: EVENT_EDITOR_FILE_CHANGED.to_string(),
391                scopes: vec!["block:removed".to_string()],
392                allscopes: false,
393            },
394        );
395
396        let pool = FsWatchPool::new();
397        let watcher = EditorFileWatcher::new(pool, broker);
398
399        let dir = std::env::temp_dir().join("agentmux_editor_watch_removed_test");
400        let _ = std::fs::remove_dir_all(&dir);
401        std::fs::create_dir_all(&dir).unwrap();
402        let file = dir.join("watched.md");
403        std::fs::write(&file, "v1").unwrap();
404
405        watcher.watch_path(&file, "removed");
406        tokio::time::sleep(Duration::from_millis(200)).await;
407        std::fs::remove_file(&file).unwrap();
408
409        // Give the pool's fs backend time to deliver (and this watcher time
410        // to wrongly act on) the Removed event before asserting silence.
411        tokio::time::sleep(Duration::from_millis(500)).await;
412        assert!(
413            client.events.lock().unwrap().is_empty(),
414            "a Removed event must not trigger EVENT_EDITOR_FILE_CHANGED"
415        );
416
417        let _ = std::fs::remove_dir_all(&dir);
418    }
419}
agentmux_srv/backend/editor_file_watcher.rs

agentmux_srv/backend/
editor_file_watcher.rs
