From 4f000bd216836b91d15b09154ddde05a4e24370d Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 13 May 2026 13:07:30 +0100 Subject: [PATCH] Disable Codex skill watcher lifecycle --- notes/fresh-workspace-handoff.md | 4 + state/ledgers.msgpack | Bin 95603 -> 95945 bytes state/map.yaml | 2 +- vendor/codex/codex-rs/core/src/lib.rs | 1 - vendor/codex/codex-rs/core/src/session/mod.rs | 24 ---- .../codex-rs/core/src/session/session.rs | 2 - vendor/codex/codex-rs/core/src/skills.rs | 13 -- .../codex/codex-rs/core/src/skills_watcher.rs | 121 +----------------- .../codex/codex-rs/core/src/thread_manager.rs | 60 +-------- 9 files changed, 11 insertions(+), 216 deletions(-) diff --git a/notes/fresh-workspace-handoff.md b/notes/fresh-workspace-handoff.md index f02c4f3c..b0d345f8 100644 --- a/notes/fresh-workspace-handoff.md +++ b/notes/fresh-workspace-handoff.md @@ -795,6 +795,10 @@ effective plugin skill roots, filesystem skill metadata, disabled-path state, or skill error conversion helpers. The route may exist for old clients; it is not an authority. +The Codex skill watcher is also a no-op marker now. Thread startup no longer +registers skill roots with a file watcher, clears skill caches from watcher +events, or starts a session listener that emits `SkillsUpdateAvailable`. + The Phase 6 freshness slice is landed. It exposes read-only `thread/epiphany/freshness` from live retrieval summaries plus graph frontier/churn state and, for loaded threads, watcher-backed invalidation diff --git a/state/ledgers.msgpack b/state/ledgers.msgpack index 3bb8f6d7be776b22190851a818d69134c6968b57..db584ad489a9e35db5228a861d582dfb75e67878 100644 GIT binary patch delta 257 zcmezTiuL4M)(Of?W&Imfr!g}AYTm=ReGen!!eY^uMJ2^MjSP&;bPY^(4UIz#jjRmJ zr!Qb+)T$jdJWV%_`#D}|Db zqSV9`h2oOLqLR`Apr}GleqK6ILs4paW^qYsQ8Cn@qWt`lVjYOWVz{wsnLrbB^79LT zQb5BKfyx~q7AGeH?JZVFE6UG>t4=LT%>x;dU!;(ln^{t<5Dao-acDtGVo9oFSz=~R TVp2}(bk0)7R915) { - let mut rx = self.services.skills_watcher.subscribe(); - let weak_sess = Arc::downgrade(self); - tokio::spawn(async move { - loop { - match rx.recv().await { - Ok(SkillsWatcherEvent::SkillsChanged { .. }) => { - let Some(sess) = weak_sess.upgrade() else { - break; - }; - let event = Event { - id: sess.next_internal_sub_id(), - msg: EventMsg::SkillsUpdateAvailable, - }; - sess.send_event_raw(event).await; - } - Err(tokio::sync::broadcast::error::RecvError::Closed) => break, - Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue, - } - } - }); - } - pub(crate) fn get_tx_event(&self) -> Sender { self.tx_event.clone() } diff --git a/vendor/codex/codex-rs/core/src/session/session.rs b/vendor/codex/codex-rs/core/src/session/session.rs index c64b1c30..0f193c3a 100644 --- a/vendor/codex/codex-rs/core/src/session/session.rs +++ b/vendor/codex/codex-rs/core/src/session/session.rs @@ -824,8 +824,6 @@ impl Session { sess.send_event_raw(event).await; } - // Start the watcher after SessionConfigured so it cannot emit earlier events. - sess.start_skills_watcher_listener(); let mut required_mcp_servers: Vec = mcp_servers .iter() .filter(|(_, server)| server.enabled && server.required) diff --git a/vendor/codex/codex-rs/core/src/skills.rs b/vendor/codex/codex-rs/core/src/skills.rs index 0681d096..0909622b 100644 --- a/vendor/codex/codex-rs/core/src/skills.rs +++ b/vendor/codex/codex-rs/core/src/skills.rs @@ -3,7 +3,6 @@ use std::collections::HashSet; use std::env; use std::sync::Arc; -use crate::config::Config; use crate::session::session::Session; use crate::session::turn_context::TurnContext; use codex_analytics::InvocationType; @@ -43,18 +42,6 @@ pub use codex_core_skills::render; pub use codex_core_skills::render::SkillRenderSideEffects; pub use codex_core_skills::system; -pub(crate) fn skills_load_input_from_config( - config: &Config, - effective_skill_roots: Vec, -) -> SkillsLoadInput { - SkillsLoadInput::new( - config.cwd.clone(), - effective_skill_roots, - config.config_layer_stack.clone(), - config.bundled_skills_enabled(), - ) -} - pub(crate) async fn resolve_skill_dependencies_for_turn( sess: &Arc, turn_context: &Arc, diff --git a/vendor/codex/codex-rs/core/src/skills_watcher.rs b/vendor/codex/codex-rs/core/src/skills_watcher.rs index 9473282b..cd4eed18 100644 --- a/vendor/codex/codex-rs/core/src/skills_watcher.rs +++ b/vendor/codex/codex-rs/core/src/skills_watcher.rs @@ -1,124 +1,9 @@ -//! Skills-specific watcher built on top of the generic [`FileWatcher`]. +//! Compatibility marker for the disabled Codex skill watcher. -use std::path::PathBuf; -use std::sync::Arc; -use std::time::Duration; - -use tokio::runtime::Handle; -use tokio::sync::broadcast; -use tracing::warn; - -use crate::SkillsManager; -use crate::config::Config; -use crate::file_watcher::FileWatcher; -use crate::file_watcher::FileWatcherSubscriber; -use crate::file_watcher::Receiver; -use crate::file_watcher::ThrottledWatchReceiver; -use crate::file_watcher::WatchPath; -use crate::file_watcher::WatchRegistration; -use crate::plugins::PluginsManager; -use crate::skills_load_input_from_config; - -#[cfg(not(test))] -const WATCHER_THROTTLE_INTERVAL: Duration = Duration::from_secs(10); -#[cfg(test)] -const WATCHER_THROTTLE_INTERVAL: Duration = Duration::from_millis(50); - -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum SkillsWatcherEvent { - SkillsChanged { paths: Vec }, -} - -pub(crate) struct SkillsWatcher { - subscriber: FileWatcherSubscriber, - tx: broadcast::Sender, -} +pub(crate) struct SkillsWatcher; impl SkillsWatcher { - pub(crate) fn new(file_watcher: &Arc) -> Self { - let (subscriber, rx) = file_watcher.add_subscriber(); - let (tx, _) = broadcast::channel(128); - let skills_watcher = Self { - subscriber, - tx: tx.clone(), - }; - Self::spawn_event_loop(rx, tx); - skills_watcher - } - pub(crate) fn noop() -> Self { - Self::new(&Arc::new(FileWatcher::noop())) - } - - pub(crate) fn subscribe(&self) -> broadcast::Receiver { - self.tx.subscribe() - } - - pub(crate) async fn register_config( - &self, - config: &Config, - skills_manager: &SkillsManager, - plugins_manager: &PluginsManager, - fs: Option>, - ) -> WatchRegistration { - let plugin_outcome = plugins_manager.plugins_for_config(config).await; - let effective_skill_roots = plugin_outcome.effective_skill_roots(); - let skills_input = skills_load_input_from_config(config, effective_skill_roots); - let roots = skills_manager - .skill_roots_for_config(&skills_input, fs) - .await - .into_iter() - .map(|root| WatchPath { - path: root.path.into_path_buf(), - recursive: true, - }) - .collect(); - self.subscriber.register_paths(roots) - } - - fn spawn_event_loop(rx: Receiver, tx: broadcast::Sender) { - let mut rx = ThrottledWatchReceiver::new(rx, WATCHER_THROTTLE_INTERVAL); - if let Ok(handle) = Handle::try_current() { - handle.spawn(async move { - while let Some(event) = rx.recv().await { - let _ = tx.send(SkillsWatcherEvent::SkillsChanged { paths: event.paths }); - } - }); - } else { - warn!("skills watcher listener skipped: no Tokio runtime available"); - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - use pretty_assertions::assert_eq; - use tokio::time::Duration; - use tokio::time::timeout; - - #[tokio::test] - async fn forwards_file_watcher_events() { - let file_watcher = Arc::new(FileWatcher::noop()); - let skills_watcher = SkillsWatcher::new(&file_watcher); - let mut rx = skills_watcher.subscribe(); - let _registration = skills_watcher - .subscriber - .register_path(PathBuf::from("/tmp/skill"), /*recursive*/ true); - - file_watcher - .send_paths_for_test(vec![PathBuf::from("/tmp/skill/SKILL.md")]) - .await; - - let event = timeout(Duration::from_secs(2), rx.recv()) - .await - .expect("skills watcher event") - .expect("broadcast recv"); - assert_eq!( - event, - SkillsWatcherEvent::SkillsChanged { - paths: vec![PathBuf::from("/tmp/skill/SKILL.md")], - } - ); + Self } } diff --git a/vendor/codex/codex-rs/core/src/thread_manager.rs b/vendor/codex/codex-rs/core/src/thread_manager.rs index 4b8e9cc5..87cfbe29 100644 --- a/vendor/codex/codex-rs/core/src/thread_manager.rs +++ b/vendor/codex/codex-rs/core/src/thread_manager.rs @@ -2,7 +2,6 @@ use crate::SkillsManager; use crate::agent::AgentControl; use crate::codex_thread::CodexThread; use crate::config::Config; -use crate::file_watcher::FileWatcher; use crate::mcp::McpManager; use crate::plugins::PluginsManager; use crate::rollout::RolloutRecorder; @@ -13,7 +12,6 @@ use crate::session::CodexSpawnOk; use crate::session::INITIAL_SUBMIT_ID; use crate::shell_snapshot::ShellSnapshot; use crate::skills_watcher::SkillsWatcher; -use crate::skills_watcher::SkillsWatcherEvent; use crate::tasks::interrupted_turn_history_marker; use codex_analytics::AnalyticsEventsClient; use codex_app_server_protocol::ThreadHistoryBuilder; @@ -55,8 +53,6 @@ use std::sync::Arc; use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; use std::time::Duration; -use tokio::runtime::Handle; -use tokio::runtime::RuntimeFlavor; use tokio::sync::RwLock; use tokio::sync::broadcast; use tracing::warn; @@ -90,45 +86,8 @@ impl Drop for TempCodexHomeGuard { } } -fn build_skills_watcher(skills_manager: Arc) -> Arc { - if should_use_test_thread_manager_behavior() - && let Ok(handle) = Handle::try_current() - && handle.runtime_flavor() == RuntimeFlavor::CurrentThread - { - // The real watcher spins background tasks that can starve the - // current-thread test runtime and cause event waits to time out. - warn!("using noop skills watcher under current-thread test runtime"); - return Arc::new(SkillsWatcher::noop()); - } - - let file_watcher = match FileWatcher::new() { - Ok(file_watcher) => Arc::new(file_watcher), - Err(err) => { - warn!("failed to initialize file watcher: {err}"); - Arc::new(FileWatcher::noop()) - } - }; - let skills_watcher = Arc::new(SkillsWatcher::new(&file_watcher)); - - let mut rx = skills_watcher.subscribe(); - let skills_manager = Arc::clone(&skills_manager); - if let Ok(handle) = Handle::try_current() { - handle.spawn(async move { - loop { - match rx.recv().await { - Ok(SkillsWatcherEvent::SkillsChanged { .. }) => { - skills_manager.clear_cache(); - } - Err(broadcast::error::RecvError::Closed) => break, - Err(broadcast::error::RecvError::Lagged(_)) => continue, - } - } - }); - } else { - warn!("skills watcher listener skipped: no Tokio runtime available"); - } - - skills_watcher +fn build_skills_watcher(_skills_manager: Arc) -> Arc { + Arc::new(SkillsWatcher::noop()) } /// Represents a newly created Codex thread (formerly called a conversation), including the first event @@ -920,20 +879,7 @@ impl ThreadManagerState { parent_trace: Option, user_shell_override: Option, ) -> CodexResult { - let environment = self.environment_manager.default_environment(); - let watch_registration = match environment.as_ref() { - Some(environment) if !environment.is_remote() => { - self.skills_watcher - .register_config( - &config, - self.skills_manager.as_ref(), - self.plugins_manager.as_ref(), - Some(environment.get_filesystem()), - ) - .await - } - Some(_) | None => crate::file_watcher::WatchRegistration::default(), - }; + let watch_registration = crate::file_watcher::WatchRegistration::default(); let CodexSpawnOk { codex, thread_id, .. } = Codex::spawn(CodexSpawnArgs {