agentmux_srv/backend/fs_watch/

pool.rs

// Copyright 2026, AgentMux Corp.
// SPDX-License-Identifier: Apache-2.0

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use notify::{EventKind, RecursiveMode, Watcher};
use tokio::sync::{broadcast, mpsc};

use super::recovery::{HealthState, WatchBackend, FsWatchHealth, RETRY_BACKOFF, HEALTH_SWEEP_INTERVAL};

/// Broadcast channel capacity. Generous relative to real fs-event volume
/// (file-save bursts, not high-frequency streams) — a slow consumer would
/// need to fall behind by this many raw events before `RecvError::Lagged`,
/// at which point it should resync from scratch rather than assume it saw
/// everything (same "wake signal, not a guaranteed delivery log" contract
/// the existing three watchers already have via their own debounce logic).
const EVENT_CHANNEL_CAPACITY: usize = 1024;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FsWatchEventKind {
    Created,
    Modified,
    Removed,
    Other,
}

impl From<EventKind> for FsWatchEventKind {
    fn from(kind: EventKind) -> Self {
        match kind {
            EventKind::Create(_) => FsWatchEventKind::Created,
            EventKind::Modify(_) => FsWatchEventKind::Modified,
            EventKind::Remove(_) => FsWatchEventKind::Removed,
            _ => FsWatchEventKind::Other,
        }
    }
}

#[derive(Debug, Clone)]
pub struct FsWatchEvent {
    pub path: PathBuf,
    pub kind: FsWatchEventKind,
}

/// Handle returned by `subscribe_file`/`subscribe_dir`. Holding this alive
/// is not required (unlike the raw `notify::Watcher`, which must be kept
/// alive by its owner) — the pool itself owns the underlying watcher.
/// `unsubscribe` takes this by value so a caller can't accidentally
/// unsubscribe the same handle twice.
#[derive(Debug, Clone)]
pub struct Subscription {
    /// What's actually registered with the OS watcher — the parent
    /// directory for a file subscription (see module doc: this is the
    /// baked-in atomic-rename-safety rule), or the directory itself for a
    /// directory subscription.
    watch_target: PathBuf,
}

struct WatchEntry {
    refcount: usize,
    mode: RecursiveMode,
}

struct Inner {
    watcher: Option<Box<dyn Watcher + Send>>,
    targets: HashMap<PathBuf, WatchEntry>,
    /// Polls WSL share paths (see [`Inner::watcher_for`]); built on first use.
    share_poller: Option<Box<dyn Watcher + Send>>,
    /// Where the share poller's events go: the same stream as `watcher`'s.
    raw_tx: mpsc::UnboundedSender<notify::Result<notify::Event>>,
}

/// How often a WSL share folder is polled for changes.
const SHARE_POLL_INTERVAL: Duration = Duration::from_secs(2);

impl Inner {
    /// The watcher for `target`. A WSL distro's folder through
    /// `\\wsl.localhost` (`remote::wsl_fs`) gets a polling watcher: Windows
    /// delivers no change notifications for that share (checked on Windows 11:
    /// none for a change made in the distro, and none for one made through the
    /// share), so a native watch there would succeed and stay silent.
    fn watcher_for(&mut self, target: &Path) -> Option<&mut Box<dyn Watcher + Send>> {
        if crate::backend::remote::wsl_fs::share_of(target).is_none() {
            return self.watcher.as_mut();
        }
        if self.share_poller.is_none() {
            let tx = self.raw_tx.clone();
            let config = notify::Config::default().with_poll_interval(SHARE_POLL_INTERVAL);
            match notify::PollWatcher::new(move |res: notify::Result<notify::Event>| { let _ = tx.send(res); }, config) {
                Ok(w) => self.share_poller = Some(Box::new(w)),
                Err(e) => tracing::warn!(error = %e, "fs_watch: no polling watcher for WSL folders"),
            }
        }
        self.share_poller.as_mut()
    }
}

pub struct FsWatchPool {
    inner: Mutex<Inner>,
    health: HealthState,
    backend: WatchBackend,
    tx: broadcast::Sender<FsWatchEvent>,
}

