Reapply "Move skills watcher to app-server" (#21652)

## Why

PR #21460 reverted the earlier move of skills change watching from
`codex-core` into app-server. This reapplies that boundary change so
app-server owns client-facing `skills/changed` notifications and core no
longer carries the watcher.

## What

- Restore the app-server `SkillsWatcher` and register it from thread
listener setup.
- Remove the core-owned skills watcher and its core live-reload
integration surface.
- Restore app-server coverage for `skills/changed` notifications after a
watched skill file changes.

## Validation

- `cargo test -p codex-app-server --test all
suite::v2::skills_list::skills_changed_notification_is_emitted_after_skill_change
-- --exact --nocapture`
- `cargo test -p codex-core --lib --no-run`

codex-rs/app-server/src/bespoke_event_handling.rs

@@ -49,7 +49,6 @@ use codex_app_server_protocol::RawResponseItemCompletedNotification;
 use codex_app_server_protocol::RequestId;
 use codex_app_server_protocol::ServerNotification;
 use codex_app_server_protocol::ServerRequestPayload;
-use codex_app_server_protocol::SkillsChangedNotification;
 use codex_app_server_protocol::ThreadGoalUpdatedNotification;
 use codex_app_server_protocol::ThreadItem;
 use codex_app_server_protocol::ThreadRealtimeClosedNotification;
@@ -194,13 +193,6 @@ pub(crate) async fn apply_bespoke_event_handling(
             )
             .await;
         }
