From c04c877d154b4d428dd59280c6d4ff87b519b41f Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 19 May 2026 14:04:07 +0100 Subject: [PATCH] Own Epiphany invalidation watcher --- epiphany-codex-bridge/Cargo.toml | 1 + epiphany-codex-bridge/src/invalidation.rs | 85 +++++++++-------------- vendor/codex/codex-rs/Cargo.lock | 1 + 3 files changed, 36 insertions(+), 51 deletions(-) diff --git a/epiphany-codex-bridge/Cargo.toml b/epiphany-codex-bridge/Cargo.toml index 92ebedee..9031eff3 100644 --- a/epiphany-codex-bridge/Cargo.toml +++ b/epiphany-codex-bridge/Cargo.toml @@ -19,6 +19,7 @@ codex-utils-absolute-path = { path = "../vendor/codex/codex-rs/utils/absolute-pa epiphany-core = { path = "../epiphany-core" } epiphany-openai-adapter = { path = "../epiphany-openai-adapter" } epiphany-openai-codex-spine = { path = "../epiphany-openai-codex-spine" } +notify = "8.2.0" serde_json = "1" serde = { version = "1", features = ["derive"] } tokio = { version = "1", features = ["rt", "sync"] } diff --git a/epiphany-codex-bridge/src/invalidation.rs b/epiphany-codex-bridge/src/invalidation.rs index d5a60caf..79f16459 100644 --- a/epiphany-codex-bridge/src/invalidation.rs +++ b/epiphany-codex-bridge/src/invalidation.rs @@ -1,23 +1,18 @@ -use codex_core::file_watcher::FileWatcher; -use codex_core::file_watcher::FileWatcherSubscriber; -use codex_core::file_watcher::ThrottledWatchReceiver; -use codex_core::file_watcher::WatchPath; -use codex_core::file_watcher::WatchRegistration; use codex_utils_absolute_path::AbsolutePathBuf; +use notify::RecursiveMode; +use notify::Watcher; use std::collections::HashMap; use std::path::PathBuf; use std::sync::Arc; -use std::time::Duration; use std::time::SystemTime; use std::time::UNIX_EPOCH; use tokio::sync::Mutex as AsyncMutex; +use tokio::sync::mpsc; use tokio::sync::oneshot; use tracing::warn; use crate::reorient::EpiphanyFreshnessWatcherSnapshot; -const INVALIDATION_WATCH_DEBOUNCE: Duration = Duration::from_millis(200); - #[derive(Clone, Debug, PartialEq, Eq)] pub struct EpiphanyInvalidationSnapshot { pub available: bool, @@ -39,8 +34,6 @@ pub fn epiphany_freshness_watcher_snapshot( #[derive(Clone)] pub struct EpiphanyInvalidationManager { - available: bool, - file_watcher: Arc, state: Arc>>, } @@ -48,8 +41,7 @@ struct WatchEntry { workspace_root: AbsolutePathBuf, latest: Arc>, terminate_tx: oneshot::Sender>, - _subscriber: FileWatcherSubscriber, - _registration: WatchRegistration, + _watcher: notify::RecommendedWatcher, } #[derive(Clone, Debug, Default, PartialEq, Eq)] @@ -60,28 +52,12 @@ struct LatestInvalidation { impl EpiphanyInvalidationManager { pub fn new() -> Self { - match FileWatcher::new() { - Ok(file_watcher) => Self { - available: true, - file_watcher: Arc::new(file_watcher), - state: Arc::new(AsyncMutex::new(HashMap::new())), - }, - Err(err) => { - warn!("epiphany invalidation manager falling back to noop core watcher: {err}"); - Self { - available: false, - file_watcher: Arc::new(FileWatcher::noop()), - state: Arc::new(AsyncMutex::new(HashMap::new())), - } - } + Self { + state: Arc::new(AsyncMutex::new(HashMap::new())), } } pub async fn ensure_thread_watch(&self, thread_id: &str, workspace_root: &AbsolutePathBuf) { - if !self.available { - return; - } - let existing_entry = { let mut state = self.state.lock().await; match state.get(thread_id) { @@ -95,12 +71,30 @@ impl EpiphanyInvalidationManager { stop_watch_entry(entry).await; } - let (subscriber, rx) = self.file_watcher.add_subscriber(); - let registration = subscriber.register_paths(vec![WatchPath { - path: workspace_root.to_path_buf(), - recursive: true, - }]); let latest = Arc::new(AsyncMutex::new(LatestInvalidation::default())); + let (event_tx, mut event_rx) = mpsc::unbounded_channel::>(); + let mut watcher = + match notify::recommended_watcher(move |result: notify::Result| { + match result { + Ok(event) => { + let _ = event_tx.send(event.paths); + } + Err(err) => warn!("epiphany invalidation watcher event failed: {err}"), + } + }) { + Ok(watcher) => watcher, + Err(err) => { + warn!("epiphany invalidation manager could not create watcher: {err}"); + return; + } + }; + if let Err(err) = watcher.watch(workspace_root.as_path(), RecursiveMode::Recursive) { + warn!( + "epiphany invalidation manager could not watch `{}`: {err}", + workspace_root.display() + ); + return; + } let (terminate_tx, terminate_rx) = oneshot::channel(); let workspace_root_path = workspace_root.to_path_buf(); @@ -108,25 +102,23 @@ impl EpiphanyInvalidationManager { workspace_root: workspace_root.clone(), latest: Arc::clone(&latest), terminate_tx, - _subscriber: subscriber, - _registration: registration, + _watcher: watcher, }; self.state.lock().await.insert(thread_id.to_string(), entry); tokio::spawn(async move { - let mut rx = ThrottledWatchReceiver::new(rx, INVALIDATION_WATCH_DEBOUNCE); let mut terminate_rx = terminate_rx; let mut done_tx = None; loop { - let event = tokio::select! { + let paths = tokio::select! { biased; result = &mut terminate_rx => { done_tx = result.ok(); break; } - event = rx.recv() => match event { - Some(event) => event, + paths = event_rx.recv() => match paths { + Some(paths) => paths, None => break, }, }; @@ -135,7 +127,7 @@ impl EpiphanyInvalidationManager { let mut latest = latest.lock().await; latest.observed_at_unix_seconds = Some(observed_at_unix_seconds); latest.changed_paths = - normalize_changed_paths(event.paths, workspace_root_path.as_path()); + normalize_changed_paths(paths, workspace_root_path.as_path()); } if let Some(done_tx) = done_tx { @@ -145,15 +137,6 @@ impl EpiphanyInvalidationManager { } pub async fn snapshot(&self, thread_id: &str) -> EpiphanyInvalidationSnapshot { - if !self.available { - return EpiphanyInvalidationSnapshot { - available: false, - workspace_root: None, - observed_at_unix_seconds: None, - changed_paths: Vec::new(), - }; - } - let (workspace_root, latest) = { let state = self.state.lock().await; match state.get(thread_id) { diff --git a/vendor/codex/codex-rs/Cargo.lock b/vendor/codex/codex-rs/Cargo.lock index bf3fe852..377fb1af 100644 --- a/vendor/codex/codex-rs/Cargo.lock +++ b/vendor/codex/codex-rs/Cargo.lock @@ -5090,6 +5090,7 @@ dependencies = [ "epiphany-core", "epiphany-openai-adapter", "epiphany-openai-codex-spine", + "notify", "serde", "serde_json", "tokio",