From d8787363b12b5ab191abd7d59c6e57c523ea9eeb Mon Sep 17 00:00:00 2001 From: Johannes Schickling Date: Sun, 23 Aug 2026 18:20:06 +0200 Subject: [PATCH 1/2] fix(run): bound catalog declaration watches --- src/run.rs | 34 ++++++- src/watch.rs | 261 +++++++++++++++++++++++++++++++++++++++++++++++---- 2 files changed, 275 insertions(+), 20 deletions(-) diff --git a/src/run.rs b/src/run.rs index 3cb36f62..b0ccc273 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,6 +2000,21 @@ fn drain(rx: &Receiver<()>) { while rx.try_recv().is_ok() {} } +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. @@ -2026,7 +2041,7 @@ fn up_loop_until( .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 = best_effort_catalog_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 +2068,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 { @@ -2443,6 +2461,18 @@ mod tests { ); } + #[cfg(target_os = "linux")] + #[test] + fn catalog_watcher_failure_selects_timer_fallback() { + let parent = tempfile::tempdir().unwrap(); + let missing_catalog = parent.path().join("missing"); + let (tx, _rx) = channel(); + assert!( + best_effort_catalog_watcher(&missing_catalog, tx).is_none(), + "watch installation failure must leave the timer-only path selected" + ); + } + // ── 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 6f4a133d..f954bac8 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( @@ -25,28 +28,140 @@ pub(crate) fn watch_recursive_mutations( 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. The catalog also contains unbounded Resource payloads, so a +/// recursive subscription can spend all startup time walking data that the callback would later +/// ignore. Keep one non-recursive watch per declaration-space directory instead. [`refresh`] is +/// called after each reconciliation pass so a newly created directory becomes watched before +/// later edits inside it. +/// +/// The boundary intentionally inherits [`agent_spec::is_catalog_path`]: outside the canonical +/// `agents//` layout, a `resources`-named directory remains declaration space until +/// an adjacent `agent.kdl` establishes that it is Agent state. +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_count(&self) -> usize { + self.watched.len() + } +} + /// 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) +} + +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) +} + +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 { @@ -156,6 +271,56 @@ mod tests { assert!(!is_declaration_path(root, &root.join("team/rendered.kdl"))); } + #[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 linux_reads_are_silent_but_real_mutations_wake() { @@ -208,4 +373,64 @@ mod tests { |dir| std::fs::remove_file(dir.join("removed")).unwrap(), ); } + + #[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_count(), + 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.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 e4f9853a2f915d8da538cdbb6ccf99087772c91e Mon Sep 17 00:00:00 2001 From: Johannes Schickling Date: Sun, 23 Aug 2026 22:48:23 +0200 Subject: [PATCH 2/2] fix(run): preserve timer fallback without a watcher --- src/run.rs | 81 ++++++++++++++++++++++++++++++++++++++++++------------ 1 file changed, 63 insertions(+), 18 deletions(-) diff --git a/src/run.rs b/src/run.rs index b0ccc273..ef1fdc69 100644 --- a/src/run.rs +++ b/src/run.rs @@ -2000,6 +2000,36 @@ fn drain(rx: &Receiver<()>) { while rx.try_recv().is_ok() {} } +#[derive(Debug, PartialEq, Eq)] +enum ReconcileWake { + Change, + Interval, + Stop, +} + +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) => {} + // A failed best-effort watcher drops the channel's sender. Disconnection must preserve + // timer polling rather than turning the outer supervisor loop into a busy loop. + Err(RecvTimeoutError::Disconnected) => std::thread::sleep(slice), + } + } +} + fn best_effort_catalog_watcher( root: &Path, tx: Sender<()>, @@ -2092,22 +2122,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; } } @@ -2463,14 +2479,43 @@ mod tests { #[cfg(target_os = "linux")] #[test] - fn catalog_watcher_failure_selects_timer_fallback() { + fn catalog_watcher_failure_waits_for_timer_while_changes_still_wake_early() { let parent = tempfile::tempdir().unwrap(); let missing_catalog = parent.path().join("missing"); - let (tx, _rx) = channel(); + let (tx, rx) = channel(); assert!( best_effort_catalog_watcher(&missing_catalog, tx).is_none(), "watch installation failure must leave the timer-only path selected" ); + let started = Instant::now(); + assert_eq!( + wait_for_reconcile(&rx, Duration::from_millis(20), &AtomicBool::new(false)), + ReconcileWake::Interval, + "a disconnected watcher channel must wait for the timer instead of spinning" + ); + assert!( + started.elapsed() >= Duration::from_millis(15), + "timer fallback must consume the interval instead of returning immediately" + ); + + 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 (tx, rx) = channel(); + let _watcher = best_effort_catalog_watcher(catalog.path(), tx) + .expect("valid declaration catalog must be watched"); + std::fs::write( + &spec, + r#"agent "live" { host "test-host"; command "changed" }"#, + ) + .unwrap(); + assert_eq!( + wait_for_reconcile(&rx, Duration::from_secs(1), &AtomicBool::new(false)), + ReconcileWake::Change, + "an ordinary declaration mutation must wake before the timer" + ); } // ── liveness debounce (R21c): a transient `pty list` not-alive flicker under load must not