impl FsWatchPool {
    /// Construct the pool and start its background event-bridging and
    /// health-sweep tasks. Always succeeds — if even the `PollWatcher`
    /// fallback fails to construct (practically never happens; it's pure
    /// userspace polling), the pool still comes up with `watcher: None` and
    /// every subscribe call fails cleanly into `degraded_paths` rather than
    /// panicking or requiring every caller to handle an `Option<Arc<Self>>`
    /// at the top level. Live-update is additive everywhere it's used.
    pub fn new() -> Arc<Self> {
        let (raw_tx, mut raw_rx) = mpsc::unbounded_channel::<notify::Result<notify::Event>>();
        let (watcher, backend) = construct_watcher(raw_tx.clone());

        let (broadcast_tx, _) = broadcast::channel(EVENT_CHANNEL_CAPACITY);

        let this = Arc::new(Self {
            inner: Mutex::new(Inner { watcher, targets: HashMap::new(), share_poller: None, raw_tx }),
            health: HealthState::default(),
            backend,
            tx: broadcast_tx,
        });

        // Bridge notify's sync callback (already funneled into raw_rx by
        // construct_watcher's closure) into the broadcast stream. Mirrors
        // the sync-callback -> mpsc -> async-task shape all three existing
        // watchers already use, just shared once here instead of three
        // times.
        let bridge = this.clone();
        tokio::spawn(async move {
            while let Some(res) = raw_rx.recv().await {
                match res {
                    Ok(event) => {
                        let kind = FsWatchEventKind::from(event.kind);
                        if matches!(kind, FsWatchEventKind::Other) {
                            continue;
                        }
                        for path in event.paths {
                            let _ = bridge.tx.send(FsWatchEvent { path, kind });
                        }
                    }
                    Err(e) => bridge.handle_backend_error(e),
                }
            }
        });

        // Periodic self-healing sweep, scoped to already-degraded targets —
        // see `sweep()`'s own doc comment for the full rationale (a
        // healthy-looking target is deliberately left untouched, to avoid
        // both a handle leak and a real event-coverage gap notify's Windows
        // backend would otherwise cost on every tick).
        let sweeper = this.clone();
        tokio::spawn(async move {
            let mut tick = tokio::time::interval(HEALTH_SWEEP_INTERVAL);
            loop {
                tick.tick().await;
                sweeper.sweep();
            }
        });

        this
    }

    /// Watch the file at `path` (specifically: its parent directory,
    /// non-recursively — see module doc). Returns a fresh `Subscription`
    /// even when the underlying OS watch is currently failing (tracked in
    /// `degraded_paths` instead) — a degraded subscription still becomes
    /// live automatically once retry/sweep succeeds, with no action needed
    /// from the caller.
    pub fn subscribe_file(self: &Arc<Self>, path: &Path) -> Subscription {
        let canonical = path.canonicalize().unwrap_or_else(|_| path.to_path_buf());
        let target = canonical.parent().map(Path::to_path_buf).unwrap_or(canonical.clone());
        self.subscribe_target(target, RecursiveMode::NonRecursive)
    }

    /// Watch the directory at `path` directly (non-recursive — matching
    /// `media_file_watcher.rs`'s existing behavior; extension/content
    /// filtering is the caller's concern, not the pool's).
    pub fn subscribe_dir(self: &Arc<Self>, path: &Path) -> Subscription {
        let canonical = path.canonicalize().unwrap_or_else(|_| path.to_path_buf());
        self.subscribe_target(canonical, RecursiveMode::NonRecursive)
    }

    /// Watch `path` and everything under it (widget packages: an edit deep in
    /// a package folder reloads it).
    pub fn subscribe_dir_recursive(self: &Arc<Self>, path: &Path) -> Subscription {
        let canonical = path.canonicalize().unwrap_or_else(|_| path.to_path_buf());
        self.subscribe_target(canonical, RecursiveMode::Recursive)
    }

    fn subscribe_target(self: &Arc<Self>, target: PathBuf, mode: RecursiveMode) -> Subscription {
        let is_new = {
            let mut inner = self.inner.lock().unwrap();
            let entry = inner.targets.entry(target.clone()).or_insert_with(|| WatchEntry { refcount: 0, mode });
            entry.refcount += 1;
            entry.refcount == 1
        };

        if is_new {
            self.start_watch(target.clone(), mode);
        }

        Subscription { watch_target: target }
    }

    /// Stop watching on behalf of `sub`. Idempotent-safe: unsubscribing the
    /// same handle twice would underflow the refcount, so this takes `self`
    /// by shared ref but `sub` by value specifically to discourage reuse —
    /// callers that need multiple independent subscriptions to the same
    /// path should call `subscribe_file`/`subscribe_dir` once per logical
    /// subscriber, matching how `editor_file_watcher.rs` already tracks one
    /// entry per `block_id` today.
    pub fn unsubscribe(&self, sub: Subscription) {
        let should_unwatch = {
            let mut inner = self.inner.lock().unwrap();
            let Some(entry) = inner.targets.get_mut(&sub.watch_target) else {
                return;
            };
            entry.refcount = entry.refcount.saturating_sub(1);
            let empty = entry.refcount == 0;
            if empty {
                inner.targets.remove(&sub.watch_target);
            }
            empty
        };

        if should_unwatch {
            let mut inner = self.inner.lock().unwrap();
            if let Some(w) = inner.watcher_for(&sub.watch_target) {
                let _ = w.unwatch(&sub.watch_target);
            }
            drop(inner);
            self.health.clear_degraded(&sub.watch_target);
        }
    }

