fix(watch): bound sidecar delivery watches away from Resource payloads

The DING sidecar and the codex/opencode delivery pumps recursively
watched each agent dir. A recursive registration eagerly allocates one
inotify watch per directory BEFORE any callback filtering, and agent dirs
contain `resources/` payload trees — the same unbounded-traversal class
the supervisor hit in #314, one level down.

- watch_delivery_inputs now makes two shallow subscriptions: a
  non-recursive watch on the agent dir (status file) plus a recursive
  watch on resources/inbox, which is st2-owned and bounded by delivery
  traffic rather than payloads.
- run_ding anchors its watch on the inbox itself and creates it up front,
  instead of falling back to a recursive walk over the whole agent dir
  when the inbox did not exist yet.
- All remaining recursive registrations construct their watcher with
  follow_symlinks(false): a symlink must never carry traversal outside
  the watched boundary.

Closes #314 (sidecar instance).

Co-authored-by: schickling-assistant <schickling-assistant@users.noreply.github.com>
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

Author: schickling-assistant

src/ding/mod.rs

@@ -951,13 +951,13 @@ pub fn run_ding(
     stop: &AtomicBool,
 ) -> anyhow::Result<()> {
     // Arm the watcher before seeding. The timer remains the correctness fallback if watching fails.
+    // The subscription is the inbox itself, never a recursive walk over the agent dir: Resource
+    // payload trees live under `resources/` beside the inbox, and eager registration would pay
+    // one inotify watch per payload directory before any filtering ever ran. The inbox is
+    // created up front so the watch anchors on it directly (senders create it on demand anyway).
+    let _ = std::fs::create_dir_all(inbox_dir);
     let (tx, rx) = channel::<()>();
-    let watch_at = if inbox_dir.exists() {
-        inbox_dir
-    } else {
-        inbox_dir.parent().unwrap_or(inbox_dir)
-    };
-    let _watcher = crate::watch::watch_recursive_mutations(watch_at, tx);
+    let _watcher = crate::watch::watch_recursive_mutations(inbox_dir, tx);
 
     let mut seen = HashSet::new();
     let backlog = new_arrivals(inbox_dir, &mut seen);

src/watch.rs

@@ -14,11 +14,15 @@ use notify::event::{ModifyKind, RenameMode};
 use notify::{Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
 
 /// Watch a directory recursively, forwarding only events that can change reconciled state.
+///
+/// Recursive registration eagerly walks the tree (one inotify watch per directory), so this is
+/// only sound over st2-owned, payload-free directories. Symlink following is disabled: a link
+/// must never carry traversal outside the watched boundary.
 pub(crate) fn watch_recursive_mutations(
     dir: &Path,
     tx: Sender<()>,
 ) -> Option<notify::RecommendedWatcher> {
-    let mut watcher = notify::recommended_watcher(move |result: notify::Result<Event>| {
+    let mut watcher = recommended_watcher(move |result: notify::Result<Event>| {
         if result.is_ok_and(|event| is_mutation(&event)) {
             let _ = tx.send(());
         }
@@ -37,23 +41,43 @@ pub(crate) fn watch_delivery_inputs(
     tx: Sender<()>,
 ) -> Option<notify::RecommendedWatcher> {
     let inbox = agent_dir.join("resources").join("inbox");
+    let inbox_for_callback = inbox.clone();
     let status = agent_dir.join("status");
-    let mut watcher = notify::recommended_watcher(move |result: notify::Result<Event>| {
+    let mut watcher = recommended_watcher(move |result: notify::Result<Event>| {
         if result.is_ok_and(|event| {
             is_mutation(&event)
                 && event
                     .paths
                     .iter()
-                    .any(|path| path.starts_with(&inbox) || *path == status)
+                    .any(|path| path.starts_with(&inbox_for_callback) || *path == status)
         }) {
             let _ = tx.send(());
         }
     })
     .ok()?;
-    watcher.watch(agent_dir, RecursiveMode::Recursive).ok()?;
+    // Two shallow subscriptions instead of one recursive walk over the agent dir: Resource
+    // payload trees live beside the inbox under `resources/`, and a recursive subscription pays
+    // one inotify watch per payload directory before any filtering ever runs. The agent-dir watch
+    // stays shallow because everything beside `status` is runtime state; the inbox itself is
+    // st2-owned and message-flat, so its recursion is bounded by delivery traffic, not payloads.
+    let inbox_for_watch = inbox;
+    watcher.watch(agent_dir, RecursiveMode::NonRecursive).ok()?;
+    watcher
+        .watch(&inbox_for_watch, RecursiveMode::Recursive)
+        .ok()?;
     Some(watcher)
 }
 
+/// A [`notify::RecommendedWatcher`] that never follows symlinks while walking.
+fn recommended_watcher(
+    handler: impl FnMut(notify::Result<Event>) + Send + 'static,
+) -> notify::Result<notify::RecommendedWatcher> {
+    notify::RecommendedWatcher::new(
+        handler,
+        notify::Config::default().with_follow_symlinks(false),
+    )
+}
+
 /// A shallow subscription over the directories that can contain declarations.
 ///
 /// `notify` implements a recursive Linux watch by eagerly walking the entire tree and allocating
@@ -336,6 +360,40 @@ mod tests {
             .expect("status rename must wake");
     }
 
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn delivery_watch_is_bounded_by_inputs_not_payload_depth() {
+        use std::sync::mpsc::channel;
+
+        fn inotify_watch_count() -> usize {
+            std::fs::read_dir("/proc/self/fdinfo")
+                .unwrap()
+                .flatten()
+                .map(|entry| std::fs::read_to_string(entry.path()).unwrap_or_default())
+                .filter(|content| content.contains("inotify wd:"))
+                .map(|content| content.lines().count())
+                .sum()
+        }
+
+        let dir = tempfile::tempdir().unwrap();
+        let agent_dir = dir.path();
+        let payload = agent_dir.join("resources/worktree/node_modules");
+        for i in 0..300 {
+            std::fs::create_dir_all(payload.join(format!("pkg-{i}/dist/sub"))).unwrap();
+        }
+        std::fs::create_dir_all(agent_dir.join("resources/inbox")).unwrap();
+
+        let before = inotify_watch_count();
+        let (tx, _rx) = channel();
+        let _watcher = watch_delivery_inputs(agent_dir, tx).expect("start delivery watcher");
+        let delta = inotify_watch_count() - before;
+        assert!(
+            delta < 32,
+            "payload depth must not drive watch allocation: {delta} watches for a \
+             900-directory payload tree"
+        );
+    }
+
     #[cfg(target_os = "linux")]
     #[test]
     fn linux_reads_are_silent_but_real_mutations_wake() {