openlatch-client 0.6.3

use std::collections::{BTreeMap, BTreeSet};
use std::ffi::OsString;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

use notify::{EventKind, RecommendedWatcher, RecursiveMode};
use notify_debouncer_full::{
    new_debouncer, new_debouncer_opt, DebounceEventResult, Debouncer, NoCache, RecommendedCache,
};
use tokio::sync::{mpsc, Notify};

use super::reconciler::request::ReconcileRequest;

/// Tamper-plane debounce window. Short enough to feel reactive yet long
/// enough that debouncer-full's file-id correlation collapses Vim /
/// JetBrains / VS Code atomic-save patterns into a single logical event.
const TAMPER_DEBOUNCE_MS: u64 = 500;

/// Watch every path a target's tamper-evidence depends on, and ask for a
/// reconcile pass when one changes.
///
/// A **file** is watched through its parent directory, filtered on its name,
/// so an editor's atomic save — which replaces the file — is still seen. A
/// **directory** (Cline's `Hooks/`, the plugin's directory) is watched itself,
/// so an edit to any file inside it triggers; its parent is watched too, so
/// deleting the directory triggers as well. Paths whose directory does not
/// exist are skipped; `Err` only when nothing at all could be watched.
pub fn spawn_watcher(
    paths: &[PathBuf],
    tx: mpsc::Sender<ReconcileRequest>,
) -> Result<Debouncer<RecommendedWatcher, RecommendedCache>, notify::Error> {
    let mut watch_dirs: BTreeSet<PathBuf> = BTreeSet::new();
    let mut names: BTreeSet<OsString> = BTreeSet::new();
    let mut any_in: BTreeSet<PathBuf> = BTreeSet::new();
    for path in paths {
        if let (Some(dir), Some(name)) = (path.parent(), path.file_name()) {
            if dir.is_dir() {
                watch_dirs.insert(dir.to_path_buf());
                names.insert(name.to_os_string());
            }
        }
        if path.is_dir() {
            watch_dirs.insert(path.clone());
            any_in.insert(path.clone());
            // FSEvents reports `/private/var/...` for a watch set on `/var/...`.
            if let Ok(resolved) = std::fs::canonicalize(path) {
                any_in.insert(resolved);
            }
        }
    }
    if watch_dirs.is_empty() {
        return Err(notify::Error::generic(&format!(
            "none of {paths:?} has a directory to watch"
        )));
    }

    let mut debouncer = new_debouncer(
        Duration::from_millis(TAMPER_DEBOUNCE_MS),
        None,
        move |res: DebounceEventResult| {
            if let Ok(events) = res {
                let hit = events
                    .iter()
                    .any(|ev| event_targets_settings(ev, &names, &any_in));
                if hit && tx.try_send(ReconcileRequest::Fs).is_err() {
                    // Reconcile channel is full or closed. Fail-open: the
                    // 30s poll fallback will catch up. Per architecture.md.
                    tracing::warn!("reconcile channel full — skipping fs-triggered reconcile");
                }
            }
        },
    )?;

    for dir in &watch_dirs {
        debouncer.watch(dir, RecursiveMode::NonRecursive)?;
    }

    Ok(debouncer)
}

fn event_targets_settings(
    ev: &notify_debouncer_full::DebouncedEvent,
    names: &BTreeSet<OsString>,
    any_in: &BTreeSet<PathBuf>,
) -> bool {
    let is_relevant_kind = matches!(
        ev.event.kind,
        EventKind::Modify(_) | EventKind::Create(_) | EventKind::Remove(_)
    );
    if !is_relevant_kind {
        return false;
    }
    ev.event.paths.iter().any(|p| {
        p.file_name().is_some_and(|n| names.contains(n))
            || p.parent().is_some_and(|dir| any_in.contains(dir))
    })
}

/// How long a file must be quiet before a change to it triggers a wiring pass.
///
/// An editor saves by writing a temp file and renaming it, and some save twice
/// in quick succession; waiting for the dust to settle turns that burst into one
/// pass reading the final bytes.
pub const FILE_TRIGGER_DEBOUNCE: Duration = Duration::from_millis(500);