    /// Raw change events for every currently-watched target, pool-wide.
    /// Each call returns an independent receiver — subscribe to this once
    /// per domain module (not once per file), and filter down to the paths
    /// that module actually cares about, same as today's per-watcher
    /// `Inner.watched_paths` filtering.
    pub fn events(&self) -> broadcast::Receiver<FsWatchEvent> {
        self.tx.subscribe()
    }

    #[allow(dead_code)] // Diagnostics surface, SPEC_SHARED_FS_WATCHER_FRAMEWORK_2026_08_07.md §4 point 4.
    pub fn health(&self) -> FsWatchHealth {
        let active_watches = self.inner.lock().unwrap().targets.len();
        FsWatchHealth {
            backend: self.backend,
            active_watches,
            degraded_paths: self.health.snapshot(),
        }
    }

    /// First attempt at a new target, plus a bounded retry-with-backoff if
    /// it fails — see `RETRY_BACKOFF`'s doc comment. Runs the retries on a
    /// spawned task so `subscribe_file`/`subscribe_dir` never blocks on it.
    fn start_watch(self: &Arc<Self>, target: PathBuf, mode: RecursiveMode) {
        let first_attempt = self.try_watch(&target, mode);
        if first_attempt.is_ok() {
            return;
        }
        let err = first_attempt.unwrap_err();
        tracing::warn!(path = %target.display(), error = %err, "fs_watch: initial watch failed, retrying with backoff");
        self.health.mark_degraded(target.clone(), err);

        let this = self.clone();
        tokio::spawn(async move {
            for delay in RETRY_BACKOFF {
                tokio::time::sleep(delay).await;
                // The subscription may have been torn down (refcount back to
                // zero) while we were waiting — don't resurrect a watch
                // nobody wants anymore.
                if !this.inner.lock().unwrap().targets.contains_key(&target) {
                    return;
                }
                match this.try_watch(&target, mode) {
                    Ok(()) => {
                        this.health.clear_degraded(&target);
                        tracing::info!(path = %target.display(), "fs_watch: watch recovered after retry");
                        return;
                    }
                    Err(e) => {
                        this.health.mark_degraded(target.clone(), e);
                    }
                }
            }
            tracing::warn!(
                path = %target.display(),
                "fs_watch: still degraded after all retries — will keep retrying via the periodic health sweep"
            );
        });
    }

    fn try_watch(&self, target: &Path, mode: RecursiveMode) -> Result<(), String> {
        let mut inner = self.inner.lock().unwrap();
        let Some(watcher) = inner.watcher_for(target) else {
            return Err("no fs watcher backend available".to_string());
        };
        watcher.watch(target, mode).map_err(|e| e.to_string())
    }

    /// Handle an async error surfaced by the `notify` backend's own callback
    /// (piped through `raw_rx` in `Self::new`'s bridge task). Extracted out
    /// of that task so a test can drive a real `notify::Error` through it
    /// directly (reagent P2 on #2722: the original three sweep regression
    /// tests all simulated degradation via a direct `health.mark_degraded`
    /// call, so none of them actually exercised this fn or proved that
    /// `notify::Error::paths` lines up with the exact `PathBuf` key
    /// `inner.targets` uses — the assumption `sweep()`'s `is_degraded()`
    /// filter depends on to ever re-arm a target that failed this way).
    ///
    /// `notify::Error` carries the affected path(s) when it can attribute
    /// one; mark each degraded so the next sweep re-arms it. A path-less
    /// error (rare — not attributable to any specific target) can only be
    /// logged.
    fn handle_backend_error(&self, e: notify::Error) {
        tracing::warn!(error = %e, paths = ?e.paths, "fs_watch: backend reported an error");
        for path in &e.paths {
            self.health.mark_degraded(path.clone(), e.to_string());
        }
    }

