From 994ce4e3121d9385e190e4c61b08abbd1807a493 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Wed, 8 Jul 2026 15:24:20 +0100 Subject: [PATCH] Expose thread-selected skills in skills list --- .../codex_app_server_protocol.schemas.json | 30 ++++ .../codex_app_server_protocol.v2.schemas.json | 30 ++++ .../schema/json/v2/SkillsListResponse.json | 30 ++++ .../schema/typescript/v2/SkillsListParams.ts | 8 +- .../typescript/v2/SkillsListResponse.ts | 2 +- .../typescript/v2/ThreadSkillMetadata.ts | 9 ++ .../schema/typescript/v2/index.ts | 1 + .../src/protocol/common.rs | 2 + .../src/protocol/v2/plugin.rs | 34 ++++- .../src/protocol/v2/tests.rs | 18 ++- codex-rs/app-server/README.md | 16 ++- codex-rs/app-server/src/message_processor.rs | 8 ++ .../request_processors/catalog_processor.rs | 85 +++++++++++- codex-rs/app-server/tests/suite/v2/mod.rs | 2 + .../suite/v2/selected_capability_proxy.rs | 131 ++++++++++++++++++ .../suite/v2/selected_capability_stack.rs | 74 +++++++++- .../app-server/tests/suite/v2/skills_list.rs | 57 +++++--- codex-rs/core/src/codex_thread.rs | 8 ++ codex-rs/ext/skills/src/catalog.rs | 3 +- codex-rs/ext/skills/src/lib.rs | 1 + codex-rs/ext/skills/src/provider/executor.rs | 25 +++- codex-rs/ext/skills/src/state.rs | 63 +++++++-- codex-rs/ext/skills/tests/skills_extension.rs | 37 ++++- codex-rs/tui/src/app/background_requests.rs | 1 + codex-rs/tui/src/app/thread_routing.rs | 1 + 25 files changed, 626 insertions(+), 50 deletions(-) create mode 100644 codex-rs/app-server-protocol/schema/typescript/v2/ThreadSkillMetadata.ts create mode 100644 codex-rs/app-server/tests/suite/v2/selected_capability_proxy.rs 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 b7526b080d..ccac2bba93 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 @@ -19483,6 +19483,36 @@ "title": "ThreadShellCommandResponse", "type": "object" }, + "ThreadSkillMetadata": { + "properties": { + "description": { + "type": "string" + }, + "enabled": { + "type": "boolean" + }, + "name": { + "type": "string" + }, + "resource": { + "description": "Stable executor-owned locator used for display and deduplication.", + "type": "string" + }, + "shortDescription": { + "type": [ + "string", + "null" + ] + } + }, + "required": [ + "description", + "enabled", + "name", + "resource" + ], + "type": "object" + }, "ThreadSortKey": { "enum": [ "created_at", 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 e1afc7883a..3e723c1cd7 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 @@ -17262,6 +17262,36 @@ "title": "ThreadShellCommandResponse", "type": "object" }, + "ThreadSkillMetadata": { + "properties": { + "description": { + "type": "string" + }, + "enabled": { + "type": "boolean" + }, + "name": { + "type": "string" + }, + "resource": { + "description": "Stable executor-owned locator used for display and deduplication.", + "type": "string" + }, + "shortDescription": { + "type": [ + "string", + "null" + ] + } + }, + "required": [ + "description", + "enabled", + "name", + "resource" + ], + "type": "object" + }, "ThreadSortKey": { "enum": [ "created_at", diff --git a/codex-rs/app-server-protocol/schema/json/v2/SkillsListResponse.json b/codex-rs/app-server-protocol/schema/json/v2/SkillsListResponse.json index 6c72bfbb68..c74a1918f6 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/SkillsListResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/SkillsListResponse.json @@ -209,6 +209,36 @@ "skills" ], "type": "object" + }, + "ThreadSkillMetadata": { + "properties": { + "description": { + "type": "string" + }, + "enabled": { + "type": "boolean" + }, + "name": { + "type": "string" + }, + "resource": { + "description": "Stable executor-owned locator used for display and deduplication.", + "type": "string" + }, + "shortDescription": { + "type": [ + "string", + "null" + ] + } + }, + "required": [ + "description", + "enabled", + "name", + "resource" + ], + "type": "object" } }, "properties": { diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/SkillsListParams.ts b/codex-rs/app-server-protocol/schema/typescript/v2/SkillsListParams.ts index 4adeb38b3b..6ea8655f57 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/SkillsListParams.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/SkillsListParams.ts @@ -2,12 +2,10 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. -export type SkillsListParams = { -/** +export type SkillsListParams = {/** * When empty, defaults to the current session working directory. */ -cwds?: Array, -/** +cwds?: Array, /** * When true, bypass the skills cache and re-scan skills from disk. */ -forceReload?: boolean, }; +forceReload?: boolean}; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/SkillsListResponse.ts b/codex-rs/app-server-protocol/schema/typescript/v2/SkillsListResponse.ts index a27c288a94..376b1457cd 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/SkillsListResponse.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/SkillsListResponse.ts @@ -3,4 +3,4 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. import type { SkillsListEntry } from "./SkillsListEntry"; -export type SkillsListResponse = { data: Array, }; +export type SkillsListResponse = {data: Array}; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadSkillMetadata.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadSkillMetadata.ts new file mode 100644 index 0000000000..190caaa982 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadSkillMetadata.ts @@ -0,0 +1,9 @@ +// GENERATED CODE! DO NOT MODIFY BY HAND! + +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +export type ThreadSkillMetadata = { name: string, description: string, shortDescription: string | null, +/** + * Stable executor-owned locator used for display and deduplication. + */ +resource: string, enabled: boolean, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts index 070d21eab8..a7fb3267b4 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts @@ -453,6 +453,7 @@ export type { ThreadSettings } from "./ThreadSettings"; export type { ThreadSettingsUpdatedNotification } from "./ThreadSettingsUpdatedNotification"; export type { ThreadShellCommandParams } from "./ThreadShellCommandParams"; export type { ThreadShellCommandResponse } from "./ThreadShellCommandResponse"; +export type { ThreadSkillMetadata } from "./ThreadSkillMetadata"; export type { ThreadSortKey } from "./ThreadSortKey"; export type { ThreadSource } from "./ThreadSource"; export type { ThreadSourceKind } from "./ThreadSourceKind"; diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 27a8b0205d..384a93fbf4 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -662,6 +662,7 @@ client_request_definitions! { }, SkillsList => "skills/list" { params: v2::SkillsListParams, + inspect_params: true, serialization: global_shared_read("config"), response: v2::SkillsListResponse, }, @@ -1865,6 +1866,7 @@ mod tests { params: v2::SkillsListParams { cwds: Vec::new(), force_reload: false, + thread_id: None, }, }; assert_eq!( 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 9bd53b5d1a..46bb5c7909 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/plugin.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/plugin.rs @@ -3,6 +3,7 @@ use super::HookEventName; use super::HookHandlerType; use super::HookSource; use super::HookTrustStatus; +use codex_experimental_api_macros::ExperimentalApi; use codex_protocol::protocol::SkillDependencies as CoreSkillDependencies; use codex_protocol::protocol::SkillInterface as CoreSkillInterface; use codex_protocol::protocol::SkillMetadata as CoreSkillMetadata; @@ -15,7 +16,9 @@ use serde::Serialize; use std::path::PathBuf; use ts_rs::TS; -#[derive(Serialize, Deserialize, Debug, Clone, Default, PartialEq, JsonSchema, TS)] +#[derive( + Serialize, Deserialize, Debug, Clone, Default, PartialEq, JsonSchema, TS, ExperimentalApi, +)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] pub struct SkillsListParams { @@ -26,13 +29,38 @@ pub struct SkillsListParams { /// When true, bypass the skills cache and re-scan skills from disk. #[serde(default, skip_serializing_if = "std::ops::Not::not")] pub force_reload: bool, + + /// Optional thread whose selected executor skills should be included. + #[experimental("skills/list.threadId")] + #[ts(optional = nullable)] + pub thread_id: Option, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS, ExperimentalApi)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct SkillsListResponse { + pub data: Vec, + /// Skills contributed by selected capability roots for the requested thread. + #[experimental("skills/list.threadSkills")] + #[serde(default)] + pub thread_skills: Vec, + /// Warnings encountered while discovering the requested thread's executor skills. + #[experimental("skills/list.threadSkillWarnings")] + #[serde(default)] + pub thread_skill_warnings: Vec, } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] -pub struct SkillsListResponse { - pub data: Vec, +pub struct ThreadSkillMetadata { + pub name: String, + pub description: String, + pub short_description: Option, + /// Stable executor-owned locator used for display and deduplication. + pub resource: String, + pub enabled: bool, } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] diff --git a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs index d35d39fe46..037cdc6d21 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs @@ -3033,24 +3033,40 @@ fn skills_list_params_serialization_uses_force_reload() { serde_json::to_value(SkillsListParams { cwds: Vec::new(), force_reload: false, + thread_id: None, }) .unwrap(), - json!({}), + json!({ "threadId": null }), ); assert_eq!( serde_json::to_value(SkillsListParams { cwds: vec![PathBuf::from("/repo")], force_reload: true, + thread_id: None, }) .unwrap(), json!({ "cwds": ["/repo"], "forceReload": true, + "threadId": null, }), ); } +#[test] +fn skills_list_response_defaults_experimental_fields() { + assert_eq!( + serde_json::from_value::(json!({ "data": [] })) + .expect("stable response should deserialize"), + SkillsListResponse { + data: Vec::new(), + thread_skills: Vec::new(), + thread_skill_warnings: Vec::new(), + } + ); +} + #[test] fn skills_extra_roots_set_params_serialization_uses_extra_roots() { assert_eq!( diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 6ee6baf2dc..0ef51cdfcd 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -1637,7 +1637,8 @@ Example: $skill-creator Add a new skill for triaging flaky CI and include step-by-step usage. ``` -Use `skills/list` to fetch the available skills (optionally scoped by `cwds`, with `forceReload`). +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. 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. @@ -1645,7 +1646,8 @@ Use `skills/extraRoots/set` to replace additional standalone skill roots for the ```json { "method": "skills/list", "id": 25, "params": { "cwds": ["/Users/me/project", "/Users/me/other-project"], - "forceReload": true + "forceReload": true, + "threadId": "67e8c0d2-5f31-4f0d-9f70-4b529c6fdd01" } } { "id": 25, "result": { "data": [{ @@ -1666,7 +1668,15 @@ Use `skills/extraRoots/set` to replace additional standalone skill roots for the } ], "errors": [] - }] + }], + "threadSkills": [{ + "name": "deploy", + "description": "Deploy through the selected executor", + "shortDescription": null, + "resource": "skill://executor-plugin@1/workspace/plugin/skills/deploy/SKILL.md", + "enabled": true + }], + "threadSkillWarnings": [] } } ``` diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 62eff72660..2b20152a16 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -257,6 +257,13 @@ impl MessageProcessor { restriction_product, ), ); + let executor_skill_inspection_provider: Arc = + Arc::new( + codex_skills_extension::ExecutorSkillProvider::new_for_inspection( + Arc::clone(&environment_manager_for_extensions), + restriction_product, + ), + ); let goal_service = Arc::new(GoalService::new()); let thread_manager = Arc::new_cyclic(|thread_manager| { ThreadManager::new( @@ -336,6 +343,7 @@ impl MessageProcessor { Arc::clone(&config), config_manager.clone(), Arc::clone(&workspace_settings_cache), + executor_skill_inspection_provider, ); let command_exec_processor = CommandExecRequestProcessor::new( arg0_paths.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 6cb4385151..4ce6fe76cd 100644 --- a/codex-rs/app-server/src/request_processors/catalog_processor.rs +++ b/codex-rs/app-server/src/request_processors/catalog_processor.rs @@ -11,6 +11,7 @@ pub(crate) struct CatalogRequestProcessor { pub(super) config: Arc, pub(super) config_manager: ConfigManager, pub(super) workspace_settings_cache: Arc, + executor_skill_providers: codex_skills_extension::SkillProviders, } const SKILLS_LIST_CWD_CONCURRENCY: usize = 5; @@ -97,6 +98,7 @@ fn errors_to_info( } impl CatalogRequestProcessor { + #[allow(clippy::too_many_arguments)] pub(crate) fn new( outgoing: Arc, skills_watcher: Arc, @@ -105,6 +107,7 @@ impl CatalogRequestProcessor { config: Arc, config_manager: ConfigManager, workspace_settings_cache: Arc, + executor_skill_provider: Arc, ) -> Self { Self { outgoing, @@ -114,6 +117,8 @@ impl CatalogRequestProcessor { config, config_manager, workspace_settings_cache, + executor_skill_providers: codex_skills_extension::SkillProviders::new() + .with_executor_provider(executor_skill_provider), } } @@ -489,12 +494,20 @@ impl CatalogRequestProcessor { &self, params: SkillsListParams, ) -> Result { - let SkillsListParams { cwds, force_reload } = params; + let SkillsListParams { + cwds, + force_reload, + thread_id, + } = params; let cwds = if cwds.is_empty() { vec![self.config.cwd.to_path_buf()] } else { cwds }; + let (thread_skills, thread_skill_warnings) = match thread_id.as_deref() { + Some(thread_id) => self.thread_skills(thread_id).await?, + None => (Vec::new(), Vec::new()), + }; let config = self.load_latest_config(/*fallback_cwd*/ None).await?; let auth = self.auth_manager.auth().await; @@ -507,7 +520,13 @@ impl CatalogRequestProcessor { .thread_manager .environment_manager() .default_environment() - .map(|environment| environment.get_filesystem()); + .map(|environment| { + if thread_id.is_some() { + environment.get_filesystem_without_reconnect() + } else { + environment.get_filesystem() + } + }); let mut data = futures::stream::iter(cwds.into_iter().enumerate()) .map(|(index, cwd)| { let config = &config; @@ -570,7 +589,67 @@ impl CatalogRequestProcessor { .await; data.sort_unstable_by_key(|(index, _)| *index); let data = data.into_iter().map(|(_, entry)| entry).collect(); - Ok(SkillsListResponse { data }) + Ok(SkillsListResponse { + data, + thread_skills, + thread_skill_warnings, + }) + } + + async fn thread_skills( + &self, + thread_id: &str, + ) -> Result< + ( + Vec, + Vec, + ), + JSONRPCErrorError, + > { + let thread_id = ThreadId::from_string(thread_id) + .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?; + let thread = self + .thread_manager + .get_thread(thread_id) + .await + .map_err(|_| invalid_request(format!("thread not found: {thread_id}")))?; + let Some(state) = + thread.extension_thread_data::() + else { + return Ok((Vec::new(), Vec::new())); + }; + let roots_status = thread.inspect_selected_capability_roots(); + let catalog = state + .inspect_executor_catalog( + &self.executor_skill_providers, + codex_skills_extension::provider::SkillListQuery { + turn_id: thread_id.to_string(), + executor_roots: roots_status.ready_roots, + host_snapshot: None, + include_host_skills: false, + include_bundled_skills: false, + include_orchestrator_skills: false, + mcp_resources: None, + }, + ) + .await; + let mut warnings = roots_status.warnings; + warnings.extend(catalog.warnings); + let skills = catalog + .entries + .into_iter() + .map(|entry| { + let resource = entry.rendered_path().to_string(); + codex_app_server_protocol::ThreadSkillMetadata { + name: entry.name, + description: entry.description, + short_description: entry.short_description, + resource, + enabled: entry.enabled, + } + }) + .collect(); + Ok((skills, warnings)) } async fn skills_extra_roots_set_response( diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index 3d55079dc6..f15ff029e6 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/codex-rs/app-server/tests/suite/v2/mod.rs @@ -59,6 +59,8 @@ mod request_validation; mod review; mod safety_check_downgrade; #[cfg(not(target_os = "windows"))] +mod selected_capability_proxy; +#[cfg(not(target_os = "windows"))] mod selected_capability_stack; mod selected_environment; mod skills_list; diff --git a/codex-rs/app-server/tests/suite/v2/selected_capability_proxy.rs b/codex-rs/app-server/tests/suite/v2/selected_capability_proxy.rs new file mode 100644 index 0000000000..cca0a212b2 --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/selected_capability_proxy.rs @@ -0,0 +1,131 @@ +use std::time::Duration; + +use anyhow::Context; +use anyhow::Result; +use tokio::io::copy_bidirectional; +use tokio::net::TcpListener; +use tokio::net::TcpStream; +use tokio::sync::oneshot; +use tokio::task::JoinHandle; +use tokio::time::timeout; + +const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); + +pub(super) struct DisconnectableWebSocketProxy { + websocket_url: String, + pause_tx: Option>, + blocked_connection_rx: Option>, + resume_tx: Option>, + task: JoinHandle<()>, +} + +impl DisconnectableWebSocketProxy { + pub(super) async fn start(upstream_url: &str) -> Result { + let upstream = upstream_url + .strip_prefix("ws://") + .context("exec-server websocket URL must use ws://")? + .to_string(); + let listener = TcpListener::bind("127.0.0.1:0").await?; + let websocket_url = format!("ws://{}", listener.local_addr()?); + let (pause_tx, pause_rx) = oneshot::channel(); + let (blocked_connection_tx, blocked_connection_rx) = oneshot::channel(); + let (resume_tx, resume_rx) = oneshot::channel(); + let task = tokio::spawn(run_proxy( + listener, + upstream, + pause_rx, + blocked_connection_tx, + resume_rx, + )); + Ok(Self { + websocket_url, + pause_tx: Some(pause_tx), + blocked_connection_rx: Some(blocked_connection_rx), + resume_tx: Some(resume_tx), + task, + }) + } + + pub(super) fn websocket_url(&self) -> &str { + &self.websocket_url + } + + pub(super) async fn pause_and_disconnect(&mut self) -> Result<()> { + self.pause_tx + .take() + .context("disconnectable websocket proxy is already paused")? + .send(()) + .map_err(|_| anyhow::anyhow!("disconnectable websocket proxy stopped"))?; + let blocked_connection_rx = self + .blocked_connection_rx + .take() + .context("disconnectable websocket proxy is already paused")?; + timeout(CONNECT_TIMEOUT, blocked_connection_rx) + .await + .context("timed out waiting for client reconnect attempt")? + .context("disconnectable websocket proxy stopped")?; + Ok(()) + } + + pub(super) fn resume(&mut self) -> Result<()> { + self.resume_tx + .take() + .context("disconnectable websocket proxy is already resumed")? + .send(()) + .map_err(|_| anyhow::anyhow!("disconnectable websocket proxy stopped"))?; + Ok(()) + } +} + +impl Drop for DisconnectableWebSocketProxy { + fn drop(&mut self) { + self.task.abort(); + } +} + +async fn run_proxy( + listener: TcpListener, + upstream: String, + pause_rx: oneshot::Receiver<()>, + blocked_connection_tx: oneshot::Sender<()>, + mut resume_rx: oneshot::Receiver<()>, +) { + let Ok((mut downstream, _)) = listener.accept().await else { + return; + }; + let Ok(mut upstream_stream) = TcpStream::connect(&upstream).await else { + return; + }; + tokio::select! { + _ = copy_bidirectional(&mut downstream, &mut upstream_stream) => return, + _ = pause_rx => {} + } + drop(downstream); + drop(upstream_stream); + + let mut blocked_connection_tx = Some(blocked_connection_tx); + loop { + tokio::select! { + _ = &mut resume_rx => break, + accepted = listener.accept() => { + let Ok((blocked, _)) = accepted else { + return; + }; + drop(blocked); + if let Some(blocked_connection_tx) = blocked_connection_tx.take() { + let _ = blocked_connection_tx.send(()); + } + } + } + } + + loop { + let Ok((mut downstream, _)) = listener.accept().await else { + return; + }; + let Ok(mut upstream_stream) = TcpStream::connect(&upstream).await else { + continue; + }; + let _ = copy_bidirectional(&mut downstream, &mut upstream_stream).await; + } +} 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 9e8983840b..c95c9240a3 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,8 +16,11 @@ 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::SkillsListParams; +use codex_app_server_protocol::SkillsListResponse; use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadResumeResponse; +use codex_app_server_protocol::ThreadSkillMetadata; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::TurnEnvironmentParams; @@ -47,6 +50,7 @@ use tokio::time::timeout; use super::app_list::connector_tool; use super::app_list::start_apps_server_with_delays; +use super::selected_capability_proxy::DisconnectableWebSocketProxy; const READ_TIMEOUT: Duration = Duration::from_secs(20); const EXECUTOR_ID: &str = "executor-1"; @@ -166,7 +170,9 @@ async fn selected_capability_stack_tracks_environment_availability_and_resume() let mut exec_server = spawn_exec_server(fixture.codex_home.path(), &fixture.exec_server_url).await?; - add_environment(&mut app_server, &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_selected_mcp_server(&mut app_server, &thread_id).await?; run_turn( @@ -187,6 +193,24 @@ async fn selected_capability_stack_tracks_environment_availability_and_resume() .await?; assert_eq!(first_mcp_pid, wait_for_pid_file(&fixture.pid_file).await?); + let inspection_thread_id = start_thread( + &mut app_server, + fixture.selected_root.clone(), + fixture.environment_cwd.clone(), + ) + .await?; + exec_server_proxy.pause_and_disconnect().await?; + assert_eq!( + timeout( + Duration::from_secs(2), + list_thread_skills(&mut app_server, &inspection_thread_id), + ) + .await + .context("skills/list waited for executor recovery")??, + (Vec::new(), Vec::new()) + ); + exec_server_proxy.resume()?; + exec_server.kill().await?; drop(app_server); std::fs::remove_file(&fixture.pid_file)?; @@ -226,10 +250,18 @@ async fn selected_capability_stack_tracks_environment_availability_and_resume() latest_selected_skill_update(&requests[4]) .is_some_and(|text| text.contains(NO_SELECTED_SKILLS_MESSAGE)) ); + 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"))); 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_selected_mcp_server(&mut app_server, &thread_id).await?; + assert_eq!( + list_thread_skills(&mut app_server, &thread_id).await?, + (vec![fixture.thread_skill.clone()], Vec::new()) + ); run_turn( &mut app_server, @@ -439,6 +471,7 @@ struct SelectedCapabilityFixture { pid_file: std::path::PathBuf, exec_server_url: String, selected_root: SelectedCapabilityRoot, + thread_skill: ThreadSkillMetadata, environment_cwd: AbsolutePathBuf, } @@ -505,8 +538,9 @@ fn selected_capability_fixture( manifest_dir.join("plugin.json"), r#"{"name":"executor-demo","apps":"./.app.json","interface":{"displayName":"Executor Demo"}}"#, )?; + let skill_path = skill_dir.join("SKILL.md"); std::fs::write( - skill_dir.join("SKILL.md"), + &skill_path, format!( "---\nname: deploy\ndescription: {SKILL_DESCRIPTION}\n---\n\n{SKILL_BODY_MARKER}\n" ), @@ -538,6 +572,17 @@ fn selected_capability_fixture( path: PathUri::from_host_native_path(plugin.path())?, }, }; + let normalized_skill_path = skill_path.to_string_lossy().replace('\\', "/"); + let thread_skill = ThreadSkillMetadata { + name: SKILL_NAME.to_string(), + description: SKILL_DESCRIPTION.to_string(), + short_description: None, + resource: format!( + "skill://{PLUGIN_ID}/{}", + normalized_skill_path.trim_start_matches('/') + ), + enabled: true, + }; let environment_cwd = AbsolutePathBuf::try_from(plugin.path().to_path_buf())?; Ok(SelectedCapabilityFixture { codex_home, @@ -545,6 +590,7 @@ fn selected_capability_fixture( pid_file, exec_server_url, selected_root, + thread_skill, environment_cwd, }) } @@ -711,6 +757,30 @@ async fn add_environment(app_server: &mut TestAppServer, exec_server_url: &str) Ok(()) } +async fn list_thread_skills( + app_server: &mut TestAppServer, + thread_id: &str, +) -> Result<(Vec, Vec)> { + let request_id = app_server + .send_skills_list_request(SkillsListParams { + cwds: Vec::new(), + force_reload: true, + thread_id: Some(thread_id.to_string()), + }) + .await?; + let response = timeout( + READ_TIMEOUT, + app_server.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + let SkillsListResponse { + thread_skills, + thread_skill_warnings, + .. + } = to_response(response)?; + Ok((thread_skills, thread_skill_warnings)) +} + 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 c17a54b844..720124f74e 100644 --- a/codex-rs/app-server/tests/suite/v2/skills_list.rs +++ b/codex-rs/app-server/tests/suite/v2/skills_list.rs @@ -210,6 +210,7 @@ enabled = true .send_skills_list_request(SkillsListParams { cwds: vec![cwd.path().to_path_buf()], force_reload: true, + thread_id: None, }) .await?; let initial_skills_list_response: JSONRPCResponse = timeout( @@ -217,7 +218,7 @@ enabled = true mcp.read_stream_until_response_message(RequestId::Integer(initial_skills_list_request_id)), ) .await??; - let SkillsListResponse { data } = to_response(initial_skills_list_response)?; + let SkillsListResponse { data, .. } = to_response(initial_skills_list_response)?; assert!(data.iter().any(|entry| { entry .skills @@ -241,6 +242,7 @@ enabled = true .send_skills_list_request(SkillsListParams { cwds: vec![cwd.path().to_path_buf()], force_reload: true, + thread_id: None, }) .await?; let skills_list_response: JSONRPCResponse = timeout( @@ -248,7 +250,7 @@ enabled = true mcp.read_stream_until_response_message(RequestId::Integer(skills_list_request_id)), ) .await??; - let SkillsListResponse { data } = to_response(skills_list_response)?; + let SkillsListResponse { data, .. } = to_response(skills_list_response)?; assert!(data.iter().all(|entry| { entry @@ -358,6 +360,7 @@ async fn skills_list_loads_remote_installed_plugin_skills_from_cache() -> Result .send_skills_list_request(SkillsListParams { cwds: vec![cwd.path().to_path_buf()], force_reload: true, + thread_id: None, }) .await?; let stale_skills_list_response: JSONRPCResponse = timeout( @@ -365,7 +368,7 @@ async fn skills_list_loads_remote_installed_plugin_skills_from_cache() -> Result mcp.read_stream_until_response_message(RequestId::Integer(stale_skills_list_request_id)), ) .await??; - let SkillsListResponse { data } = to_response(stale_skills_list_response)?; + let SkillsListResponse { data, .. } = to_response(stale_skills_list_response)?; assert_eq!(data.len(), 1); assert!( data[0] @@ -403,12 +406,13 @@ async fn skills_list_loads_remote_installed_plugin_skills_from_cache() -> Result .await??; let _: PluginListResponse = to_response(plugin_list_response)?; - let SkillsListResponse { data } = timeout(DEFAULT_TIMEOUT, async { + let SkillsListResponse { data, .. } = timeout(DEFAULT_TIMEOUT, async { loop { let skills_list_request_id = mcp .send_skills_list_request(SkillsListParams { cwds: vec![cwd.path().to_path_buf()], force_reload: false, + thread_id: None, }) .await?; let skills_list_response: JSONRPCResponse = timeout( @@ -488,6 +492,7 @@ async fn skills_list_excludes_plugin_skills_when_workspace_codex_plugins_disable .send_skills_list_request(SkillsListParams { cwds: vec![repo_root.path().to_path_buf()], force_reload: true, + thread_id: None, }) .await?; @@ -496,7 +501,7 @@ async fn skills_list_excludes_plugin_skills_when_workspace_codex_plugins_disable mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; - let SkillsListResponse { data } = to_response(response)?; + let SkillsListResponse { data, .. } = to_response(response)?; assert_eq!(data.len(), 1); assert!( data[0] @@ -539,6 +544,7 @@ async fn skills_list_skips_cwd_roots_when_environment_disabled() -> Result<()> { .send_skills_list_request(SkillsListParams { cwds: vec![cwd.path().to_path_buf()], force_reload: true, + thread_id: None, }) .await?; @@ -547,7 +553,7 @@ async fn skills_list_skips_cwd_roots_when_environment_disabled() -> Result<()> { mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; - let SkillsListResponse { data } = to_response(response)?; + let SkillsListResponse { data, .. } = to_response(response)?; assert_eq!(data.len(), 1); assert_eq!(data[0].cwd, cwd.path().to_path_buf()); assert_eq!(data[0].errors, Vec::new()); @@ -583,6 +589,7 @@ async fn skills_list_accepts_relative_cwds() -> Result<()> { .send_skills_list_request(SkillsListParams { cwds: vec![relative_cwd.clone()], force_reload: true, + thread_id: None, }) .await?; @@ -591,7 +598,7 @@ async fn skills_list_accepts_relative_cwds() -> Result<()> { mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; - let SkillsListResponse { data } = to_response(response)?; + let SkillsListResponse { data, .. } = to_response(response)?; assert_eq!(data.len(), 1); assert_eq!(data[0].cwd, relative_cwd); assert_eq!(data[0].errors, Vec::new()); @@ -618,6 +625,7 @@ async fn skills_list_preserves_requested_cwd_order() -> Result<()> { second_cwd.path().to_path_buf(), ], force_reload: true, + thread_id: None, }) .await?; @@ -626,7 +634,7 @@ async fn skills_list_preserves_requested_cwd_order() -> Result<()> { mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; - let SkillsListResponse { data } = to_response(response)?; + let SkillsListResponse { data, .. } = to_response(response)?; assert_eq!( data.iter() .map(|entry| entry.cwd.clone()) @@ -656,6 +664,7 @@ async fn skills_list_uses_cached_result_until_force_reload() -> Result<()> { .send_skills_list_request(SkillsListParams { cwds: vec![cwd.path().to_path_buf()], force_reload: false, + thread_id: None, }) .await?; let first_response: JSONRPCResponse = timeout( @@ -663,7 +672,9 @@ async fn skills_list_uses_cached_result_until_force_reload() -> Result<()> { mcp.read_stream_until_response_message(RequestId::Integer(first_request_id)), ) .await??; - let SkillsListResponse { data: first_data } = to_response(first_response)?; + let SkillsListResponse { + data: first_data, .. + } = to_response(first_response)?; assert_eq!(first_data.len(), 1); assert!( first_data[0] @@ -683,6 +694,7 @@ async fn skills_list_uses_cached_result_until_force_reload() -> Result<()> { .send_skills_list_request(SkillsListParams { cwds: vec![cwd.path().to_path_buf()], force_reload: false, + thread_id: None, }) .await?; let second_response: JSONRPCResponse = timeout( @@ -690,7 +702,9 @@ async fn skills_list_uses_cached_result_until_force_reload() -> Result<()> { mcp.read_stream_until_response_message(RequestId::Integer(second_request_id)), ) .await??; - let SkillsListResponse { data: second_data } = to_response(second_response)?; + let SkillsListResponse { + data: second_data, .. + } = to_response(second_response)?; assert_eq!(second_data.len(), 1); assert!( second_data[0] @@ -703,6 +717,7 @@ async fn skills_list_uses_cached_result_until_force_reload() -> Result<()> { .send_skills_list_request(SkillsListParams { cwds: vec![cwd.path().to_path_buf()], force_reload: true, + thread_id: None, }) .await?; let third_response: JSONRPCResponse = timeout( @@ -710,7 +725,9 @@ async fn skills_list_uses_cached_result_until_force_reload() -> Result<()> { mcp.read_stream_until_response_message(RequestId::Integer(third_request_id)), ) .await??; - let SkillsListResponse { data: third_data } = to_response(third_response)?; + let SkillsListResponse { + data: third_data, .. + } = to_response(third_response)?; assert_eq!(third_data.len(), 1); assert!( third_data[0] @@ -758,6 +775,7 @@ async fn skills_extra_roots_set_updates_process_runtime_roots() -> Result<()> { .send_skills_list_request(SkillsListParams { cwds: vec![cwd.path().to_path_buf()], force_reload: false, + thread_id: None, }) .await?; let skills_response: JSONRPCResponse = timeout( @@ -765,7 +783,7 @@ async fn skills_extra_roots_set_updates_process_runtime_roots() -> Result<()> { mcp.read_stream_until_response_message(RequestId::Integer(skills_request_id)), ) .await??; - let SkillsListResponse { data } = to_response(skills_response)?; + let SkillsListResponse { data, .. } = to_response(skills_response)?; assert_eq!(data.len(), 1); assert_eq!(data[0].errors, Vec::new()); assert!( @@ -793,6 +811,7 @@ async fn skills_extra_roots_set_updates_process_runtime_roots() -> Result<()> { .send_skills_list_request(SkillsListParams { cwds: vec![cwd.path().to_path_buf()], force_reload: false, + thread_id: None, }) .await?; let skills_response: JSONRPCResponse = timeout( @@ -800,7 +819,7 @@ async fn skills_extra_roots_set_updates_process_runtime_roots() -> Result<()> { mcp.read_stream_until_response_message(RequestId::Integer(skills_request_id)), ) .await??; - let SkillsListResponse { data } = to_response(skills_response)?; + let SkillsListResponse { data, .. } = to_response(skills_response)?; assert_eq!(data.len(), 1); assert_eq!(data[0].errors, Vec::new()); assert!( @@ -826,6 +845,7 @@ async fn skills_extra_roots_set_updates_process_runtime_roots() -> Result<()> { .send_skills_list_request(SkillsListParams { cwds: vec![cwd.path().to_path_buf()], force_reload: false, + thread_id: None, }) .await?; let skills_response: JSONRPCResponse = timeout( @@ -833,7 +853,7 @@ async fn skills_extra_roots_set_updates_process_runtime_roots() -> Result<()> { mcp.read_stream_until_response_message(RequestId::Integer(skills_request_id)), ) .await??; - let SkillsListResponse { data } = to_response(skills_response)?; + let SkillsListResponse { data, .. } = to_response(skills_response)?; assert_eq!(data.len(), 1); assert_eq!(data[0].errors, Vec::new()); assert!( @@ -854,6 +874,7 @@ async fn skills_extra_roots_set_updates_process_runtime_roots() -> Result<()> { .send_skills_list_request(SkillsListParams { cwds: vec![cwd.path().to_path_buf()], force_reload: false, + thread_id: None, }) .await?; let skills_response: JSONRPCResponse = timeout( @@ -861,7 +882,7 @@ async fn skills_extra_roots_set_updates_process_runtime_roots() -> Result<()> { mcp.read_stream_until_response_message(RequestId::Integer(skills_request_id)), ) .await??; - let SkillsListResponse { data } = to_response(skills_response)?; + let SkillsListResponse { data, .. } = to_response(skills_response)?; assert_eq!(data.len(), 1); assert_eq!(data[0].errors, Vec::new()); assert!( @@ -899,6 +920,7 @@ async fn skills_changed_notification_is_emitted_after_skill_change() -> Result<( .send_skills_list_request(SkillsListParams { cwds: vec![codex_home.path().to_path_buf()], force_reload: true, + thread_id: None, }) .await?; let initial_skills_response: JSONRPCResponse = timeout( @@ -906,7 +928,7 @@ async fn skills_changed_notification_is_emitted_after_skill_change() -> Result<( mcp.read_stream_until_response_message(RequestId::Integer(initial_skills_request_id)), ) .await??; - let SkillsListResponse { data } = to_response(initial_skills_response)?; + let SkillsListResponse { data, .. } = to_response(initial_skills_response)?; assert_eq!(data.len(), 1); assert!( data[0] @@ -975,6 +997,7 @@ async fn skills_changed_notification_is_emitted_after_skill_change() -> Result<( .send_skills_list_request(SkillsListParams { cwds: vec![codex_home.path().to_path_buf()], force_reload: false, + thread_id: None, }) .await?; let updated_skills_response: JSONRPCResponse = timeout( @@ -982,7 +1005,7 @@ async fn skills_changed_notification_is_emitted_after_skill_change() -> Result<( mcp.read_stream_until_response_message(RequestId::Integer(updated_skills_request_id)), ) .await??; - let SkillsListResponse { data } = to_response(updated_skills_response)?; + let SkillsListResponse { data, .. } = to_response(updated_skills_response)?; assert_eq!(data.len(), 1); assert!( data[0] diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 257a32733b..645deb3125 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -630,6 +630,14 @@ impl CodexThread { self.codex.thread_environment_selections().await } + /// Returns one typed value owned by a thread extension. + pub fn extension_thread_data(&self) -> Option> + where + T: std::any::Any + Send + Sync, + { + self.codex.session.services.thread_extension_data.get::() + } + /// 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/ext/skills/src/catalog.rs b/codex-rs/ext/skills/src/catalog.rs index 59977a47ca..b8dfa0ef6d 100644 --- a/codex-rs/ext/skills/src/catalog.rs +++ b/codex-rs/ext/skills/src/catalog.rs @@ -166,7 +166,8 @@ impl SkillCatalogEntry { self } - pub(crate) fn rendered_path(&self) -> &str { + /// Returns the stable locator rendered to clients and the model. + pub fn rendered_path(&self) -> &str { self.display_path .as_deref() .unwrap_or_else(|| self.main_prompt.as_str()) diff --git a/codex-rs/ext/skills/src/lib.rs b/codex-rs/ext/skills/src/lib.rs index 4a004ff9ac..c7245af974 100644 --- a/codex-rs/ext/skills/src/lib.rs +++ b/codex-rs/ext/skills/src/lib.rs @@ -19,3 +19,4 @@ pub use provider::OrchestratorSkillProvider; pub use provider::SkillProvider; pub use sources::SkillProviderSource; pub use sources::SkillProviders; +pub use state::SkillsThreadState; diff --git a/codex-rs/ext/skills/src/provider/executor.rs b/codex-rs/ext/skills/src/provider/executor.rs index 9938035e92..6943b165db 100644 --- a/codex-rs/ext/skills/src/provider/executor.rs +++ b/codex-rs/ext/skills/src/provider/executor.rs @@ -27,6 +27,13 @@ use crate::provider::SkillSearchRequest; pub struct ExecutorSkillProvider { environment_manager: Arc, restriction_product: Option, + environment_access: EnvironmentAccess, +} + +#[derive(Clone, Copy, Debug)] +enum EnvironmentAccess { + Reconnect, + ReadyOnly, } impl ExecutorSkillProvider { @@ -37,6 +44,19 @@ impl ExecutorSkillProvider { Self { environment_manager, restriction_product, + environment_access: EnvironmentAccess::Reconnect, + } + } + + /// Creates a provider whose skill discovery never starts or reconnects an environment. + pub fn new_for_inspection( + environment_manager: Arc, + restriction_product: Option, + ) -> Self { + Self { + environment_manager, + restriction_product, + environment_access: EnvironmentAccess::ReadyOnly, } } } @@ -60,7 +80,10 @@ impl SkillProvider for ExecutorSkillProvider { )); continue; }; - let file_system = environment.get_filesystem(); + let file_system = match self.environment_access { + EnvironmentAccess::Reconnect => environment.get_filesystem(), + EnvironmentAccess::ReadyOnly => environment.get_filesystem_without_reconnect(), + }; let outcome = load_environment_skills_from_root( file_system.as_ref(), &path, diff --git a/codex-rs/ext/skills/src/state.rs b/codex-rs/ext/skills/src/state.rs index 2094f34496..fea3847fdd 100644 --- a/codex-rs/ext/skills/src/state.rs +++ b/codex-rs/ext/skills/src/state.rs @@ -25,7 +25,7 @@ use crate::sources::SkillProviders; const MAX_CACHED_ORCHESTRATOR_RESOURCES: usize = 100; const MAX_CACHED_ORCHESTRATOR_CONTENT_BYTES: usize = 8 * 1024 * 1024; -pub(crate) struct SkillsThreadState { +pub struct SkillsThreadState { config: Mutex, orchestrator_skills_available: bool, executor_cache: Mutex>, @@ -89,6 +89,33 @@ impl SkillsThreadState { catalog } + /// Reads executor catalogs without populating the thread's model-facing cache. + /// + /// Roots already cached by a model step reuse that snapshot. Uncached roots are discovered + /// for this read only, so a transient discovery failure cannot affect a later model step. + #[tracing::instrument( + name = "skills.executor.catalog_inspection", + level = "info", + skip_all, + fields(root_count = query.executor_roots.len()) + )] + pub async fn inspect_executor_catalog( + &self, + providers: &SkillProviders, + mut query: SkillListQuery, + ) -> SkillCatalog { + let roots = std::mem::take(&mut query.executor_roots); + let mut catalog = SkillCatalog::default(); + for root in roots { + query.executor_roots = vec![root.clone()]; + catalog.extend( + self.inspect_executor_root_catalog(providers, root, query.clone()) + .await, + ); + } + catalog + } + pub(crate) async fn orchestrator_catalog_snapshot( &self, mcp_resources: Option<&McpResourceClient>, @@ -170,14 +197,8 @@ impl SkillsThreadState { root: SelectedCapabilityRoot, query: SkillListQuery, ) -> SkillCatalog { - if let Some(cached) = self - .executor_cache - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .iter() - .find(|cached| cached.root == root) - { - return cached.catalog.clone(); + if let Some(cached) = self.cached_executor_root_catalog(&root) { + return cached; } let discovered = providers.list_executor_for_turn(query).await; @@ -194,6 +215,30 @@ impl SkillsThreadState { }); discovered } + + async fn inspect_executor_root_catalog( + &self, + providers: &SkillProviders, + root: SelectedCapabilityRoot, + query: SkillListQuery, + ) -> SkillCatalog { + if let Some(cached) = self.cached_executor_root_catalog(&root) { + return cached; + } + + let discovered = providers.list_executor_for_turn(query).await; + self.cached_executor_root_catalog(&root) + .unwrap_or(discovered) + } + + fn cached_executor_root_catalog(&self, root: &SelectedCapabilityRoot) -> Option { + self.executor_cache + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .iter() + .find(|cached| &cached.root == root) + .map(|cached| cached.catalog.clone()) + } } struct CachedExecutorCatalog { diff --git a/codex-rs/ext/skills/tests/skills_extension.rs b/codex-rs/ext/skills/tests/skills_extension.rs index 76d0df7954..48cffae3f1 100644 --- a/codex-rs/ext/skills/tests/skills_extension.rs +++ b/codex-rs/ext/skills/tests/skills_extension.rs @@ -33,6 +33,7 @@ use codex_protocol::protocol::TurnEnvironmentSelection; use codex_protocol::user_input::UserInput; use codex_skills_extension::SkillProviders; use codex_skills_extension::SkillsExtensionConfig; +use codex_skills_extension::SkillsThreadState; use codex_skills_extension::catalog::SkillAuthority; use codex_skills_extension::catalog::SkillCatalog; use codex_skills_extension::catalog::SkillCatalogEntry; @@ -146,7 +147,8 @@ async fn installed_extension_uses_host_service_snapshot() -> TestResult { } #[tokio::test] -async fn selected_executor_catalog_follows_step_availability_and_reuses_its_cache() -> TestResult { +async fn selected_executor_catalog_inspection_is_cache_neutral_and_tracks_availability() +-> TestResult { let read_requests = Arc::new(Mutex::new(Vec::new())); let list_calls = Arc::new(AtomicUsize::new(0)); let executor_provider = Arc::new(StaticSkillProvider { @@ -161,11 +163,11 @@ async fn selected_executor_catalog_follows_step_availability_and_reuses_its_cach }, read_requests: Arc::clone(&read_requests), list_calls: Some(Arc::clone(&list_calls)), - fail_first_list: false, + fail_first_list: true, }); let providers = SkillProviders::new().with_executor_provider(executor_provider); let mut builder = ExtensionRegistryBuilder::new(); - install_with_providers(&mut builder, providers, skills_extension_config); + install_with_providers(&mut builder, providers.clone(), skills_extension_config); let registry = builder.build(); let session_store = ExtensionData::new("session"); @@ -190,6 +192,33 @@ async fn selected_executor_catalog_follows_step_availability_and_reuses_its_cach }) .await; + let state = thread_store + .get::() + .ok_or("skills thread state should be initialized")?; + let inspected = state + .inspect_executor_catalog( + &providers, + SkillListQuery { + turn_id: "inspection".to_string(), + executor_roots: selected_roots.clone(), + host_snapshot: None, + include_host_skills: false, + include_bundled_skills: false, + include_orchestrator_skills: false, + mcp_resources: None, + }, + ) + .await; + assert_eq!( + SkillCatalog { + entries: Vec::new(), + warnings: vec![ + "executor skills unavailable: temporary orchestrator failure".to_string() + ], + }, + inspected + ); + let prompt_fragments = registry.context_contributors()[0] .contribute_thread_context(&session_store, &thread_store) .await; @@ -290,7 +319,7 @@ async fn selected_executor_catalog_follows_step_availability_and_reuses_its_cach .render_diff(PreviousWorldStateSection::Known(&unavailable_snapshot)) .ok_or("restored skills should render")?; assert!(restored_fragment.body().contains("lint-fix")); - assert_eq!(1, list_calls.load(Ordering::Relaxed)); + assert_eq!(2, list_calls.load(Ordering::Relaxed)); let mut listing_disabled_config = config.clone(); listing_disabled_config.include_instructions = false; diff --git a/codex-rs/tui/src/app/background_requests.rs b/codex-rs/tui/src/app/background_requests.rs index 6e2fd0ecbd..b1b5c1e0ca 100644 --- a/codex-rs/tui/src/app/background_requests.rs +++ b/codex-rs/tui/src/app/background_requests.rs @@ -838,6 +838,7 @@ pub(super) async fn fetch_skills_list( params: SkillsListParams { cwds: vec![cwd], force_reload: true, + thread_id: None, }, }) .await diff --git a/codex-rs/tui/src/app/thread_routing.rs b/codex-rs/tui/src/app/thread_routing.rs index b149fd53f3..3423e23a83 100644 --- a/codex-rs/tui/src/app/thread_routing.rs +++ b/codex-rs/tui/src/app/thread_routing.rs @@ -682,6 +682,7 @@ impl App { .skills_list(codex_app_server_protocol::SkillsListParams { cwds: cwds.clone(), force_reload: *force_reload, + thread_id: None, }) .await, "failed to refresh skills",