/// A live watch on a set of files, notifying one [`tokio::sync::Notify`] when
/// any of them changes. Dropping it stops the watch.
pub struct FileTrigger {
    _debouncer: Debouncer<RecommendedWatcher, NoCache>,
    watched: Vec<PathBuf>,
}

impl FileTrigger {
    /// The files this trigger was armed for.
    pub fn watched(&self) -> &[PathBuf] {
        &self.watched
    }
}

/// Watch `files` and wake `trigger` whenever one is created, modified, renamed
/// over or removed.
///
/// Each file's PARENT directory is watched, non-recursively, and events are
/// filtered to the watched file names: an editor's atomic save replaces the
/// file, which a watch on the file itself would lose.
///
/// **No file-id cache, on every platform.** The debouncer's recommended cache
/// resolves a file id for every file in a watched directory, and on Windows that
/// OPENS each one. Cline's data directory holds `secrets.json` beside
/// `globalState.json`, and PRD C-6 forbids opening it — so the cache is
/// `NoCache`, and rename correlation, which only affects how events are merged,
/// is given up.
///
/// A file whose parent directory does not exist yet is skipped; the caller
/// re-arms on its next full pass, which is how a directory the agent creates
/// later gets watched. Returns `Ok(None)` when no file could be watched.
///
/// **Blocks** for as long as the OS takes to start the watch (FSEvents: about a
/// third of a second per directory), and dropping the trigger blocks too: call
/// both from a blocking context, never on a runtime worker.
pub fn spawn_file_trigger(
    files: &[PathBuf],
    trigger: Arc<Notify>,
) -> Result<Option<FileTrigger>, notify::Error> {
    let mut names: BTreeMap<PathBuf, BTreeSet<OsString>> = BTreeMap::new();
    for file in files {
        let (Some(dir), Some(name)) = (file.parent(), file.file_name()) else {
            continue;
        };
        if !dir.is_dir() {
            continue;
        }
        names
            .entry(dir.to_path_buf())
            .or_default()
            .insert(name.to_os_string());
    }
    if names.is_empty() {
        return Ok(None);
    }
    // The backend may report a directory by its resolved path: macOS FSEvents
    // names `/private/var/...` for a watch set on `/var/...`. Both spellings
    // are keys, so a symlinked data directory still triggers.
    let mut filter = names.clone();
    for (dir, files) in &names {
        if let Ok(resolved) = std::fs::canonicalize(dir) {
            if resolved != *dir {
                filter
                    .entry(resolved)
                    .or_default()
                    .extend(files.iter().cloned());
            }
        }
    }
    let mut debouncer = new_debouncer_opt::<_, RecommendedWatcher, NoCache>(
        FILE_TRIGGER_DEBOUNCE,
        None,
        move |res: DebounceEventResult| {
            let Ok(events) = res else {
                return;
            };
            let hit = events.iter().any(|ev| {
                matches!(
                    ev.event.kind,
                    EventKind::Modify(_) | EventKind::Create(_) | EventKind::Remove(_)
                ) && ev
                    .event
                    .paths
                    .iter()
                    .any(|p| match (p.parent(), p.file_name()) {
                        (Some(dir), Some(name)) => {
                            filter.get(dir).is_some_and(|wanted| wanted.contains(name))
                        }
                        _ => false,
                    })
            });
            if hit {
                trigger.notify_one();
            }
        },
        NoCache,
        notify::Config::default(),
    )?;
    let mut watched = Vec::new();
    for (dir, files) in &names {
        debouncer.watch(dir, RecursiveMode::NonRecursive)?;
        watched.extend(files.iter().map(|f| dir.join(f)));
    }
    Ok(Some(FileTrigger {
        _debouncer: debouncer,
        watched,
    }))
}

pub fn spawn_poll_fallback(tx: mpsc::Sender<ReconcileRequest>) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        let mut interval = tokio::time::interval(Duration::from_secs(30));
        interval.tick().await;

        loop {
            interval.tick().await;
            if tx.send(ReconcileRequest::Poll).await.is_err() {
                break;
            }
        }
    })
}