    /// Re-arm every currently-*degraded* target — the backstop that lets a
    /// path recover from a failed `watch()` attempt (at `start_watch` time
    /// or a prior sweep) without a dedicated retry loop running forever.
    ///
    /// **Deliberately does NOT touch a target that already looks healthy**
    /// (codex P2 on #2722, revising this fn's first version): the original
    /// fix here unconditionally called `unwatch()` then `watch()` on every
    /// subscribed target every tick, reasoning that this was needed to
    /// detect a watch that died *silently* (no error ever surfaced). Two
    /// problems with that, both real:
    /// 1. **The handle leak this fn exists to fix** — `notify` 7.0.0's
    ///    Windows backend (`add_watch` in the vendored `windows.rs`) opens a
    ///    brand-new `CreateFileW` handle + `CreateSemaphoreW` on *every*
    ///    `watch()` call, unconditionally, and silently drops the previous
    ///    entry's handles with no cleanup — see
    ///    `docs/status/STATUS_FS_WATCH_SWEEP_HANDLE_LEAK_2026_08_22.md`.
    /// 2. **A real coverage gap** — `unwatch()` then `watch()` leaves the
    ///    target genuinely unwatched for the gap between the two calls.
    ///    `native_memory_drift`'s slow-path reconciliation sweep backstops a
    ///    missed event there, but `config_watcher_fs`, `EditorFileWatcher`,
    ///    and `MediaFileWatcher` all rely solely on the broadcast stream —
    ///    a create/modify landing in that gap would go unnoticed until some
    ///    *later*, unrelated change on the same path happened to trigger a
    ///    refresh.
    ///
    /// Skipping healthy targets fixes both: nothing is ever double-watched
    /// (no leak) and a healthy watch is never touched (no gap). The
    /// **honestly-bounded trade-off**: a watch that fails *silently* — dies
    /// without `notify` or the OS ever reporting an error on it — is no
    /// longer self-healed by this sweep, only a watch that's already known
    /// degraded (an explicit prior failure). Accepted because a truly
    /// silent death has no confirmed occurrence in this codebase's history
    /// (the scenarios `HEALTH_SWEEP_INTERVAL`'s own doc comment lists —
    /// inotify instance-limit churn, flaky network mounts — are Linux-
    /// flavored concerns on a Windows-primary codebase), against a
    /// certain, systematic cost (leak or gap, every target, every tick) for
    /// every other watch the alternative would have imposed.
    ///
    /// **Race with concurrent `unsubscribe()` — closed by re-checking under
    /// one held lock** (reagent P1, second review round on #2722): the
    /// degraded-target list above is a snapshot; without re-verification, a
    /// concurrent `unsubscribe()` dropping the last subscriber for one of
    /// these targets between the snapshot and this loop's `watch()` call
    /// would remove it from `targets` and `unwatch()` it, while this sweep's
    /// in-flight `watch()` call still re-established a native watch for the
    /// now-untracked path — orphaning exactly the handle pair this fn exists
    /// to stop leaking, just via a race instead of per-tick re-watching.
    /// Each target's re-check + `unwatch()` + `watch()` now happens under a
    /// single `inner` lock acquisition (not `try_watch()`, which takes its
    /// own lock and would deadlock if called while this one is held) — a
    /// concurrent `unsubscribe()` either finishes its own removal entirely
    /// before this runs (this loop then sees the target is gone and skips
    /// it) or entirely after (its own `unwatch()` then correctly tears down
    /// the watch this loop just re-armed) — `Mutex` serializes the two, so
    /// no interleaving in between is possible.
    fn sweep(&self) {
        let targets: Vec<PathBuf> = {
            let inner = self.inner.lock().unwrap();
            inner
                .targets
                .keys()
                .filter(|p| self.health.is_degraded(p))
                .cloned()
                .collect()
        };
        for target in targets {
            self.rearm_if_still_subscribed(target);
        }
    }

    /// The per-target half of a sweep: re-verify `target` is still
    /// subscribed, then `unwatch()` + `watch()` it, all under one held
    /// `inner` lock (see `sweep()`'s doc comment for why the re-check and
    /// the single lock both matter). Split out from `sweep()`'s loop body
    /// so a test can call it directly with a `target` captured *before* a
    /// real `unsubscribe()` ran — deterministically reproducing "sweep
    /// already decided to process this target, then it was removed before
    /// the action ran" without needing to win a real thread-scheduling
    /// race.
    fn rearm_if_still_subscribed(&self, target: PathBuf) {
        let mut inner = self.inner.lock().unwrap();
        // Re-check: still subscribed? A concurrent unsubscribe() may have
        // removed it since the caller's snapshot — don't resurrect a watch
        // nobody wants anymore.
        let Some(mode) = inner.targets.get(&target).map(|e| e.mode) else {
            return;
        };
        let Some(w) = inner.watcher_for(&target) else {
            return;
        };
        // Best-effort: a never-established watch has nothing to unwatch —
        // the watch() call below (re)establishes it regardless of whether
        // this succeeds.
        let _ = w.unwatch(&target);
        let result = w.watch(&target, mode).map_err(|e| e.to_string());
        drop(inner);
        match result {
            Ok(()) => self.health.clear_degraded(&target),
            Err(e) => self.health.mark_degraded(target, e),
        }
    }
}

