diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 75d72d360f..adf4f724f8 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -1388,6 +1388,7 @@ dependencies = [ "libc", "maplit", "mcp-types", + "notify", "once_cell", "openssl-sys", "os_info", diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index e32b060902..dac093c5c8 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -601,6 +601,7 @@ server_notification_definitions! { ContextCompacted => "thread/compacted" (v2::ContextCompactedNotification), DeprecationNotice => "deprecationNotice" (v2::DeprecationNoticeNotification), ConfigWarning => "configWarning" (v2::ConfigWarningNotification), + SkillsListUpdated => "skills/list/updated" (v2::SkillsListUpdatedNotification), /// Notifies the user of world-writable directories on Windows, which cannot be protected by the sandbox. WindowsWorldWritableWarning => "windows/worldWritableWarning" (v2::WindowsWorldWritableWarningNotification), diff --git a/codex-rs/app-server-protocol/src/protocol/v2.rs b/codex-rs/app-server-protocol/src/protocol/v2.rs index e4b3b90589..8b5d55bc91 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2.rs @@ -2367,6 +2367,11 @@ pub struct ContextCompactedNotification { pub turn_id: String, } +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct SkillsListUpdatedNotification {} + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 95d62d3f90..aebfcff355 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -1,5 +1,7 @@ use std::path::PathBuf; use std::sync::Arc; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::Ordering; use crate::codex_message_processor::CodexMessageProcessor; use crate::config_api::ConfigApi; @@ -19,7 +21,9 @@ use codex_app_server_protocol::JSONRPCRequest; use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ServerNotification; +use codex_app_server_protocol::SkillsListUpdatedNotification; use codex_core::AuthManager; +use codex_core::FileWatcherEvent; use codex_core::ThreadManager; use codex_core::config::Config; use codex_core::config_loader::LoaderOverrides; @@ -38,6 +42,7 @@ pub(crate) struct MessageProcessor { codex_message_processor: CodexMessageProcessor, config_api: ConfigApi, initialized: bool, + initialized_flag: Arc, config_warnings: Vec, } @@ -64,6 +69,31 @@ impl MessageProcessor { auth_manager.clone(), SessionSource::VSCode, )); + // Watch for on-disk skill changes and reinject the updated skills into + // subsequent requests. + let initialized_flag = Arc::new(AtomicBool::new(false)); + let mut skills_updates_rx = thread_manager.subscribe_file_watcher(); + let outgoing_for_skills = Arc::clone(&outgoing); + let initialized_for_skills = Arc::clone(&initialized_flag); + tokio::spawn(async move { + loop { + match skills_updates_rx.recv().await { + Ok(FileWatcherEvent::SkillsChanged { .. }) => { + if !initialized_for_skills.load(Ordering::SeqCst) { + continue; + } + outgoing_for_skills + .send_server_notification(ServerNotification::SkillsListUpdated( + SkillsListUpdatedNotification {}, + )) + .await; + } + Ok(FileWatcherEvent::AgentsChanged { .. }) => {} + Err(broadcast::error::RecvError::Closed) => break, + Err(broadcast::error::RecvError::Lagged(_)) => continue, + } + } + }); let codex_message_processor = CodexMessageProcessor::new( auth_manager, thread_manager, @@ -80,6 +110,7 @@ impl MessageProcessor { codex_message_processor, config_api, initialized: false, + initialized_flag, config_warnings, } } @@ -161,6 +192,7 @@ impl MessageProcessor { self.outgoing.send_response(request_id, response).await; self.initialized = true; + self.initialized_flag.store(true, Ordering::SeqCst); if !self.config_warnings.is_empty() { for notification in self.config_warnings.drain(..) { self.outgoing diff --git a/codex-rs/core/Cargo.toml b/codex-rs/core/Cargo.toml index c151356e3d..d35c284a33 100644 --- a/codex-rs/core/Cargo.toml +++ b/codex-rs/core/Cargo.toml @@ -55,6 +55,7 @@ indoc = { workspace = true } keyring = { workspace = true, features = ["crypto-rust"] } libc = { workspace = true } mcp-types = { workspace = true } +notify = { workspace = true } once_cell = { workspace = true } os_info = { workspace = true } rand = { workspace = true } diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 41e148e113..dd6ae29a78 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -110,6 +110,8 @@ use crate::error::Result as CodexResult; use crate::exec::StreamOutput; use crate::exec_policy::ExecPolicyUpdateError; use crate::feedback_tags; +use crate::file_watcher::FileWatcher; +use crate::file_watcher::FileWatcherEvent; use crate::instructions::UserInstructions; use crate::mcp::CODEX_APPS_MCP_SERVER_NAME; use crate::mcp::auth::compute_auth_statuses; @@ -247,6 +249,7 @@ impl Codex { auth_manager: Arc, models_manager: Arc, skills_manager: Arc, + file_watcher: Arc, conversation_history: InitialHistory, session_source: SessionSource, agent_control: AgentControl, @@ -296,7 +299,7 @@ impl Codex { // Resolve base instructions for the session. Priority order: // 1. config.base_instructions override // 2. conversation history => session_meta.base_instructions - // 3. base_intructions for current model + // 3. base_instructions for current model let model_info = models_manager.get_model_info(model.as_str(), &config).await; let base_instructions = config .base_instructions @@ -346,6 +349,7 @@ impl Codex { conversation_history, session_source_clone, skills_manager, + file_watcher, agent_control, ) .await @@ -427,6 +431,8 @@ pub(crate) struct Session { pending_mcp_server_refresh_config: Mutex>, pub(crate) active_turn: Mutex>, pub(crate) services: SessionServices, + agents_changed: Arc, + agents_watch_dirs: Vec, next_internal_sub_id: AtomicU64, } @@ -576,6 +582,54 @@ impl Session { per_turn_config } + // Build the directories we watch for `AGENTS.md` changes: project search + // roots (falling back to `cwd`) plus the user's `codex_home`. + fn build_agents_watch_dirs(config: &Config) -> Vec { + let mut dirs = match crate::project_doc::project_doc_search_dirs(config) { + Ok(dirs) => dirs, + Err(err) => { + warn!("failed to compute AGENTS.md search dirs: {err}"); + vec![config.cwd.clone()] + } + }; + dirs.push(config.codex_home.clone()); + dirs + } + + fn start_file_watcher_listener(self: &Arc) { + let mut rx = self.services.file_watcher.subscribe(); + let agents_changed = Arc::clone(&self.agents_changed); + let agents_watch_dirs = self.agents_watch_dirs.clone(); + let sess = Arc::clone(self); + tokio::spawn(async move { + loop { + match rx.recv().await { + Ok(FileWatcherEvent::AgentsChanged { paths }) => { + if paths + .iter() + .any(|path| agents_watch_dirs.iter().any(|root| path.starts_with(root))) + && !agents_changed.swap(true, Ordering::SeqCst) + { + info!( + "AGENTS change detected; will refresh instructions next turn: {:?}", + paths + ); + } + } + Ok(FileWatcherEvent::SkillsChanged { .. }) => { + let event = Event { + id: sess.next_internal_sub_id_with_prefix("skills-update"), + 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, + } + } + }); + } + #[allow(clippy::too_many_arguments)] fn make_turn_context( auth_manager: Option>, @@ -643,6 +697,7 @@ impl Session { initial_history: InitialHistory, session_source: SessionSource, skills_manager: Arc, + file_watcher: Arc, agent_control: AgentControl, ) -> anyhow::Result> { debug!( @@ -806,6 +861,7 @@ impl Session { ); } let state = SessionState::new(session_configuration.clone()); + let agents_watch_dirs = Self::build_agents_watch_dirs(&config); let services = SessionServices { mcp_connection_manager: Arc::new(RwLock::new(McpConnectionManager::default())), @@ -821,6 +877,7 @@ impl Session { models_manager: Arc::clone(&models_manager), tool_approvals: Mutex::new(ApprovalStore::default()), skills_manager, + file_watcher, agent_control, }; @@ -833,9 +890,13 @@ impl Session { pending_mcp_server_refresh_config: Mutex::new(None), active_turn: Mutex::new(None), services, + agents_changed: Arc::new(AtomicBool::new(false)), + agents_watch_dirs, next_internal_sub_id: AtomicU64::new(0), }); + sess.start_file_watcher_listener(); + // Dispatch the SessionConfiguredEvent first and then report any errors. // If resuming, include converted initial messages in the payload so UIs can render them immediately. let initial_messages = initial_history.get_event_msgs(); @@ -908,10 +969,14 @@ impl Session { } fn next_internal_sub_id(&self) -> String { + self.next_internal_sub_id_with_prefix("auto-compact") + } + + fn next_internal_sub_id_with_prefix(&self, prefix: &str) -> String { let id = self .next_internal_sub_id .fetch_add(1, std::sync::atomic::Ordering::SeqCst); - format!("auto-compact-{id}") + format!("{prefix}-{id}") } async fn get_total_token_usage(&self) -> i64 { @@ -1665,6 +1730,32 @@ impl Session { state.session_configuration.collaboration_mode.clone() } + // If `AGENTS.md` changed, reload skills, recompute user instructions, and + // update session state; otherwise return `None`. + pub(crate) async fn refresh_user_instructions_if_needed(&self) -> Option { + if !self.agents_changed.swap(false, Ordering::SeqCst) { + return None; + } + + let config = { + let state = self.state.lock().await; + Arc::clone(&state.session_configuration.original_config_do_not_use) + }; + let skills_outcome = self.services.skills_manager.skills_for_config(&config); + for err in &skills_outcome.errors { + error!( + "failed to load skill {}: {}", + err.path.display(), + err.message + ); + } + let enabled_skills = skills_outcome.enabled_skills(); + let user_instructions = get_user_instructions(&config, Some(&enabled_skills)).await; + let mut state = self.state.lock().await; + state.session_configuration.user_instructions = user_instructions.clone(); + user_instructions + } + async fn send_raw_response_items(&self, turn_context: &TurnContext, items: &[ResponseItem]) { for item in items { self.send_event( @@ -2259,6 +2350,7 @@ mod handlers { use crate::codex::spawn_review_thread; use crate::config::Config; + use crate::instructions::UserInstructions; use crate::mcp::auth::compute_auth_statuses; use crate::mcp::collect_mcp_snapshot_from_manager; @@ -2408,6 +2500,7 @@ mod handlers { .collaboration_mode .clone(); let next_collaboration_mode = updates.collaboration_mode.clone(); + let refreshed_user_instructions = sess.refresh_user_instructions_if_needed().await; let Ok(current_context) = sess.new_turn_with_sub_id(sub_id, updates).await else { // new_turn_with_sub_id already emits the error event. return; @@ -2419,12 +2512,21 @@ mod handlers { // Attempt to inject input into current task if let Err(items) = sess.inject_input(items).await { - let update_items = sess.build_settings_update_items( + let mut update_items = sess.build_settings_update_items( previous_context.as_ref(), ¤t_context, &previous_collaboration_mode, next_collaboration_mode.as_ref(), ); + if let Some(user_instructions) = refreshed_user_instructions { + update_items.push( + UserInstructions { + text: user_instructions, + directory: current_context.cwd.to_string_lossy().into_owned(), + } + .into(), + ); + } if !update_items.is_empty() { sess.record_conversation_items(¤t_context, &update_items) .await; @@ -4350,6 +4452,7 @@ mod tests { let state = SessionState::new(session_configuration.clone()); let skills_manager = Arc::new(SkillsManager::new(config.codex_home.clone())); + let file_watcher = Arc::new(FileWatcher::noop()); let services = SessionServices { mcp_connection_manager: Arc::new(RwLock::new(McpConnectionManager::default())), mcp_startup_cancellation_token: Mutex::new(CancellationToken::new()), @@ -4364,6 +4467,7 @@ mod tests { models_manager: Arc::clone(&models_manager), tool_approvals: Mutex::new(ApprovalStore::default()), skills_manager, + file_watcher, agent_control, }; @@ -4387,6 +4491,8 @@ mod tests { pending_mcp_server_refresh_config: Mutex::new(None), active_turn: Mutex::new(None), services, + agents_changed: Arc::new(AtomicBool::new(false)), + agents_watch_dirs: Session::build_agents_watch_dirs(&config), next_internal_sub_id: AtomicU64::new(0), }; @@ -4459,6 +4565,7 @@ mod tests { let state = SessionState::new(session_configuration.clone()); let skills_manager = Arc::new(SkillsManager::new(config.codex_home.clone())); + let file_watcher = Arc::new(FileWatcher::noop()); let services = SessionServices { mcp_connection_manager: Arc::new(RwLock::new(McpConnectionManager::default())), mcp_startup_cancellation_token: Mutex::new(CancellationToken::new()), @@ -4473,6 +4580,7 @@ mod tests { models_manager: Arc::clone(&models_manager), tool_approvals: Mutex::new(ApprovalStore::default()), skills_manager, + file_watcher, agent_control, }; @@ -4496,6 +4604,8 @@ mod tests { pending_mcp_server_refresh_config: Mutex::new(None), active_turn: Mutex::new(None), services, + agents_changed: Arc::new(AtomicBool::new(false)), + agents_watch_dirs: Session::build_agents_watch_dirs(&config), next_internal_sub_id: AtomicU64::new(0), }); @@ -4636,7 +4746,7 @@ mod tests { } #[tokio::test] - async fn abort_gracefuly_emits_turn_aborted_only() { + async fn abort_gracefully_emits_turn_aborted_only() { let (sess, tc, rx) = make_session_and_context_with_rx().await; let input = vec![UserInput::Text { text: "hello".to_string(), diff --git a/codex-rs/core/src/codex_delegate.rs b/codex-rs/core/src/codex_delegate.rs index 5e1de2ef1f..16d3cf0dac 100644 --- a/codex-rs/core/src/codex_delegate.rs +++ b/codex-rs/core/src/codex_delegate.rs @@ -54,6 +54,7 @@ pub(crate) async fn run_codex_thread_interactive( auth_manager, models_manager, Arc::clone(&parent_session.services.skills_manager), + Arc::clone(&parent_session.services.file_watcher), initial_history.unwrap_or(InitialHistory::New), SessionSource::SubAgent(SubAgentSource::Review), parent_session.services.agent_control.clone(), diff --git a/codex-rs/core/src/file_watcher.rs b/codex-rs/core/src/file_watcher.rs new file mode 100644 index 0000000000..59cf0c3daa --- /dev/null +++ b/codex-rs/core/src/file_watcher.rs @@ -0,0 +1,224 @@ +//! Watches AGENTS and skill roots for changes and broadcasts coarse-grained +//! `FileWatcherEvent`s that higher-level components react to on the next turn. + +use std::collections::HashMap; +use std::collections::HashSet; +use std::path::Path; +use std::path::PathBuf; +use std::sync::Arc; +use std::sync::Mutex; +use std::sync::RwLock; + +use notify::Event; +use notify::RecommendedWatcher; +use notify::RecursiveMode; +use notify::Watcher; +use tokio::runtime::Handle; +use tokio::sync::broadcast; +use tokio::sync::mpsc; +use tracing::warn; + +use crate::config::Config; +use crate::project_doc::DEFAULT_PROJECT_DOC_FILENAME; +use crate::project_doc::LOCAL_PROJECT_DOC_FILENAME; +use crate::project_doc::project_doc_search_dirs; +use crate::skills::loader::skill_roots_from_layer_stack; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum FileWatcherEvent { + AgentsChanged { paths: Vec }, + SkillsChanged { paths: Vec }, +} + +struct WatchState { + skills_roots: HashSet, +} + +struct FileWatcherInner { + watcher: RecommendedWatcher, + watched_paths: HashMap, +} + +pub(crate) struct FileWatcher { + inner: Option>, + state: Arc>, + tx: broadcast::Sender, +} + +impl FileWatcher { + pub(crate) fn new(codex_home: PathBuf) -> notify::Result { + let (raw_tx, raw_rx) = mpsc::unbounded_channel(); + let raw_tx_clone = raw_tx; + let watcher = notify::recommended_watcher(move |res| { + let _ = raw_tx_clone.send(res); + })?; + let inner = FileWatcherInner { + watcher, + watched_paths: HashMap::new(), + }; + let (tx, _) = broadcast::channel(128); + let state = Arc::new(RwLock::new(WatchState { + skills_roots: HashSet::new(), + })); + let file_watcher = Self { + inner: Some(Mutex::new(inner)), + state: Arc::clone(&state), + tx: tx.clone(), + }; + file_watcher.spawn_event_loop(raw_rx, state, tx); + file_watcher.watch_agents_root(codex_home.clone()); + file_watcher.register_skills_root(codex_home.join("skills")); + Ok(file_watcher) + } + + pub(crate) fn noop() -> Self { + let (tx, _) = broadcast::channel(1); + Self { + inner: None, + state: Arc::new(RwLock::new(WatchState { + skills_roots: HashSet::new(), + })), + tx, + } + } + + pub(crate) fn subscribe(&self) -> broadcast::Receiver { + self.tx.subscribe() + } + + pub(crate) fn register_config(&self, config: &Config) { + self.watch_agents_root(config.codex_home.clone()); + + match project_doc_search_dirs(config) { + Ok(dirs) => { + for dir in dirs { + self.watch_path(dir, RecursiveMode::NonRecursive); + } + } + Err(err) => { + warn!("failed to determine AGENTS.md search dirs: {err}"); + } + } + + let roots = skill_roots_from_layer_stack(&config.config_layer_stack); + for root in roots { + self.register_skills_root(root.path); + } + } + + fn spawn_event_loop( + &self, + mut raw_rx: mpsc::UnboundedReceiver>, + state: Arc>, + tx: broadcast::Sender, + ) { + if let Ok(handle) = Handle::try_current() { + handle.spawn(async move { + while let Some(res) = raw_rx.recv().await { + match res { + Ok(event) => { + let (agents_paths, skills_paths) = classify_event(&event, &state); + if !agents_paths.is_empty() { + let _ = tx.send(FileWatcherEvent::AgentsChanged { + paths: agents_paths, + }); + } + if !skills_paths.is_empty() { + let _ = tx.send(FileWatcherEvent::SkillsChanged { + paths: skills_paths, + }); + } + } + Err(err) => { + warn!("file watcher error: {err}"); + } + } + } + }); + } else { + warn!("file watcher loop skipped: no Tokio runtime available"); + } + } + + fn watch_agents_root(&self, root: PathBuf) { + self.watch_path(root, RecursiveMode::NonRecursive); + } + + fn register_skills_root(&self, root: PathBuf) { + { + let mut state = match self.state.write() { + Ok(state) => state, + Err(err) => err.into_inner(), + }; + state.skills_roots.insert(root.clone()); + } + self.watch_path(root, RecursiveMode::Recursive); + } + + fn watch_path(&self, path: PathBuf, mode: RecursiveMode) { + let Some(inner) = &self.inner else { + return; + }; + let Some(watch_path) = nearest_existing_ancestor(&path) else { + return; + }; + let mut guard = match inner.lock() { + Ok(guard) => guard, + Err(err) => err.into_inner(), + }; + if let Some(existing) = guard.watched_paths.get(&watch_path) { + if *existing == RecursiveMode::Recursive || *existing == mode { + return; + } + if let Err(err) = guard.watcher.unwatch(&watch_path) { + warn!("failed to unwatch {}: {err}", watch_path.display()); + } + } + if let Err(err) = guard.watcher.watch(&watch_path, mode) { + warn!("failed to watch {}: {err}", watch_path.display()); + return; + } + guard.watched_paths.insert(watch_path, mode); + } +} + +fn classify_event(event: &Event, state: &RwLock) -> (Vec, Vec) { + let mut agents_paths = Vec::new(); + let mut skills_paths = Vec::new(); + let skills_roots = match state.read() { + Ok(state) => state.skills_roots.clone(), + Err(err) => err.into_inner().skills_roots.clone(), + }; + + for path in &event.paths { + if is_agents_path(path) { + agents_paths.push(path.clone()); + } + if is_skills_path(path, &skills_roots) { + skills_paths.push(path.clone()); + } + } + + (agents_paths, skills_paths) +} + +fn is_agents_path(path: &Path) -> bool { + let Some(name) = path.file_name().and_then(|name| name.to_str()) else { + return false; + }; + name == DEFAULT_PROJECT_DOC_FILENAME || name == LOCAL_PROJECT_DOC_FILENAME +} + +fn is_skills_path(path: &Path, roots: &HashSet) -> bool { + roots.iter().any(|root| path.starts_with(root)) +} + +fn nearest_existing_ancestor(path: &Path) -> Option { + let mut cursor = path; + loop { + if cursor.exists() { + return Some(cursor.to_path_buf()); + } + cursor = cursor.parent()?; + } +} diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 010ba3874a..53b5b7aa80 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -31,6 +31,7 @@ pub mod exec; pub mod exec_env; mod exec_policy; pub mod features; +mod file_watcher; mod flags; pub mod git_info; pub mod instructions; @@ -125,6 +126,7 @@ pub use command_safety::is_safe_command; pub use exec_policy::ExecPolicyError; pub use exec_policy::check_execpolicy_for_warnings; pub use exec_policy::load_exec_policy; +pub use file_watcher::FileWatcherEvent; pub use safety::get_platform_sandbox; pub use safety::is_windows_elevated_sandbox_enabled; pub use safety::set_windows_elevated_sandbox_enabled; diff --git a/codex-rs/core/src/project_doc.rs b/codex-rs/core/src/project_doc.rs index c763755af5..588ced63ad 100644 --- a/codex-rs/core/src/project_doc.rs +++ b/codex-rs/core/src/project_doc.rs @@ -148,6 +148,31 @@ pub async fn read_project_docs(config: &Config) -> std::io::Result std::io::Result> { + let search_dirs = project_doc_search_dirs(config)?; + let mut found: Vec = Vec::new(); + let candidate_filenames = candidate_filenames(config); + for d in search_dirs { + for name in &candidate_filenames { + let candidate = d.join(name); + match std::fs::symlink_metadata(&candidate) { + Ok(md) => { + let ft = md.file_type(); + // Allow regular files and symlinks; opening will later fail for dangling links. + if ft.is_file() || ft.is_symlink() { + found.push(candidate); + break; + } + } + Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue, + Err(e) => return Err(e), + } + } + } + + Ok(found) +} + +pub(crate) fn project_doc_search_dirs(config: &Config) -> std::io::Result> { let mut dir = config.cwd.clone(); if let Ok(canon) = normalize_path(&dir) { dir = canon; @@ -192,27 +217,7 @@ pub fn discover_project_doc_paths(config: &Config) -> std::io::Result = Vec::new(); - let candidate_filenames = candidate_filenames(config); - for d in search_dirs { - for name in &candidate_filenames { - let candidate = d.join(name); - match std::fs::symlink_metadata(&candidate) { - Ok(md) => { - let ft = md.file_type(); - // Allow regular files and symlinks; opening will later fail for dangling links. - if ft.is_file() || ft.is_symlink() { - found.push(candidate); - break; - } - } - Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue, - Err(e) => return Err(e), - } - } - } - - Ok(found) + Ok(search_dirs) } fn candidate_filenames<'a>(config: &'a Config) -> Vec<&'a str> { diff --git a/codex-rs/core/src/skills/manager.rs b/codex-rs/core/src/skills/manager.rs index 8aa6986882..ca3ec584b7 100644 --- a/codex-rs/core/src/skills/manager.rs +++ b/codex-rs/core/src/skills/manager.rs @@ -6,6 +6,7 @@ use std::sync::RwLock; use codex_utils_absolute_path::AbsolutePathBuf; use toml::Value as TomlValue; +use tracing::info; use tracing::warn; use crate::config::Config; @@ -119,8 +120,17 @@ impl SkillsManager { pub fn clear_cache(&self) { match self.cache_by_cwd.write() { - Ok(mut cache) => cache.clear(), - Err(err) => err.into_inner().clear(), + Ok(mut cache) => { + let cleared = cache.len(); + cache.clear(); + info!("skills cache cleared ({} entries)", cleared); + } + Err(err) => { + let mut cache = err.into_inner(); + let cleared = cache.len(); + cache.clear(); + info!("skills cache cleared ({} entries)", cleared); + } } } } diff --git a/codex-rs/core/src/state/service.rs b/codex-rs/core/src/state/service.rs index cd1f1c0498..0e19727ee2 100644 --- a/codex-rs/core/src/state/service.rs +++ b/codex-rs/core/src/state/service.rs @@ -4,6 +4,7 @@ use crate::AuthManager; use crate::RolloutRecorder; use crate::agent::AgentControl; use crate::exec_policy::ExecPolicyManager; +use crate::file_watcher::FileWatcher; use crate::mcp_connection_manager::McpConnectionManager; use crate::models_manager::manager::ModelsManager; use crate::skills::SkillsManager; @@ -29,5 +30,6 @@ pub(crate) struct SessionServices { pub(crate) otel_manager: OtelManager, pub(crate) tool_approvals: Mutex, pub(crate) skills_manager: Arc, + pub(crate) file_watcher: Arc, pub(crate) agent_control: AgentControl, } diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 9b633ded37..01d93210df 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -11,6 +11,8 @@ use crate::codex_thread::CodexThread; use crate::config::Config; use crate::error::CodexErr; use crate::error::Result as CodexResult; +use crate::file_watcher::FileWatcher; +use crate::file_watcher::FileWatcherEvent; use crate::models_manager::manager::ModelsManager; use crate::protocol::Event; use crate::protocol::EventMsg; @@ -31,12 +33,44 @@ use std::path::PathBuf; use std::sync::Arc; #[cfg(any(test, feature = "test-support"))] use tempfile::TempDir; +use tokio::runtime::Handle; use tokio::sync::RwLock; use tokio::sync::broadcast; use tracing::warn; const THREAD_CREATED_CHANNEL_CAPACITY: usize = 1024; +fn build_file_watcher(codex_home: PathBuf, skills_manager: Arc) -> Arc { + let file_watcher = match FileWatcher::new(codex_home) { + Ok(file_watcher) => Arc::new(file_watcher), + Err(err) => { + warn!("failed to initialize file watcher: {err}"); + Arc::new(FileWatcher::noop()) + } + }; + + let mut rx = file_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(FileWatcherEvent::SkillsChanged { .. }) => { + skills_manager.clear_cache(); + } + Ok(FileWatcherEvent::AgentsChanged { .. }) => {} + Err(broadcast::error::RecvError::Closed) => break, + Err(broadcast::error::RecvError::Lagged(_)) => continue, + } + } + }); + } else { + warn!("file watcher listener skipped: no Tokio runtime available"); + } + + file_watcher +} + /// Represents a newly created Codex thread (formerly called a conversation), including the first event /// (which is [`EventMsg::SessionConfigured`]). pub struct NewThread { @@ -62,6 +96,7 @@ pub(crate) struct ThreadManagerState { auth_manager: Arc, models_manager: Arc, skills_manager: Arc, + file_watcher: Arc, session_source: SessionSource, #[cfg(any(test, feature = "test-support"))] #[allow(dead_code)] @@ -76,15 +111,15 @@ impl ThreadManager { session_source: SessionSource, ) -> Self { let (thread_created_tx, _) = broadcast::channel(THREAD_CREATED_CHANNEL_CAPACITY); + let skills_manager = Arc::new(SkillsManager::new(codex_home.clone())); + let file_watcher = build_file_watcher(codex_home.clone(), Arc::clone(&skills_manager)); Self { state: Arc::new(ThreadManagerState { threads: Arc::new(RwLock::new(HashMap::new())), thread_created_tx, - models_manager: Arc::new(ModelsManager::new( - codex_home.clone(), - auth_manager.clone(), - )), - skills_manager: Arc::new(SkillsManager::new(codex_home)), + models_manager: Arc::new(ModelsManager::new(codex_home, auth_manager.clone())), + skills_manager, + file_watcher, auth_manager, session_source, #[cfg(any(test, feature = "test-support"))] @@ -116,16 +151,19 @@ impl ThreadManager { ) -> Self { let auth_manager = AuthManager::from_auth_for_testing(auth); let (thread_created_tx, _) = broadcast::channel(THREAD_CREATED_CHANNEL_CAPACITY); + let skills_manager = Arc::new(SkillsManager::new(codex_home.clone())); + let file_watcher = build_file_watcher(codex_home.clone(), Arc::clone(&skills_manager)); Self { state: Arc::new(ThreadManagerState { threads: Arc::new(RwLock::new(HashMap::new())), thread_created_tx, models_manager: Arc::new(ModelsManager::with_provider( - codex_home.clone(), + codex_home, auth_manager.clone(), provider, )), - skills_manager: Arc::new(SkillsManager::new(codex_home)), + skills_manager, + file_watcher, auth_manager, session_source: SessionSource::Exec, #[cfg(any(test, feature = "test-support"))] @@ -143,6 +181,10 @@ impl ThreadManager { self.state.skills_manager.clone() } + pub fn subscribe_file_watcher(&self) -> broadcast::Receiver { + self.state.file_watcher.subscribe() + } + pub fn get_models_manager(&self) -> Arc { self.state.models_manager.clone() } @@ -380,6 +422,7 @@ impl ThreadManagerState { session_source: SessionSource, dynamic_tools: Vec, ) -> CodexResult { + self.file_watcher.register_config(&config); let CodexSpawnOk { codex, thread_id, .. } = Codex::spawn( @@ -387,6 +430,7 @@ impl ThreadManagerState { auth_manager, Arc::clone(&self.models_manager), Arc::clone(&self.skills_manager), + Arc::clone(&self.file_watcher), initial_history, session_source, agent_control,