Cache empty remote skill discovery with filesystem watches (#57598)

## Why

Repeated repository skill discovery performs remote metadata requests even when no `.agents/skills` directories exist. Reuse verified empty results to avoid those round trips while watching for changes that require a new scan.

## What changed

- Add the Linux executor capability `fs_watch_missing`, the `fs/watchMissing` request, and `fs/missingChanged` notifications.
- Register watches in the background and reuse empty results only after the executor acknowledges coverage. Key reuse by filesystem, working directory, and project root markers.
- Invalidate observations on relevant filesystem or mount changes, watch supersession, or connection loss. Clear cached observations on forced reloads and cache resets.
- Retain live discovery while registration is pending or watching is unsupported, unavailable, or cannot verify the observation.

## Testing

Add tests for empty-result reuse and invalidation, skill directory creation before watch acknowledgement, unsupported executors, idle WebSocket notifications, reconnection, case aliases, and bind mounts and unmounts. Include an ignored manual latency comparison for repeated empty remote scans.

GitOrigin-RevId: 3ad499decefb184a99c068afa01d0dcdd8605c7f

codex-rs/exec-server/src/server/handler.rs

@@ -79,6 +79,9 @@ use crate::server::session_registry::SessionHandle;
 use crate::server::session_registry::SessionRegistry;
 use crate::telemetry::ExecutorRegistration;
 
+#[path = "missing_watch.rs"]
+mod missing_watch;
+
 pub(crate) struct ExecServerHandler {
     pub(super) executor_registration: Option<Arc<ExecutorRegistration>>,
     session_registry: Arc<SessionRegistry>,
@@ -86,6 +89,8 @@ pub(crate) struct ExecServerHandler {
     notifications: RpcNotificationSender,
     session: StdMutex<Option<SessionHandle>>,
     active_body_stream_ids: Mutex<HashSet<String>>,
+    // Only the newest observation on this physical connection can be reused.
+    missing_watch: Arc<StdMutex<missing_watch::MissingWatch>>,
     background_task_shutdown: CancellationToken,
     background_tasks: TaskTracker,
     file_system: FileSystemHandler,
@@ -110,6 +115,7 @@ impl ExecServerHandler {
             notifications,
             session: StdMutex::new(None),
             active_body_stream_ids: Mutex::new(HashSet::new()),
+            missing_watch: Arc::default(),
             background_task_shutdown: CancellationToken::new(),
             background_tasks: TaskTracker::new(),
             file_system: FileSystemHandler::new(runtime_paths.clone()),
@@ -395,6 +401,24 @@ impl ExecServerHandler {
         self.file_system.create_directory(params).await
     }
 
+    pub(crate) async fn fs_watch_missing(
+        &self,
+        params: crate::protocol::FsWatchMissingParams,
+    ) -> Result<crate::protocol::FsWatchMissingResponse, JSONRPCErrorError> {
+        self.require_initialized_for("filesystem")?;
+        let watch = Arc::clone(&self.missing_watch);
+        let notifications = self.notifications.clone();
+        let shutdown = self.background_task_shutdown.clone();
+        tokio::task::spawn_blocking(move || {
+            let Ok(mut active) = watch.try_lock() else {
+                return Ok(crate::protocol::FsWatchMissingResponse { watching: false });
+            };
+            active.register(params, notifications, shutdown)
+        })
+        .await
+        .map_err(|error| crate::rpc::internal_error(error.to_string()))?
+    }
+
     pub(crate) async fn fs_get_metadata(
         &self,
         params: FsGetMetadataParams,

codex-rs/exec-server/src/server/missing_watch.rs

@@ -0,0 +1,207 @@
+//! Watches one missing-path observation per physical executor connection. Changes,
+//! supersession and transport loss invalidate the observation without a clock deadline.
+
+use codex_exec_server_protocol::JSONRPCErrorError;
+use codex_utils_absolute_path::AbsolutePathBuf;
+use std::collections::BTreeSet;
+use std::future::Future;
+use std::io;
+#[cfg(target_os = "linux")]
+use std::os::unix::fs::OpenOptionsExt as _;
+use std::path::Path;
+use tokio_util::sync::CancellationToken;
+use tokio_util::task::AbortOnDropHandle;
+
+use crate::protocol::FS_MISSING_CHANGED_METHOD;
+use crate::protocol::FsMissingChangedNotification;
+use crate::protocol::FsWatchMissingParams;
+use crate::protocol::FsWatchMissingResponse;
+use crate::rpc::RpcNotificationSender;
+
+use notify::Event;
+use notify::EventKind;
+use notify::RecursiveMode;
+use notify::Watcher;
+#[cfg(target_os = "linux")]
+use tokio::io::Interest;
+#[cfg(target_os = "linux")]
+use tokio::io::unix::AsyncFd;
+use tokio::sync::oneshot;
+
+const MAX_WATCH_PATHS: usize = 64;
+
+#[derive(Default)]
+pub(super) struct MissingWatch {
+    id: u64,
+    listener: Option<AbortOnDropHandle<()>>,
+}
+
+impl MissingWatch {
+    pub(super) fn register(
+        &mut self,
+        params: FsWatchMissingParams,
+        notifications: RpcNotificationSender,
+        shutdown: CancellationToken,
+    ) -> Result<FsWatchMissingResponse, JSONRPCErrorError> {
+        let not_watching = FsWatchMissingResponse { watching: false };
+        if params.watch_id <= self.id {
+            return Ok(not_watching);
+        }
+        self.id = params.watch_id;
+        self.listener.take();
+        if !cfg!(target_os = "linux")
+            || params.missing.is_empty()
+            || params.missing.len() > MAX_WATCH_PATHS
+        {
+            return Ok(not_watching);
+        }
+        let watch_id = params.watch_id;
+        let watch = match watch_missing(params) {
+            Ok(Some(watch)) => watch,
+            Ok(None) => return Ok(not_watching),
+            Err(error) => {
+                tracing::debug!(%error, "metadata watch unavailable; retaining live discovery");
+                return Ok(not_watching);
+            }
+        };
+        let changed = FsMissingChangedNotification { watch_id };
+        self.listener = Some(AbortOnDropHandle::new(tokio::spawn(async move {
+            tokio::select! {
+                _ = shutdown.cancelled() => return,
+                () = watch => {}
+            }
+            tracing::debug!(
+                watch_id = changed.watch_id,
+                "invalidating watched missing paths"
+            );
+            let _ = notifications
+                .notify(FS_MISSING_CHANGED_METHOD, &changed)
+                .await;
+        })));
+        Ok(FsWatchMissingResponse { watching: true })
+    }
+}
+
+/// Observe directory and mount changes before verifying the paths. Ancestor changes,
+/// mount changes, watcher errors and queue overflows invalidate the observation.
+fn watch_missing(
+    params: FsWatchMissingParams,
+) -> io::Result<Option<impl Future<Output = ()> + Send>> {
+    let missing = params
+        .missing
+        .iter()
+        .map(|path| path.to_abs_path().map(AbsolutePathBuf::into_path_buf))
+        .collect::<io::Result<Vec<_>>>()?;
+    let present = params
+        .present
+        .map(|path| path.to_abs_path().map(AbsolutePathBuf::into_path_buf))
+        .transpose()?;
+    let mut parents = BTreeSet::new();
+    let targets: Vec<_> = missing.iter().chain(&present).cloned().collect();
+    for target in &targets {
+        for parent in target.ancestors().skip(1) {
+            parents.insert(parent.to_owned());
+            if parents.len() > MAX_WATCH_PATHS {
+                return Ok(None);
+            }
+        }
+    }
+    // inotify stays attached to a covered directory. Arm mount-table notifications
+    // before installing directory watches and verifying the observed paths.
+    #[cfg(target_os = "linux")]
+    let mounts = std::fs::OpenOptions::new()
+        .read(true)
+        .custom_flags(libc::O_NONBLOCK)
+        .open("/proc/self/mounts")
+        .and_then(|file| AsyncFd::with_interest(file, Interest::PRIORITY))
+        .inspect_err(
+            |error| tracing::debug!(%error, "mount changes unavailable for missing-path watch"),
+        )
+        .ok();
+    let (changed, invalidated) = oneshot::channel();
+    let mut changed = Some(changed);
+    let mut watcher = notify::recommended_watcher(move |event: notify::Result<Event>| {
+        if event.as_ref().map_or(true, |event| {
+            event.need_rescan()
+                || (!matches!(event.kind, EventKind::Access(_))
+                    && (event.paths.is_empty()
+                        || event.paths.iter().any(|path| {
+                            targets
+                                .iter()
+                                .any(|target| event_may_affect_target(path, target))
+                        })))
+        }) && let Some(changed) = changed.take()
+        {
+            let _ = changed.send(());
+        }
+    })
+    .map_err(io::Error::other)?;
+
+    // Ancestors sort before their descendants. If a missing parent is created during
+    // setup its already-watched parent invalidates the observation. Reject symlinks:
+    // their targets can change outside any of these directories.
+    for parent in parents {
+        match std::fs::symlink_metadata(&parent) {
+            Ok(metadata) if metadata.is_dir() => {
+                // inotify can accept a watch on a network mount yet never see changes
+                // made by another host. Only reuse observations on local Linux filesystems.
+                #[cfg(target_os = "linux")]
+                if !matches!(
+                    rustix::fs::statfs(&parent)?.f_type as _,
+                    libc::EXT4_SUPER_MAGIC
+                        | 0x5846_5342 /* XFS_SUPER_MAGIC from linux/magic.h; absent in libc on musl */
+                        | libc::BTRFS_SUPER_MAGIC
+                        | libc::TMPFS_MAGIC
+                        | libc::OVERLAYFS_SUPER_MAGIC
+                ) {
+                    return Ok(None);
+                }
+                watcher
+                    .watch(&parent, RecursiveMode::NonRecursive)
+                    .map_err(io::Error::other)?;
+            }
+            Ok(_) => return Ok(None),
+            Err(error) if error.kind() == io::ErrorKind::NotFound => {}
+            Err(error) => return Err(error),
+        }
+    }
+    for path in missing {
+        match std::fs::symlink_metadata(path) {
+            Ok(_) => return Ok(None),
+            Err(error) if error.kind() == io::ErrorKind::NotFound => {}
+            Err(error) => return Err(error),
+        }
+    }
+    if let Some(path) = present {
+        match std::fs::symlink_metadata(path) {
+            Ok(metadata) if !metadata.is_symlink() => {}
+            Ok(_) => return Ok(None),
+            Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None),
+            Err(error) => return Err(error),
+        }
+    }
+    Ok(Some(async move {
+        let _directory_watcher = watcher;
+        #[cfg(target_os = "linux")]
+        if let Some(mounts) = mounts {
+            tokio::select! {
+                _ = invalidated => {}
+                _ = mounts.ready(Interest::PRIORITY) => {}
+            }
+            return;
+        }
+        let _ = invalidated.await;
+    }))
+}
+
+/// Events preserve the spelling that was written, even in casefolded directories.
+/// When either component is non-ASCII, let the next scan resolve Unicode aliases
+/// using the filesystem's own rules.
+fn event_may_affect_target(path: &Path, target: &Path) -> bool {
+    let mut components = target.iter();
+    path.iter().all(|name| {
+        components.next().is_some_and(|expected| {
+            name.eq_ignore_ascii_case(expected) || !name.is_ascii() || !expected.is_ascii()
+        })
+    })
+}

codex-rs/exec-server/src/server/registry.rs

@@ -187,6 +187,12 @@ pub(crate) fn build_router() -> RpcRouter<ExecServerHandler> {
             handler.fs_create_directory(params).await
         },
     );
+    router.request(
+        crate::protocol::FS_WATCH_MISSING_METHOD,
+        |handler: Arc<ExecServerHandler>, params: crate::protocol::FsWatchMissingParams| async move {
+            handler.fs_watch_missing(params).await
+        },
+    );
     router.request(
         FS_GET_METADATA_METHOD,
         |handler: Arc<ExecServerHandler>, params: WireFsGetMetadataParams| async move {