From dd0fcc7bee04fb6cd6a18da46121a2aaf6bbfe73 Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Mon, 24 Aug 2026 19:28:30 +0200 Subject: [PATCH 1/3] fix(run): bound catalog watches to declaration space The supervisor registered one recursive inotify watch over the whole catalog root before its first reconcile. notify eagerly walks the entire tree and allocates one watch per directory BEFORE any callback filtering, and follows symlinks by default, so Resource payload trees (37GB / 151k directories on dev3) dominated startup and could exhaust kernel limits. When installation then failed, the dropped channel sender made RecvTimeoutError::Disconnected look like an immediate wake, turning the nominal 30s timer into a tight full-reconcile loop (~48 passes/min, ~70% of a core, measured live). - Replace the recursive registration with CatalogDeclarationWatcher: one non-recursive watch per declaration-space directory, discovered without descending into Resource payloads (gated by is_catalog_path), refreshed after each pass so new directories become watched. - Treat directory topology mutations as wakes so create/rename/remove still reconcile immediately; file mutations filter through is_declaration_path as before. - Diagnose watcher installation failure once instead of discarding the error behind Option. - Wait on an absolute deadline where Disconnection sleeps the slice instead of waking, so the timer fallback actually honors the interval while stop stays responsive in bounded slices. - Drive the real supervisor loop in tests with an injectable watcher factory: installation failure must stay on timer cadence (the spin was only covered at helper level before), and a live watcher must wake on declaration mutation long before the timer. Closes #314, closes #328. Verified against a synthetic oversized catalog (2051 payload dirs plus a symlinked external tree): 7 inotify watches installed, first reconcile immediate, zero CPU ticks across a 3s idle window, SIGINT responsive. Co-authored-by: schickling-assistant agent-identity: unknown agent-persona: generalist agent-supervisor: unavailable agent-tool: OMP agent-tool-version: 18.0.3 agent-runtime: OMP 18.0.3 tooling-profile: dotfiles@f33cd9c-dirty --- src/run.rs | 195 ++++++++++++++++++++++++++++++++++---- src/watch.rs | 263 +++++++++++++++++++++++++++++++++++++++++++++++---- 2 files changed, 420 insertions(+), 38 deletions(-) diff --git a/src/run.rs b/src/run.rs index 3cb36f62..2ab2eaac 100644 --- a/src/run.rs +++ b/src/run.rs @@ -21,7 +21,7 @@ use std::os::unix::process::CommandExt as _; use std::path::{Path, PathBuf}; use std::process::{Child, ChildStdin, Command, Output, Stdio}; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::mpsc::{Receiver, RecvTimeoutError, channel}; +use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender, channel}; use std::time::{Duration, Instant}; use anyhow::Context as _; @@ -2000,9 +2000,56 @@ fn drain(rx: &Receiver<()>) { while rx.try_recv().is_ok() {} } +#[derive(Debug, PartialEq, Eq)] +enum ReconcileWake { + Change, + Interval, + Stop, +} + +/// Wait for the next reconciliation trigger: a declaration change, the timer fallback, or a stop. +/// Stop stays responsive in bounded slices; a disconnected watcher channel must never masquerade +/// as a change — that turned the nominal timer fallback into a tight full-catalog reconcile loop. +fn wait_for_reconcile(rx: &Receiver<()>, interval: Duration, stop: &AtomicBool) -> ReconcileWake { + let deadline = Instant::now() + interval; + loop { + if stop.load(Ordering::SeqCst) { + return ReconcileWake::Stop; + } + let Some(remaining) = deadline.checked_duration_since(Instant::now()) else { + return ReconcileWake::Interval; + }; + let slice = remaining.min(Duration::from_millis(250)); + match rx.recv_timeout(slice) { + Ok(()) => { + drain(rx); // coalesce a burst of events into one pass + return ReconcileWake::Change; + } + Err(RecvTimeoutError::Timeout) => {} + Err(RecvTimeoutError::Disconnected) => std::thread::sleep(slice), + } + } +} + +/// Best-effort supervisor watch: installation failure is diagnosed once, then reconciliation +/// continues on the timer alone instead of silently losing immediate wakeups forever. +fn best_effort_catalog_watcher( + root: &Path, + tx: Sender<()>, +) -> Option { + match crate::watch::watch_catalog_declarations(root, tx) { + Ok(watcher) => Some(watcher), + Err(error) => { + eprintln!( + "st2: cannot watch catalog declarations: {error}; immediate catalog changes are unavailable, continuing with timer polling." + ); + None + } + } +} + /// The supervisor loop: reconcile on a timer AND on folder changes until interrupted. The fs-watch is /// best-effort; the `interval` timer is the always-on fallback. `on_report` is called once per pass -/// (e.g. to log a summary). Returns when a stop signal arrives — agents are left running. pub fn up_loop( root: &Path, this_host: &str, @@ -2011,7 +2058,15 @@ pub fn up_loop( on_report: impl FnMut(&UpReport), ) -> anyhow::Result<()> { install_signal_handler(); - up_loop_until(root, this_host, runner, interval, &STOP, on_report) + up_loop_until( + root, + this_host, + runner, + interval, + &STOP, + best_effort_catalog_watcher, + on_report, + ) } fn up_loop_until( @@ -2020,13 +2075,14 @@ fn up_loop_until( runner: &dyn Runner, interval: Duration, stop: &AtomicBool, + install_watcher: impl FnOnce(&Path, Sender<()>) -> Option, mut on_report: impl FnMut(&UpReport), ) -> anyhow::Result<()> { crate::event::publish_owner_binding(root, this_host) .context("publish machine-local stream owner binding")?; let task_context = TaskCompileContext::current(root.to_path_buf())?; let (tx, rx) = channel::<()>(); - let _watcher = crate::watch::watch_catalog_declarations(root, tx); + let mut watcher = install_watcher(root, tx); let mut cap = FlappingCap::default(); // Carries per-id liveness across passes so a transient `pty list` flicker under load isn't // destructively GC'd (R21c). Fresh throwaway in `up_once` — a single pass has no flicker to absorb. @@ -2053,6 +2109,9 @@ fn up_loop_until( ); pre.absorb(report); report = pre; + if let Some(watcher) = &mut watcher { + watcher.refresh(); + } // A recovered task that crash-loops again is a new crash-loop, so it must be able to surface // again. Leaving the id in the dedup set would make every park after the first one silent. for id in &report.unparked { @@ -2074,22 +2133,8 @@ fn up_loop_until( break; } - // Sleep the interval in 250ms slices, waking early on a folder change or a stop signal. - let slices = (interval.as_millis() / 250).max(1); - for _ in 0..slices { - if stop.load(Ordering::SeqCst) { - break; - } - match rx.recv_timeout(Duration::from_millis(250)) { - Ok(()) => { - drain(&rx); // coalesce a burst of events into one pass - break; - } - Err(RecvTimeoutError::Timeout) => {} - Err(RecvTimeoutError::Disconnected) => break, - } - } - if stop.load(Ordering::SeqCst) { + // Wait for a declaration change or the timer fallback in stop-responsive slices. + if wait_for_reconcile(&rx, interval, stop) == ReconcileWake::Stop { break; } } @@ -2171,6 +2216,7 @@ mod tests { use std::cell::{Cell, RefCell}; use std::collections::{BTreeMap, BTreeSet}; use std::ffi::OsStr; + use std::sync::atomic::AtomicUsize; #[cfg(target_os = "linux")] fn linux_process_state(pid: i32) -> Option { @@ -2432,6 +2478,7 @@ mod tests { }, Duration::from_secs(60), &stop, + best_effort_catalog_watcher, |_| passes += 1, ) .unwrap(); @@ -2443,6 +2490,114 @@ mod tests { ); } + #[cfg(target_os = "linux")] + #[test] + fn failed_watch_installation_keeps_supervisor_on_timer_cadence() { + let catalog = tempfile::tempdir().unwrap(); + let agent = catalog.path().join("agents/test-host/live"); + std::fs::create_dir_all(&agent).unwrap(); + std::fs::write( + agent.join("agent.kdl"), + r#"agent "live" { host "test-host"; command "x" }"#, + ) + .unwrap(); + let stop = AtomicBool::new(false); + let mut passes = 0usize; + + std::thread::scope(|scope| { + scope.spawn(|| { + std::thread::sleep(Duration::from_millis(350)); + stop.store(true, Ordering::SeqCst); + }); + up_loop_until( + catalog.path(), + "test-host", + &SpawnCountingRunner::default(), + Duration::from_millis(100), + &stop, + |_, _| None, // watcher installation fails, as it did on dev3's oversized catalog + |_| passes += 1, + ) + .unwrap(); + }); + + assert!( + (2..=6).contains(&passes), + "a disconnected watcher channel must fall back to timer cadence, not spin: \ + {passes} passes in ~350ms at a 100ms interval" + ); + } + + #[cfg(target_os = "linux")] + #[test] + fn supervisor_still_wakes_on_declaration_change_with_live_watcher() { + let catalog = tempfile::tempdir().unwrap(); + let agent = catalog.path().join("agents/test-host/live"); + std::fs::create_dir_all(&agent).unwrap(); + let spec = agent.join("agent.kdl"); + std::fs::write(&spec, r#"agent "live" { host "test-host"; command "x" }"#).unwrap(); + let stop = AtomicBool::new(false); + let passes = std::sync::Arc::new(AtomicUsize::new(0)); + let observed = passes.clone(); + let started = Instant::now(); + + std::thread::scope(|scope| { + scope.spawn(|| { + std::thread::sleep(Duration::from_millis(200)); + std::fs::write( + &spec, + r#"agent "live" { host "test-host"; command "changed" }"#, + ) + .unwrap(); + }); + scope.spawn({ + let stop = &stop; + let passes = &passes; + move || { + let deadline = Instant::now() + Duration::from_secs(10); + while passes.load(Ordering::SeqCst) < 2 + && !stop.load(Ordering::SeqCst) + && Instant::now() < deadline + { + std::thread::sleep(Duration::from_millis(10)); + } + stop.store(true, Ordering::SeqCst); + } + }); + up_loop_until( + catalog.path(), + "test-host", + &SpawnCountingRunner::default(), + Duration::from_secs(60), + &stop, + best_effort_catalog_watcher, + |_| { + passes.fetch_add(1, Ordering::SeqCst); + }, + ) + .unwrap(); + }); + + assert!( + started.elapsed() < Duration::from_secs(10), + "a declaration mutation must wake the supervisor long before the 60s timer" + ); + assert!(observed.load(Ordering::SeqCst) >= 2); + } + + #[cfg(target_os = "linux")] + #[test] + fn disconnected_watcher_channel_waits_out_the_interval_instead_of_spinning() { + let (_tx, rx) = channel::<()>(); + let started = Instant::now(); + assert_eq!( + wait_for_reconcile(&rx, Duration::from_millis(80), &AtomicBool::new(false)), + ReconcileWake::Interval, + "disconnection must be treated as silence, not as a change" + ); + assert!(started.elapsed() >= Duration::from_millis(75)); + } + // ── liveness debounce (R21c): a transient `pty list` not-alive flicker under load must not // destructively GC/relaunch a HEALTHY agent; a stable death must still be reaped ────────────── diff --git a/src/watch.rs b/src/watch.rs index ca4ede1c..92f93110 100644 --- a/src/watch.rs +++ b/src/watch.rs @@ -5,10 +5,13 @@ //! read emits another event, and the next timed wait returns immediately. Only mutations are useful //! wakeups; timer polling remains the fallback for anything a backend cannot classify. -use std::path::Path; +use std::collections::BTreeSet; +use std::fs; +use std::path::{Path, PathBuf}; use std::sync::mpsc::Sender; -use notify::{Event, EventKind, RecursiveMode, Watcher}; +use notify::event::{ModifyKind, RenameMode}; +use notify::{Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher}; /// Watch a directory recursively, forwarding only events that can change reconciled state. pub(crate) fn watch_recursive_mutations( @@ -51,28 +54,142 @@ pub(crate) fn watch_delivery_inputs( Some(watcher) } +/// A shallow subscription over the directories that can contain declarations. +/// +/// `notify` implements a recursive Linux watch by eagerly walking the entire tree and allocating +/// one inotify watch per directory BEFORE the first callback ever runs. A production catalog also +/// contains unbounded Resource payload trees, so a recursive subscription spends all of startup +/// walking data the callback would later ignore — and fails outright once the walk exhausts +/// kernel limits. Keep one non-recursive watch per declaration-space directory instead; +/// [`CatalogDeclarationWatcher::refresh`] runs after each reconcile pass so a newly created +/// directory becomes watched before later edits inside it. +pub(crate) struct CatalogDeclarationWatcher { + root: PathBuf, + watcher: RecommendedWatcher, + watched: BTreeSet, + failed: BTreeSet, +} + +impl CatalogDeclarationWatcher { + fn new(root: &Path, tx: Sender<()>) -> notify::Result { + let callback_root = root.to_path_buf(); + let watcher = notify::recommended_watcher(move |result: notify::Result| { + if result.is_ok_and(|event| should_wake_catalog(&callback_root, &event)) { + let _ = tx.send(()); + } + })?; + let mut this = Self { + root: root.to_path_buf(), + watcher, + watched: BTreeSet::new(), + failed: BTreeSet::new(), + }; + this.watcher.watch(root, RecursiveMode::NonRecursive)?; + this.watched.insert(root.to_path_buf()); + this.refresh(); + Ok(this) + } + + /// Reconcile the shallow subscriptions with the current declaration namespace. + pub(crate) fn refresh(&mut self) { + let desired = declaration_watch_dirs(&self.root); + + for stale in self + .watched + .difference(&desired) + .cloned() + .collect::>() + { + // Backends normally discard a watch when its directory disappears. An explicit + // best-effort unwatch also handles moves that leave the watched inode alive elsewhere. + let _ = self.watcher.unwatch(&stale); + self.watched.remove(&stale); + } + for added in desired + .difference(&self.watched) + .cloned() + .collect::>() + { + match self.watcher.watch(&added, RecursiveMode::NonRecursive) { + Ok(()) => { + self.failed.remove(&added); + self.watched.insert(added); + } + Err(error) if self.failed.insert(added.clone()) => { + eprintln!( + "st2: cannot watch catalog declaration directory '{}': {error}; immediate changes below it are unavailable, continuing with timer polling.", + added.display() + ); + } + Err(_) => {} + } + } + } + + #[cfg(test)] + fn watched_dirs(&self) -> &BTreeSet { + &self.watched + } +} + /// Watch only declaration inputs for the supervisor. Runtime state (PTY registry, bus, logs, -/// locks, inboxes, and generated materializations) must never wake reconciliation. +/// locks, inboxes, and generated materializations) must never be traversed or wake reconciliation. pub(crate) fn watch_catalog_declarations( root: &Path, tx: Sender<()>, -) -> Option { - let root = root.to_path_buf(); - let callback_root = root.clone(); - let mut watcher = notify::recommended_watcher(move |result: notify::Result| { - if result.is_ok_and(|event| { - is_mutation(&event) - && event - .paths - .iter() - .any(|path| is_declaration_path(&callback_root, path)) - }) { - let _ = tx.send(()); +) -> notify::Result { + CatalogDeclarationWatcher::new(root, tx) +} + +/// Every directory that can hold declarations, discovered WITHOUT descending into Resource +/// payloads: recursion is gated by [`agent_spec::is_catalog_path`], so a declaration parent's +/// `resources`/`archive`/`inbox` children and `.git`/`.st2` control dirs prune the walk. +fn declaration_watch_dirs(root: &Path) -> BTreeSet { + fn collect(root: &Path, dir: &Path, out: &mut BTreeSet) { + out.insert(dir.to_path_buf()); + let Ok(entries) = fs::read_dir(dir) else { + return; + }; + for entry in entries.flatten() { + let path = entry.path(); + if entry.file_type().is_ok_and(|kind| kind.is_dir()) + && agent_spec::is_catalog_path(root, &path) + { + collect(root, &path, out); + } } + } + + let mut dirs = BTreeSet::new(); + collect(root, root, &mut dirs); + dirs +} + +fn should_wake_catalog(root: &Path, event: &Event) -> bool { + if !is_mutation(event) { + return false; + } + event.paths.iter().any(|path| { + is_declaration_path(root, path) + || (is_directory_topology_mutation(event) && agent_spec::is_catalog_path(root, path)) }) - .ok()?; - watcher.watch(&root, RecursiveMode::Recursive).ok()?; - Some(watcher) +} + +/// Directory create/remove/rename must wake even though no `agent.kdl` path exists yet — the +/// created directory may receive one next, and the watcher needs `refresh` anyway. +fn is_directory_topology_mutation(event: &Event) -> bool { + matches!( + event.kind, + EventKind::Create(_) + | EventKind::Remove(_) + | EventKind::Modify(ModifyKind::Name( + RenameMode::Any + | RenameMode::From + | RenameMode::To + | RenameMode::Both + | RenameMode::Other + )) + ) } fn is_declaration_path(root: &Path, path: &Path) -> bool { @@ -271,4 +388,114 @@ mod tests { |dir| std::fs::remove_file(dir.join("removed")).unwrap(), ); } + + #[test] + fn declaration_watch_tree_stops_before_resource_payloads() { + let catalog = tempfile::tempdir().unwrap(); + let outside = tempfile::tempdir().unwrap(); + let root = catalog.path(); + let agent = root.join("agents/h/live"); + std::fs::create_dir_all(agent.join("resources/cache/a/b/c/d/e")).unwrap(); + std::fs::create_dir_all(agent.join("inbox/archive/a/b/c")).unwrap(); + std::fs::create_dir_all(outside.path().join("worktree/node_modules/a/b/c/d/e")).unwrap(); + #[cfg(unix)] + std::os::unix::fs::symlink( + outside.path().join("worktree"), + agent.join("resources/worktree"), + ) + .unwrap(); + std::fs::write( + agent.join("agent.kdl"), + r#"agent "live" { host "h"; command "x" }"#, + ) + .unwrap(); + + let watched = declaration_watch_dirs(root); + assert_eq!( + watched, + [ + root.to_path_buf(), + root.join("agents"), + root.join("agents/h"), + agent, + ] + .into_iter() + .collect() + ); + assert!( + watched.iter().all(|path| !path.starts_with(outside.path())), + "Resource worktree links must not escape the catalog watch boundary" + ); + } + + #[test] + fn watcher_installation_preserves_backend_errors() { + let parent = tempfile::tempdir().unwrap(); + let missing_catalog = parent.path().join("missing"); + let (tx, _rx) = std::sync::mpsc::channel(); + assert!( + watch_catalog_declarations(&missing_catalog, tx).is_err(), + "a missing root must return the installation error" + ); + } + + #[cfg(target_os = "linux")] + #[test] + fn catalog_watch_is_bounded_and_refreshes_new_declaration_directories() { + use std::sync::mpsc::channel; + use std::time::Duration; + + let catalog = tempfile::tempdir().unwrap(); + let root = catalog.path(); + let agent = root.join("agents/h/live"); + std::fs::create_dir_all(agent.join("resources/cache/a/b/c/d/e")).unwrap(); + std::fs::write( + agent.join("agent.kdl"), + r#"agent "live" { host "h"; command "x" }"#, + ) + .unwrap(); + + let (tx, rx) = channel(); + let mut watcher = watch_catalog_declarations(root, tx).expect("start catalog watcher"); + assert_eq!( + watcher.watched_dirs().len(), + 4, + "Resource depth must not add watches" + ); + + std::fs::write( + agent.join("agent.kdl"), + r#"agent "live" { host "h"; command "changed" }"#, + ) + .unwrap(); + rx.recv_timeout(Duration::from_secs(1)) + .expect("declaration update must wake"); + while rx.try_recv().is_ok() {} + + let nested = root.join("teams/new/live"); + std::fs::create_dir_all(&nested).unwrap(); + std::fs::write( + nested.join("agent.kdl"), + r#"agent "live" { host "new"; command "x" }"#, + ) + .unwrap(); + rx.recv_timeout(Duration::from_secs(1)) + .expect("new declaration directory must wake its watched ancestor"); + while rx.try_recv().is_ok() {} + watcher.refresh(); + assert!(watcher.watched_dirs().contains(&nested)); + + std::fs::write( + nested.join("agent.kdl"), + r#"agent "live" { host "new"; command "changed" }"#, + ) + .unwrap(); + rx.recv_timeout(Duration::from_secs(1)) + .expect("declaration update after refresh must wake"); + while rx.try_recv().is_ok() {} + + std::fs::remove_file(nested.join("agent.kdl")).unwrap(); + rx.recv_timeout(Duration::from_secs(1)) + .expect("declaration removal must wake"); + } } From 5fddf7ff750cf9f84bfc21cea6125492f9140003 Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Tue, 25 Aug 2026 01:01:04 +0200 Subject: [PATCH 2/3] fix(watch): directory-scoped topology wakes and inode-aware refresh MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-ups on the declaration-space watcher: - Topology wakes matched every create/remove/rename in declaration space, so a scratch, log, or editor-swap FILE could wake a full-catalog reconcile — the churn this watcher exists to prevent. Classify by entry type where the backend provides it (Linux does), check the live tree for untyped rename arrivals, and leave rename-away silent (an in-catalog rename emits the arrived side; a full removal is bounded by the timer plus refresh). - An inotify watch dies with its inode, but `watched` tracked names only: a directory deleted and recreated at the same pathname was never resubscribed until the timer. Track (dev, ino) identity per watched directory and force re-registration when a replacement differs. Co-authored-by: schickling-assistant agent-identity: unknown agent-persona: generalist agent-supervisor: unavailable agent-tool: OMP agent-tool-version: 18.0.3 agent-runtime: OMP 18.0.3 tooling-profile: dotfiles@f33cd9c-dirty --- src/watch.rs | 215 +++++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 174 insertions(+), 41 deletions(-) diff --git a/src/watch.rs b/src/watch.rs index 92f93110..0f2be12d 100644 --- a/src/watch.rs +++ b/src/watch.rs @@ -5,12 +5,12 @@ //! read emits another event, and the next timed wait returns immediately. Only mutations are useful //! wakeups; timer polling remains the fallback for anything a backend cannot classify. -use std::collections::BTreeSet; +use std::collections::{BTreeMap, BTreeSet}; use std::fs; use std::path::{Path, PathBuf}; use std::sync::mpsc::Sender; -use notify::event::{ModifyKind, RenameMode}; +use notify::event::{CreateKind, ModifyKind, RemoveKind, RenameMode}; use notify::{Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher}; /// Watch a directory recursively, forwarding only events that can change reconciled state. @@ -66,10 +66,29 @@ pub(crate) fn watch_delivery_inputs( pub(crate) struct CatalogDeclarationWatcher { root: PathBuf, watcher: RecommendedWatcher, - watched: BTreeSet, + watched: BTreeMap, failed: BTreeSet, } +/// Identity of a watched directory. An inotify watch attaches to an INODE, not a name, so a +/// directory deleted and recreated at the same pathname is a DIFFERENT directory: matching on +/// identity alone lets `refresh` force re-registration for replacements. +#[cfg(unix)] +type DirIdentity = (u64, u64); +#[cfg(not(unix))] +type DirIdentity = (); + +#[cfg(unix)] +fn dir_identity(path: &Path) -> Option { + use std::os::unix::fs::MetadataExt; + fs::metadata(path).ok().map(|meta| (meta.dev(), meta.ino())) +} + +#[cfg(not(unix))] +fn dir_identity(path: &Path) -> Option { + fs::metadata(path).ok().map(|_| ()) +} + impl CatalogDeclarationWatcher { fn new(root: &Path, tx: Sender<()>) -> notify::Result { let callback_root = root.to_path_buf(); @@ -81,11 +100,13 @@ impl CatalogDeclarationWatcher { let mut this = Self { root: root.to_path_buf(), watcher, - watched: BTreeSet::new(), + watched: BTreeMap::new(), failed: BTreeSet::new(), }; this.watcher.watch(root, RecursiveMode::NonRecursive)?; - this.watched.insert(root.to_path_buf()); + if let Some(identity) = dir_identity(root) { + this.watched.insert(root.to_path_buf(), identity); + } this.refresh(); Ok(this) } @@ -94,26 +115,40 @@ impl CatalogDeclarationWatcher { pub(crate) fn refresh(&mut self) { let desired = declaration_watch_dirs(&self.root); - for stale in self - .watched - .difference(&desired) - .cloned() - .collect::>() - { + // An inotify watch dies with its inode. A directory deleted and recreated at the same + // pathname keeps its key here, so compare identities, not names: a replacement directory + // must be force-registered or edits below it would wait for the timer fallback. + let mut stale = Vec::new(); + self.watched + .retain(|path, identity| match desired.get(path) { + Some(fresh) if *fresh == Some(*identity) => true, + _ => { + stale.push(path.clone()); + false + } + }); + for path in &stale { // Backends normally discard a watch when its directory disappears. An explicit // best-effort unwatch also handles moves that leave the watched inode alive elsewhere. - let _ = self.watcher.unwatch(&stale); - self.watched.remove(&stale); + let _ = self.watcher.unwatch(path); } - for added in desired - .difference(&self.watched) - .cloned() - .collect::>() - { + for added in desired.into_keys() { + if self.watched.contains_key(&added) { + continue; + } match self.watcher.watch(&added, RecursiveMode::NonRecursive) { Ok(()) => { self.failed.remove(&added); - self.watched.insert(added); + match dir_identity(&added) { + Some(identity) => { + self.watched.insert(added, identity); + } + // Vanished between registration and stat: leave it unrecorded so the + // next refresh retries from scratch. + None => { + let _ = self.watcher.unwatch(&added); + } + } } Err(error) if self.failed.insert(added.clone()) => { eprintln!( @@ -127,8 +162,8 @@ impl CatalogDeclarationWatcher { } #[cfg(test)] - fn watched_dirs(&self) -> &BTreeSet { - &self.watched + fn watched_dirs(&self) -> BTreeSet { + self.watched.keys().cloned().collect() } } @@ -144,9 +179,9 @@ pub(crate) fn watch_catalog_declarations( /// Every directory that can hold declarations, discovered WITHOUT descending into Resource /// payloads: recursion is gated by [`agent_spec::is_catalog_path`], so a declaration parent's /// `resources`/`archive`/`inbox` children and `.git`/`.st2` control dirs prune the walk. -fn declaration_watch_dirs(root: &Path) -> BTreeSet { - fn collect(root: &Path, dir: &Path, out: &mut BTreeSet) { - out.insert(dir.to_path_buf()); +fn declaration_watch_dirs(root: &Path) -> BTreeMap> { + fn collect(root: &Path, dir: &Path, out: &mut BTreeMap>) { + out.insert(dir.to_path_buf(), dir_identity(dir)); let Ok(entries) = fs::read_dir(dir) else { return; }; @@ -160,7 +195,7 @@ fn declaration_watch_dirs(root: &Path) -> BTreeSet { } } - let mut dirs = BTreeSet::new(); + let mut dirs = BTreeMap::new(); collect(root, root, &mut dirs); dirs } @@ -175,21 +210,21 @@ fn should_wake_catalog(root: &Path, event: &Event) -> bool { }) } -/// Directory create/remove/rename must wake even though no `agent.kdl` path exists yet — the -/// created directory may receive one next, and the watcher needs `refresh` anyway. +/// Directory-level topology changes must wake even though no `agent.kdl` path exists yet — the +/// created directory may receive one next, and the watcher needs `refresh` anyway. FILE-level +/// create/remove/rename stays silent: a scratch, log, or editor-swap file in declaration space +/// must not wake a full-catalog reconcile. Linux classifies creates/removals by entry type; +/// renames carry no type, so the arrived side is checked against the live tree. A rename AWAY +/// from the catalog has no surviving entry to inspect and stays silent here — an in-catalog +/// rename also emits the arrived side, and a full removal is bounded by the timer plus the +/// next reconciliation's `refresh`. fn is_directory_topology_mutation(event: &Event) -> bool { - matches!( - event.kind, - EventKind::Create(_) - | EventKind::Remove(_) - | EventKind::Modify(ModifyKind::Name( - RenameMode::Any - | RenameMode::From - | RenameMode::To - | RenameMode::Both - | RenameMode::Other - )) - ) + match &event.kind { + EventKind::Create(CreateKind::Folder) => true, + EventKind::Remove(RemoveKind::Folder | RemoveKind::Any | RemoveKind::Other) => true, + EventKind::Modify(ModifyKind::Name(_)) => event.paths.iter().any(|p| p.is_dir()), + _ => false, + } } fn is_declaration_path(root: &Path, path: &Path) -> bool { @@ -412,7 +447,7 @@ mod tests { let watched = declaration_watch_dirs(root); assert_eq!( - watched, + watched.keys().cloned().collect::>(), [ root.to_path_buf(), root.join("agents"), @@ -423,11 +458,47 @@ mod tests { .collect() ); assert!( - watched.iter().all(|path| !path.starts_with(outside.path())), + watched.keys().all(|path| !path.starts_with(outside.path())), "Resource worktree links must not escape the catalog watch boundary" ); } + #[test] + fn topology_wakes_are_directory_scoped() { + use notify::event::{AccessKind, AccessMode}; + + let file_create = Event::new(EventKind::Create(CreateKind::File)) + .add_path(PathBuf::from("/cat/agents/h/live/scratch.log")); + let dir_create = Event::new(EventKind::Create(CreateKind::Folder)) + .add_path(PathBuf::from("/cat/agents/h/newdir")); + let file_remove = Event::new(EventKind::Remove(RemoveKind::File)) + .add_path(PathBuf::from("/cat/agents/h/live/scratch.log")); + let dir_remove = Event::new(EventKind::Remove(RemoveKind::Folder)) + .add_path(PathBuf::from("/cat/agents/h/gone")); + let file_rename_in = Event::new(EventKind::Modify(ModifyKind::Name(RenameMode::To))) + .add_path(PathBuf::from("/cat/agents/h/live/renamed.txt")); + let rename_away = Event::new(EventKind::Modify(ModifyKind::Name(RenameMode::From))) + .add_path(PathBuf::from("/cat/agents/h/live/.swap.swp")); + + for quiet in [file_create, file_remove, file_rename_in, rename_away] { + assert!( + !is_directory_topology_mutation(&quiet), + "{quiet:?}: a FILE topology event must not wake the catalog" + ); + } + for loud in [dir_create, dir_remove] { + assert!( + is_directory_topology_mutation(&loud), + "{loud:?}: a DIRECTORY-level topology event must wake the catalog" + ); + } + + // Sanity: access events never wake regardless of classification. + let read = Event::new(EventKind::Access(AccessKind::Open(AccessMode::Read))) + .add_path(PathBuf::from("/cat/agents/h/newdir")); + assert!(!is_directory_topology_mutation(&read)); + } + #[test] fn watcher_installation_preserves_backend_errors() { let parent = tempfile::tempdir().unwrap(); @@ -498,4 +569,66 @@ mod tests { rx.recv_timeout(Duration::from_secs(1)) .expect("declaration removal must wake"); } + + #[cfg(target_os = "linux")] + #[test] + fn non_declaration_file_topology_stays_silent() { + use std::sync::mpsc::channel; + use std::time::Duration; + + let catalog = tempfile::tempdir().unwrap(); + let agent = catalog.path().join("agents/h/live"); + std::fs::create_dir_all(&agent).unwrap(); + std::fs::write( + agent.join("agent.kdl"), + r#"agent "live" { host "h"; command "x" }"#, + ) + .unwrap(); + + let (tx, rx) = channel(); + let _watcher = watch_catalog_declarations(catalog.path(), tx).expect("start watcher"); + + std::fs::write(agent.join("scratch.log"), "noise").unwrap(); + std::fs::rename(agent.join("scratch.log"), agent.join("scratch2.log")).unwrap(); + std::fs::remove_file(agent.join("scratch2.log")).unwrap(); + assert!( + rx.recv_timeout(Duration::from_millis(300)).is_err(), + "scratch-file create/rename/remove must not wake a full-catalog reconcile" + ); + } + + #[cfg(target_os = "linux")] + #[test] + fn replaced_declaration_directory_is_resubscribed_on_refresh() { + use std::sync::mpsc::channel; + use std::time::Duration; + + let catalog = tempfile::tempdir().unwrap(); + let root = catalog.path(); + let agent = root.join("agents/h/live"); + std::fs::create_dir_all(&agent).unwrap(); + std::fs::write( + agent.join("agent.kdl"), + r#"agent "live" { host "h"; command "x" }"#, + ) + .unwrap(); + + let (tx, rx) = channel(); + let mut watcher = watch_catalog_declarations(root, tx).expect("start watcher"); + + // Delete and recreate at the SAME pathname: the backend watch died with the old inode + // while `watched` still holds the name, so only identity comparison can catch this. + std::fs::remove_dir_all(&agent).unwrap(); + std::fs::create_dir_all(&agent).unwrap(); + watcher.refresh(); + while rx.try_recv().is_ok() {} + + std::fs::write( + agent.join("agent.kdl"), + r#"agent "live" { host "h"; command "changed" }"#, + ) + .unwrap(); + rx.recv_timeout(Duration::from_secs(1)) + .expect("a replaced directory must be resubscribed by refresh"); + } } From 6b6d83676affa8be1a8dfe164c437056e9e9afd9 Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Tue, 25 Aug 2026 08:26:02 +0200 Subject: [PATCH 3/3] fix(watch): share the watch set so removals invalidate eagerly Identity comparison alone cannot catch every replacement: deleting and recreating a directory at the same pathname can reuse the old (dev, ino), so refresh would see no change while the backend watch had died with the original inode. Share the tracked-directory map with the notify callback and drop entries the moment the backend reports their directory removed or renamed away (including descendants), making resubscription at the next refresh deterministic instead of trusting a stat race. agent-identity: unknown agent-persona: generalist agent-supervisor: unavailable agent-tool: OMP agent-tool-version: 18.0.3 agent-runtime: OMP 18.0.3 tooling-profile: dotfiles@f33cd9c-dirty --- src/watch.rs | 77 ++++++++++++++++++++++++++++++++++++++-------------- 1 file changed, 57 insertions(+), 20 deletions(-) diff --git a/src/watch.rs b/src/watch.rs index 0f2be12d..f9d535ae 100644 --- a/src/watch.rs +++ b/src/watch.rs @@ -9,6 +9,7 @@ use std::collections::{BTreeMap, BTreeSet}; use std::fs; use std::path::{Path, PathBuf}; use std::sync::mpsc::Sender; +use std::sync::{Arc, Mutex}; use notify::event::{CreateKind, ModifyKind, RemoveKind, RenameMode}; use notify::{Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher}; @@ -66,7 +67,7 @@ pub(crate) fn watch_delivery_inputs( pub(crate) struct CatalogDeclarationWatcher { root: PathBuf, watcher: RecommendedWatcher, - watched: BTreeMap, + watched: Arc>>>, failed: BTreeSet, } @@ -92,21 +93,29 @@ fn dir_identity(path: &Path) -> Option { impl CatalogDeclarationWatcher { fn new(root: &Path, tx: Sender<()>) -> notify::Result { let callback_root = root.to_path_buf(); + let watched = Arc::new(Mutex::new(BTreeMap::new())); + let invalidator = Arc::clone(&watched); let watcher = notify::recommended_watcher(move |result: notify::Result| { - if result.is_ok_and(|event| should_wake_catalog(&callback_root, &event)) { - let _ = tx.send(()); + if let Ok(event) = result { + invalidate_removed_dirs(&invalidator, &event); + if should_wake_catalog(&callback_root, &event) { + let _ = tx.send(()); + } } })?; let mut this = Self { root: root.to_path_buf(), watcher, - watched: BTreeMap::new(), + watched, failed: BTreeSet::new(), }; this.watcher.watch(root, RecursiveMode::NonRecursive)?; - if let Some(identity) = dir_identity(root) { - this.watched.insert(root.to_path_buf(), identity); - } + let mut watched = this + .watched + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + watched.insert(root.to_path_buf(), dir_identity(root)); + drop(watched); this.refresh(); Ok(this) } @@ -114,26 +123,26 @@ impl CatalogDeclarationWatcher { /// Reconcile the shallow subscriptions with the current declaration namespace. pub(crate) fn refresh(&mut self) { let desired = declaration_watch_dirs(&self.root); + let mut watched = self.watched.lock().unwrap_or_else(|p| p.into_inner()); // An inotify watch dies with its inode. A directory deleted and recreated at the same - // pathname keeps its key here, so compare identities, not names: a replacement directory - // must be force-registered or edits below it would wait for the timer fallback. + // pathname can even reuse the identity, so `invalidate_removed_dirs` drops removals from + // this set eagerly; identity comparison here then catches anything the events missed. let mut stale = Vec::new(); - self.watched - .retain(|path, identity| match desired.get(path) { - Some(fresh) if *fresh == Some(*identity) => true, - _ => { - stale.push(path.clone()); - false - } - }); + watched.retain(|path, identity| match desired.get(path) { + Some(fresh) if *fresh == *identity => true, + _ => { + stale.push(path.clone()); + false + } + }); for path in &stale { // Backends normally discard a watch when its directory disappears. An explicit // best-effort unwatch also handles moves that leave the watched inode alive elsewhere. let _ = self.watcher.unwatch(path); } for added in desired.into_keys() { - if self.watched.contains_key(&added) { + if watched.contains_key(&added) { continue; } match self.watcher.watch(&added, RecursiveMode::NonRecursive) { @@ -141,7 +150,7 @@ impl CatalogDeclarationWatcher { self.failed.remove(&added); match dir_identity(&added) { Some(identity) => { - self.watched.insert(added, identity); + watched.insert(added, Some(identity)); } // Vanished between registration and stat: leave it unrecorded so the // next refresh retries from scratch. @@ -163,7 +172,35 @@ impl CatalogDeclarationWatcher { #[cfg(test)] fn watched_dirs(&self) -> BTreeSet { - self.watched.keys().cloned().collect() + self.watched + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .keys() + .cloned() + .collect() + } +} + +/// Eagerly drop tracked directories the moment the backend reports them removed or renamed +/// away — an inotify watch dies with its inode, and a same-pathname replacement can reuse the +/// old identity, so only event-time invalidation makes the next [`refresh`] re-register +/// deterministically instead of trusting a stat race. +fn invalidate_removed_dirs(watched: &Mutex>>, event: &Event) { + let torn_down = matches!( + &event.kind, + EventKind::Remove(RemoveKind::Folder | RemoveKind::Any | RemoveKind::Other) + | EventKind::Modify(ModifyKind::Name(RenameMode::From)) + ); + if !torn_down { + return; + } + let mut watched = watched + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + for path in &event.paths { + watched.remove(path); + // Removing or moving away a parent retires every watch beneath it too. + watched.retain(|tracked, _| !tracked.starts_with(path)); } }