Deduplicate cloud and executor skill listings before budgeting (#49127)

## Why

Skills available from both cloud and executor providers consume duplicate catalog space, leaving less room for descriptions of unique skills.

## What changed

- Prefer model-visible cloud skills over executor listings with the same `plugin:skill` name before allocating the metadata budget.
- Preserve aliases from the full visible executor inventory so filtering duplicates does not renumber remaining packages. Continue choosing full locators when they fit the budget better.
- Keep executor packages available through `skills.list` and `skills.read`, and retain executor listings when cloud skills are disabled.

## Testing

Add regression tests and a model-request snapshot covering description retention, stable aliases across turns, executor availability changes, package reads, and full-locator selection after deduplication.

GitOrigin-RevId: 477e4cf16b8968dbe61438dd5af1eca9cbfa78c8

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

@@ -474,6 +474,10 @@ pub(crate) struct RenderedSkillCatalogs {
     pub(crate) host: Option<AvailableSkillsRender>,
 }
 
+#[path = "render_dedup.rs"]
+mod dedup;
+pub(crate) use dedup::PreparedSkillCatalog;
+
 impl AvailableSkillsRender {
     /// Count the metadata charged by the allocator, including aliases and omission notices.
     pub(crate) fn metadata_cost(
@@ -522,139 +526,65 @@ pub(crate) fn render_available_skills(
     budget: SkillMetadataBudget,
     include_skills_usage_instructions: bool,
 ) -> Option<AvailableSkillsRender> {
-    let mut entries = catalog
-        .entries
-        .iter()
-        .filter(|entry| entry.is_model_visible())
-        .collect::<Vec<_>>();
-    policy.order_entries(&mut entries);
-    if entries.is_empty() {
-        return None;
-    }
-
-    let absolute = render_catalog(
-        entries
-            .iter()
-            .map(|entry| SkillLine::new(entry, policy))
-            .collect(),
-        budget,
-        Vec::new(),
-        SkillPromptKind::Unaliased,
-        policy,
-    );
-    let selected = if let Some(aliased) =
-        build_aliased_catalog(&entries, policy, budget, include_skills_usage_instructions)
-        && aliased_render_is_better(
-            &aliased,
-            &absolute,
-            budget,
-            include_skills_usage_instructions,
-        ) {
-        aliased
-    } else {
-        absolute
-    };
-
-    Some(AvailableSkillsRender {
-        prompt_kind: selected.prompt_kind,
-        skill_root_lines: selected.skill_root_lines,
-        skill_lines: selected.skill_lines,
-        preserve_empty_fragment: policy == SkillCatalogRenderPolicy::CoreCompatible,
-        report: selected.report,
-    })
+    PreparedSkillCatalog::new(catalog, policy).render(budget, include_skills_usage_instructions)
 }
 
-pub(crate) fn render_combined_available_skills(
-    executor_catalog: &SkillCatalog,
-    cloud_catalog: &SkillCatalog,
-    host_catalog: &SkillCatalog,
+pub(crate) fn render_prepared_skill_catalogs(
+    executor: &PreparedSkillCatalog<'_>,
+    cloud: &PreparedSkillCatalog<'_>,
+    host: &PreparedSkillCatalog<'_>,
     budget: SkillMetadataBudget,
     include_skills_usage_instructions: bool,
 ) -> RenderedSkillCatalogs {
-    let mut executor_entries = executor_catalog
-        .entries
-        .iter()
-        .filter(|entry| entry.is_model_visible())
-        .collect::<Vec<_>>();
-    let mut cloud_entries = cloud_catalog
-        .entries
-        .iter()
-        .filter(|entry| entry.is_model_visible())
-        .collect::<Vec<_>>();
-    let mut host_entries = host_catalog
-        .entries
-        .iter()
-        .filter(|entry| entry.is_model_visible())
-        .collect::<Vec<_>>();
-    SkillCatalogRenderPolicy::ExtensionCompatible.order_entries(&mut executor_entries);
-    SkillCatalogRenderPolicy::ExtensionCompatible.order_entries(&mut cloud_entries);
-    SkillCatalogRenderPolicy::CoreCompatible.order_entries(&mut host_entries);
     let nonempty_catalog_count = [
-        !executor_entries.is_empty(),
-        !cloud_entries.is_empty(),
-        !host_entries.is_empty(),
+        !executor.entries.is_empty(),
+        !cloud.entries.is_empty(),
+        !host.entries.is_empty(),
     ]
     .into_iter()
     .filter(|nonempty| *nonempty)
     .count();
     if nonempty_catalog_count <= 1 {
         return RenderedSkillCatalogs {
-            executor: render_available_skills(
-                executor_catalog,
-                SkillCatalogRenderPolicy::ExtensionCompatible,
-                budget,
-                include_skills_usage_instructions,
-            ),
-            cloud: render_available_skills(
-                cloud_catalog,
-                SkillCatalogRenderPolicy::ExtensionCompatible,
-                budget,
-                include_skills_usage_instructions,
-            ),
-            host: render_available_skills(
-                host_catalog,
-                SkillCatalogRenderPolicy::CoreCompatible,
-                budget,
-                include_skills_usage_instructions,
-            ),
+            executor: executor.render(budget, include_skills_usage_instructions),
+            cloud: cloud.render(budget, include_skills_usage_instructions),
+            host: host.render(budget, include_skills_usage_instructions),
         };
     }
 
-    let extension_policy = SkillCatalogRenderPolicy::ExtensionCompatible;
-    let host_policy = SkillCatalogRenderPolicy::CoreCompatible;
     let absolute = render_combined_lines(
-        CatalogLines::unaliased(&executor_entries, extension_policy),
-        CatalogLines::unaliased(&cloud_entries, extension_policy),
-        CatalogLines::unaliased(&host_entries, host_policy),
+        executor.unaliased(),
+        cloud.unaliased(),
+        host.unaliased(),
         budget,
     );
 
     let mut selected = absolute;
     let host_only_aliases = build_aliased_combined_catalog(
-        CatalogLines::unaliased(&executor_entries, extension_policy),
-        CatalogLines::unaliased(&cloud_entries, extension_policy),
-        CatalogLines::aliased(&host_entries, host_policy),
+        executor.unaliased(),
+        cloud.unaliased(),
+        host.aliased(),
         budget,
         include_skills_usage_instructions,
     );
     let executor_only_aliases = build_aliased_combined_catalog(
-        CatalogLines::aliased(&executor_entries, extension_policy),
-        CatalogLines::unaliased(&cloud_entries, extension_policy),
-        CatalogLines::unaliased(&host_entries, host_policy),
+        executor.aliased(),
+        cloud.unaliased(),
+        host.unaliased(),
         budget,
         include_skills_usage_instructions,
     );
     let cloud_only_aliases = build_aliased_combined_catalog(
-        CatalogLines::unaliased(&executor_entries, extension_policy),
-        CatalogLines::aliased(&cloud_entries, extension_policy),
-        CatalogLines::unaliased(&host_entries, host_policy),
+        executor.unaliased(),
+        cloud.aliased(),
+        host.unaliased(),
         budget,
         include_skills_usage_instructions,
     );
     let all_source_aliases = build_aliased_combined_catalog(
-        CatalogLines::aliased(&executor_entries, extension_policy),
-        CatalogLines::aliased(&cloud_entries, extension_policy),
-        CatalogLines::aliased(&host_entries, host_policy),
+        executor.aliased(),
+        cloud.aliased(),
+        host.aliased(),
         budget,
         include_skills_usage_instructions,
     );
@@ -709,8 +639,12 @@ impl<'a> CatalogLines<'a> {
         }
     }
 
-    fn aliased(entries: &[&'a SkillCatalogEntry], policy: SkillCatalogRenderPolicy) -> Self {
-        let Some(plan) = build_alias_plan(entries) else {
+    fn aliased(
+        entries: &[&'a SkillCatalogEntry],
+        policy: SkillCatalogRenderPolicy,
+        alias_plan: Option<&AliasPlan>,
+    ) -> Self {
+        let Some(plan) = alias_plan.filter(|_| !entries.is_empty()) else {
             return Self::unaliased(entries, policy);
         };
 
@@ -725,7 +659,7 @@ impl<'a> CatalogLines<'a> {
                     SkillLine::with_locator(
                         entry,
                         policy,
-                        render_skill_locator_with_aliases(entry, &plan),
+                        render_skill_locator_with_aliases(entry, plan),
                     )
                 })
                 .collect(),
@@ -1014,12 +948,11 @@ fn available_skills_fragment(
 }
 
 fn build_aliased_catalog(
-    entries: &[&SkillCatalogEntry],
+    catalog: CatalogLines<'_>,
     policy: SkillCatalogRenderPolicy,
     budget: SkillMetadataBudget,
     include_skills_usage_instructions: bool,
 ) -> Option<RenderedCatalog> {
-    let catalog = CatalogLines::aliased(entries, policy);
     if catalog.root_lines.is_empty() {
         return None;
     }

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

@@ -0,0 +1,103 @@
+//! Prefers cloud skills before budgeting while retaining aliases from the full visible inventory.
+
+use std::collections::HashSet;
+
+use super::AvailableSkillsRender;
+use super::CatalogLines;
+use super::SkillCatalogRenderPolicy;
+use super::SkillMetadataBudget;
+use super::aliased_render_is_better;
+use super::build_alias_plan;
+use super::build_aliased_catalog;
+use super::render_catalog;
+use crate::aliases::AliasPlan;
+use crate::catalog::SkillCatalog;
+use crate::catalog::SkillCatalogEntry;
+use crate::catalog_prompt::SkillPromptKind;
+
+pub(crate) struct PreparedSkillCatalog<'a> {
+    pub(super) entries: Vec<&'a SkillCatalogEntry>,
+    alias_plan: Option<AliasPlan>,
+    policy: SkillCatalogRenderPolicy,
+}
+
+impl<'a> PreparedSkillCatalog<'a> {
+    pub(crate) fn new(catalog: &'a SkillCatalog, policy: SkillCatalogRenderPolicy) -> Self {
+        let mut entries = catalog
+            .entries
+            .iter()
+            .filter(|entry| entry.is_model_visible())
+            .collect::<Vec<_>>();
+        policy.order_entries(&mut entries);
+        let alias_plan = build_alias_plan(&entries);
+        Self {
+            entries,
+            alias_plan,
+            policy,
+        }
+    }
+
+    pub(crate) fn prefer_cloud_skills(&mut self, cloud: &SkillCatalog) {
+        let cloud_names: HashSet<_> = cloud
+            .entries
+            .iter()
+            .filter(|entry| entry.is_model_visible())
+            .map(|entry| entry.name.as_str())
+            .filter(|name| {
+                name.split_once(':')
+                    .is_some_and(|(plugin, skill)| !plugin.is_empty() && !skill.is_empty())
+            })
+            .collect();
+
+        self.entries
+            .retain(|entry| !cloud_names.contains(entry.name.as_str()));
+    }
+
+    pub(super) fn unaliased(&self) -> CatalogLines<'a> {
+        CatalogLines::unaliased(&self.entries, self.policy)
+    }
+
+    pub(super) fn aliased(&self) -> CatalogLines<'a> {
+        CatalogLines::aliased(&self.entries, self.policy, self.alias_plan.as_ref())
+    }
+
+    pub(crate) fn render(
+        &self,
+        budget: SkillMetadataBudget,
+        include_skills_usage_instructions: bool,
+    ) -> Option<AvailableSkillsRender> {
+        if self.entries.is_empty() {
+            return None;
+        }
+        let absolute = render_catalog(
+            self.unaliased().skills,
+            budget,
+            Vec::new(),
+            SkillPromptKind::Unaliased,
+            self.policy,
+        );
+        let selected = if let Some(aliased) = build_aliased_catalog(
+            self.aliased(),
+            self.policy,
+            budget,
+            include_skills_usage_instructions,
+        ) && aliased_render_is_better(
+            &aliased,
+            &absolute,
+            budget,
+            include_skills_usage_instructions,
+        ) {
+            aliased
+        } else {
+            absolute
+        };
+
+        Some(AvailableSkillsRender {
+            prompt_kind: selected.prompt_kind,
+            skill_root_lines: selected.skill_root_lines,
+            skill_lines: selected.skill_lines,
+            preserve_empty_fragment: self.policy == SkillCatalogRenderPolicy::CoreCompatible,
+            report: selected.report,
+        })
+    }
+}

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

@@ -20,6 +20,25 @@ use crate::catalog_prompt::render_available_skills_body;
 use crate::loader::HostSkillRoot;
 use crate::loader::load_and_merge_host_skill_roots;
 
+fn render_combined_available_skills(
+    executor_catalog: &SkillCatalog,
+    cloud_catalog: &SkillCatalog,
+    host_catalog: &SkillCatalog,
+    budget: SkillMetadataBudget,
+    include_skills_usage_instructions: bool,
+) -> RenderedSkillCatalogs {
+    render_prepared_skill_catalogs(
+        &PreparedSkillCatalog::new(
+            executor_catalog,
+            SkillCatalogRenderPolicy::ExtensionCompatible,
+        ),
+        &PreparedSkillCatalog::new(cloud_catalog, SkillCatalogRenderPolicy::ExtensionCompatible),
+        &PreparedSkillCatalog::new(host_catalog, SkillCatalogRenderPolicy::CoreCompatible),
+        budget,
+        include_skills_usage_instructions,
+    )
+}
+
 #[test]
 fn skill_prompt_contents_are_bounded_at_utf8_boundaries() {
     let contents = format!("{}é", "a".repeat(MAX_SKILL_PROMPT_BYTES - 1));
@@ -30,6 +49,61 @@ fn skill_prompt_contents_are_bounded_at_utf8_boundaries() {
     assert_eq!(truncated, true);
 }
 
+#[test]
+fn prebudget_dedup_can_choose_full_locators_over_preserved_aliases() {
+    let hidden_root = format!("skill://executor/{}", "long-package-root/".repeat(40));
+    let mut hidden = entry(
+        "demo:hidden",
+        "Hidden duplicate.",
+        /*short_description*/ None,
+    )
+    .with_alias_root(hidden_root.clone());
+    hidden.authority.kind = SkillSourceKind::Executor;
+    hidden.id = SkillPackageId(format!("{hidden_root}/hidden"));
+    let mut visible = entry(
+        "other:visible",
+        "Keep this description.",
+        /*short_description*/ None,
+    )
+    .with_alias_root("skill://short");
+    visible.authority.kind = SkillSourceKind::Executor;
+    visible.id = SkillPackageId("skill://short/visible".to_string());
+    let executor = SkillCatalog {
+        entries: vec![hidden.clone(), visible],
+        warnings: Vec::new(),
+    };
+    hidden.authority.kind = SkillSourceKind::Cloud;
+    let cloud = SkillCatalog {
+        entries: vec![hidden],
+        warnings: Vec::new(),
+    };
+    let mut prepared =
+        PreparedSkillCatalog::new(&executor, SkillCatalogRenderPolicy::ExtensionCompatible);
+    prepared.prefer_cloud_skills(&cloud);
+    let rendered = prepared
+        .render(
+            SkillMetadataBudget::Characters(300),
+            /*include_skills_usage_instructions*/ false,
+        )
+        .expect("remaining executor skill should render");
+    assert!(rendered.skill_root_lines.is_empty());
+    assert_eq!(
+        rendered.skill_lines,
+        vec![
+            "- other:visible: Keep this description. (executor package: skill://short/visible)"
+                .to_string(),
+        ]
+    );
+    assert_eq!(
+        rendered.report,
+        SkillRenderReport {
+            total_count: 1,
+            included_count: 1,
+            ..Default::default()
+        }
+    );
+}
+
 fn entry(name: &str, description: &str, short_description: Option<&str>) -> SkillCatalogEntry {
     entry_with_path(
         name,

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

@@ -15,11 +15,12 @@ use crate::catalog::SkillCatalog;
 use crate::provider::SkillListQuery;
 use crate::provider::attribute_executor_plugins;
 use crate::render::AvailableSkillsRender;
+use crate::render::PreparedSkillCatalog;
 use crate::render::RenderedSkillCatalogs;
 use crate::render::SkillCatalogRenderPolicy;
 use crate::render::SkillMetadataBudget;
 use crate::render::render_available_skills;
-use crate::render::render_combined_available_skills;
+use crate::render::render_prepared_skill_catalogs;
 use crate::render::skill_metadata_budget;
 use crate::render_observability::CatalogSurface;
 use crate::render_observability::record_catalog_render;
@@ -265,6 +266,11 @@ impl<'a> CatalogContext<'a> {
     ) -> [RenderedCatalogContribution; 3] {
         let total_limit = self.metadata_budget.limit();
         let default_cloud_limit = total_limit - total_limit / FILESYSTEM_BUDGET_DIVISOR;
+        let mut executor = PreparedSkillCatalog::new(
+            &catalogs.executor.catalog,
+            SkillCatalogRenderPolicy::ExtensionCompatible,
+        );
+        executor.prefer_cloud_skills(&catalogs.cloud.catalog);
         // Identify the full visible cloud inventory before budgeting, including entries
         // a previous cap omitted. Executor readiness and provider warnings are not inputs.
         let cloud_metadata = catalogs
@@ -307,14 +313,17 @@ impl<'a> CatalogContext<'a> {
             });
         allocation.cloud_limit = allocation.cloud_limit.min(default_cloud_limit);
         let (mut rendered, mut filesystem_budget) =
-            self.render_with_cloud_limit(&catalogs, allocation.cloud_limit);
+            self.render_with_cloud_limit(&catalogs, &executor, allocation.cloud_limit);
 
-        if let Some(cloud_limit) =
-            self.rebalance_to_retain_all_skills(&catalogs, &rendered, allocation.cloud_limit)
-        {
+        if let Some(cloud_limit) = self.rebalance_to_retain_all_skills(
+            &catalogs,
+            &executor,
+            &rendered,
+            allocation.cloud_limit,
+        ) {
             allocation.cloud_limit = cloud_limit;
             (rendered, filesystem_budget) =
-                self.render_with_cloud_limit(&catalogs, allocation.cloud_limit);
+                self.render_with_cloud_limit(&catalogs, &executor, allocation.cloud_limit);
         }
 
         [
@@ -341,6 +350,7 @@ impl<'a> CatalogContext<'a> {
     fn rebalance_to_retain_all_skills(
         &self,
         catalogs: &CatalogContributions,
+        executor: &PreparedSkillCatalog<'_>,
         rendered: &RenderedSkillCatalogs,
         current_cloud_limit: usize,
     ) -> Option<usize> {
@@ -352,10 +362,16 @@ impl<'a> CatalogContext<'a> {
             return None;
         }
 
-        let shared = render_combined_available_skills(
-            &catalogs.executor.catalog,
-            &catalogs.cloud.catalog,
-            &catalogs.host.catalog,
+        let shared = render_prepared_skill_catalogs(
+            executor,
+            &PreparedSkillCatalog::new(
+                &catalogs.cloud.catalog,
+                SkillCatalogRenderPolicy::ExtensionCompatible,
+            ),
+            &PreparedSkillCatalog::new(
+                &catalogs.host.catalog,
+                SkillCatalogRenderPolicy::CoreCompatible,
+            ),
             self.metadata_budget,
             self.include_usage,
         );
@@ -378,6 +394,7 @@ impl<'a> CatalogContext<'a> {
     fn render_with_cloud_limit(
         &self,
         catalogs: &CatalogContributions,
+        executor: &PreparedSkillCatalog<'_>,
         cloud_limit: usize,
     ) -> (RenderedSkillCatalogs, SkillMetadataBudget) {
         if !self.config.include_instructions {
@@ -398,10 +415,16 @@ impl<'a> CatalogContext<'a> {
         let filesystem_budget = self
             .metadata_budget
             .with_limit(self.metadata_budget.limit().saturating_sub(cloud_cost));
-        let mut rendered = render_combined_available_skills(
-            &catalogs.executor.catalog,
-            &SkillCatalog::default(),
-            &catalogs.host.catalog,
+        let mut rendered = render_prepared_skill_catalogs(
+            executor,
+            &PreparedSkillCatalog::new(
+                &SkillCatalog::default(),
+                SkillCatalogRenderPolicy::ExtensionCompatible,
+            ),
+            &PreparedSkillCatalog::new(
+                &catalogs.host.catalog,
+                SkillCatalogRenderPolicy::CoreCompatible,
+            ),
             filesystem_budget,
             self.include_usage,
         );

codex-rs/ext/skills/tests/skills_extension.rs

@@ -89,6 +89,9 @@ type TestResult = Result<(), Box<dyn std::error::Error>>;
 #[path = "skills_extension/shadow_task_context_tests.rs"]
 mod shadow_task_context_tests;
 
+#[path = "skills_extension/presentation_dedup_tests.rs"]
+mod presentation_dedup_tests;
+
 static NEXT_CODEX_HOME_ID: AtomicUsize = AtomicUsize::new(0);
 const SKILLS_INTRO_WITH_ABSOLUTE_PATHS: &str = "A skill is a set of instructions provided through a `SKILL.md` source. Below is the list of skills that can be used. Each entry includes a name, description, and source locator. `file` locators are on the host filesystem, `executor package` locators are owned by their execution environment, `cloud package` locators are opaque package identifiers, and `custom resource` locators use their provider's access mechanism.";
 const DEMO_SKILL_CONTENTS: &str =

codex-rs/ext/skills/tests/skills_extension/presentation_dedup_tests.rs

@@ -0,0 +1,239 @@
+//! Pre-budget deduplication preserves source packages and aliases across executor readiness changes.
+
+use super::*;
+use codex_exec_server::EnvironmentManager;
+use pretty_assertions::assert_eq;
+
+#[tokio::test]
+async fn cloud_preference_preserves_aliases_reads_and_executor_fallback() -> TestResult {
+    for (cloud_enabled, all_overlap) in [(true, false), (false, false), (true, true)] {
+        let description = format!("{} UNIQUE_DESCRIPTION_END", "Skill guidance. ".repeat(60));
+        let executor_roots = [
+            format!("skill://executor-a/{}", "a".repeat(180)),
+            format!("skill://executor-b/{}", "b".repeat(180)),
+        ];
+        let mut executor_entries = Vec::new();
+        for (plugin, root) in ["demo", "other"].into_iter().zip(&executor_roots) {
+            for index in 0..3 {
+                let package = format!("{root}/s{index}");
+                let mut entry = test_entry(
+                    SkillSourceKind::Executor,
+                    "skills",
+                    &package,
+                    &format!("{package}/SKILL.md"),
+                );
+                entry.name = format!("{plugin}:s{index}");
+                entry.description = description.clone();
+                executor_entries.push(entry.with_alias_root(root));
+            }
+        }
+        let mut cloud_entries = Vec::new();
+        for plugin in if all_overlap {
+            vec!["demo", "other"]
+        } else {
+            vec!["demo"]
+        } {
+            for index in 0..3 {
+                let mut entry = test_entry(
+                    SkillSourceKind::Cloud,
+                    "codex_apps",
+                    &format!("cloud/{plugin}/s{index}"),
+                    &format!("cloud/{plugin}/s{index}/SKILL.md"),
+                );
+                entry.name = format!("{plugin}:s{index}");
+                cloud_entries.push(entry);
+            }
+        }
+        let provider = |entries| {
+            Arc::new(StaticSkillProvider {
+                catalog: SkillCatalog {
+                    entries,
+                    warnings: Vec::new(),
+                },
+                read_requests: Arc::new(Mutex::new(Vec::new())),
+                list_calls: None,
+                fail_first_list: false,
+            })
+        };
+        let providers = SkillProviders::new()
+            .with_executor_provider(provider(executor_entries.clone()))
+            .with_cloud_provider(provider(cloud_entries));
+        let mut builder = ExtensionRegistryBuilder::new();
+        install_with_providers(&mut builder, providers, skills_extension_config);
+        let registry = builder.build();
+        let session_store = ExtensionData::new("session");
+        let thread_store = ExtensionData::new("thread");
+        let config = TestConfig {
+            cloud_skill_enabled: cloud_enabled,
+            ..default_config()
+        };
+        registry.thread_lifecycle_contributors()[0]
+            .on_thread_start(ThreadStartInput {
+                config: &config,
+                session_source: &SessionSource::Cli,
+                persistent_thread_state_available: true,
+                environments: &[],
+                mcp_resource_client: None,
+                extension_metrics: None,
+                session_store: &session_store,
+                thread_store: &thread_store,
+            })
+            .await;
+        let roots = [SelectedCapabilityRoot {
+            id: "skills".to_string(),
+            location: CapabilityRootLocation::Environment {
+                environment_id: "local".to_string(),
+                path: PathUri::parse("file:///skills")?,
+            },
+        }];
+        let resolved_roots = EnvironmentManager::default_for_tests()
+            .resolve_selected_capability_roots(&roots, &Default::default())
+            .await;
+        assert_eq!(resolved_roots.len(), 1);
+        let mut previous = serde_json::Map::new();
+        let mut initial_cloud_body = None;
+        let model_info = ModelInfo {
+            context_window: Some(70_000),
+            ..catalog_model_info()
+        };
+        for (stage, ready) in [
+            ("before", false),
+            ("connected", true),
+            ("disconnected", false),
+        ] {
+            start_registered_turn(&registry, &session_store, &thread_store, stage).await;
+            let turn_store = ExtensionData::new(stage);
+            let sections = registry.context_contributors()[0]
+                .contribute_world_state(WorldStateContributionInput {
+                    previous_world_state: Some(&previous),
+                    model_info: &model_info,
+                    thread_id: codex_protocol::ThreadId::new(),
+                    turn_id: stage,
+                    environments: &[],
+                    ready_selected_capability_roots: if ready { &roots } else { &[] },
+                    executor_capability_discovery: None,
+                    extension_metrics: None,
+                    session_store: &session_store,
+                    thread_store: &thread_store,
+                    turn_store: &turn_store,
+                })
+                .await;
+            let executor_body = world_state_section(&sections, "skills").snapshot()["body"]
+                .as_str()
+                .unwrap_or_default()
+                .to_string();
+            let cloud_body = world_state_section(&sections, "cloud_skills").snapshot()["body"]
+                .as_str()
+                .unwrap_or_default()
+                .to_string();
+            for index in 0..3 {
+                let listing = format!("- demo:s{index}:");
+                assert_eq!(executor_body.contains(&listing), ready && !cloud_enabled);
+                assert_eq!(cloud_body.contains(&listing), cloud_enabled);
+                assert_eq!(
+                    executor_body.contains(&format!("- other:s{index}:")),
+                    ready && !all_overlap
+                );
+            }
+            assert_eq!(
+                executor_body.matches("UNIQUE_DESCRIPTION_END").count(),
+                if ready && cloud_enabled && !all_overlap {
+                    3
+                } else {
+                    0
+                }
+            );
+            if stage == "before" {
+                initial_cloud_body = Some(cloud_body.clone());
+            } else {
+                assert_eq!(Some(&cloud_body), initial_cloud_body.as_ref());
+            }
+            if !ready || all_overlap {
+                assert!(executor_body.is_empty());
+            } else {
+                assert!(executor_body.contains("(executor package: e1/s0)"));
+                for (index, root) in executor_roots.iter().enumerate() {
+                    assert!(executor_body.contains(&format!("- `e{index}` = `{root}`")));
+                }
+            }
+            previous = sections
+                .iter()
+                .map(|section| (section.id().to_string(), section.snapshot().clone()))
+                .collect();
+            if !ready {
+                continue;
+            }
+            turn_store.insert(resolved_roots.clone());
+            let tools = registry.tool_contributors()[0].tools_for_step(
+                &session_store,
+                &thread_store,
+                &turn_store,
+            );
+            for (name, arguments, expected_resource) in [
+                (
+                    "list",
+                    serde_json::json!({"authority": {"kind": "executor"}}),
+                    None,
+                ),
+                (
+                    "read",
+                    serde_json::json!({"package": "e0/s0"}),
+                    Some(executor_entries[0].main_prompt.as_str()),
+                ),
+                (
+                    "read",
+                    serde_json::json!({"package": "e1/s0"}),
+                    Some(executor_entries[3].main_prompt.as_str()),
+                ),
+                (
+                    "read",
+                    serde_json::json!({"package": executor_entries[0].id.0}),
+                    Some(executor_entries[0].main_prompt.as_str()),
+                ),
+            ] {
+                let tool = tools
+                    .iter()
+                    .find(|tool| tool.tool_name().name == name)
+                    .ok_or("missing tool")?;
+                let payload = ToolPayload::Function {
+                    arguments: arguments.to_string(),
+                };
+                let output = tool
+                    .handle(ToolCall {
+                        turn_id: stage.to_string(),
+                        call_id: "call".to_string(),
+                        tool_name: tool.tool_name(),
+                        model: "test".to_string(),
+                        codex_turn_metadata: None,
+                        truncation_policy: TruncationPolicy::Bytes(20_000),
+                        source: ToolCallSource::Direct,
+                        conversation_history: ConversationHistory::default(),
+                        turn_item_emitter: Arc::new(NoopTurnItemEmitter),
+                        environments: Vec::new(),
+                        payload: payload.clone(),
+                    })
+                    .await?;
+                let response = output
+                    .post_tool_use_response("call", &payload)
+                    .ok_or("missing response")?;
+                if let Some(resource) = expected_resource {
+                    assert_eq!(response["resource"], resource);
+                } else {
+                    assert_eq!(
+                        response["skills"]
+                            .as_array()
+                            .ok_or("missing skills")?
+                            .iter()
+                            .map(|skill| skill["package"].as_str())
+                            .collect::<Vec<_>>(),
+                        executor_entries
+                            .iter()
+                            .map(|entry| Some(entry.id.0.as_str()))
+                            .collect::<Vec<_>>()
+                    );
+                }
+            }
+        }
+    }
+    Ok(())
+}