/// Build the watcher backend: prefer the native OS backend
/// (inotify/FSEvents/ReadDirectoryChangesW); fall back to `notify`'s
/// stat-based `PollWatcher` if the native backend can't even construct
/// (practically never on the platforms AgentMux ships for, but real on some
/// restricted/containerized environments). Returns `(None, Native)` only if
/// both fail to construct — recorded as `WatchBackend::Native` since that
/// was the attempted backend; every `subscribe_*` call will immediately
/// fail into `degraded_paths` until the process restarts, matching the
/// existing three watchers' "live-update becomes unavailable, nothing else
/// breaks" posture, just now with a health signal instead of silence.
fn construct_watcher(
    tx: mpsc::UnboundedSender<notify::Result<notify::Event>>,
) -> (Option<Box<dyn Watcher + Send>>, WatchBackend) {
    let tx_native = tx.clone();
    match notify::recommended_watcher(move |res: notify::Result<notify::Event>| {
        let _ = tx_native.send(res);
    }) {
        Ok(w) => return (Some(Box::new(w)), WatchBackend::Native),
        Err(e) => {
            tracing::warn!(error = %e, "fs_watch: native backend unavailable, falling back to polling");
        }
    }

    let poll_config = notify::Config::default().with_poll_interval(Duration::from_secs(2));
    match notify::PollWatcher::new(move |res: notify::Result<notify::Event>| { let _ = tx.send(res); }, poll_config) {
        Ok(w) => (Some(Box::new(w)), WatchBackend::Polling),
        Err(e) => {
            tracing::error!(error = %e, "fs_watch: polling fallback also failed to construct — live-update unavailable this session");
            (None, WatchBackend::Native)
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::time::Duration as StdDuration;

    /// A WSL share folder is polled, whatever the native backend is, and every
    /// other path keeps the native watcher.
    #[cfg(windows)]
    #[test]
    fn a_wsl_share_folder_gets_the_polling_watcher() {
        let (raw_tx, _raw_rx) = mpsc::unbounded_channel();
        let mut inner = Inner { watcher: None, targets: HashMap::new(), share_poller: None, raw_tx };
        assert!(inner.watcher_for(Path::new(r"C:\Users\me\proj")).is_none(), "no native backend here");
        assert!(inner.watcher_for(Path::new(r"\\?\UNC\wsl.localhost\Ubuntu\home\u")).is_some());
        assert!(inner.share_poller.is_some(), "built on first use");
        assert!(inner.watcher_for(Path::new(r"\\wsl$\Debian\srv")).is_some(), "and shared");
    }

    /// End to end on a Windows machine with WSL: a file written inside a
    /// distro shows up as an event for its folder watched through the share.
    /// Needs a distro, so it is not run on CI. Run manually with the distro to
    /// use: `WSL_TEST_DISTRO=Ubuntu cargo test -p agentmux-srv --bin
    /// agentmux-srv a_change_inside_a_distro_reaches_a_share_watch -- --ignored`
    #[cfg(windows)]
    #[tokio::test]
    #[ignore]
    async fn a_change_inside_a_distro_reaches_a_share_watch() {
        let distro = std::env::var("WSL_TEST_DISTRO").unwrap_or_else(|_| "Ubuntu".to_string());
        let dir = crate::backend::remote::wsl_fs::share_path(&distro, "/tmp");
        // What `canonicalize` really returns for a share path is still one: the
        // editor's write rule and the protected folders read that form.
        let canonical = Path::new(&dir).canonicalize().expect("the share path resolves");
        assert_eq!(
            crate::backend::remote::wsl_fs::split_share(&canonical.to_string_lossy()).map(|s| s.linux),
            Some("/tmp".to_string()),
            "{}",
            canonical.display()
        );
        let pool = FsWatchPool::new();
        let mut events = pool.events();
        let _sub = pool.subscribe_dir(Path::new(&dir));
        let name = format!("am-share-watch-{}", std::process::id());
        let status = std::process::Command::new("wsl.exe")
            .args(["-d", &distro, "--exec", "sh", "-c", &format!("echo x > /tmp/{name}")])
            .status()
            .expect("wsl.exe runs");
        assert!(status.success());
        let seen = tokio::time::timeout(StdDuration::from_secs(10), async {
            loop {
                if let Ok(ev) = events.recv().await {
                    if ev.path.to_string_lossy().ends_with(&name) {
                        return ev;
                    }
                }
            }
        })
        .await;
        let _ = std::process::Command::new("wsl.exe")
            .args(["-d", &distro, "--exec", "rm", "-f", &format!("/tmp/{name}")])
            .status();
        assert!(seen.is_ok(), "no event for {name} in {dir} within 10s");
    }

    #[tokio::test]
    async fn subscribe_unsubscribe_refcounting_does_not_panic() {
        let pool = FsWatchPool::new();
        let tmp = std::env::temp_dir().join("agentmux_fs_watch_pool_test");
        std::fs::create_dir_all(&tmp).unwrap();

        let sub1 = pool.subscribe_dir(&tmp);
        let sub2 = pool.subscribe_dir(&tmp);
        assert_eq!(pool.health().active_watches, 1, "two subscriptions to the same dir share one OS watch");

        pool.unsubscribe(sub1);
        assert_eq!(pool.health().active_watches, 1, "still watched — sub2 is still active");

        pool.unsubscribe(sub2);
        assert_eq!(pool.health().active_watches, 0, "last subscriber gone -> watch torn down");

        let _ = std::fs::remove_dir_all(&tmp);
    }

    /// Regression test for reagent's P2 on #2722: the round-2 fix
    /// (`handle_backend_error`, née an inline branch in the bridge task)
    /// only helps `sweep()` re-arm a target if `notify::Error::paths`
    /// actually lines up with the exact `PathBuf` key `inner.targets` uses
    /// — otherwise `mark_degraded` records a path `is_degraded()` will
    /// faithfully report as degraded, but that never matches any key
    /// `sweep()`'s `targets.keys().filter(is_degraded)` iterates, so the
    /// re-arm silently never happens for any *real* backend error. Uses a
    /// real subscription's `watch_target` (produced by the same
    /// `canonicalize()` path a live `notify` callback's error would need to
    /// match) rather than a hand-built path, so this actually exercises
    /// that assumption instead of just asserting `handle_backend_error`
    /// calls `mark_degraded` (which the original three sweep tests already
    /// covered indirectly by calling `mark_degraded` directly).
    #[tokio::test]
    async fn handle_backend_error_degrades_the_exact_target_key_sweep_looks_up() {
        let pool = FsWatchPool::new();
        let dir = std::env::temp_dir().join("agentmux_fs_watch_pool_backend_error_test");
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();

        let sub = pool.subscribe_dir(&dir);
        let target = sub.watch_target.clone();
        assert!(!pool.health.is_degraded(&target), "precondition: not degraded yet");

        pool.handle_backend_error(notify::Error::generic("simulated backend error").add_path(target.clone()));

        assert!(
            pool.health.is_degraded(&target),
            "handle_backend_error() must mark the exact watch_target key degraded \
             so sweep()'s is_degraded() filter (which iterates inner.targets' own \
             keys) actually picks it up — a mismatched path here would silently \
             defeat the whole reagent-P1 fix for any real notify backend error"
        );

        pool.unsubscribe(sub);
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[tokio::test]
    async fn subscribe_file_watches_the_parent_directory() {
        let pool = FsWatchPool::new();
        let dir = std::env::temp_dir().join("agentmux_fs_watch_pool_file_test");
        std::fs::create_dir_all(&dir).unwrap();
        let file = dir.join("thing.md");
        std::fs::write(&file, "hello").unwrap();

        let sub = pool.subscribe_file(&file);
        assert_eq!(sub.watch_target, dir.canonicalize().unwrap());

        pool.unsubscribe(sub);
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[tokio::test]
    async fn a_change_event_is_observable_on_the_broadcast_stream() {
        let pool = FsWatchPool::new();
        let dir = std::env::temp_dir().join("agentmux_fs_watch_pool_event_test");
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        let file = dir.join("watched.md");
        std::fs::write(&file, "v1").unwrap();

        let mut events = pool.events();
        let sub = pool.subscribe_file(&file);
        // Give the native backend a moment to actually register before we
        // write — otherwise this is a flaky "wrote before the watch was live"
        // race, not a real assertion about the pool's behavior.
        tokio::time::sleep(StdDuration::from_millis(200)).await;
        std::fs::write(&file, "v2").unwrap();

        let seen = tokio::time::timeout(StdDuration::from_secs(5), async {
            loop {
                match events.recv().await {
                    Ok(ev) if ev.path.file_name() == file.file_name() => return true,
                    Ok(_) => continue,
                    Err(_) => return false,
                }
            }
        })
        .await
        .unwrap_or(false);

        assert!(seen, "expected to observe a change event for the watched file");

        pool.unsubscribe(sub);
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[tokio::test]
    async fn health_reports_the_constructed_backend() {
        let pool = FsWatchPool::new();
        // On every platform this actually runs on in CI, the native backend
        // constructs successfully — this just pins that expectation rather
        // than the (currently untestable-without-platform-mocking) fallback
        // path, which recovery.rs's own unit tests cover at the bookkeeping
        // level instead.
        assert_eq!(pool.health().backend, WatchBackend::Native);
    }

    /// Regression test for
    /// `docs/status/STATUS_FS_WATCH_SWEEP_HANDLE_LEAK_2026_08_22.md`, healthy
    /// side: `sweep()` used to call `watch()` unconditionally on every
    /// subscribed target on every `HEALTH_SWEEP_INTERVAL` tick, leaking a
    /// File + Semaphore handle pair per call (see companion test below for
    /// why) — the fix (codex P2 on #2722) is to skip a target entirely once
    /// it's healthy, not just to `unwatch()` before re-`watch()`-ing it (that
    /// still leaked nothing, but cost every other watcher a real event-
    /// coverage gap). Simulates many sweep ticks on a healthy target and
    /// asserts both that handle count does not grow *and* that a healthy
    /// watch is never touched at all.
    #[cfg(target_os = "windows")]
    #[tokio::test]
    async fn sweep_leaves_a_healthy_target_untouched() {
        let pool = FsWatchPool::new();
        let tmp = std::env::temp_dir().join("agentmux_fs_watch_pool_sweep_healthy_test");
        let _ = std::fs::remove_dir_all(&tmp);
        std::fs::create_dir_all(&tmp).unwrap();

        let sub = pool.subscribe_dir(&tmp);
        // Let the initial watch settle before measuring.
        tokio::time::sleep(StdDuration::from_millis(100)).await;
        assert!(!pool.health.is_degraded(&sub.watch_target), "a freshly-established watch must be healthy");

        let pid = std::process::id();
        let before = crate::backend::sysinfo::process_handle_count(pid)
            .expect("own handle count must be queryable");

        const SWEEPS: u32 = 200;
        for _ in 0..SWEEPS {
            pool.sweep();
        }

        let after = crate::backend::sysinfo::process_handle_count(pid)
            .expect("own handle count must be queryable");
        let grew_by = after.saturating_sub(before);

        pool.unsubscribe(sub);
        let _ = std::fs::remove_dir_all(&tmp);

        assert!(
            grew_by < SWEEPS / 2,
            "handle count grew by {grew_by} over {SWEEPS} sweep() calls on a \
             healthy target (before={before}, after={after}) — a healthy \
             target must never be re-watched at all"
        );
    }

    /// Companion to the test above: exercises the actual `unwatch()` +
    /// `watch()` re-arm path that runs for a *degraded* target, proving that
    /// path itself doesn't leak either (the original bug — `notify` 7.0.0's
    /// Windows backend opens a new `CreateFileW` + `CreateSemaphoreW` on
    /// every `watch()` call, unconditionally, and its internal map's
    /// `insert()` silently drops the previous entry's handles with no
    /// cleanup). Forces the target into `degraded` state directly before
    /// every sweep so each call really re-arms it, then asserts handle
    /// count does not grow linearly with sweep count.
    #[cfg(target_os = "windows")]
    #[tokio::test]
    async fn sweep_does_not_leak_a_handle_pair_per_call_for_a_degraded_target() {
        let pool = FsWatchPool::new();
        let tmp = std::env::temp_dir().join("agentmux_fs_watch_pool_sweep_degraded_test");
        let _ = std::fs::remove_dir_all(&tmp);
        std::fs::create_dir_all(&tmp).unwrap();

        let sub = pool.subscribe_dir(&tmp);
        tokio::time::sleep(StdDuration::from_millis(100)).await;
        let target = sub.watch_target.clone();

        let pid = std::process::id();
        let before = crate::backend::sysinfo::process_handle_count(pid)
            .expect("own handle count must be queryable");

        const SWEEPS: u32 = 200;
        for _ in 0..SWEEPS {
            // Force degraded so this tick's sweep() actually re-arms the
            // watch (unwatch() + watch()), same as it would for a real
            // failure — clear_degraded() on success would otherwise make
            // only the first sweep do real work.
            pool.health.mark_degraded(target.clone(), "simulated for test".to_string());
            pool.sweep();
        }

        let after = crate::backend::sysinfo::process_handle_count(pid)
            .expect("own handle count must be queryable");
        let grew_by = after.saturating_sub(before);

        pool.unsubscribe(sub);
        let _ = std::fs::remove_dir_all(&tmp);

        assert!(
            grew_by < SWEEPS / 2,
            "handle count grew by {grew_by} over {SWEEPS} sweep() calls on a \
             degraded target (before={before}, after={after}) — the \
             unwatch()-before-watch() re-arm path must not leak a \
             File+Semaphore pair per call"
        );
    }

    /// Regression test for reagent's second-round P1 on #2722: `sweep()`
    /// snapshots the degraded-target list, then acts on each one later.
    /// Without re-verification, a concurrent `unsubscribe()` dropping the
    /// last subscriber for one of these targets in that window would remove
    /// it from `targets` (and `unwatch()` it), while this sweep's own
    /// in-flight `watch()` call still re-established a native watch for the
    /// now-untracked path — orphaning the handle pair forever, since nothing
    /// is left to `unwatch()` it again.
    ///
    /// **Deterministic, not a timing-based race** (an earlier version of
    /// this test tried to actually race `sweep()` against `unsubscribe()`
    /// on separate tasks — unreliable: `unsubscribe()`'s critical section is
    /// so much shorter than `spawn_blocking`'s own scheduling latency that
    /// it essentially always completed entirely before or after `sweep()`
    /// ran, never in the middle, so it couldn't reliably exercise the
    /// vulnerable window either way). Instead, `sweep()`'s per-target action
    /// is split into its own method
    /// ([`rearm_if_still_subscribed`](Self::rearm_if_still_subscribed)) so
    /// this test can call it directly with a target captured *before* a
    /// real `unsubscribe()` runs — deterministically reproducing "sweep
    /// already decided to process this target, then it was removed before
    /// the action ran" without needing to win any race at all.
    /// **Asserts the guard directly, not via OS handle counts.** An earlier
    /// version of this test measured `process_handle_count()` across 200
    /// rounds and asserted the total grew by less than half that. The
    /// measurement is process-GLOBAL, so in a full `cargo test` run the
    /// ~3400 other tests executing in parallel contribute their own file,
    /// socket and thread churn to the same window — it failed in CI and
    /// locally at `grew_by=103` against a threshold of 100 while passing in
    /// isolation, i.e. it was measuring the test suite, not the code under
    /// test. (It also had to be `#[cfg(target_os = "windows")]`, since
    /// `process_handle_count` is Windows-only, so two of three platforms
    /// got no protection at all.)
    ///
    /// The guard's observable signature is used instead:
    /// `rearm_if_still_subscribed()` only ever touches degraded state on a
    /// path that actually reaches `watch()` — `clear_degraded` on success,
    /// `mark_degraded` on failure. Returning early, as it must for an
    /// unsubscribed target, leaves the entry untouched. So the target is
    /// marked degraded with a sentinel reason *after* the unsubscribe, and
    /// the assertion is that the reason is still exactly that sentinel:
    /// deterministic, unaffected by anything else running, and portable.
    ///
    /// **The REASON, not just the presence of a degraded entry** (Codex P2
    /// on #3235, fixed here). Asserting only `is_degraded()` conflated two
    /// different outcomes, because `mark_degraded` is also what runs when
    /// `watch()` FAILS — so a bypassed guard whose `watch()` call errored
    /// (permissions, resource exhaustion, an unavailable backend) left the
    /// flag set and passed the test, which is precisely the
    /// passes-under-broken-logic failure mode #3235 set out to remove.
    /// Comparing the reason separates all three cases: sentinel intact =
    /// returned early; entry gone = `watch()` succeeded; any other string =
    /// `watch()` failed. The last two both mean the guard was bypassed.
    ///
    /// The `watcher.is_some()` precondition closes the remaining vacuous
    /// path — a pool whose watcher failed to construct returns early at a
    /// *different* guard, so the probe would hold for a reason unrelated to
    /// what this test checks.
    #[tokio::test]
    async fn sweep_does_not_resurrect_a_target_unsubscribed_after_being_snapshotted() {
        let pool = FsWatchPool::new();
        let dir = std::env::temp_dir().join("agentmux_fs_watch_pool_sweep_race_test");
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();

        // A watcher that failed to construct makes `rearm_if_still_subscribed()`
        // return early at its `inner.watcher` guard regardless of the
        // targets re-check this test is about — which would pass the probe
        // below vacuously. `FsWatchPool::new()` documents `watcher: None` as
        // reachable, so assert it explicitly rather than silently skipping.
        assert!(
            pool.inner.lock().unwrap().watcher.is_some(),
            "precondition: the pool must have a live watcher, or the stale-rearm \
             path short-circuits before the re-check this test exercises"
        );

        let sub = pool.subscribe_dir(&dir);
        // Captured now, exactly as sweep()'s own snapshot phase would —
        // before the real unsubscribe() below runs.
        let target = sub.watch_target.clone();

        // The real teardown — exactly what a concurrent unsubscribe() does,
        // landing (for this test) strictly *after* the target was
        // "snapshotted" above but strictly *before* the re-arm action below.
        pool.unsubscribe(sub);

        // Mark degraded AFTER the unsubscribe, which clears the flag for a
        // target it drops. This is the probe — see the doc comment: only a
        // call that reaches `watch()` touches this entry again, and the
        // REASON distinguishes which way it went.
        const PROBE_REASON: &str = "probe sentinel — must survive a correctly-guarded stale rearm";
        pool.health.mark_degraded(target.clone(), PROBE_REASON.to_string());

        // sweep()'s per-target action, called directly with the stale
        // (now-unsubscribed) target — the exact call sweep()'s loop would
        // have made had it reached this target after losing the race.
        pool.rearm_if_still_subscribed(target.clone());

        // The REASON, not merely the presence of a degraded entry — see the
        // doc comment for why the boolean alone is not a sufficient probe.
        let reason_after = pool
            .health
            .snapshot()
            .into_iter()
            .find(|(p, _)| p == &target)
            .map(|(_, reason)| reason);
        let still_untracked = !pool.inner.lock().unwrap().targets.contains_key(&target);

        let _ = std::fs::remove_dir_all(&dir);

        assert!(
            still_untracked,
            "precondition: the target must still be absent from `targets` after \
             unsubscribe — otherwise this test isn't exercising the stale-rearm \
             path at all"
        );
        assert_eq!(
            reason_after.as_deref(),
            Some(PROBE_REASON),
            "rearm_if_still_subscribed() reached watch() for a target that was \
             unsubscribed after being snapshotted. The degraded reason is the probe: \
             it is still `{PROBE_REASON}` only if the function returned early without \
             touching health state. `None` means watch() SUCCEEDED (clear_degraded); \
             any other string means watch() FAILED (mark_degraded overwrote the \
             reason with the backend error). Either way the guard was bypassed and a \
             native watch nobody tracks was re-established, orphaning the handle pair \
             with nothing left to ever unwatch() it"
        );
    }
}