-        EventMsg::SkillsUpdateAvailable => {
-            outgoing
-                .send_server_notification(ServerNotification::SkillsChanged(
-                    SkillsChangedNotification {},
-                ))
-                .await;
-        }
         EventMsg::McpStartupUpdate(update) => {
             let (status, error) = match update.status {
                 codex_protocol::protocol::McpStartupStatus::Starting => {

codex-rs/app-server/src/lib.rs

@@ -94,6 +94,7 @@ mod outgoing_message;
 mod request_processors;
 mod request_serialization;
 mod server_request_error;
+mod skills_watcher;
 mod thread_state;
 mod thread_status;
 mod transport;

codex-rs/app-server/src/message_processor.rs

@@ -36,6 +36,7 @@ use crate::request_processors::WindowsSandboxRequestProcessor;
 use crate::request_serialization::QueuedInitializedRequest;
 use crate::request_serialization::RequestSerializationQueueKey;
 use crate::request_serialization::RequestSerializationQueues;
+use crate::skills_watcher::SkillsWatcher;
 use crate::thread_state::ConnectionCapabilities;
 use crate::thread_state::ThreadStateManager;
 use crate::transport::AppServerTransport;
@@ -314,6 +315,7 @@ impl MessageProcessor {
         thread_manager
             .plugins_manager()
             .set_analytics_events_client(analytics_events_client.clone());
+        let skills_watcher = SkillsWatcher::new(thread_manager.skills_manager(), outgoing.clone());
 
         let pending_thread_unloads = Arc::new(Mutex::new(HashSet::new()));
         let thread_watch_manager =
@@ -405,6 +407,7 @@ impl MessageProcessor {
             Arc::clone(&thread_list_state_permit),
             thread_goal_processor.clone(),
             state_db,
+            Arc::clone(&skills_watcher),
         );
         let turn_processor = TurnRequestProcessor::new(
             auth_manager.clone(),
@@ -418,6 +421,7 @@ impl MessageProcessor {
             thread_state_manager,
             thread_watch_manager,
             thread_list_state_permit,
+            Arc::clone(&skills_watcher),
         );
         if matches!(plugin_startup_tasks, crate::PluginStartupTasks::Start) {
             // Keep plugin startup warmups aligned at app-server startup.

codex-rs/app-server/src/request_processors.rs

@@ -11,6 +11,7 @@ use crate::outgoing_message::ConnectionRequestId;
 use crate::outgoing_message::OutgoingMessageSender;
 use crate::outgoing_message::RequestContext;
 use crate::outgoing_message::ThreadScopedOutgoingMessageSender;
+use crate::skills_watcher::SkillsWatcher;
 use crate::thread_status::ThreadWatchManager;
 use crate::thread_status::resolve_thread_status;
 use chrono::Duration as ChronoDuration;

codex-rs/app-server/src/request_processors/thread_lifecycle.rs

@@ -12,6 +12,7 @@ pub(super) struct ListenerTaskContext {
     pub(super) thread_list_state_permit: Arc<Semaphore>,
     pub(super) fallback_model_provider: String,
     pub(super) codex_home: PathBuf,
+    pub(super) skills_watcher: Arc<SkillsWatcher>,
 }
 
 struct UnloadingState {
@@ -226,12 +227,22 @@ pub(super) async fn ensure_listener_task_running(
             "thread {conversation_id} is closing; retry after the thread is closed"
         )));
     };
+    let config = conversation.config().await;
+    let environments = conversation.environment_selections().await;
+    let watch_registration = listener_task_context
+        .skills_watcher
+        .register_thread_config(
+            config.as_ref(),
+            listener_task_context.thread_manager.as_ref(),
+            &environments,
+        )
+        .await;
     let (mut listener_command_rx, listener_generation) = {
         let mut thread_state = thread_state.lock().await;
         if thread_state.listener_matches(&conversation) {
             return Ok(());
         }
-        thread_state.set_listener(cancel_tx, &conversation)
+        thread_state.set_listener(cancel_tx, &conversation, watch_registration)
     };
     let ListenerTaskContext {
         outgoing,
@@ -242,6 +253,7 @@ pub(super) async fn ensure_listener_task_running(
         thread_list_state_permit,
         fallback_model_provider,
         codex_home,
+        ..
     } = listener_task_context;
     let outgoing_for_task = Arc::clone(&outgoing);
     tokio::spawn(async move {

codex-rs/app-server/src/request_processors/thread_processor.rs

@@ -317,6 +317,7 @@ pub(crate) struct ThreadRequestProcessor {
     pub(super) thread_goal_processor: ThreadGoalRequestProcessor,
     pub(super) state_db: Option<StateDbHandle>,
     pub(super) background_tasks: TaskTracker,
+    pub(super) skills_watcher: Arc<SkillsWatcher>,
 }
 
 impl ThreadRequestProcessor {
@@ -335,6 +336,7 @@ impl ThreadRequestProcessor {
         thread_list_state_permit: Arc<Semaphore>,
         thread_goal_processor: ThreadGoalRequestProcessor,
         state_db: Option<StateDbHandle>,
+        skills_watcher: Arc<SkillsWatcher>,
     ) -> Self {
         Self {
             auth_manager,
@@ -351,6 +353,7 @@ impl ThreadRequestProcessor {
             thread_goal_processor,
             state_db,
             background_tasks: TaskTracker::new(),
+            skills_watcher,
         }
     }
 
@@ -752,6 +755,7 @@ impl ThreadRequestProcessor {
             thread_list_state_permit: self.thread_list_state_permit.clone(),
             fallback_model_provider: self.config.model_provider_id.clone(),
             codex_home: self.config.codex_home.to_path_buf(),
+            skills_watcher: Arc::clone(&self.skills_watcher),
         }
     }
 
@@ -849,6 +853,7 @@ impl ThreadRequestProcessor {
             thread_list_state_permit: self.thread_list_state_permit.clone(),
             fallback_model_provider: self.config.model_provider_id.clone(),
             codex_home: self.config.codex_home.to_path_buf(),
+            skills_watcher: Arc::clone(&self.skills_watcher),
         };
         let request_trace = request_context.request_trace();
         let config_manager = self.config_manager.clone();
@@ -1049,7 +1054,6 @@ impl ThreadRequestProcessor {
                 .collect()
         };
         let core_dynamic_tool_count = core_dynamic_tools.len();
-
         let NewThread {
             thread_id,
             thread,

codex-rs/app-server/src/request_processors/turn_processor.rs

@@ -13,6 +13,7 @@ pub(crate) struct TurnRequestProcessor {
     thread_state_manager: ThreadStateManager,
     thread_watch_manager: ThreadWatchManager,
     thread_list_state_permit: Arc<Semaphore>,
+    skills_watcher: Arc<SkillsWatcher>,
 }
 
 impl TurnRequestProcessor {
@@ -29,6 +30,7 @@ impl TurnRequestProcessor {
         thread_state_manager: ThreadStateManager,
         thread_watch_manager: ThreadWatchManager,
         thread_list_state_permit: Arc<Semaphore>,
+        skills_watcher: Arc<SkillsWatcher>,
     ) -> Self {
         Self {
             auth_manager,
@@ -42,6 +44,7 @@ impl TurnRequestProcessor {
             thread_state_manager,
             thread_watch_manager,
             thread_list_state_permit,
+            skills_watcher,
         }
     }
 
@@ -1087,6 +1090,7 @@ impl TurnRequestProcessor {
             thread_list_state_permit: self.thread_list_state_permit.clone(),
             fallback_model_provider: self.config.model_provider_id.clone(),
             codex_home: self.config.codex_home.to_path_buf(),
+            skills_watcher: Arc::clone(&self.skills_watcher),
         }
     }
 

codex-rs/app-server/src/skills_watcher.rs

@@ -0,0 +1,112 @@
+use std::sync::Arc;
+use std::time::Duration;
+
+use crate::outgoing_message::OutgoingMessageSender;
+use codex_app_server_protocol::ServerNotification;
+use codex_app_server_protocol::SkillsChangedNotification;
+use codex_core::ThreadManager;
+use codex_core::config::Config;
+use codex_core::file_watcher::FileWatcher;
+use codex_core::file_watcher::FileWatcherSubscriber;
+use codex_core::file_watcher::Receiver;
+use codex_core::file_watcher::ThrottledWatchReceiver;
+use codex_core::file_watcher::WatchPath;
+use codex_core::file_watcher::WatchRegistration;
+use codex_core::skills::SkillsLoadInput;
+use codex_core::skills::SkillsManager;
+use codex_protocol::protocol::TurnEnvironmentSelection;
+use tracing::warn;
+
+#[cfg(not(test))]
+const WATCHER_THROTTLE_INTERVAL: Duration = Duration::from_secs(10);
+#[cfg(test)]
+const WATCHER_THROTTLE_INTERVAL: Duration = Duration::from_millis(50);
+
+pub(crate) struct SkillsWatcher {
+    subscriber: FileWatcherSubscriber,
+}
+
+impl SkillsWatcher {
+    pub(crate) fn new(
+        skills_manager: Arc<SkillsManager>,
+        outgoing: Arc<OutgoingMessageSender>,
+    ) -> Arc<Self> {
+        let file_watcher = match FileWatcher::new() {
+            Ok(file_watcher) => Arc::new(file_watcher),
+            Err(err) => {
+                warn!("failed to initialize skills file watcher: {err}");
+                Arc::new(FileWatcher::noop())
+            }
+        };
+        let (subscriber, rx) = file_watcher.add_subscriber();
+        Self::spawn_event_loop(rx, skills_manager, outgoing);
+        Arc::new(Self { subscriber })
+    }
+
+    pub(crate) async fn register_thread_config(
+        &self,
+        config: &Config,
+        thread_manager: &ThreadManager,
+        environments: &[TurnEnvironmentSelection],
+    ) -> WatchRegistration {
+        let Some(environment_selection) = environments.first() else {
+            return WatchRegistration::default();
+        };
+        let Some(environment) = thread_manager
+            .environment_manager()
+            .get_environment(&environment_selection.environment_id)
+        else {
+            warn!(
+                "failed to register skills watcher for unknown environment `{}`",
+                environment_selection.environment_id
+            );
+            return WatchRegistration::default();
+        };
+        if environment.is_remote() {
+            return WatchRegistration::default();
+        }
+
+        let plugins_input = config.plugins_config_input();
+        let plugins_manager = thread_manager.plugins_manager();
+        let plugin_outcome = plugins_manager.plugins_for_config(&plugins_input).await;
+        let skills_input = SkillsLoadInput::new(
+            config.cwd.clone(),
+            plugin_outcome.effective_plugin_skill_roots(),
+            config.config_layer_stack.clone(),
+            config.bundled_skills_enabled(),
+        );
+        let roots = thread_manager
+            .skills_manager()
+            .skill_roots_for_config(&skills_input, Some(environment.get_filesystem()))
+            .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,
+        skills_manager: Arc<SkillsManager>,
+        outgoing: Arc<OutgoingMessageSender>,
+    ) {
+        let mut rx = ThrottledWatchReceiver::new(rx, WATCHER_THROTTLE_INTERVAL);
+        let Ok(handle) = tokio::runtime::Handle::try_current() else {
+            warn!("skills watcher listener skipped: no Tokio runtime available");
+            return;
+        };
+        handle.spawn(async move {
+            while rx.recv().await.is_some() {
+                skills_manager.clear_cache();
+                outgoing
+                    .send_server_notification(ServerNotification::SkillsChanged(
+                        SkillsChangedNotification {},
+                    ))
+                    .await;
+            }
+        });
+    }
+}

codex-rs/app-server/src/thread_state.rs

@@ -7,6 +7,7 @@ use codex_app_server_protocol::Turn;
 use codex_app_server_protocol::TurnError;
 use codex_core::CodexThread;
 use codex_core::ThreadConfigSnapshot;
+use codex_core::file_watcher::WatchRegistration;
 use codex_protocol::ThreadId;
 use codex_protocol::protocol::EventMsg;
 use codex_protocol::protocol::RolloutItem;
@@ -77,6 +78,7 @@ pub(crate) struct ThreadState {
     listener_command_tx: Option<mpsc::UnboundedSender<ThreadListenerCommand>>,
     current_turn_history: ThreadHistoryBuilder,
     listener_thread: Option<Weak<CodexThread>>,
+    watch_registration: WatchRegistration,
 }
 
 impl ThreadState {
@@ -91,6 +93,7 @@ impl ThreadState {
         &mut self,
         cancel_tx: oneshot::Sender<()>,
         conversation: &Arc<CodexThread>,
+        watch_registration: WatchRegistration,
     ) -> (mpsc::UnboundedReceiver<ThreadListenerCommand>, u64) {
         if let Some(previous) = self.cancel_tx.replace(cancel_tx) {
             let _ = previous.send(());
@@ -99,6 +102,7 @@ impl ThreadState {
         let (listener_command_tx, listener_command_rx) = mpsc::unbounded_channel();
         self.listener_command_tx = Some(listener_command_tx);
         self.listener_thread = Some(Arc::downgrade(conversation));
+        self.watch_registration = watch_registration;
         (listener_command_rx, self.listener_generation)
     }
 
@@ -109,6 +113,7 @@ impl ThreadState {
         self.listener_command_tx = None;
         self.current_turn_history.reset();
         self.listener_thread = None;
+        self.watch_registration = WatchRegistration::default();
     }
 
     pub(crate) fn set_experimental_raw_events(&mut self, enabled: bool) {

codex-rs/app-server/tests/suite/v2/skills_list.rs

@@ -4,8 +4,10 @@ use anyhow::Context;
 use anyhow::Result;
 use app_test_support::ChatGptAuthFixture;
 use app_test_support::McpProcess;
+use app_test_support::create_mock_responses_server_repeating_assistant;
 use app_test_support::to_response;
 use app_test_support::write_chatgpt_auth;
+use app_test_support::write_mock_responses_config_toml_with_chatgpt_base_url;
 use codex_app_server_protocol::JSONRPCResponse;
 use codex_app_server_protocol::PluginListParams;
 use codex_app_server_protocol::PluginListResponse;
@@ -573,11 +575,39 @@ async fn skills_list_uses_cached_result_until_force_reload() -> Result<()> {
 
 #[tokio::test]
 async fn skills_changed_notification_is_emitted_after_skill_change() -> Result<()> {
+    let server = create_mock_responses_server_repeating_assistant("Done").await;
     let codex_home = TempDir::new()?;
+    write_mock_responses_config_toml_with_chatgpt_base_url(
+        codex_home.path(),
+        &server.uri(),
+        &server.uri(),
+    )?;
     write_skill(&codex_home, "demo")?;
 
-    let mut mcp = McpProcess::new(codex_home.path()).await?;
+    let mut mcp =
+        McpProcess::new_with_env(codex_home.path(), &[(CODEX_EXEC_SERVER_URL_ENV_VAR, None)])
+            .await?;
     timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;
+    let initial_skills_request_id = mcp
+        .send_skills_list_request(SkillsListParams {
+            cwds: vec![codex_home.path().to_path_buf()],
+            force_reload: true,
+        })
+        .await?;
+    let initial_skills_response: JSONRPCResponse = timeout(
+        DEFAULT_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(initial_skills_request_id)),
+    )
+    .await??;
+    let SkillsListResponse { data } = to_response(initial_skills_response)?;
+    assert_eq!(data.len(), 1);
+    assert!(
+        data[0]
+            .skills
+            .iter()
+            .any(|skill| { skill.name == "demo" && skill.description == "demo description" })
+    );
+
     let thread_start_request_id = mcp
         .send_thread_start_request(ThreadStartParams {
             model: None,
@@ -630,5 +660,24 @@ async fn skills_changed_notification_is_emitted_after_skill_change() -> Result<(
     let notification: SkillsChangedNotification = serde_json::from_value(params)?;
 
     assert_eq!(notification, SkillsChangedNotification {});
+    let updated_skills_request_id = mcp
+        .send_skills_list_request(SkillsListParams {
+            cwds: vec![codex_home.path().to_path_buf()],
+            force_reload: false,
+        })
+        .await?;
+    let updated_skills_response: JSONRPCResponse = timeout(
+        DEFAULT_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(updated_skills_request_id)),
+    )
+    .await??;
+    let SkillsListResponse { data } = to_response(updated_skills_response)?;
+    assert_eq!(data.len(), 1);
+    assert!(
+        data[0]
+            .skills
+            .iter()
+            .any(|skill| skill.name == "demo" && skill.description == "updated")
+    );
     Ok(())
 }