Disable Codex skill watcher lifecycle

This commit is contained in:
Meta Vens 2026-05-13 13:07:30 +01:00
commit 4f000bd216
9 changed files with 11 additions and 216 deletions

View file

@ -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

Binary file not shown.

File diff suppressed because one or more lines are too long

View file

@ -121,7 +121,6 @@ pub(crate) use skills::injection;
pub(crate) use skills::manager;
pub(crate) use skills::maybe_emit_implicit_skill_invocation;
pub(crate) use skills::resolve_skill_dependencies_for_turn;
pub(crate) use skills::skills_load_input_from_config;
mod skills_watcher;
mod stream_events_utils;
pub mod test_support;

View file

@ -274,7 +274,6 @@ use crate::session_startup_prewarm::SessionStartupPrewarmHandle;
use crate::shell;
use crate::shell_snapshot::ShellSnapshot;
use crate::skills_watcher::SkillsWatcher;
use crate::skills_watcher::SkillsWatcherEvent;
use crate::state::ActiveTurn;
use crate::state::MailboxDeliveryPhase;
use crate::state::PendingRequestPermissions;
@ -937,29 +936,6 @@ impl Session {
self.out_of_band_elicitation_paused.send_replace(paused);
}
fn start_skills_watcher_listener(self: &Arc<Self>) {
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<Event> {
self.tx_event.clone()
}

View file

@ -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<String> = mcp_servers
.iter()
.filter(|(_, server)| server.enabled && server.required)

View file

@ -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<AbsolutePathBuf>,
) -> 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<Session>,
turn_context: &Arc<TurnContext>,

View file

@ -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<PathBuf> },
}
pub(crate) struct SkillsWatcher {
subscriber: FileWatcherSubscriber,
tx: broadcast::Sender<SkillsWatcherEvent>,
}
pub(crate) struct SkillsWatcher;
impl SkillsWatcher {
pub(crate) fn new(file_watcher: &Arc<FileWatcher>) -> 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<SkillsWatcherEvent> {
self.tx.subscribe()
}
pub(crate) async fn register_config(
&self,
config: &Config,
skills_manager: &SkillsManager,
plugins_manager: &PluginsManager,
fs: Option<Arc<dyn codex_exec_server::ExecutorFileSystem>>,
) -> 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<SkillsWatcherEvent>) {
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
}
}

View file

@ -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<SkillsManager>) -> Arc<SkillsWatcher> {
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<SkillsManager>) -> Arc<SkillsWatcher> {
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<W3cTraceContext>,
user_shell_override: Option<crate::shell::Shell>,
) -> CodexResult<NewThread> {
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 {