diff --git a/Cargo.lock b/Cargo.lock index ea1fa94a..f46b8446 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -435,7 +435,7 @@ dependencies = [ "metrics", "metrics-exporter-prometheus", "mockito", - "notify", + "notify 0.2.0", "reqwest", "serde", "serde_json", @@ -852,6 +852,16 @@ dependencies = [ "alloc-stdlib", ] +[[package]] +name = "bstr" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63044e1ae8e69f3b5a92c736ca6269b8d12fa7efe39bf34ddb06d102cf0e2cab" +dependencies = [ + "memchr", + "serde", +] + [[package]] name = "bumpalo" version = "3.20.2" @@ -1027,7 +1037,7 @@ dependencies = [ "memory", "mockito", "monitor", - "notify", + "notify 0.2.0", "orchestrator", "prettytable-rs", "reqwest", @@ -1309,7 +1319,7 @@ dependencies = [ "bitflags 2.11.0", "crossterm_winapi", "futures-core", - "mio", + "mio 1.1.1", "parking_lot", "rustix 0.38.44", "signal-hook", @@ -2386,6 +2396,17 @@ dependencies = [ "winapi", ] +[[package]] +name = "filetime" +version = "0.2.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f98844151eee8917efc50bd9e8318cb963ae8b297431495d3f758616ea5c57db" +dependencies = [ + "cfg-if", + "libc", + "libredox", +] + [[package]] name = "find-msvc-tools" version = "0.1.9" @@ -2481,6 +2502,15 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "fsevent-sys" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76ee7a02da4d231650c7cea31349b889be2f45ddb3ef3032d2ec8185f6313fd2" +dependencies = [ + "libc", +] + [[package]] name = "fsst" version = "3.0.1" @@ -2676,6 +2706,19 @@ version = "0.3.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0cc23270f6e1808e30a928bdc84dea0b9b4136a8bc82338574f23baf47bbd280" +[[package]] +name = "globset" +version = "0.4.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52dfc19153a48bde0cbd630453615c8151bce3a5adfac7a0aebfbf0a1e1f57e3" +dependencies = [ + "aho-corasick", + "bstr", + "log", + "regex-automata", + "regex-syntax", +] + [[package]] name = "h2" version = "0.4.13" @@ -3183,6 +3226,26 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "inotify" +version = "0.9.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8069d3ec154eb856955c1c0fbffefbf5f3c40a104ec912d4797314c1801abff" +dependencies = [ + "bitflags 1.3.2", + "inotify-sys", + "libc", +] + +[[package]] +name = "inotify-sys" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e05c02b5e89bff3b946cedeca278abc628fe811e604f027c45a8aa3cf793d0eb" +dependencies = [ + "libc", +] + [[package]] name = "ipnet" version = "2.12.0" @@ -3321,6 +3384,26 @@ dependencies = [ "serde_json", ] +[[package]] +name = "kqueue" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eac30106d7dce88daf4a3fcb4879ea939476d5074a9b7ddd0fb97fa4bed5596a" +dependencies = [ + "kqueue-sys", + "libc", +] + +[[package]] +name = "kqueue-sys" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed9625ffda8729b85e45cf04090035ac368927b8cebc34898e7c120f52e4838b" +dependencies = [ + "bitflags 1.3.2", + "libc", +] + [[package]] name = "lance" version = "3.0.1" @@ -4248,6 +4331,18 @@ dependencies = [ "simd-adler32", ] +[[package]] +name = "mio" +version = "0.8.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4a650543ca06a924e8b371db273b2756685faae30f8487da1b56505a8f78b0c" +dependencies = [ + "libc", + "log", + "wasi", + "windows-sys 0.48.0", +] + [[package]] name = "mio" version = "1.1.1" @@ -4439,6 +4534,25 @@ dependencies = [ "uuid", ] +[[package]] +name = "notify" +version = "6.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6205bd8bb1e454ad2e27422015fb5e4f2bcc7e08fa8f27058670d208324a4d2d" +dependencies = [ + "bitflags 2.11.0", + "crossbeam-channel", + "filetime", + "fsevent-sys", + "inotify", + "kqueue", + "libc", + "log", + "mio 0.8.11", + "walkdir", + "windows-sys 0.48.0", +] + [[package]] name = "ntapi" version = "0.4.3" @@ -4653,10 +4767,12 @@ dependencies = [ "croner", "directories", "futures", + "globset", "hex", "hmac", "metrics", "metrics-exporter-prometheus", + "notify 6.1.1", "reqwest", "sea-orm", "sea-orm-migration", @@ -6225,7 +6341,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b75a19a7a740b25bc7944bdee6172368f988763b744e3d4dfe753f6b4ece40cc" dependencies = [ "libc", - "mio", + "mio 1.1.1", "signal-hook", ] @@ -7084,7 +7200,7 @@ checksum = "27ad5e34374e03cfffefc301becb44e9dc3c17584f414349ebe29ed26661822d" dependencies = [ "bytes", "libc", - "mio", + "mio 1.1.1", "parking_lot", "pin-project-lite", "signal-hook-registry", @@ -8380,7 +8496,7 @@ dependencies = [ "colored 2.2.0", "communicate", "memory", - "notify", + "notify 0.2.0", "orchestrator", "tokio", ] diff --git a/crates/orchestrator/Cargo.toml b/crates/orchestrator/Cargo.toml index 56f0083e..c6e14624 100644 --- a/crates/orchestrator/Cargo.toml +++ b/crates/orchestrator/Cargo.toml @@ -54,6 +54,10 @@ hex = "0.4" # Cron expression parsing croner = { workspace = true } +# Filesystem change watching (FileWatch trigger) +notify = "6" +globset = "0.4" + # TOML config file parsing (for agentd config file support) toml = { workspace = true } diff --git a/crates/orchestrator/src/scheduler/strategy.rs b/crates/orchestrator/src/scheduler/strategy.rs index 185faa78..9f3af685 100644 --- a/crates/orchestrator/src/scheduler/strategy.rs +++ b/crates/orchestrator/src/scheduler/strategy.rs @@ -11,7 +11,10 @@ use crate::scheduler::types::{DispatchStatus, Task}; use async_trait::async_trait; use chrono::Utc; use croner::Cron; +use globset::{Glob, GlobSet, GlobSetBuilder}; +use notify::Watcher as _; use std::collections::HashMap; +use std::path::PathBuf; use std::sync::Arc; use std::time::Duration; use tokio::sync::{broadcast, mpsc, watch}; @@ -837,6 +840,288 @@ impl TriggerStrategy for IdleStrategy { } } +// --------------------------------------------------------------------------- +// FileWatchStrategy +// --------------------------------------------------------------------------- + +/// Convert a `notify::EventKind` to a short lowercase string suitable for +/// template variables and `source_id` generation. +fn event_kind_to_str(kind: ¬ify::EventKind) -> &'static str { + match kind { + notify::EventKind::Create(_) => "create", + notify::EventKind::Modify(_) => "modify", + notify::EventKind::Remove(_) => "delete", + notify::EventKind::Access(_) => "access", + notify::EventKind::Other => "other", + notify::EventKind::Any => "any", + } +} + +/// A [`TriggerStrategy`] that watches filesystem paths for changes using the +/// OS-native notification API (inotify on Linux, FSEvents on macOS, ReadDirectoryChangesW +/// on Windows) via the `notify` crate. +/// +/// When a matching filesystem event occurs the strategy fires a synthetic +/// [`Task`] carrying the changed file's path, name, containing directory, and +/// event kind in its metadata. +/// +/// # Pattern Filtering +/// +/// If `patterns` is provided, only paths matching at least one glob (using +/// `globset` syntax) will produce tasks. An empty or absent patterns list +/// accepts every path. +/// +/// # Event Kind Filtering +/// +/// `event_kinds` contains lowercase strings: `"create"`, `"modify"`, +/// `"delete"`, `"access"`, `"rename"`. An empty vec accepts all kinds. +/// +/// # Debouncing +/// +/// After the first matching event the strategy waits `debounce_ms` milliseconds +/// for more events. Any additional events within the window reset the timer. +/// When the window expires the strategy fires a single task based on the +/// *first* event seen in the burst, reducing duplicate dispatches on rapid +/// file changes (e.g., editor atomic writes). +/// +/// # Shutdown +/// +/// Both the first-event wait and the debounce phase respect the shutdown +/// signal and return an empty vec immediately when it fires. +/// +/// # Example +/// +/// ```rust,ignore +/// use orchestrator::scheduler::strategy::FileWatchStrategy; +/// +/// let strategy = FileWatchStrategy::new( +/// vec!["/tmp/watched".into()], +/// Some(vec!["**/*.toml".into()]), +/// vec!["create".into(), "modify".into()], +/// 200, // 200 ms debounce +/// )?; +/// ``` +pub struct FileWatchStrategy { + /// The OS-level filesystem watcher — must be kept alive for the lifetime of + /// this strategy or events will stop being delivered. + _watcher: notify::RecommendedWatcher, + /// Async receiver end of the mpsc bridge from the sync `notify` callback. + rx: mpsc::UnboundedReceiver>, + /// Optional glob patterns — `None` means accept every path. + include_patterns: Option, + /// Event kinds to accept. Empty means accept all. + event_kinds: Vec, + /// Debounce window in milliseconds (0 = no debounce). + debounce_ms: u64, + /// Watched root paths (stored for metadata / diagnostics). + #[allow(dead_code)] + watch_paths: Vec, +} + +impl FileWatchStrategy { + /// Create a new file-watch strategy. + /// + /// * `paths` — directories or files to watch recursively. + /// * `patterns` — optional glob patterns restricting which paths produce tasks. + /// * `event_kinds` — which event kinds to accept (`"create"`, `"modify"`, + /// `"delete"`, `"access"`). An empty vec accepts all. + /// * `debounce_ms` — debounce window in milliseconds (0 = no debounce). + pub fn new( + paths: Vec, + patterns: Option>, + event_kinds: Vec, + debounce_ms: u64, + ) -> anyhow::Result { + let (tx, rx) = mpsc::unbounded_channel(); + + // Build the OS-native watcher. The callback bridges sync notify events + // into the async mpsc channel. + let mut watcher = notify::RecommendedWatcher::new( + move |res: notify::Result| { + // Ignore send errors — receiver may have been dropped during shutdown. + let _ = tx.send(res); + }, + notify::Config::default(), + ) + .map_err(|e| anyhow::anyhow!("Failed to create filesystem watcher: {}", e))?; + + // Register each path for recursive watching. + for path in &paths { + watcher + .watch(path, notify::RecursiveMode::Recursive) + .map_err(|e| anyhow::anyhow!("Failed to watch path {:?}: {}", path, e))?; + } + + // Build the optional glob set. + let include_patterns = match patterns { + None => None, + Some(pats) if pats.is_empty() => None, + Some(pats) => { + let mut builder = GlobSetBuilder::new(); + for pat in &pats { + let glob = Glob::new(pat) + .map_err(|e| anyhow::anyhow!("Invalid glob pattern '{}': {}", pat, e))?; + builder.add(glob); + } + Some( + builder + .build() + .map_err(|e| anyhow::anyhow!("Failed to build glob set: {}", e))?, + ) + } + }; + + Ok(Self { + _watcher: watcher, + rx, + include_patterns, + event_kinds, + debounce_ms, + watch_paths: paths, + }) + } + + /// Returns `true` if the event passes the kind and pattern filters. + fn matches_event(&self, event: ¬ify::Event) -> bool { + // Filter by event kind. + if !self.event_kinds.is_empty() { + let kind_str = event_kind_to_str(&event.kind); + if !self.event_kinds.iter().any(|k| k == kind_str) { + return false; + } + } + + // Filter by glob patterns. + if let Some(ref patterns) = self.include_patterns { + if !event.paths.iter().any(|p| patterns.is_match(p)) { + return false; + } + } + + true + } + + /// Build a synthetic [`Task`] from a filesystem event. + /// + /// The task's metadata includes: + /// + /// | key | value | + /// |--------------|----------------------------| + /// | `file_path` | full path of changed file | + /// | `file_name` | basename of changed file | + /// | `file_dir` | parent directory | + /// | `event_type` | `"create"` / `"modify"` / … | + /// | `timestamp` | RFC 3339 fire time | + fn build_task(&self, event: ¬ify::Event) -> Task { + let path = event.paths.first().map(|p| p.display().to_string()).unwrap_or_default(); + let file_name = event + .paths + .first() + .and_then(|p| p.file_name()) + .map(|n| n.to_string_lossy().into_owned()) + .unwrap_or_default(); + let file_dir = event + .paths + .first() + .and_then(|p| p.parent()) + .map(|d| d.display().to_string()) + .unwrap_or_default(); + let kind_str = event_kind_to_str(&event.kind); + let timestamp = Utc::now().to_rfc3339(); + + let mut metadata = HashMap::new(); + metadata.insert("file_path".to_string(), path.clone()); + metadata.insert("file_name".to_string(), file_name); + metadata.insert("file_dir".to_string(), file_dir); + metadata.insert("event_type".to_string(), kind_str.to_string()); + metadata.insert("timestamp".to_string(), timestamp); + + Task { + source_id: format!("file:{}:{}", path, kind_str), + title: format!("File {} event: {}", kind_str, path), + body: String::new(), + url: String::new(), + labels: vec![], + assignee: None, + metadata, + } + } +} + +#[async_trait] +impl TriggerStrategy for FileWatchStrategy { + async fn next_tasks(&mut self, shutdown: &watch::Receiver) -> anyhow::Result> { + let mut shutdown = shutdown.clone(); + + // ── Phase 1: wait for the first matching event ────────────────────── + let first_event = loop { + tokio::select! { + maybe_event = self.rx.recv() => { + match maybe_event { + Some(Ok(event)) if self.matches_event(&event) => break event, + Some(Ok(_)) => { + // Non-matching event — keep waiting. + } + Some(Err(e)) => { + warn!(%e, "FileWatchStrategy: watcher error"); + } + None => { + // Watcher was dropped — signal done. + return Ok(vec![]); + } + } + } + _ = shutdown.changed() => { + if *shutdown.borrow() { + return Ok(vec![]); + } + } + } + }; + + // ── Phase 2: debounce — drain additional events within the window ─── + if self.debounce_ms > 0 { + let debounce = Duration::from_millis(self.debounce_ms); + loop { + // A fresh sleep each iteration means arriving events reset the timer. + let sleep = tokio::time::sleep(debounce); + tokio::pin!(sleep); + + tokio::select! { + _ = &mut sleep => { + // Window expired — ready to fire. + break; + } + maybe_event = self.rx.recv() => { + match maybe_event { + Some(Ok(_)) => { + // Another event arrived — loop restarts with a fresh sleep. + } + Some(Err(e)) => { + warn!(%e, "FileWatchStrategy: watcher error during debounce"); + } + None => break, + } + } + _ = shutdown.changed() => { + if *shutdown.borrow() { + return Ok(vec![]); + } + } + } + } + } + + let task = self.build_task(&first_event); + info!( + path = %first_event.paths.first().map(|p| p.display().to_string()).unwrap_or_default(), + kind = %event_kind_to_str(&first_event.kind), + "FileWatchStrategy: filesystem event fired task" + ); + Ok(vec![task]) + } +} + // --------------------------------------------------------------------------- // Tests // ---------------------------------------------------------------------------