From a2a0e0978b2ae4e9991d39613564c5632469ae93 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Wed, 8 Jul 2026 15:30:03 +0100 Subject: [PATCH] Notify clients when thread skills change --- .../schema/json/ServerNotification.json | 3 +- .../codex_app_server_protocol.schemas.json | 3 +- .../codex_app_server_protocol.v2.schemas.json | 3 +- .../json/v2/SkillsChangedNotification.json | 3 +- .../v2/SkillsChangedNotification.ts | 4 +- .../src/protocol/common.rs | 1 + .../src/protocol/v2/plugin.rs | 10 ++- codex-rs/app-server/README.md | 6 +- codex-rs/app-server/src/extensions.rs | 27 +++++- codex-rs/app-server/src/mcp_refresh.rs | 1 + codex-rs/app-server/src/message_processor.rs | 23 ++++- .../request_processors/catalog_processor.rs | 2 +- .../environment_processor.rs | 83 ++++++++++++++++++- codex-rs/app-server/src/skills_watcher.rs | 2 +- codex-rs/app-server/src/transport_tests.rs | 38 +++++++++ .../suite/v2/selected_capability_stack.rs | 81 ++++++++++++------ .../app-server/tests/suite/v2/skills_list.rs | 4 +- .../codex-experimental-api-macros/src/lib.rs | 33 ++++++-- codex-rs/core/src/codex_thread.rs | 6 ++ codex-rs/exec-server/src/client.rs | 62 ++++++++++---- codex-rs/exec-server/src/client_recovery.rs | 18 +++- codex-rs/exec-server/src/environment.rs | 32 ++++++- codex-rs/ext/skills/src/extension.rs | 21 ++++- codex-rs/ext/skills/src/lib.rs | 1 + codex-rs/ext/skills/src/state.rs | 17 ++++ .../tui/src/app/app_server_event_targets.rs | 4 +- codex-rs/tui/src/app/tests.rs | 2 +- codex-rs/tui/src/chatwidget/protocol.rs | 8 +- 28 files changed, 420 insertions(+), 78 deletions(-) diff --git a/codex-rs/app-server-protocol/schema/json/ServerNotification.json b/codex-rs/app-server-protocol/schema/json/ServerNotification.json index e025a61a33..7bbfaa2b39 100644 --- a/codex-rs/app-server-protocol/schema/json/ServerNotification.json +++ b/codex-rs/app-server-protocol/schema/json/ServerNotification.json @@ -3453,7 +3453,8 @@ "type": "object" }, "SkillsChangedNotification": { - "description": "Notification emitted when watched local skill files change.\n\nTreat this as an invalidation signal and re-run `skills/list` with the client's current parameters when refreshed skill metadata is needed.", + "description": "Notification emitted when the available skills catalog changes.\n\nTreat this as an invalidation signal and re-run `skills/list` with the client's current parameters when refreshed skill metadata is needed.", + "properties": {}, "type": "object" }, "SpendControlLimitSnapshot": { diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index ccac2bba93..c310032dd0 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -16758,7 +16758,8 @@ }, "SkillsChangedNotification": { "$schema": "http://json-schema.org/draft-07/schema#", - "description": "Notification emitted when watched local skill files change.\n\nTreat this as an invalidation signal and re-run `skills/list` with the client's current parameters when refreshed skill metadata is needed.", + "description": "Notification emitted when the available skills catalog changes.\n\nTreat this as an invalidation signal and re-run `skills/list` with the client's current parameters when refreshed skill metadata is needed.", + "properties": {}, "title": "SkillsChangedNotification", "type": "object" }, diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index 3e723c1cd7..8841f03a87 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -14537,7 +14537,8 @@ }, "SkillsChangedNotification": { "$schema": "http://json-schema.org/draft-07/schema#", - "description": "Notification emitted when watched local skill files change.\n\nTreat this as an invalidation signal and re-run `skills/list` with the client's current parameters when refreshed skill metadata is needed.", + "description": "Notification emitted when the available skills catalog changes.\n\nTreat this as an invalidation signal and re-run `skills/list` with the client's current parameters when refreshed skill metadata is needed.", + "properties": {}, "title": "SkillsChangedNotification", "type": "object" }, diff --git a/codex-rs/app-server-protocol/schema/json/v2/SkillsChangedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/SkillsChangedNotification.json index cb67d81602..8801e501bc 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/SkillsChangedNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/SkillsChangedNotification.json @@ -1,6 +1,7 @@ { "$schema": "http://json-schema.org/draft-07/schema#", - "description": "Notification emitted when watched local skill files change.\n\nTreat this as an invalidation signal and re-run `skills/list` with the client's current parameters when refreshed skill metadata is needed.", + "description": "Notification emitted when the available skills catalog changes.\n\nTreat this as an invalidation signal and re-run `skills/list` with the client's current parameters when refreshed skill metadata is needed.", + "properties": {}, "title": "SkillsChangedNotification", "type": "object" } \ No newline at end of file diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/SkillsChangedNotification.ts b/codex-rs/app-server-protocol/schema/typescript/v2/SkillsChangedNotification.ts index 23ed93a5ec..fbbe5659a6 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/SkillsChangedNotification.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/SkillsChangedNotification.ts @@ -3,9 +3,9 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. /** - * Notification emitted when watched local skill files change. + * Notification emitted when the available skills catalog changes. * * Treat this as an invalidation signal and re-run `skills/list` with the * client's current parameters when refreshed skill metadata is needed. */ -export type SkillsChangedNotification = Record; +export type SkillsChangedNotification = {}; diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 384a93fbf4..0d17f3a25b 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -1620,6 +1620,7 @@ server_notification_definitions! { ThreadDeleted => "thread/deleted" (v2::ThreadDeletedNotification), ThreadUnarchived => "thread/unarchived" (v2::ThreadUnarchivedNotification), ThreadClosed => "thread/closed" (v2::ThreadClosedNotification), + #[experimental(nested)] SkillsChanged => "skills/changed" (v2::SkillsChangedNotification), ThreadNameUpdated => "thread/name/updated" (v2::ThreadNameUpdatedNotification), ThreadGoalUpdated => "thread/goal/updated" (v2::ThreadGoalUpdatedNotification), diff --git a/codex-rs/app-server-protocol/src/protocol/v2/plugin.rs b/codex-rs/app-server-protocol/src/protocol/v2/plugin.rs index 46bb5c7909..fa0b807e8e 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/plugin.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/plugin.rs @@ -915,11 +915,15 @@ impl From for SkillScope { } } } -#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS, ExperimentalApi)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] -/// Notification emitted when watched local skill files change. +/// Notification emitted when the available skills catalog changes. /// /// Treat this as an invalidation signal and re-run `skills/list` with the /// client's current parameters when refreshed skill metadata is needed. -pub struct SkillsChangedNotification {} +pub struct SkillsChangedNotification { + /// The affected thread, or `None` for a process-wide local skill change. + #[experimental("skills/changed.threadId")] + pub thread_id: Option, +} diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 0ef51cdfcd..8ff36ffc52 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -217,7 +217,7 @@ Example with notification opt-out: - `plugin/installed` — list installed plugin rows plus any explicitly requested local install-suggestion plugin names, without fetching the broader remote catalog. Remote rows include nullable `installPolicySource`; local rows return `null`. Mention surfaces can use this narrower view when they need plugin mention payloads rather than plugin-page discovery data (**under development; do not call from production clients yet**). - `plugin/read` — read one plugin by `marketplacePath` plus `pluginName`, returning marketplace info, a list-style `summary`, manifest descriptions/interface metadata, and bundled skills/hooks/apps/MCP server names. Remote plugin details expose the canonical `shareUrl` supplied by the remote catalog when available; it is `null` for local plugins or when the catalog omits it. This field is separate from `summary.shareContext`, which continues to describe user and workspace sharing state. Returned plugin skills include their current `enabled` state after local config filtering; bundled hooks are returned as lightweight declaration summaries keyed for correlation with `hooks/list`. Use `plugin/install`'s `appsNeedingAuth` to drive post-install authentication and `app/list`'s `isAccessible` to determine current connector accessibility (**under development; do not call from production clients yet**). - `plugin/skill/read` — read remote plugin skill markdown on demand by `remoteMarketplaceName`, `remotePluginId`, and `skillName`. This lets clients preview uninstalled remote plugin skills without downloading the plugin bundle. -- `skills/changed` — notification emitted when watched local skill files change. +- `skills/changed` — notification emitted when local or thread-selected skills change. - `app/list` — list available apps. - `remoteControl/enable` — experimental; enable remote control for the current app-server process and return the current remote-control status snapshot. By default, any missing enrollment is completed before the response and the preference is persisted for the current app-server client scope. Pass `ephemeral: true` to enable remote control only for the current process without changing the persisted preference. - `remoteControl/disable` — experimental; disable remote control for the current app-server process and return the current remote-control status snapshot. By default, the disabled preference is persisted for the current app-server client scope. Pass `ephemeral: true` to disable only for the current process without changing the persisted preference. This does not revoke already enrolled controller devices. @@ -1640,7 +1640,7 @@ $skill-creator Add a new skill for triaging flaky CI and include step-by-step us Use `skills/list` to fetch the available skills (optionally scoped by `cwds`, with `forceReload`). Pass the experimental `threadId` field to also receive the selected executor skills currently available to that thread in `threadSkills`; executor discovery problems are returned in `threadSkillWarnings`. `threadId` does not change `data`; that field continues to use `cwds`, or the app-server default working directory when `cwds` is empty. `skills/list` might reuse a cached skills result per `cwd`; setting `forceReload` to `true` refreshes the result from disk. -The server also emits `skills/changed` notifications when watched local skill files change. Treat this as an invalidation signal and re-run `skills/list` with your current params when needed. +The server also emits `skills/changed` notifications when watched local skill files or a selected executor catalog changes. A thread-scoped notification includes `threadId`; re-run `skills/list` with that same ID to refresh the invocation UI. A notification with `threadId: null` invalidates process-wide local skill state. Use `skills/extraRoots/set` to replace additional standalone skill roots for the current app-server process. These roots use the same layout as other standalone skill roots: each root contains skill directories, and each skill directory contains `SKILL.md`. Missing roots are accepted and load no skills until they exist. This setting is lost when app-server exits. ```json @@ -1683,7 +1683,7 @@ Use `skills/extraRoots/set` to replace additional standalone skill roots for the ```json { "method": "skills/changed", - "params": {} + "params": { "threadId": "67e8c0d2-5f31-4f0d-9f70-4b529c6fdd01" } } ``` diff --git a/codex-rs/app-server/src/extensions.rs b/codex-rs/app-server/src/extensions.rs index ed040aa8d9..7bc7f87f98 100644 --- a/codex-rs/app-server/src/extensions.rs +++ b/codex-rs/app-server/src/extensions.rs @@ -3,6 +3,7 @@ use std::sync::Weak; use codex_analytics::AnalyticsEventsClient; use codex_app_server_protocol::ServerNotification; +use codex_app_server_protocol::SkillsChangedNotification; use codex_app_server_protocol::ThreadGoal; use codex_app_server_protocol::ThreadGoalUpdatedNotification; use codex_core::NewThread; @@ -28,6 +29,27 @@ use crate::outgoing_message::OutgoingMessageSender; use crate::thread_state::ThreadListenerCommand; use crate::thread_state::ThreadStateManager; +pub(crate) async fn send_thread_skills_changed( + outgoing: &OutgoingMessageSender, + thread_state_manager: &ThreadStateManager, + thread_id: ThreadId, +) { + let connection_ids = thread_state_manager + .subscribed_connection_ids(thread_id) + .await; + if connection_ids.is_empty() { + return; + } + outgoing + .send_server_notification_to_connections( + &connection_ids, + ServerNotification::SkillsChanged(SkillsChangedNotification { + thread_id: Some(thread_id.to_string()), + }), + ) + .await; +} + pub(crate) struct ThreadExtensionDependencies { pub(crate) event_sink: Arc, pub(crate) auth_manager: Arc, @@ -37,6 +59,7 @@ pub(crate) struct ThreadExtensionDependencies { pub(crate) goal_service: Arc, pub(crate) environment_manager: Arc, pub(crate) executor_skill_provider: Arc, + pub(crate) skills_catalog_changed: Arc, /// Process-scoped persistence backend for extensions that need stored thread history. pub(crate) thread_store: Arc, } @@ -57,6 +80,7 @@ where goal_service, environment_manager, executor_skill_provider, + skills_catalog_changed, thread_store: _thread_store, } = dependencies; let mut builder = ExtensionRegistryBuilder::::with_event_sink(event_sink); @@ -84,7 +108,7 @@ where .with_orchestrator_provider(Arc::new( codex_skills_extension::OrchestratorSkillProvider::new(), )); - codex_skills_extension::install_with_providers( + codex_skills_extension::install_with_providers_and_catalog_updates( &mut builder, skill_providers, |config: &Config| codex_skills_extension::SkillsExtensionConfig { @@ -92,6 +116,7 @@ where bundled_skills_enabled: config.bundled_skills_enabled(), orchestrator_skills_enabled: config.orchestrator_skills_enabled, }, + move |thread_id| skills_catalog_changed(thread_id), ); Arc::new(builder.build()) } diff --git a/codex-rs/app-server/src/mcp_refresh.rs b/codex-rs/app-server/src/mcp_refresh.rs index 848d201f2e..07ce8a24fe 100644 --- a/codex-rs/app-server/src/mcp_refresh.rs +++ b/codex-rs/app-server/src/mcp_refresh.rs @@ -239,6 +239,7 @@ mod tests { goal_service: Arc::new(codex_goal_extension::GoalService::new()), environment_manager: Arc::clone(&environment_manager), executor_skill_provider: Arc::clone(&executor_skill_provider), + skills_catalog_changed: Arc::new(|_| {}), thread_store: Arc::clone(&thread_store), }, ), diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 2b20152a16..611f705ebe 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -12,6 +12,7 @@ use crate::error_code::invalid_request; use crate::extensions::ThreadExtensionDependencies; use crate::extensions::app_server_extension_event_sink; use crate::extensions::guardian_agent_spawner; +use crate::extensions::send_thread_skills_changed; use crate::extensions::thread_extensions; use crate::fs_watch::FsWatchManager; use crate::outgoing_message::ConnectionId; @@ -264,6 +265,17 @@ impl MessageProcessor { restriction_product, ), ); + let skills_catalog_changed: Arc = { + let outgoing = outgoing.clone(); + let thread_state_manager = thread_state_manager.clone(); + Arc::new(move |thread_id| { + let outgoing = outgoing.clone(); + let thread_state_manager = thread_state_manager.clone(); + tokio::spawn(async move { + send_thread_skills_changed(&outgoing, &thread_state_manager, thread_id).await; + }); + }) + }; let goal_service = Arc::new(GoalService::new()); let thread_manager = Arc::new_cyclic(|thread_manager| { ThreadManager::new( @@ -285,6 +297,7 @@ impl MessageProcessor { goal_service: Arc::clone(&goal_service), environment_manager: Arc::clone(&environment_manager_for_extensions), executor_skill_provider: Arc::clone(&executor_skill_provider), + skills_catalog_changed: Arc::clone(&skills_catalog_changed), thread_store: Arc::clone(&thread_store), }, ), @@ -428,7 +441,7 @@ impl MessageProcessor { Arc::clone(&config), config_manager.clone(), pending_thread_unloads, - thread_state_manager, + thread_state_manager.clone(), thread_watch_manager, thread_list_state_permit, Arc::clone(&skills_watcher), @@ -463,8 +476,12 @@ impl MessageProcessor { arg0_paths, codex_home: config.codex_home.to_path_buf(), }); - let environment_processor = - EnvironmentRequestProcessor::new(thread_manager.environment_manager()); + let environment_processor = EnvironmentRequestProcessor::new( + thread_manager.environment_manager(), + Arc::clone(&thread_manager), + outgoing.clone(), + thread_state_manager, + ); let fs_processor = FsRequestProcessor::new( Arc::clone(&environment_manager_for_requests), FsWatchManager::new(outgoing.clone()), diff --git a/codex-rs/app-server/src/request_processors/catalog_processor.rs b/codex-rs/app-server/src/request_processors/catalog_processor.rs index cccdff72e8..4068940707 100644 --- a/codex-rs/app-server/src/request_processors/catalog_processor.rs +++ b/codex-rs/app-server/src/request_processors/catalog_processor.rs @@ -671,7 +671,7 @@ impl CatalogRequestProcessor { .set_extra_roots(extra_roots); self.outgoing .send_server_notification(ServerNotification::SkillsChanged( - codex_app_server_protocol::SkillsChangedNotification {}, + codex_app_server_protocol::SkillsChangedNotification { thread_id: None }, )) .await; Ok(SkillsExtraRootsSetResponse {}) diff --git a/codex-rs/app-server/src/request_processors/environment_processor.rs b/codex-rs/app-server/src/request_processors/environment_processor.rs index 635e1732ec..483f389820 100644 --- a/codex-rs/app-server/src/request_processors/environment_processor.rs +++ b/codex-rs/app-server/src/request_processors/environment_processor.rs @@ -1,32 +1,109 @@ use super::*; +use codex_exec_server::Environment; use std::time::Duration; #[derive(Clone)] pub(crate) struct EnvironmentRequestProcessor { environment_manager: Arc, + thread_manager: Arc, + outgoing: Arc, + thread_state_manager: ThreadStateManager, } impl EnvironmentRequestProcessor { - pub(crate) fn new(environment_manager: Arc) -> Self { - Self { + pub(crate) fn new( + environment_manager: Arc, + thread_manager: Arc, + outgoing: Arc, + thread_state_manager: ThreadStateManager, + ) -> Self { + let processor = Self { environment_manager, + thread_manager, + outgoing, + thread_state_manager, + }; + for (environment_id, environment) in processor.environment_manager.registered_environments() + { + processor.notify_selected_threads_on_readiness_changes(environment_id, environment); } + processor } pub(crate) async fn environment_add( &self, params: EnvironmentAddParams, ) -> Result, JSONRPCErrorError> { + let environment_id = params.environment_id; self.environment_manager .upsert_environment( - params.environment_id, + environment_id.clone(), params.exec_server_url, params.connect_timeout_ms.map(Duration::from_millis), ) .map_err(|err| invalid_request(err.to_string()))?; + let environment = self + .environment_manager + .get_environment(&environment_id) + .ok_or_else(|| internal_error("upserted environment is unavailable"))?; + self.notify_selected_threads_on_readiness_changes(environment_id, environment); Ok(Some(EnvironmentAddResponse {}.into())) } + fn notify_selected_threads_on_readiness_changes( + &self, + environment_id: String, + environment: Arc, + ) { + let Some(mut readiness_changed) = environment.observe_readiness() else { + return; + }; + let startup_finished = environment.startup_finished(); + let thread_manager = Arc::downgrade(&self.thread_manager); + let outgoing = Arc::downgrade(&self.outgoing); + let thread_state_manager = self.thread_state_manager.clone(); + tokio::spawn(async move { + if !startup_finished && readiness_changed.changed().await.is_err() { + return; + } + loop { + let (Some(thread_manager), Some(outgoing)) = + (thread_manager.upgrade(), outgoing.upgrade()) + else { + return; + }; + for thread_id in thread_manager.list_thread_ids().await { + let Ok(thread) = thread_manager.get_thread(thread_id).await else { + continue; + }; + let selected_environment = + thread.selected_capability_roots().iter().any(|root| { + matches!( + &root.location, + codex_protocol::capabilities::CapabilityRootLocation::Environment { + environment_id: selected_environment_id, + .. + } if selected_environment_id == &environment_id + ) + }); + if selected_environment { + crate::extensions::send_thread_skills_changed( + &outgoing, + &thread_state_manager, + thread_id, + ) + .await; + } + } + drop(thread_manager); + drop(outgoing); + if readiness_changed.changed().await.is_err() { + return; + } + } + }); + } + pub(crate) async fn environment_info( &self, params: EnvironmentInfoParams, diff --git a/codex-rs/app-server/src/skills_watcher.rs b/codex-rs/app-server/src/skills_watcher.rs index 637fe963ea..e687f2e21d 100644 --- a/codex-rs/app-server/src/skills_watcher.rs +++ b/codex-rs/app-server/src/skills_watcher.rs @@ -147,7 +147,7 @@ impl SkillsWatcher { skills_service.clear_cache(); outgoing .send_server_notification(ServerNotification::SkillsChanged( - SkillsChangedNotification {}, + SkillsChangedNotification { thread_id: None }, )) .await; } diff --git a/codex-rs/app-server/src/transport_tests.rs b/codex-rs/app-server/src/transport_tests.rs index 439284a176..effec54258 100644 --- a/codex-rs/app-server/src/transport_tests.rs +++ b/codex-rs/app-server/src/transport_tests.rs @@ -2,6 +2,7 @@ use super::*; use codex_app_server_protocol::ConfigWarningNotification; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ServerNotification; +use codex_app_server_protocol::SkillsChangedNotification; use codex_app_server_protocol::ThreadRealtimeStartedNotification; use codex_protocol::protocol::RealtimeConversationVersion; use codex_utils_absolute_path::AbsolutePathBuf; @@ -182,6 +183,43 @@ async fn experimental_notifications_are_dropped_without_capability() { writer_rx.try_recv().is_err(), "experimental notifications should not reach clients without capability" ); + + route_outgoing_envelope( + &mut connections, + OutgoingEnvelope::ToConnection { + connection_id, + message: OutgoingMessage::AppServerNotification(ServerNotification::SkillsChanged( + SkillsChangedNotification { + thread_id: Some("thread-1".to_string()), + }, + )), + write_complete_tx: None, + }, + ) + .await; + assert!(writer_rx.try_recv().is_err()); + + route_outgoing_envelope( + &mut connections, + OutgoingEnvelope::ToConnection { + connection_id, + message: OutgoingMessage::AppServerNotification(ServerNotification::SkillsChanged( + SkillsChangedNotification { thread_id: None }, + )), + write_complete_tx: None, + }, + ) + .await; + let message = writer_rx + .recv() + .await + .expect("global skill changes should remain stable"); + assert!(matches!( + message.message, + OutgoingMessage::AppServerNotification(ServerNotification::SkillsChanged( + SkillsChangedNotification { thread_id: None } + )) + )); } #[tokio::test] diff --git a/codex-rs/app-server/tests/suite/v2/selected_capability_stack.rs b/codex-rs/app-server/tests/suite/v2/selected_capability_stack.rs index 50e2faab42..91718fd1a6 100644 --- a/codex-rs/app-server/tests/suite/v2/selected_capability_stack.rs +++ b/codex-rs/app-server/tests/suite/v2/selected_capability_stack.rs @@ -16,6 +16,7 @@ use codex_app_server_protocol::ListMcpServerStatusResponse; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::SelectedCapabilityRoot; use codex_app_server_protocol::ServerRequest; +use codex_app_server_protocol::SkillsChangedNotification; use codex_app_server_protocol::SkillsListParams; use codex_app_server_protocol::SkillsListResponse; use codex_app_server_protocol::ThreadResumeParams; @@ -98,11 +99,6 @@ async fn selected_capability_stack_tracks_environment_availability_and_resume() let response_mock = responses::mount_sse_sequence( &responses_server, vec![ - responses::sse(vec![ - responses::ev_response_created("environment-unavailable"), - responses::ev_assistant_message("unavailable-message", "Waiting"), - responses::ev_completed("environment-unavailable"), - ]), responses::sse(vec![ responses::ev_response_created("environment-available-call"), responses::ev_function_call_with_namespace( @@ -157,22 +153,17 @@ async fn selected_capability_stack_tracks_environment_availability_and_resume() fixture.environment_cwd.clone(), ) .await?; - - run_turn( - &mut app_server, - &thread_id, - "Inspect the current capabilities", - fixture.environment_cwd.clone(), - ) - .await?; - let initial_requests = response_mock.requests(); - assert_selected_capabilities_absent(&initial_requests[0]); + let (thread_skills, warnings) = list_thread_skills(&mut app_server, &thread_id).await?; + assert_eq!(thread_skills, Vec::new()); + assert_eq!(warnings.len(), 1); + assert!(warnings[0].contains(&format!("environment `{EXECUTOR_ID}` is unavailable"))); let mut exec_server = spawn_exec_server(fixture.codex_home.path(), &fixture.exec_server_url).await?; let mut exec_server_proxy = DisconnectableWebSocketProxy::start(&fixture.exec_server_url).await?; add_environment(&mut app_server, exec_server_proxy.websocket_url()).await?; + wait_for_thread_skills_changed(&mut app_server, &thread_id).await?; wait_for_selected_mcp_server(&mut app_server, &thread_id).await?; run_turn( @@ -182,6 +173,7 @@ async fn selected_capability_stack_tracks_environment_availability_and_resume() fixture.environment_cwd.clone(), ) .await?; + wait_for_thread_skills_changed(&mut app_server, &thread_id).await?; let first_mcp_pid = wait_for_pid_file(&fixture.pid_file).await?; run_turn( @@ -200,6 +192,7 @@ async fn selected_capability_stack_tracks_environment_availability_and_resume() ) .await?; exec_server_proxy.pause_and_disconnect().await?; + wait_for_thread_skills_changed(&mut app_server, &inspection_thread_id).await?; assert_eq!( timeout( Duration::from_secs(2), @@ -210,6 +203,7 @@ async fn selected_capability_stack_tracks_environment_availability_and_resume() (Vec::new(), Vec::new()) ); exec_server_proxy.resume()?; + wait_for_thread_skills_changed(&mut app_server, &inspection_thread_id).await?; wait_for_selected_mcp_server(&mut app_server, &inspection_thread_id).await?; let request_id = app_server @@ -229,6 +223,10 @@ async fn selected_capability_stack_tracks_environment_availability_and_resume() data.iter() .any(|entry| { entry.skills.iter().any(|skill| skill.name == SKILL_NAME) }) ); + assert_eq!( + list_thread_skills(&mut app_server, &inspection_thread_id).await?, + (vec![fixture.thread_skill.clone()], Vec::new()) + ); exec_server.kill().await?; drop(app_server); @@ -263,10 +261,10 @@ async fn selected_capability_stack_tracks_environment_availability_and_resume() ) .await?; let requests = response_mock.requests(); - assert_eq!(5, requests.len()); - assert_selected_plugin_tools_absent(&requests[4]); + assert_eq!(4, requests.len()); + assert_selected_plugin_tools_absent(&requests[3]); assert!( - latest_selected_skill_update(&requests[4]) + latest_selected_skill_update(&requests[3]) .is_some_and(|text| text.contains(NO_SELECTED_SKILLS_MESSAGE)) ); let (thread_skills, warnings) = list_thread_skills(&mut app_server, &thread_id).await?; @@ -276,6 +274,7 @@ async fn selected_capability_stack_tracks_environment_availability_and_resume() exec_server = spawn_exec_server(fixture.codex_home.path(), &fixture.exec_server_url).await?; add_environment(&mut app_server, &fixture.exec_server_url).await?; + wait_for_thread_skills_changed(&mut app_server, &thread_id).await?; wait_for_selected_mcp_server(&mut app_server, &thread_id).await?; assert_eq!( list_thread_skills(&mut app_server, &thread_id).await?, @@ -293,16 +292,16 @@ async fn selected_capability_stack_tracks_environment_availability_and_resume() assert_ne!(first_mcp_pid, resumed_mcp_pid); let requests = response_mock.requests(); - assert_eq!(6, requests.len()); - for request in &requests[1..4] { + assert_eq!(5, requests.len()); + for request in &requests[..3] { assert_selected_skill_is_injected(request, /*expected_count*/ 1); assert_selected_plugin_tools(request); assert_plugin_guidance_count(request, /*expected_count*/ 1); } - assert_plugin_guidance_count(&requests[4], /*expected_count*/ 1); - assert_selected_skill_is_injected(&requests[5], /*expected_count*/ 2); - assert_selected_plugin_tools(&requests[5]); - let output = requests[2].function_call_output(MCP_CALL_ID); + assert_plugin_guidance_count(&requests[3], /*expected_count*/ 1); + assert_selected_skill_is_injected(&requests[4], /*expected_count*/ 2); + assert_selected_plugin_tools(&requests[4]); + let output = requests[1].function_call_output(MCP_CALL_ID); let output = output["output"] .as_str() .expect("MCP function output should be text"); @@ -800,6 +799,40 @@ async fn list_thread_skills( Ok((thread_skills, thread_skill_warnings)) } +async fn wait_for_thread_skills_changed( + app_server: &mut TestAppServer, + thread_id: &str, +) -> Result<()> { + let notification = timeout( + READ_TIMEOUT, + app_server.read_stream_until_matching_notification( + "thread-scoped skills/changed", + |notification| { + notification.method == "skills/changed" + && notification + .params + .as_ref() + .and_then(|params| params.get("threadId")) + .and_then(serde_json::Value::as_str) + == Some(thread_id) + }, + ), + ) + .await??; + let notification: SkillsChangedNotification = serde_json::from_value( + notification + .params + .context("skills/changed params missing")?, + )?; + assert_eq!( + notification, + SkillsChangedNotification { + thread_id: Some(thread_id.to_string()) + } + ); + Ok(()) +} + async fn wait_for_selected_mcp_server( app_server: &mut TestAppServer, thread_id: &str, diff --git a/codex-rs/app-server/tests/suite/v2/skills_list.rs b/codex-rs/app-server/tests/suite/v2/skills_list.rs index 720124f74e..f38f035e8a 100644 --- a/codex-rs/app-server/tests/suite/v2/skills_list.rs +++ b/codex-rs/app-server/tests/suite/v2/skills_list.rs @@ -60,7 +60,7 @@ async fn expect_skills_changed_notification( .params .context("skills/changed params must be present")?; let notification: SkillsChangedNotification = serde_json::from_value(params)?; - assert_eq!(notification, SkillsChangedNotification {}); + assert_eq!(notification, SkillsChangedNotification { thread_id: None }); Ok(()) } @@ -992,7 +992,7 @@ async fn skills_changed_notification_is_emitted_after_skill_change() -> Result<( .context("skills/changed params must be present")?; let notification: SkillsChangedNotification = serde_json::from_value(params)?; - assert_eq!(notification, SkillsChangedNotification {}); + assert_eq!(notification, SkillsChangedNotification { thread_id: None }); let updated_skills_request_id = mcp .send_skills_list_request(SkillsListParams { cwds: vec![codex_home.path().to_path_buf()], diff --git a/codex-rs/codex-experimental-api-macros/src/lib.rs b/codex-rs/codex-experimental-api-macros/src/lib.rs index 2bca0190ea..ce2f1aaf2f 100644 --- a/codex-rs/codex-experimental-api-macros/src/lib.rs +++ b/codex-rs/codex-experimental-api-macros/src/lib.rs @@ -163,17 +163,40 @@ fn derive_for_enum(input: &DeriveInput, data: &DataEnum) -> TokenStream { for variant in &data.variants { let variant_name = &variant.ident; - let pattern = match &variant.fields { - Fields::Named(_) => quote!(Self::#variant_name { .. }), - Fields::Unnamed(_) => quote!(Self::#variant_name ( .. )), - Fields::Unit => quote!(Self::#variant_name), - }; let reason = experimental_reason(&variant.attrs); if let Some(reason) = reason { + let pattern = match &variant.fields { + Fields::Named(_) => quote!(Self::#variant_name { .. }), + Fields::Unnamed(_) => quote!(Self::#variant_name ( .. )), + Fields::Unit => quote!(Self::#variant_name), + }; match_arms.push(quote! { #pattern => Some(#reason), }); + } else if variant.attrs.iter().any(experimental_nested_attr) { + match &variant.fields { + Fields::Unnamed(unnamed) if unnamed.unnamed.len() == 1 => { + match_arms.push(quote! { + Self::#variant_name(payload) => { + crate::experimental_api::ExperimentalApi::experimental_reason(payload) + } + }); + } + Fields::Named(_) | Fields::Unnamed(_) | Fields::Unit => { + return syn::Error::new_spanned( + variant, + "#[experimental(nested)] requires a single-field tuple variant", + ) + .to_compile_error() + .into(); + } + } } else { + let pattern = match &variant.fields { + Fields::Named(_) => quote!(Self::#variant_name { .. }), + Fields::Unnamed(_) => quote!(Self::#variant_name ( .. )), + Fields::Unit => quote!(Self::#variant_name), + }; match_arms.push(quote! { #pattern => None, }); diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 645deb3125..6c508cbcb1 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -8,6 +8,7 @@ use codex_exec_server::SelectedCapabilityRootsStatus; use codex_features::Feature; use codex_otel::SessionTelemetry; use codex_protocol::ThreadId; +use codex_protocol::capabilities::SelectedCapabilityRoot; use codex_protocol::config_types::ApprovalsReviewer; use codex_protocol::config_types::CollaborationMode; use codex_protocol::config_types::Personality; @@ -638,6 +639,11 @@ impl CodexThread { self.codex.session.services.thread_extension_data.get::() } + /// Returns the selected capability roots persisted for this thread. + pub fn selected_capability_roots(&self) -> &[SelectedCapabilityRoot] { + &self.codex.session.services.selected_capability_roots + } + /// Passively inspects the selected capability roots whose environments are ready now. pub fn inspect_selected_capability_roots(&self) -> SelectedCapabilityRootsStatus { self.codex diff --git a/codex-rs/exec-server/src/client.rs b/codex-rs/exec-server/src/client.rs index 73e49b2f54..3b6fd39a90 100644 --- a/codex-rs/exec-server/src/client.rs +++ b/codex-rs/exec-server/src/client.rs @@ -189,6 +189,7 @@ pub(crate) struct Session { struct Inner { connection: StdMutex, connection_changed: watch::Sender<()>, + readiness_changed: StdMutex>>, // The remote transport delivers one shared notification stream for every // process on the connection. Keep a local process_id -> session registry so // we can turn those connection-global notifications into process wakeups @@ -252,6 +253,8 @@ pub(crate) struct LazyRemoteExecServerClient { recovery_policy: RecoveryPolicy, // Saves the first startup result so callers share it and failures remain final. startup: Arc, + // Broadcasts readiness transitions without allowing observers to initiate them. + readiness_changed: watch::Sender<()>, // The latest successful client, replaced whenever reconnecting succeeds. current_client: Arc>>, reconnect: Arc>>>, @@ -259,10 +262,12 @@ pub(crate) struct LazyRemoteExecServerClient { impl LazyRemoteExecServerClient { pub(crate) fn new(transport_params: ExecServerTransportParams) -> Self { + let (readiness_changed, _readiness_changed_rx) = watch::channel(()); Self { transport_params, recovery_policy: RecoveryPolicy::Wait, startup: Arc::new(ConnectionAttempt::new()), + readiness_changed, current_client: Arc::new(StdMutex::new(None)), reconnect: Arc::new(StdMutex::new(None)), } @@ -309,6 +314,10 @@ impl LazyRemoteExecServerClient { self.initial_client().await.map(drop) } + pub(crate) fn observe_readiness(&self) -> watch::Receiver<()> { + self.readiness_changed.subscribe() + } + pub(crate) async fn get(&self) -> Result { if matches!(self.recovery_policy, RecoveryPolicy::FailFast) { let client = match self.cached_client() { @@ -348,21 +357,32 @@ impl LazyRemoteExecServerClient { async fn initial_client(&self) -> Result { // The first caller starts the work; every other caller waits for that same result. + let transport_params = self.transport_params.clone(); + let current_client = Arc::clone(&self.current_client); + let readiness_changed = self.readiness_changed.clone(); + let mut initialized = false; let result = self .startup - .get_or_init(|| connect_once(self.transport_params.clone())) - .await; - match result { - Ok(client) => { - let mut current_client = self - .current_client - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - if current_client.is_none() { - *current_client = Some(client.clone()); + .get_or_init(|| { + initialized = true; + async move { + let result = connect_once(transport_params).await; + if let Ok(client) = &result { + client.set_readiness_observer(Some(readiness_changed)); + *current_client + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = + Some(client.clone()); + } + result } - Ok(client.clone()) - } + }) + .await; + if initialized { + self.readiness_changed.send_replace(()); + } + match result { + Ok(client) => Ok(client.clone()), Err(error) => Err(ExecServerError::ConnectionAttempt(Arc::clone(error))), } } @@ -385,11 +405,16 @@ impl LazyRemoteExecServerClient { .get_or_init(|| async { let result = connect_once(self.transport_params.clone()).await; if let Ok(client) = &result { - *self + let mut current_client = self .current_client .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(client.clone()); + .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some(previous) = current_client.replace(client.clone()) { + previous.set_readiness_observer(/*observer*/ None); + } + client.set_readiness_observer(Some(self.readiness_changed.clone())); } + self.readiness_changed.send_replace(()); result }) .await; @@ -502,6 +527,14 @@ pub enum ExecServerError { } impl ExecServerClient { + fn set_readiness_observer(&self, observer: Option>) { + *self + .inner + .readiness_changed + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = observer; + } + fn fail_fast(&self) -> Result { self.rpc_client_without_recovery()?; Ok(Self { @@ -846,6 +879,7 @@ impl ExecServerClient { active_process_starts: 0, }), connection_changed, + readiness_changed: StdMutex::new(None), sessions: ArcSwap::from_pointee(HashMap::new()), sessions_write_lock: StdMutex::new(()), http_body_streams: ArcSwap::from_pointee(HashMap::new()), diff --git a/codex-rs/exec-server/src/client_recovery.rs b/codex-rs/exec-server/src/client_recovery.rs index a6075f3739..6933b912ce 100644 --- a/codex-rs/exec-server/src/client_recovery.rs +++ b/codex-rs/exec-server/src/client_recovery.rs @@ -310,7 +310,7 @@ impl Inner { return; } - self.notify_connection_changed(); + self.notify_readiness_changed(); let inner = Arc::clone(self); tokio::spawn(async move { inner.recover(disconnect_message).await; @@ -416,11 +416,23 @@ impl Inner { } }; if installed { - self.notify_connection_changed(); + self.notify_readiness_changed(); } installed } + fn notify_readiness_changed(&self) { + self.notify_connection_changed(); + if let Some(readiness_changed) = self + .readiness_changed + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .as_ref() + { + readiness_changed.send_replace(()); + } + } + fn notify_connection_changed(&self) { self.connection_changed.send_replace(()); } @@ -518,7 +530,7 @@ impl Inner { } }; if newly_failed { - self.notify_connection_changed(); + self.notify_readiness_changed(); fail_all_in_flight_work(self, message.clone()).await; } } diff --git a/codex-rs/exec-server/src/environment.rs b/codex-rs/exec-server/src/environment.rs index d070be6786..20cd475b97 100644 --- a/codex-rs/exec-server/src/environment.rs +++ b/codex-rs/exec-server/src/environment.rs @@ -266,6 +266,16 @@ impl EnvironmentManager { environment_ids } + /// Returns a snapshot of every currently registered environment. + pub fn registered_environments(&self) -> Vec<(String, Arc)> { + self.environments + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .iter() + .map(|(environment_id, environment)| (environment_id.clone(), Arc::clone(environment))) + .collect() + } + /// Returns the local environment instance when one is configured. pub fn try_local_environment(&self) -> Option> { self.local_environment.as_ref().map(Arc::clone) @@ -616,6 +626,13 @@ impl Environment { } } + /// Observes readiness transitions without initiating startup or recovery. + pub fn observe_readiness(&self) -> Option> { + self.remote_client + .as_ref() + .map(LazyRemoteExecServerClient::observe_readiness) + } + pub fn get_exec_backend(&self) -> Arc { Arc::clone(&self.exec_backend) } @@ -1070,7 +1087,7 @@ mod tests { } #[tokio::test] - async fn environment_manager_leaves_stdio_environment_lazy() { + async fn environment_readiness_observer_keeps_stdio_environment_lazy() { let environment = Environment::remote_with_transport( ExecServerTransportParams::StdioCommand { command: StdioExecServerCommand { @@ -1094,8 +1111,19 @@ mod tests { .expect("environment manager"); let environment = manager.get_environment("stdio").expect("stdio environment"); + let mut readiness = environment + .observe_readiness() + .expect("remote environment should expose readiness changes"); + tokio::task::yield_now().await; assert!(!environment.startup_finished()); - assert!(environment.wait_until_ready().await.is_err()); + assert!( + tokio::time::timeout(Duration::from_millis(10), readiness.changed()) + .await + .is_err() + ); + + Environment::start_connecting_for_use(&environment); + assert!(readiness.changed().await.is_ok()); assert!(environment.startup_finished()); } diff --git a/codex-rs/ext/skills/src/extension.rs b/codex-rs/ext/skills/src/extension.rs index a632ab6824..a0d4d98528 100644 --- a/codex-rs/ext/skills/src/extension.rs +++ b/codex-rs/ext/skills/src/extension.rs @@ -21,6 +21,7 @@ use codex_extension_api::TurnInputContributor; use codex_extension_api::WorldStateContributionInput; use codex_extension_api::WorldStateSectionContribution; use codex_mcp::McpResourceClient; +use codex_protocol::ThreadId; use codex_protocol::openai_models::ModelInfo; use codex_protocol::protocol::Event; use codex_protocol::protocol::EventMsg; @@ -52,6 +53,7 @@ struct SkillsExtension { providers: SkillProviders, event_sink: Arc, config_from_host: Arc SkillsExtensionConfig + Send + Sync>, + catalog_changed: Arc, } impl ThreadLifecycleContributor for SkillsExtension @@ -149,8 +151,8 @@ where return Vec::new(); }; let config = thread_state.config(); - let catalog = thread_state - .executor_catalog_snapshot( + let (catalog, changed) = thread_state + .project_executor_catalog( &self.providers, SkillListQuery { turn_id: input.turn_id.to_string(), @@ -163,6 +165,9 @@ where }, ) .await; + if changed { + (self.catalog_changed)(input.thread_id); + } input .turn_store .insert(ExecutorSkillsStepState(catalog.clone())); @@ -406,11 +411,23 @@ pub fn install_with_providers( config_from_host: impl Fn(&C) -> SkillsExtensionConfig + Send + Sync + 'static, ) where C: Send + Sync + 'static, +{ + install_with_providers_and_catalog_updates(registry, providers, config_from_host, |_| {}); +} + +pub fn install_with_providers_and_catalog_updates( + registry: &mut ExtensionRegistryBuilder, + providers: SkillProviders, + config_from_host: impl Fn(&C) -> SkillsExtensionConfig + Send + Sync + 'static, + catalog_changed: impl Fn(ThreadId) + Send + Sync + 'static, +) where + C: Send + Sync + 'static, { let extension = Arc::new(SkillsExtension { providers, event_sink: registry.event_sink(), config_from_host: Arc::new(config_from_host), + catalog_changed: Arc::new(catalog_changed), }); registry.thread_lifecycle_contributor(extension.clone()); registry.config_contributor(extension.clone()); diff --git a/codex-rs/ext/skills/src/lib.rs b/codex-rs/ext/skills/src/lib.rs index c7245af974..e80f0ad277 100644 --- a/codex-rs/ext/skills/src/lib.rs +++ b/codex-rs/ext/skills/src/lib.rs @@ -13,6 +13,7 @@ mod world_state; pub use config::SkillsExtensionConfig; pub use extension::install; pub use extension::install_with_providers; +pub use extension::install_with_providers_and_catalog_updates; pub use provider::ExecutorSkillProvider; pub use provider::HostSkillProvider; pub use provider::OrchestratorSkillProvider; diff --git a/codex-rs/ext/skills/src/state.rs b/codex-rs/ext/skills/src/state.rs index fea3847fdd..202cf641fc 100644 --- a/codex-rs/ext/skills/src/state.rs +++ b/codex-rs/ext/skills/src/state.rs @@ -29,6 +29,7 @@ pub struct SkillsThreadState { config: Mutex, orchestrator_skills_available: bool, executor_cache: Mutex>, + projected_executor_catalog: Mutex>, orchestrator_cache: Mutex>>, } @@ -38,6 +39,7 @@ impl SkillsThreadState { config: Mutex::new(config), orchestrator_skills_available, executor_cache: Mutex::new(Vec::new()), + projected_executor_catalog: Mutex::new(None), orchestrator_cache: Mutex::new(None), } } @@ -116,6 +118,21 @@ impl SkillsThreadState { catalog } + pub(crate) async fn project_executor_catalog( + &self, + providers: &SkillProviders, + query: SkillListQuery, + ) -> (SkillCatalog, bool) { + let catalog = self.executor_catalog_snapshot(providers, query).await; + let mut projected = self + .projected_executor_catalog + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let changed = projected.as_ref() != Some(&catalog); + *projected = Some(catalog.clone()); + (catalog, changed) + } + pub(crate) async fn orchestrator_catalog_snapshot( &self, mcp_resources: Option<&McpResourceClient>, diff --git a/codex-rs/tui/src/app/app_server_event_targets.rs b/codex-rs/tui/src/app/app_server_event_targets.rs index 6c55850e4d..47e70cd169 100644 --- a/codex-rs/tui/src/app/app_server_event_targets.rs +++ b/codex-rs/tui/src/app/app_server_event_targets.rs @@ -161,8 +161,8 @@ pub(super) fn server_notification_thread_target( None => return ServerNotificationThreadTarget::AppScoped, } } - ServerNotification::SkillsChanged(_) - | ServerNotification::McpServerOauthLoginCompleted(_) + ServerNotification::SkillsChanged(notification) => notification.thread_id.as_deref(), + ServerNotification::McpServerOauthLoginCompleted(_) | ServerNotification::AccountUpdated(_) | ServerNotification::AccountRateLimitsUpdated(_) | ServerNotification::AppListUpdated(_) diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index d8d63575bb..b8c177e3a0 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/codex-rs/tui/src/app/tests.rs @@ -3804,7 +3804,7 @@ async fn active_non_primary_shutdown_target_returns_none_for_non_shutdown_event( assert_eq!( app.active_non_primary_shutdown_target(&ServerNotification::SkillsChanged( - codex_app_server_protocol::SkillsChangedNotification {}, + codex_app_server_protocol::SkillsChangedNotification { thread_id: None }, )), None ); diff --git a/codex-rs/tui/src/chatwidget/protocol.rs b/codex-rs/tui/src/chatwidget/protocol.rs index f60f9c41da..5bcad983ec 100644 --- a/codex-rs/tui/src/chatwidget/protocol.rs +++ b/codex-rs/tui/src/chatwidget/protocol.rs @@ -140,8 +140,12 @@ impl ChatWidget { ); } } - ServerNotification::SkillsChanged(_) => { - self.refresh_skills_for_current_cwd(/*force_reload*/ true); + ServerNotification::SkillsChanged(notification) => { + // Thread-selected skills use executor resource locators that the TUI does not + // consume yet. Keep the existing local-skill refresh behavior unchanged. + if notification.thread_id.is_none() { + self.refresh_skills_for_current_cwd(/*force_reload*/ true); + } } ServerNotification::ModelRerouted(_) => {} ServerNotification::ModelVerification(notification) => {