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

@@ -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
     }

codex-rs/ext/skills/src/state.rs

@@ -63,11 +63,16 @@ impl SkillsThreadState {
     pub(crate) async fn orchestrator_catalog_snapshot(
         &self,
         initialize: impl Future<Output = Result<SkillCatalog, SkillProviderError>> + Send,
-    ) -> Result<SkillCatalog, SkillProviderError> {
+    ) -> 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()
     }
 }
 

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("<name>first</name>"));
+        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<Event>);
+
+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