#[cfg(test)]
mod tests {
    use super::*;

    async fn fired(trigger: &Notify, within: Duration) -> bool {
        tokio::time::timeout(within, trigger.notified())
            .await
            .is_ok()
    }

    /// An editor saves by writing a temp file and renaming it over the target;
    /// the watch is on the directory, so that save is seen.
    #[tokio::test(flavor = "multi_thread")]
    async fn an_atomic_save_of_a_watched_file_fires_the_trigger() {
        let dir = tempfile::tempdir().expect("tempdir");
        let state = dir.path().join("globalState.json");
        std::fs::write(&state, "{}").expect("seed");
        let notify = Arc::new(Notify::new());
        let _trigger = spawn_file_trigger(std::slice::from_ref(&state), notify.clone())
            .expect("watch")
            .expect("armed");
        // FSEvents on macOS starts delivering a beat after the watch is set up.
        tokio::time::sleep(Duration::from_millis(300)).await;

        let tmp = dir.path().join("globalState.json.tmp-editor");
        std::fs::write(&tmp, r#"{"actModeApiProvider":"ollama"}"#).expect("write temp");
        std::fs::rename(&tmp, &state).expect("rename over");

        assert!(fired(&notify, Duration::from_secs(5)).await);
    }

    /// A change to another file in the same directory is not a change to a
    /// watched file.
    #[tokio::test(flavor = "multi_thread")]
    async fn a_sibling_change_does_not_fire_the_trigger() {
        let dir = tempfile::tempdir().expect("tempdir");
        let state = dir.path().join("globalState.json");
        std::fs::write(&state, "{}").expect("seed");
        let notify = Arc::new(Notify::new());
        let _trigger = spawn_file_trigger(std::slice::from_ref(&state), notify.clone())
            .expect("watch")
            .expect("armed");
        tokio::time::sleep(Duration::from_millis(300)).await;

        std::fs::write(dir.path().join("taskHistory.json"), "[]").expect("sibling");
        assert!(!fired(&notify, Duration::from_millis(1500)).await);
    }

    /// A file whose directory does not exist is skipped rather than failing the
    /// watch; the caller re-arms once the agent creates it.
    #[test]
    fn a_missing_directory_is_skipped() {
        let dir = tempfile::tempdir().expect("tempdir");
        let notify = Arc::new(Notify::new());
        let trigger = spawn_file_trigger(
            &[dir.path().join("not-yet").join("globalState.json")],
            notify,
        )
        .expect("no error");
        assert!(trigger.is_none());
    }

    /// PRD C-6: watching Cline's data directory must not open the secrets file
    /// beside the state file. The debouncer's file-id cache would, on Windows;
    /// this asserts the read-avoidance where atime can show it.
    #[cfg(unix)]
    #[tokio::test(flavor = "multi_thread")]
    async fn the_watch_opens_no_sibling_file() {
        let dir = tempfile::tempdir().expect("tempdir");
        let state = dir.path().join("globalState.json");
        let secrets = dir.path().join("secrets.json");
        std::fs::write(&state, "{}").expect("seed");
        std::fs::write(&secrets, r#"{"apiKey":"SENTINEL"}"#).expect("seed secrets");
        let status = std::process::Command::new("touch")
            .args(["-a", "-t", "202001010000"])
            .arg(&secrets)
            .status()
            .expect("touch");
        assert!(status.success());
        let before = std::fs::metadata(&secrets)
            .expect("stat")
            .accessed()
            .expect("atime");

        let notify = Arc::new(Notify::new());
        let _trigger = spawn_file_trigger(std::slice::from_ref(&state), notify.clone())
            .expect("watch")
            .expect("armed");
        tokio::time::sleep(Duration::from_millis(300)).await;
        std::fs::write(&state, r#"{"a":1}"#).expect("save");
        assert!(
            fired(&notify, Duration::from_secs(5)).await,
            "positive control"
        );

        let after = std::fs::metadata(&secrets)
            .expect("stat")
            .accessed()
            .expect("atime");
        assert_eq!(before, after, "the watch opened secrets.json");
    }
}