From 6754e77792e5cb788a8ef2b4f3c3b51d0324482a Mon Sep 17 00:00:00 2001 From: jif Date: Thu, 11 Jun 2026 10:46:47 +0100 Subject: [PATCH] skills: cache remote catalog failures per thread (#27403) ## Summary - cache the first remote skill catalog outcome per thread, including failures - preserve discovery errors as catalog warnings - update the existing cache regression test to verify failed discovery is attempted once ## Why A failed or hanging `codex_apps` `resources/list` call could run once while building initial context and immediately again while contributing first-turn input. With the discovery timeout, an ordinary Apps turn could wait up to 20 seconds before inference and retry again on later turns even when no remote skill was mentioned. Caching a warning-only empty catalog preserves graceful degradation while preventing repeated synchronous discovery attempts. ## Testing - `just fmt` - Tests and Clippy not run per request; CI will validate the change. --- codex-rs/ext/skills/src/extension.rs | 9 ++---- codex-rs/ext/skills/src/state.rs | 11 +++++-- codex-rs/ext/skills/tests/skills_extension.rs | 29 +++++++++++++++---- 3 files changed, 35 insertions(+), 14 deletions(-) diff --git a/codex-rs/ext/skills/src/extension.rs b/codex-rs/ext/skills/src/extension.rs index ca09aa3c6..84d7906eb 100644 --- a/codex-rs/ext/skills/src/extension.rs +++ b/codex-rs/ext/skills/src/extension.rs @@ -250,16 +250,13 @@ impl SkillsExtension { let mut catalog = self.providers.list_for_turn(query).await; if include_orchestrator_skills { - match thread_state + let orchestrator_catalog = thread_state .orchestrator_catalog_snapshot( self.providers .list_orchestrator_for_turn(orchestrator_query), ) - .await - { - Ok(orchestrator_catalog) => catalog.extend(orchestrator_catalog), - Err(err) => catalog.warnings.push(err.message), - } + .await; + catalog.extend(orchestrator_catalog); } catalog } diff --git a/codex-rs/ext/skills/src/state.rs b/codex-rs/ext/skills/src/state.rs index 726487f5c..c69f2d341 100644 --- a/codex-rs/ext/skills/src/state.rs +++ b/codex-rs/ext/skills/src/state.rs @@ -63,11 +63,16 @@ impl SkillsThreadState { pub(crate) async fn orchestrator_catalog_snapshot( &self, initialize: impl Future> + Send, - ) -> Result { + ) -> SkillCatalog { self.orchestrator_catalog - .get_or_try_init(|| initialize) + .get_or_init(|| async { + initialize.await.unwrap_or_else(|err| SkillCatalog { + warnings: vec![err.message], + ..Default::default() + }) + }) .await - .cloned() + .clone() } } diff --git a/codex-rs/ext/skills/tests/skills_extension.rs b/codex-rs/ext/skills/tests/skills_extension.rs index 432358754..42d3fac79 100644 --- a/codex-rs/ext/skills/tests/skills_extension.rs +++ b/codex-rs/ext/skills/tests/skills_extension.rs @@ -11,11 +11,14 @@ use codex_core_skills::SkillsLoadInput; use codex_core_skills::SkillsManager; use codex_core_skills::injection::InjectedHostSkillPrompts; use codex_extension_api::ExtensionData; +use codex_extension_api::ExtensionEventSink; use codex_extension_api::ExtensionRegistryBuilder; use codex_extension_api::ThreadStartInput; use codex_extension_api::TurnInputContext; use codex_protocol::capabilities::CapabilityRootLocation; use codex_protocol::capabilities::SelectedCapabilityRoot; +use codex_protocol::protocol::Event; +use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::SKILLS_INSTRUCTIONS_OPEN_TAG; use codex_protocol::protocol::SessionSource; use codex_protocol::user_input::UserInput; @@ -243,7 +246,7 @@ async fn selected_executor_catalog_is_context_and_selected_entrypoint_is_turn_in } #[tokio::test] -async fn orchestrator_catalog_snapshot_retries_failure_then_is_reused() -> TestResult { +async fn orchestrator_catalog_snapshot_caches_failure() -> TestResult { let list_calls = Arc::new(AtomicUsize::new(0)); let providers = SkillProviders::new().with_orchestrator_provider(Arc::new(StaticSkillProvider { @@ -260,7 +263,9 @@ async fn orchestrator_catalog_snapshot_retries_failure_then_is_reused() -> TestR list_calls: Some(Arc::clone(&list_calls)), fail_first_list: true, })); - let mut builder = ExtensionRegistryBuilder::new(); + let (event_tx, event_rx) = std::sync::mpsc::channel(); + let mut builder = + ExtensionRegistryBuilder::with_event_sink(Arc::new(ChannelEventSink(event_tx))); install_with_providers(&mut builder, providers); let registry = builder.build(); let session_store = ExtensionData::new("session"); @@ -281,6 +286,13 @@ async fn orchestrator_catalog_snapshot_retries_failure_then_is_reused() -> TestR .contribute(&session_store, &thread_store) .await; assert!(initial_fragments.is_empty()); + let EventMsg::Warning(warning) = event_rx.try_recv()?.msg else { + panic!("expected warning event"); + }; + assert_eq!( + warning.message, + "orchestrator skills unavailable: temporary orchestrator failure" + ); for turn_id in ["turn-1", "turn-2"] { let fragments = registry.turn_input_contributors()[0] @@ -298,10 +310,9 @@ async fn orchestrator_catalog_snapshot_retries_failure_then_is_reused() -> TestR &ExtensionData::new(turn_id), ) .await; - assert_eq!(1, fragments.len()); - assert!(fragments[0].render().contains("first")); + assert!(fragments.is_empty()); } - assert_eq!(2, list_calls.load(Ordering::Relaxed)); + assert_eq!(1, list_calls.load(Ordering::Relaxed)); Ok(()) } @@ -476,6 +487,14 @@ struct StaticSkillProvider { fail_first_list: bool, } +struct ChannelEventSink(std::sync::mpsc::Sender); + +impl ExtensionEventSink for ChannelEventSink { + fn emit(&self, event: Event) { + let _ = self.0.send(event); + } +} + impl SkillProvider for StaticSkillProvider { fn list(&self, _query: SkillListQuery) -> SkillProviderFuture<'_, SkillCatalog> { let list_call = self