Fix fs/watch debounce batching (#24716)

## Summary

`fs/watch` was using a local debounce wrapper whose deadline was
initialized once and then reused after the first batch. Once that stale
deadline was in the past, later file changes could bypass the intended
200ms debounce and send noisier `fs/changed` notifications.

This moves the debounce wrapper into `codex-file-watcher` as
`DebouncedWatchReceiver`, resets the debounce deadline for each event
batch, preserves pending paths across cancelled receives, and updates
app-server `fs/watch` to use the shared wrapper.

Fixes #24692.

codex-rs/file-watcher/src/file_watcher_tests.rs

@@ -86,6 +86,62 @@ async fn throttled_receiver_flushes_pending_on_shutdown() {
     assert_eq!(closed, None);
 }
 
+#[tokio::test]
+async fn debounced_receiver_coalesces_each_event_batch() {
+    let (tx, rx) = watch_channel();
+    let mut debounced = DebouncedWatchReceiver::new(rx, TEST_THROTTLE_INTERVAL);
+
+    tx.add_changed_paths(&[path("a")]).await;
+    let first = timeout(TEST_THROTTLE_INTERVAL * 2, debounced.recv())
+        .await
+        .expect("first emit timeout");
+    assert_eq!(
+        first,
+        Some(FileWatcherEvent {
+            paths: vec![path("a")],
+        })
+    );
+
+    tx.add_changed_paths(&[path("c")]).await;
+    let blocked = timeout(TEST_THROTTLE_INTERVAL / 2, debounced.recv()).await;
+    assert_eq!(blocked.is_err(), true);
+
+    tx.add_changed_paths(&[path("d")]).await;
+    let second = timeout(TEST_THROTTLE_INTERVAL * 2, debounced.recv())
+        .await
+        .expect("second emit timeout");
+    assert_eq!(
+        second,
+        Some(FileWatcherEvent {
+            paths: vec![path("c"), path("d")],
+        })
+    );
+}
+
+#[tokio::test]
+async fn debounced_receiver_flushes_pending_on_shutdown() {
+    let (tx, rx) = watch_channel();
+    let mut debounced = DebouncedWatchReceiver::new(rx, TEST_THROTTLE_INTERVAL);
+
+    tx.add_changed_paths(&[path("a")]).await;
+    drop(tx);
+
+    let flushed = timeout(Duration::from_secs(1), debounced.recv())
+        .await
+        .expect("shutdown flush timeout");
+    assert_eq!(
+        flushed,
+        Some(FileWatcherEvent {
+            paths: vec![path("a")],
+        })
+    );
+
+    let closed = timeout(Duration::from_secs(1), debounced.recv())
+        .await
+        .expect("closed recv timeout");
+    assert_eq!(closed, None);
+}
+
 #[test]
 fn is_mutating_event_filters_non_mutating_event_kinds() {
     assert_eq!(

codex-rs/file-watcher/src/lib.rs

@@ -253,6 +253,49 @@ impl ThrottledWatchReceiver {
     }
 }
 
+/// Coalesces file watcher notifications that arrive within a fixed debounce
+/// window after the first event in each batch.
+pub struct DebouncedWatchReceiver {
+    rx: Receiver,
+    interval: Duration,
+    changed_paths: BTreeSet<PathBuf>,
+}
+
+impl DebouncedWatchReceiver {
+    /// Creates a debouncing wrapper around a raw watcher [`Receiver`].
+    pub fn new(rx: Receiver, interval: Duration) -> Self {
+        Self {
+            rx,
+            interval,
+            changed_paths: BTreeSet::new(),
+        }
+    }
+
+    /// Receives the next debounced event batch.
+    pub async fn recv(&mut self) -> Option<FileWatcherEvent> {
+        while self.changed_paths.is_empty() {
+            self.changed_paths.extend(self.rx.recv().await?.paths);
+        }
+        let deadline = Instant::now() + self.interval;
+
+        loop {
+            tokio::select! {
+                event = self.rx.recv() => match event {
+                    Some(event) => self.changed_paths.extend(event.paths),
+                    None => break,
+                },
+                _ = sleep_until(deadline) => break,
+            }
+        }
+
+        Some(FileWatcherEvent {
+            paths: std::mem::take(&mut self.changed_paths)
+                .into_iter()
+                .collect(),
+        })
+    }
+}
+
 /// Handle used to register watched paths for one logical consumer.
 pub struct FileWatcherSubscriber {
     id: SubscriberId,