From 8fe083b5412d4453fc0b13be128f38142466ce01 Mon Sep 17 00:00:00 2001 From: Eric Traut Date: Tue, 27 Jan 2026 00:07:12 -0800 Subject: [PATCH] Added support for live updates to skills and AGENTS Add a centralized FileWatcher in codex-core (using notify) that watches: * AGENTS.md / AGENTS.override.md and project AGENTS search dirs * Skill roots from the config layer stack (recursive) Send `AgentsChanged` and `SkillsChanged` events when relevant file system changes are detected On `SkillsChanged`: * Invalidate the skills cache immediately in ThreadManager * Emit EventMsg::SkillsUpdateAvailable to active sessions * Broadcast a new app-server notification: skills/list/updated On `AgentsChanged`: * Set a per-session agents_changed flag scoped to relevant watch dirs * On the next user turn, recompute skills + user instructions and inject an updated UserInstructions item into the event stream Wire the watcher through ThreadManager -> Codex::spawn -> SessionServices, including sub-agent sessions. Refactor project_doc discovery to expose project_doc_search_dirs for shared watch-dir computation. Add SkillsListUpdatedNotification to the app-server protocol and gate broadcast until after initialize. Testing: I did a bunch of manual testing of both AGENTS and skills updates. They work surprisingly well. Skill addition and removal are both handled seamlessly in both the UI and in the behavior of the agent on the next turn. Changes to skill details were picked up and followed on the next turn. Additions and modifications to AGENTS.md were followed well. The only failure that I found was when I deleted a major block of instructions from AGENTS.md. In that case, the model still followed those instructions from the older AGENTS.md because it was still in the context window. --- codex-rs/Cargo.lock | 1 + .../src/protocol/common.rs | 1 + .../app-server-protocol/src/protocol/v2.rs | 5 + codex-rs/app-server/src/message_processor.rs | 32 +++ codex-rs/core/Cargo.toml | 1 + codex-rs/core/src/codex.rs | 118 ++++++++- codex-rs/core/src/codex_delegate.rs | 1 + codex-rs/core/src/file_watcher.rs | 224 ++++++++++++++++++ codex-rs/core/src/lib.rs | 2 + codex-rs/core/src/project_doc.rs | 47 ++-- codex-rs/core/src/skills/manager.rs | 14 +- codex-rs/core/src/state/service.rs | 2 + codex-rs/core/src/thread_manager.rs | 58 ++++- 13 files changed, 472 insertions(+), 34 deletions(-) create mode 100644 codex-rs/core/src/file_watcher.rs 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,