From a1890b69988ff76a254bf0897ade993aaa718470 Mon Sep 17 00:00:00 2001 From: rphilizaire-openai Date: Wed, 5 Aug 2026 20:08:21 +0000 Subject: [PATCH] Project orchestrator skills through world state (#37149) ## What changed - Move the orchestrator skill catalog from thread context into its own `orchestrator_skills` world-state section so unchanged catalogs remain incremental across turns. - Discover executor, orchestrator, and host catalogs together and render them against one context-window-aware metadata budget, preserving executor entries before orchestrator and host entries under pressure. - Emit catalog metrics and budget warnings when each world-state section is rendered. ## Testing - Cover incremental orchestrator catalogs across production turns, combined catalog budgeting, disabled and unavailable providers, caching, warnings, and metrics. GitOrigin-RevId: a0fb25d39dcaf43c978aff9da7972a1206f736e4 --- codex-rs/core/tests/suite/skills_extension.rs | 108 +++- codex-rs/ext/skills/src/extension.rs | 206 +------- codex-rs/ext/skills/src/lib.rs | 1 + codex-rs/ext/skills/src/render.rs | 198 ++++--- .../ext/skills/src/render_observability.rs | 2 + codex-rs/ext/skills/src/render_tests.rs | 30 +- codex-rs/ext/skills/src/state.rs | 29 +- codex-rs/ext/skills/src/tools/mod.rs | 6 +- codex-rs/ext/skills/src/world_state.rs | 189 +++---- .../ext/skills/src/world_state_catalogs.rs | 329 ++++++++++++ codex-rs/ext/skills/tests/skills_extension.rs | 497 +++++++++++++----- 11 files changed, 1101 insertions(+), 494 deletions(-) create mode 100644 codex-rs/ext/skills/src/world_state_catalogs.rs diff --git a/codex-rs/core/tests/suite/skills_extension.rs b/codex-rs/core/tests/suite/skills_extension.rs index 4eb9ab73b5..afa081ecbc 100644 --- a/codex-rs/core/tests/suite/skills_extension.rs +++ b/codex-rs/core/tests/suite/skills_extension.rs @@ -6,6 +6,7 @@ use codex_config::ConfigLayerSource; use codex_config::ConfigLayerStack; use codex_config::ConfigRequirements; use codex_config::ConfigRequirementsToml; +use codex_core::StartThreadOptions; use codex_core::config::Config; use codex_extension_api::ExtensionEventSink; use codex_extension_api::ExtensionRegistryBuilder; @@ -66,7 +67,7 @@ struct StaticSkillProvider { main_prompt_contents: Option, } -struct ExecutorSkillProvider { +struct CatalogSkillProvider { catalog: SkillCatalog, } @@ -128,7 +129,7 @@ impl SkillProvider for StaticSkillProvider { } } -impl SkillProvider for ExecutorSkillProvider { +impl SkillProvider for CatalogSkillProvider { fn list(&self, _query: SkillListQuery) -> SkillProviderFuture<'_, SkillCatalog> { Box::pin(async { Ok(self.catalog.clone()) }) } @@ -239,7 +240,7 @@ fn catalog_extensions( let mut extensions = ExtensionRegistryBuilder::::with_event_sink(Arc::new(ChannelEventSink(event_tx))); let mut providers = - SkillProviders::new().with_executor_provider(Arc::new(ExecutorSkillProvider { + SkillProviders::new().with_executor_provider(Arc::new(CatalogSkillProvider { catalog: executor_catalog, })); if include_host_provider { @@ -1110,6 +1111,107 @@ async fn production_turn_keeps_full_executor_only_catalog_when_it_fits() -> Resu Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn production_turn_keeps_orchestrator_world_state_incremental_across_turns() -> Result<()> { + let server = responses::start_mock_server().await; + let response = responses::mount_sse_sequence( + &server, + ["resp-1", "resp-2"] + .into_iter() + .map(|response_id| { + sse(vec![ + ev_response_created(response_id), + ev_completed(response_id), + ]) + }) + .collect(), + ) + .await; + let skill_name = "orchestrator-search"; + let skill_description = "Search available company knowledge."; + let skill_resource = "skill://codex_apps/orchestrator-search/SKILL.md"; + let catalog = SkillCatalog { + entries: vec![ + SkillCatalogEntry::new( + SkillPackageId("orchestrator/orchestrator-search".to_string()), + SkillAuthority::new(SkillSourceKind::Orchestrator, CODEX_APPS_MCP_SERVER_NAME), + skill_name, + skill_description, + SkillResourceId::new(skill_resource), + ) + .with_display_path(skill_resource), + ], + warnings: Vec::new(), + }; + let mut extensions = ExtensionRegistryBuilder::::new(); + install_with_providers( + &mut extensions, + SkillProviders::new() + .with_orchestrator_provider(Arc::new(CatalogSkillProvider { catalog })), + |config: &Config| SkillsExtensionConfig { + include_instructions: config.include_skill_instructions, + bundled_skills_enabled: false, + orchestrator_skills_enabled: config.orchestrator_skills_enabled, + shadow_selection_enabled: false, + }, + ); + let mut builder = test_codex() + .with_extensions(Arc::new(extensions.build())) + .with_config(|config| { + configure_catalog_test(config); + config.orchestrator_skills_enabled = true; + }); + let test = builder.build_with_auto_env(&server).await?; + let orchestrator_thread = test + .thread_manager + .start_thread(StartThreadOptions { + environments: Some(Vec::new()), + ..StartThreadOptions::new(test.config.clone()) + }) + .await?; + + for prompt in [ + "Inspect the available skills.", + "Inspect the available skills again.", + ] { + orchestrator_thread + .thread + .submit(Op::UserInput { + items: vec![UserInput::Text { + text: prompt.to_string(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + responsesapi_client_metadata: None, + additional_context: Default::default(), + thread_settings: Default::default(), + }) + .await?; + core_test_support::wait_for_event(&orchestrator_thread.thread, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + } + + let requests = response.requests(); + assert_eq!(requests.len(), 2); + let expected_line = + format!("- {skill_name}: {skill_description} (orchestrator resource: {skill_resource})"); + for (index, request) in requests.iter().enumerate() { + let developer_texts = request.message_input_texts("developer"); + let occurrences = developer_texts + .iter() + .map(|text| text.matches(&expected_line).count()) + .sum::(); + assert_eq!( + occurrences, 1, + "request {index} should contain the orchestrator catalog exactly once: {developer_texts:?}" + ); + } + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn production_turn_keeps_full_host_and_executor_catalogs_when_they_fit() -> Result<()> { let (developer_texts, _) = rendered_catalogs( diff --git a/codex-rs/ext/skills/src/extension.rs b/codex-rs/ext/skills/src/extension.rs index 335308fce0..fff087af4a 100644 --- a/codex-rs/ext/skills/src/extension.rs +++ b/codex-rs/ext/skills/src/extension.rs @@ -52,17 +52,14 @@ use crate::render::SkillCatalogRenderPolicy; use crate::render::SkillMetadataBudget; use crate::render::SkillRenderReport; use crate::render::render_available_skills; -use crate::render::render_combined_available_skills; use crate::render::skill_metadata_budget; use crate::render::truncate_main_prompt_contents; use crate::render::truncate_utf8_to_bytes; use crate::render_observability::CatalogSurface; use crate::render_observability::record_catalog_render; -use crate::render_observability::trace_catalog_budget_pressure; use crate::selection::collect_explicit_skill_mentions; use crate::shadow_selection_experiment::ShadowSelectionExperiment; use crate::sources::SkillProviders; -use crate::state::EmittedCatalogBudgetWarnings; use crate::state::ExecutorSkillsStepState; use crate::state::HostSkillsStepState; use crate::state::SkillsSessionState; @@ -71,9 +68,8 @@ use crate::state::SkillsTurnState; use crate::tools::SkillToolAuthority; use crate::tools::skill_tools; use crate::warnings::bounded_warnings; -use crate::world_state::HostSkillsWarningEmitter; -use crate::world_state::executor_skills_world_state_section; -use crate::world_state::host_skills_world_state_section; +use crate::world_state_catalogs::CatalogContext; +use crate::world_state_catalogs::CatalogStatus; struct SkillsExtension { providers: SkillProviders, @@ -206,7 +202,7 @@ where host_snapshot: None, include_host_skills: false, include_bundled_skills: config.bundled_skills_enabled, - include_orchestrator_skills: thread_state.orchestrator_skills_enabled(), + include_orchestrator_skills: false, mcp_resources: session_store .get::() .and_then(|state| state.mcp_resources.clone()), @@ -248,186 +244,19 @@ where input: WorldStateContributionInput<'a>, ) -> ExtensionFuture<'a, Vec> { Box::pin(async move { - let Some(thread_state) = input.thread_store.get::() else { + let Some(context) = + CatalogContext::new(&self.providers, Arc::clone(&self.event_sink), input) + else { return Vec::new(); }; - let config = thread_state.config(); - let catalog = thread_state - .executor_catalog_snapshot( - &self.providers, - SkillListQuery { - turn_id: input.turn_id.to_string(), - executor_roots: input.ready_selected_capability_roots.to_vec(), - resolved_executor_roots: Vec::new(), - host_snapshot: None, - include_host_skills: false, - include_bundled_skills: config.bundled_skills_enabled, - include_orchestrator_skills: false, - mcp_resources: input - .session_store - .get::() - .and_then(|state| state.mcp_resources.clone()), - executor_capability_discovery: input.executor_capability_discovery.cloned(), - }, - ) - .await; - input - .turn_store - .insert(ExecutorSkillsStepState(catalog.clone())); - let model_info = input.thread_store.get::(); - let include_usage = model_info - .as_deref() - .is_some_and(|model_info| model_info.include_skills_usage_instructions); - let context_window = model_info - .as_deref() - .and_then(ModelInfo::resolved_context_window); - let metadata_budget = skill_metadata_budget(context_window); - let extension_metrics = input.extension_metrics.clone(); - let host_snapshot = input.turn_store.get::(); - let needs_host_catalog = config.include_instructions || config.shadow_selection_enabled; - let host_catalog = if needs_host_catalog - && let Some(host_snapshot) = host_snapshot.as_ref() - && self.providers.has_host_provider() - { - Some( - self.providers - .list_host_for_turn(SkillListQuery { - turn_id: input.turn_id.to_string(), - executor_roots: Vec::new(), - resolved_executor_roots: Vec::new(), - host_snapshot: Some(Arc::clone(host_snapshot)), - include_host_skills: true, - include_bundled_skills: false, - include_orchestrator_skills: false, - mcp_resources: None, - executor_capability_discovery: None, - }) - .await, - ) - } else { - None - }; - if let Some(host_catalog) = host_catalog.as_ref() { - input - .turn_store - .insert(HostSkillsStepState(host_catalog.clone())); - } - let host_catalog_for_rendering = if config.include_instructions { - host_catalog.as_ref() - } else { - None - }; - let shared_rendered = host_catalog_for_rendering.and_then(|host_catalog| { - (host_catalog - .entries - .iter() - .any(SkillCatalogEntry::is_model_visible) - && catalog - .entries - .iter() - .any(SkillCatalogEntry::is_model_visible)) - .then(|| render_combined_available_skills(host_catalog, &catalog, metadata_budget)) - }); - let (rendered, rendered_host) = - if let Some((host_rendered, executor_rendered)) = shared_rendered { - if let Some(host_rendered) = host_rendered.as_ref() { - trace_catalog_budget_pressure(metadata_budget, &host_rendered.report); - } - ( - render_prepared_catalog( - extension_metrics.as_deref(), - CatalogSurface::ExecutorWorldState, - include_usage, - metadata_budget, - executor_rendered, - ), - host_rendered, - ) - } else { - let rendered_host = host_catalog_for_rendering.and_then(|host_catalog| { - render_available_skills( - host_catalog, - SkillCatalogRenderPolicy::CoreCompatible, - metadata_budget, - ) - }); - if let Some(rendered_host) = rendered_host.as_ref() { - trace_catalog_budget_pressure(metadata_budget, &rendered_host.report); - } - ( - if config.include_instructions { - render_catalog( - extension_metrics.as_deref(), - CatalogSurface::ExecutorWorldState, - &catalog, - include_usage, - SkillCatalogRenderPolicy::ExtensionCompatible, - metadata_budget, - ) - } else { - RenderedCatalog::default() - }, - rendered_host, - ) - }; - let host_warning_message = rendered_host - .as_ref() - .and_then(|rendered_host| rendered_host.report.warning_message()); - let host_render_metrics = config.include_instructions.then(|| { - rendered_host - .as_ref() - .map(|rendered_host| { - let report = &rendered_host.report; - ( - report.total_count, - report.included_count, - report.omitted_count, - report.truncated_description_chars, - ) - }) - .unwrap_or_default() - }); - let host_body = rendered_host - .and_then(|rendered_host| rendered_host.into_fragment(include_usage)) - .map(|fragment| fragment.body()); - let executor_body = rendered.fragment.map(|fragment| fragment.body()); - let mut sections = vec![executor_skills_world_state_section( - executor_body, - config.include_instructions, - )]; - let emitted_warnings = input - .turn_store - .get_or_init(EmittedCatalogBudgetWarnings::default); - if host_snapshot.is_some() && self.providers.has_host_provider() { - input.turn_store.insert(HostSkillsCatalogInWorldState); - let event_sink = Arc::clone(&self.event_sink); - let emitted_warnings = Arc::clone(&emitted_warnings); - let thread_id = input.thread_store.level_id().to_string(); - let turn_id = input.turn_id.to_string(); - let warning_emitter: HostSkillsWarningEmitter = Arc::new(move |message| { - if emitted_warnings.insert(&message) { - event_sink.emit_warning(ExtensionWarning { - thread_id: thread_id.clone(), - turn_id: Some(turn_id.clone()), - message, - }); - } - }); - sections.push(host_skills_world_state_section( - host_body, - config.include_instructions, - host_render_metrics, - extension_metrics, - host_warning_message, - warning_emitter, - )); - } - if let Some(message) = rendered.warning_message - && emitted_warnings.insert(&message) - { - self.emit_warning(input.thread_store.level_id(), Some(input.turn_id), message); - } - sections + let catalogs = context.discover_catalogs().await; + + context + .render_catalogs(catalogs) + .into_iter() + .filter(|catalog| catalog.status != CatalogStatus::Unavailable) + .map(|catalog| context.build_world_state_section(catalog)) + .collect() }) } } @@ -738,17 +567,12 @@ impl SkillsExtension { ) -> SkillCatalog { let include_orchestrator_skills = query.include_orchestrator_skills; let orchestrator_query = query.clone(); - let mcp_resources = orchestrator_query.mcp_resources.clone(); query.include_orchestrator_skills = false; let mut catalog = self.providers.list_for_turn(query).await; if include_orchestrator_skills { let orchestrator_catalog = thread_state - .orchestrator_catalog_snapshot( - mcp_resources.as_deref(), - self.providers - .list_orchestrator_for_turn(orchestrator_query), - ) + .orchestrator_catalog_snapshot(&self.providers, orchestrator_query) .await; catalog.extend(orchestrator_catalog); } diff --git a/codex-rs/ext/skills/src/lib.rs b/codex-rs/ext/skills/src/lib.rs index df740b4f9b..532bb33b40 100644 --- a/codex-rs/ext/skills/src/lib.rs +++ b/codex-rs/ext/skills/src/lib.rs @@ -19,6 +19,7 @@ mod state; mod tools; mod warnings; mod world_state; +mod world_state_catalogs; pub use config::SkillsExtensionConfig; pub use extension::install; diff --git a/codex-rs/ext/skills/src/render.rs b/codex-rs/ext/skills/src/render.rs index 05c41e45ee..7a2127a081 100644 --- a/codex-rs/ext/skills/src/render.rs +++ b/codex-rs/ext/skills/src/render.rs @@ -444,6 +444,13 @@ pub(crate) struct AvailableSkillsRender { pub(crate) report: SkillRenderReport, } +#[derive(Default)] +pub(crate) struct RenderedSkillCatalogs { + pub(crate) executor: Option, + pub(crate) orchestrator: Option, + pub(crate) host: Option, +} + impl AvailableSkillsRender { pub(crate) fn into_fragment( self, @@ -508,91 +515,143 @@ pub(crate) fn render_available_skills( } pub(crate) fn render_combined_available_skills( - host_catalog: &SkillCatalog, executor_catalog: &SkillCatalog, + orchestrator_catalog: &SkillCatalog, + host_catalog: &SkillCatalog, budget: SkillMetadataBudget, -) -> (Option, Option) { - let mut host_entries = host_catalog - .entries - .iter() - .filter(|entry| entry.is_model_visible()) - .collect::>(); +) -> RenderedSkillCatalogs { let mut executor_entries = executor_catalog .entries .iter() .filter(|entry| entry.is_model_visible()) .collect::>(); - SkillCatalogRenderPolicy::CoreCompatible.order_entries(&mut host_entries); + let mut orchestrator_entries = orchestrator_catalog + .entries + .iter() + .filter(|entry| entry.is_model_visible()) + .collect::>(); + let mut host_entries = host_catalog + .entries + .iter() + .filter(|entry| entry.is_model_visible()) + .collect::>(); SkillCatalogRenderPolicy::ExtensionCompatible.order_entries(&mut executor_entries); - if host_entries.is_empty() || executor_entries.is_empty() { - return ( - render_available_skills( - host_catalog, - SkillCatalogRenderPolicy::CoreCompatible, - budget, - ), - render_available_skills( + SkillCatalogRenderPolicy::ExtensionCompatible.order_entries(&mut orchestrator_entries); + SkillCatalogRenderPolicy::CoreCompatible.order_entries(&mut host_entries); + let nonempty_catalog_count = [ + !executor_entries.is_empty(), + !orchestrator_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, ), - ); + orchestrator: render_available_skills( + orchestrator_catalog, + SkillCatalogRenderPolicy::ExtensionCompatible, + budget, + ), + host: render_available_skills( + host_catalog, + SkillCatalogRenderPolicy::CoreCompatible, + budget, + ), + }; } let absolute = render_combined_lines( - host_entries - .iter() - .map(|entry| SkillLine::new(entry, SkillCatalogRenderPolicy::CoreCompatible)) - .collect(), executor_entries .iter() .map(|entry| SkillLine::new(entry, SkillCatalogRenderPolicy::ExtensionCompatible)) .collect(), + orchestrator_entries + .iter() + .map(|entry| SkillLine::new(entry, SkillCatalogRenderPolicy::ExtensionCompatible)) + .collect(), + host_entries + .iter() + .map(|entry| SkillLine::new(entry, SkillCatalogRenderPolicy::CoreCompatible)) + .collect(), budget, Vec::new(), ); let selected = if combined_catalog_fully_rendered(&absolute) { absolute - } else if let Some(aliased) = - build_aliased_combined_catalog(&host_entries, &executor_entries, budget) - && combined_render_is_better(&aliased, &absolute, budget) + } else if let Some(aliased) = build_aliased_combined_catalog( + &executor_entries, + &orchestrator_entries, + &host_entries, + budget, + ) && combined_render_is_better(&aliased, &absolute, budget) { aliased } else { absolute }; - (Some(selected.host), Some(selected.executor)) + RenderedSkillCatalogs { + executor: Some(selected.executor), + orchestrator: Some(selected.orchestrator), + host: Some(selected.host), + } } struct CombinedAvailableSkillsRender { - host: AvailableSkillsRender, executor: AvailableSkillsRender, + orchestrator: AvailableSkillsRender, + host: AvailableSkillsRender, } fn render_combined_lines( - host_lines: Vec>, executor_lines: Vec>, + orchestrator_lines: Vec>, + host_lines: Vec>, budget: SkillMetadataBudget, host_skill_root_lines: Vec, ) -> CombinedAvailableSkillsRender { - let executor_count = executor_lines.len(); + let executor_end = executor_lines.len(); + let orchestrator_end = executor_end.saturating_add(orchestrator_lines.len()); let mut lines = executor_lines; + lines.extend(orchestrator_lines); lines.extend(host_lines); let mut allocations = allocate_skill_lines(&lines, budget); - let executor_omission_marker = - reserve_executor_omission_marker(&lines, executor_count, budget, &mut allocations); + let omission_marker = reserve_non_host_omission_marker( + &lines, + executor_end, + orchestrator_end, + budget, + &mut allocations, + ); + let (executor_omission_marker, orchestrator_omission_marker) = + if executor_end == orchestrator_end { + (omission_marker, None) + } else { + (None, omission_marker) + }; CombinedAvailableSkillsRender { executor: render_combined_group( - &lines[..executor_count], - &allocations[..executor_count], + &lines[..executor_end], + &allocations[..executor_end], Vec::new(), executor_omission_marker, ), + orchestrator: render_combined_group( + &lines[executor_end..orchestrator_end], + &allocations[executor_end..orchestrator_end], + Vec::new(), + orchestrator_omission_marker, + ), host: render_combined_group( - &lines[executor_count..], - &allocations[executor_count..], + &lines[orchestrator_end..], + &allocations[orchestrator_end..], host_skill_root_lines, /*omission_marker*/ None, ), @@ -629,8 +688,9 @@ fn render_combined_group( } fn build_aliased_combined_catalog( - host_entries: &[&SkillCatalogEntry], executor_entries: &[&SkillCatalogEntry], + orchestrator_entries: &[&SkillCatalogEntry], + host_entries: &[&SkillCatalogEntry], budget: SkillMetadataBudget, ) -> Option { let plan = build_alias_plan(host_entries, budget)?; @@ -643,6 +703,14 @@ fn build_aliased_combined_catalog( SkillMetadataBudget::Tokens(_) => SkillMetadataBudget::Tokens(adjusted_limit), SkillMetadataBudget::Characters(_) => SkillMetadataBudget::Characters(adjusted_limit), }; + let executor_lines = executor_entries + .iter() + .map(|entry| SkillLine::new(entry, SkillCatalogRenderPolicy::ExtensionCompatible)) + .collect(); + let orchestrator_lines = orchestrator_entries + .iter() + .map(|entry| SkillLine::new(entry, SkillCatalogRenderPolicy::ExtensionCompatible)) + .collect(); let host_lines = host_entries .iter() .map(|entry| { @@ -653,23 +721,21 @@ fn build_aliased_combined_catalog( ) }) .collect(); - let executor_lines = executor_entries - .iter() - .map(|entry| SkillLine::new(entry, SkillCatalogRenderPolicy::ExtensionCompatible)) - .collect(); Some(render_combined_lines( - host_lines, executor_lines, + orchestrator_lines, + host_lines, adjusted_budget, plan.skill_root_lines, )) } fn combined_catalog_fully_rendered(rendered: &CombinedAvailableSkillsRender) -> bool { - rendered.host.report.omitted_count == 0 - && rendered.host.report.truncated_description_chars == 0 - && rendered.executor.report.omitted_count == 0 - && rendered.executor.report.truncated_description_chars == 0 + [&rendered.executor, &rendered.orchestrator, &rendered.host] + .into_iter() + .all(|catalog| { + catalog.report.omitted_count == 0 && catalog.report.truncated_description_chars == 0 + }) } fn combined_render_is_better( @@ -677,27 +743,23 @@ fn combined_render_is_better( current: &CombinedAvailableSkillsRender, budget: SkillMetadataBudget, ) -> bool { - if candidate.executor.report.included_count != current.executor.report.included_count { - return candidate.executor.report.included_count > current.executor.report.included_count; - } - - let included_count = |rendered: &CombinedAvailableSkillsRender| { - rendered - .host - .report - .included_count - .saturating_add(rendered.executor.report.included_count) + let priority = |rendered: &CombinedAvailableSkillsRender| { + ( + rendered.executor.report.included_count, + rendered.orchestrator.report.included_count, + rendered.host.report.included_count, + ) }; - if included_count(candidate) != included_count(current) { - return included_count(candidate) > included_count(current); + if priority(candidate) != priority(current) { + return priority(candidate) > priority(current); } let truncated_chars = |rendered: &CombinedAvailableSkillsRender| { - rendered - .host - .report - .truncated_description_chars - .saturating_add(rendered.executor.report.truncated_description_chars) + [&rendered.executor, &rendered.orchestrator, &rendered.host] + .into_iter() + .fold(0usize, |total, catalog| { + total.saturating_add(catalog.report.truncated_description_chars) + }) }; if truncated_chars(candidate) != truncated_chars(current) { return truncated_chars(candidate) < truncated_chars(current); @@ -711,7 +773,7 @@ fn combined_available_skills_cost( budget: SkillMetadataBudget, rendered: &CombinedAvailableSkillsRender, ) -> usize { - [&rendered.host, &rendered.executor] + [&rendered.executor, &rendered.orchestrator, &rendered.host] .into_iter() .fold(0usize, |used, catalog| { let root_cost = if !catalog.skill_root_lines.is_empty() { @@ -728,14 +790,15 @@ fn combined_available_skills_cost( }) } -fn reserve_executor_omission_marker( +fn reserve_non_host_omission_marker( skill_lines: &[SkillLine<'_>], - executor_count: usize, + executor_end: usize, + orchestrator_end: usize, budget: SkillMetadataBudget, allocations: &mut [SkillLineAllocation], ) -> Option { loop { - let omitted_count = allocations[..executor_count] + let omitted_count = allocations[..orchestrator_end] .iter() .filter(|allocation| matches!(allocation, SkillLineAllocation::Omitted)) .count(); @@ -749,9 +812,10 @@ fn reserve_executor_omission_marker( return Some(marker); } - let index = (executor_count..allocations.len()) + let index = (orchestrator_end..allocations.len()) .rev() - .chain((0..executor_count).rev()) + .chain((executor_end..orchestrator_end).rev()) + .chain((0..executor_end).rev()) .find(|index| { matches!( allocations[*index], diff --git a/codex-rs/ext/skills/src/render_observability.rs b/codex-rs/ext/skills/src/render_observability.rs index 01fddfd5d9..c9720155e8 100644 --- a/codex-rs/ext/skills/src/render_observability.rs +++ b/codex-rs/ext/skills/src/render_observability.rs @@ -11,6 +11,7 @@ use crate::render::SkillRenderReport; pub(crate) enum CatalogSurface { ThreadContext, ExecutorWorldState, + OrchestratorWorldState, HostWorldState, TurnInput, } @@ -20,6 +21,7 @@ impl CatalogSurface { match self { Self::ThreadContext => "thread_context", Self::ExecutorWorldState => "executor_world_state", + Self::OrchestratorWorldState => "orchestrator_world_state", Self::HostWorldState => "host_world_state", Self::TurnInput => "turn_input", } diff --git a/codex-rs/ext/skills/src/render_tests.rs b/codex-rs/ext/skills/src/render_tests.rs index 69f0e88843..fd14fb70cb 100644 --- a/codex-rs/ext/skills/src/render_tests.rs +++ b/codex-rs/ext/skills/src/render_tests.rs @@ -448,9 +448,10 @@ fn mixed_catalog_reserves_executor_omission_marker_by_omitting_host_first() { warnings: Vec::new(), }; - let (host, executor) = render_combined_available_skills( - &host_catalog, + let RenderedSkillCatalogs { host, executor, .. } = render_combined_available_skills( &executor_catalog, + &SkillCatalog::default(), + &host_catalog, SkillMetadataBudget::Tokens(28), ); let host = host.expect("host catalog should render"); @@ -520,14 +521,34 @@ fn mixed_catalog_prefers_executor_inclusion_over_total_aliased_inclusion() { ], warnings: Vec::new(), }; + let orchestrator_resource = "skill://orchestrator/one"; + let orchestrator_catalog = SkillCatalog { + entries: vec![ + SkillCatalogEntry::new( + SkillPackageId("o1".to_string()), + SkillAuthority::new(SkillSourceKind::Orchestrator, "codex_apps"), + "o1", + "", + SkillResourceId::new(orchestrator_resource), + ) + .with_display_path(orchestrator_resource), + ], + warnings: Vec::new(), + }; - let (host, executor) = render_combined_available_skills( - &host_catalog, + let RenderedSkillCatalogs { + host, + executor, + orchestrator, + } = render_combined_available_skills( &executor_catalog, + &orchestrator_catalog, + &host_catalog, SkillMetadataBudget::Tokens(74), ); let host = host.expect("host catalog should render"); let executor = executor.expect("executor catalog should render"); + let orchestrator = orchestrator.expect("orchestrator catalog should render"); assert_eq!( executor.report, @@ -549,6 +570,7 @@ fn mixed_catalog_prefers_executor_inclusion_over_total_aliased_inclusion() { truncated_description_count: 0, } ); + assert_eq!(orchestrator.report.total_count, 1); assert_eq!(host.skill_root_lines, Vec::::new()); } diff --git a/codex-rs/ext/skills/src/state.rs b/codex-rs/ext/skills/src/state.rs index d2ea009bf1..b741ba47b8 100644 --- a/codex-rs/ext/skills/src/state.rs +++ b/codex-rs/ext/skills/src/state.rs @@ -1,6 +1,5 @@ use std::collections::HashMap; use std::collections::HashSet; -use std::future::Future; use std::sync::Arc; use std::sync::Mutex; @@ -16,7 +15,6 @@ use crate::catalog::SkillAuthority; use crate::catalog::SkillCatalog; use crate::catalog::SkillCatalogEntry; use crate::catalog::SkillPackageId; -use crate::catalog::SkillProviderError; use crate::catalog::SkillProviderResult; use crate::catalog::SkillReadResult; use crate::catalog::SkillResourceId; @@ -175,18 +173,31 @@ impl SkillsThreadState { discovered } + #[tracing::instrument( + name = "skills.orchestrator.catalog_snapshot", + level = "info", + skip_all + )] pub(crate) async fn orchestrator_catalog_snapshot( &self, - mcp_resources: Option<&McpResourceClient>, - initialize: impl Future> + Send, + providers: &SkillProviders, + query: SkillListQuery, ) -> SkillCatalog { - self.orchestrator_cache(mcp_resources) + if !query.include_orchestrator_skills { + return SkillCatalog::default(); + } + + let cache = self.orchestrator_cache(query.mcp_resources.as_deref()); + cache .catalog .get_or_init(|| async { - initialize.await.unwrap_or_else(|err| SkillCatalog { - warnings: vec![err.message], - ..Default::default() - }) + providers + .list_orchestrator_for_turn(query) + .await + .unwrap_or_else(|err| SkillCatalog { + warnings: vec![err.message], + ..Default::default() + }) }) .await .clone() diff --git a/codex-rs/ext/skills/src/tools/mod.rs b/codex-rs/ext/skills/src/tools/mod.rs index a754af42ca..6fd56e503c 100644 --- a/codex-rs/ext/skills/src/tools/mod.rs +++ b/codex-rs/ext/skills/src/tools/mod.rs @@ -88,8 +88,8 @@ impl SkillToolContext { } self.thread_state .orchestrator_catalog_snapshot( - self.mcp_resources.as_deref(), - self.providers.list_orchestrator_for_turn(SkillListQuery { + &self.providers, + SkillListQuery { turn_id: turn_id.to_string(), executor_roots: Vec::new(), resolved_executor_roots: Vec::new(), @@ -99,7 +99,7 @@ impl SkillToolContext { include_orchestrator_skills: true, mcp_resources: self.mcp_resources.clone(), executor_capability_discovery: None, - }), + }, ) .await } diff --git a/codex-rs/ext/skills/src/world_state.rs b/codex-rs/ext/skills/src/world_state.rs index d34055707f..466b8e56e5 100644 --- a/codex-rs/ext/skills/src/world_state.rs +++ b/codex-rs/ext/skills/src/world_state.rs @@ -1,6 +1,3 @@ -use std::sync::Arc; - -use codex_extension_api::ExtensionMetrics; use codex_extension_api::PreviousWorldStateSection; use codex_extension_api::RenderedWorldStateFragment; use codex_extension_api::WorldStateSectionContribution; @@ -8,65 +5,113 @@ use codex_protocol::protocol::SKILLS_INSTRUCTIONS_CLOSE_TAG; use codex_protocol::protocol::SKILLS_INSTRUCTIONS_OPEN_TAG; use serde_json::json; -use crate::render_observability::CatalogSurface; -use crate::render_observability::record_catalog_metrics; +use crate::render::SkillRenderReport; pub(crate) const SKILLS_WORLD_STATE_ID: &str = "skills"; +pub(crate) const ORCHESTRATOR_SKILLS_WORLD_STATE_ID: &str = "orchestrator_skills"; pub(crate) const HOST_SKILLS_WORLD_STATE_ID: &str = "host_skills"; const NO_EXECUTOR_SKILLS_BODY: &str = "\n## Skills update\nNo selected-environment skills are currently available.\n"; const HIDDEN_EXECUTOR_SKILLS_BODY: &str = "\n## Skills update\nSelected-environment skills are not listed automatically. Explicit skill mentions can still be resolved when available.\n"; +const NO_ORCHESTRATOR_SKILLS_BODY: &str = + "\n## Orchestrator skills update\nNo orchestrator skills are currently available.\n"; +const HIDDEN_ORCHESTRATOR_SKILLS_BODY: &str = "\n## Orchestrator skills update\nOrchestrator skills are not listed automatically. Explicit skill mentions can still be resolved when available.\n"; const NO_HOST_SKILLS_BODY: &str = "\n## Host skills update\nNo host skills are currently available.\n"; const HIDDEN_HOST_SKILLS_BODY: &str = "\n## Host skills update\nHost skills are not listed automatically. Explicit skill mentions can still be resolved when available.\n"; const OMITTED_HOST_SKILLS_BODY: &str = "\n## Host skills update\nHost skills are available but omitted from the model-visible skills list because the skills context budget was exceeded.\n"; -pub(crate) type HostSkillsWarningEmitter = Arc; +pub(crate) type CatalogRenderCallback = Box; pub(crate) fn executor_skills_world_state_section( body: Option, include_instructions: bool, + on_render: CatalogRenderCallback, ) -> WorldStateSectionContribution { - let snapshot = json!({ + skills_world_state_section( + SKILLS_WORLD_STATE_ID, + body, + include_instructions, + /*enabled*/ None, + NO_EXECUTOR_SKILLS_BODY, + HIDDEN_EXECUTOR_SKILLS_BODY, + on_render, + ) + .with_legacy_matcher(|role, text| { + role == "developer" + && text.trim_start().starts_with(SKILLS_INSTRUCTIONS_OPEN_TAG) + && text.trim_end().ends_with(SKILLS_INSTRUCTIONS_CLOSE_TAG) + }) +} + +pub(crate) fn orchestrator_skills_world_state_section( + body: Option, + include_instructions: bool, + enabled: bool, + on_render: CatalogRenderCallback, +) -> WorldStateSectionContribution { + skills_world_state_section( + ORCHESTRATOR_SKILLS_WORLD_STATE_ID, + body, + include_instructions, + Some(enabled), + NO_ORCHESTRATOR_SKILLS_BODY, + if enabled { + HIDDEN_ORCHESTRATOR_SKILLS_BODY + } else { + NO_ORCHESTRATOR_SKILLS_BODY + }, + on_render, + ) +} + +fn skills_world_state_section( + id: &'static str, + body: Option, + include_instructions: bool, + enabled: Option, + no_skills_body: &'static str, + hidden_skills_body: &'static str, + on_render: CatalogRenderCallback, +) -> WorldStateSectionContribution { + let mut snapshot = json!({ "body": body, "includeInstructions": include_instructions, }); + if let Some(enabled) = enabled { + snapshot["enabled"] = json!(enabled); + } let retained_body = body.clone(); - let contribution = - WorldStateSectionContribution::new(SKILLS_WORLD_STATE_ID, snapshot, move |previous| { - let previous_is_absent = matches!(&previous, PreviousWorldStateSection::Absent); - if let PreviousWorldStateSection::Known(previous) = &previous { - let previous_body = previous.get("body").and_then(serde_json::Value::as_str); - let previous_include_instructions = previous - .get("includeInstructions") - .and_then(serde_json::Value::as_bool); - if previous_body == body.as_deref() - && previous_include_instructions == Some(include_instructions) - { - return None; - } + let contribution = WorldStateSectionContribution::new(id, snapshot, move |previous| { + if let PreviousWorldStateSection::Known(previous) = &previous { + let previous_body = previous.get("body").and_then(serde_json::Value::as_str); + let previous_include_instructions = previous + .get("includeInstructions") + .and_then(serde_json::Value::as_bool); + let previous_enabled = previous.get("enabled").and_then(serde_json::Value::as_bool); + if previous_body == body.as_deref() + && previous_include_instructions == Some(include_instructions) + && previous_enabled == enabled + { + return None; } + } - let body = match body.as_deref() { - Some(body) => Some(body), - None if previous_is_absent => None, - None if !include_instructions => Some(HIDDEN_EXECUTOR_SKILLS_BODY), - None => Some(NO_EXECUTOR_SKILLS_BODY), - }; - body.map(|body| { - RenderedWorldStateFragment::new( - "developer", - (SKILLS_INSTRUCTIONS_OPEN_TAG, SKILLS_INSTRUCTIONS_CLOSE_TAG), - body, - ) - }) - }) - .with_legacy_matcher(|role, text| { - role == "developer" - && text.trim_start().starts_with(SKILLS_INSTRUCTIONS_OPEN_TAG) - && text.trim_end().ends_with(SKILLS_INSTRUCTIONS_CLOSE_TAG) - }); + let body = match body.as_deref() { + Some(body) => body, + None if matches!(previous, PreviousWorldStateSection::Absent) => return None, + None if !include_instructions => hidden_skills_body, + None => no_skills_body, + }; + on_render(); + + Some(RenderedWorldStateFragment::new( + "developer", + (SKILLS_INSTRUCTIONS_OPEN_TAG, SKILLS_INSTRUCTIONS_CLOSE_TAG), + body, + )) + }); match retained_body { Some(body) => contribution.with_retained_fragment_matcher(move |role, text| { role == "developer" && text.contains(&body) @@ -78,68 +123,26 @@ pub(crate) fn executor_skills_world_state_section( pub(crate) fn host_skills_world_state_section( body: Option, include_instructions: bool, - render_metrics: Option<(usize, usize, usize, usize)>, - extension_metrics: Option>, - warning_message: Option, - warning_emitter: HostSkillsWarningEmitter, + report: &SkillRenderReport, + on_render: CatalogRenderCallback, ) -> WorldStateSectionContribution { let body = body.or_else(|| { - render_metrics - .is_some_and(|(total_count, included_count, omitted_count, _)| { - total_count > 0 && included_count == 0 && omitted_count > 0 - }) + (report.included_count == 0 && report.omitted_count > 0) .then(|| OMITTED_HOST_SKILLS_BODY.to_string()) }); - let snapshot = json!({ - "body": body, - "includeInstructions": include_instructions, - }); let retained_fragment = body .as_ref() .map(|body| format!("{SKILLS_INSTRUCTIONS_OPEN_TAG}{body}{SKILLS_INSTRUCTIONS_CLOSE_TAG}")); - let contribution = - WorldStateSectionContribution::new(HOST_SKILLS_WORLD_STATE_ID, snapshot, move |previous| { - let previous_is_absent = matches!(&previous, PreviousWorldStateSection::Absent); - if let PreviousWorldStateSection::Known(previous) = &previous { - let previous_body = previous.get("body").and_then(serde_json::Value::as_str); - let previous_include_instructions = previous - .get("includeInstructions") - .and_then(serde_json::Value::as_bool); - if previous_body == body.as_deref() - && previous_include_instructions == Some(include_instructions) - { - return None; - } - } - - if let Some((total_count, included_count, omitted_count, truncated_description_chars)) = - render_metrics - { - record_catalog_metrics( - extension_metrics.as_deref(), - CatalogSurface::HostWorldState, - total_count, - included_count, - omitted_count, - truncated_description_chars, - ); - } - let body = match body.as_deref() { - Some(body) => body, - None if previous_is_absent => return None, - None if !include_instructions => HIDDEN_HOST_SKILLS_BODY, - None => NO_HOST_SKILLS_BODY, - }; - if let Some(message) = warning_message.as_ref() { - warning_emitter(message.clone()); - } - Some(RenderedWorldStateFragment::new( - "developer", - (SKILLS_INSTRUCTIONS_OPEN_TAG, SKILLS_INSTRUCTIONS_CLOSE_TAG), - body, - )) - }); + let contribution = skills_world_state_section( + HOST_SKILLS_WORLD_STATE_ID, + body, + include_instructions, + /*enabled*/ None, + NO_HOST_SKILLS_BODY, + HIDDEN_HOST_SKILLS_BODY, + on_render, + ); match retained_fragment { Some(fragment) => contribution.with_retained_fragment_matcher(move |role, text| { role == "developer" && text.contains(&fragment) diff --git a/codex-rs/ext/skills/src/world_state_catalogs.rs b/codex-rs/ext/skills/src/world_state_catalogs.rs new file mode 100644 index 0000000000..c9f5572e51 --- /dev/null +++ b/codex-rs/ext/skills/src/world_state_catalogs.rs @@ -0,0 +1,329 @@ +use std::sync::Arc; + +use codex_core_skills::HostSkillsSnapshot; +use codex_core_skills::injection::HostSkillsCatalogInWorldState; +use codex_extension_api::ContextualUserFragment; +use codex_extension_api::ExtensionEventSink; +use codex_extension_api::ExtensionWarning; +use codex_extension_api::WorldStateContributionInput; +use codex_extension_api::WorldStateSectionContribution; +use codex_protocol::openai_models::ModelInfo; + +use crate::SkillsExtensionConfig; +use crate::catalog::SkillCatalog; +use crate::provider::SkillListQuery; +use crate::render::AvailableSkillsRender; +use crate::render::RenderedSkillCatalogs; +use crate::render::SkillMetadataBudget; +use crate::render::render_combined_available_skills; +use crate::render::skill_metadata_budget; +use crate::render_observability::CatalogSurface; +use crate::render_observability::record_catalog_render; +use crate::sources::SkillProviders; +use crate::state::EmittedCatalogBudgetWarnings; +use crate::state::ExecutorSkillsStepState; +use crate::state::HostSkillsStepState; +use crate::state::SkillsSessionState; +use crate::state::SkillsThreadState; +use crate::world_state::CatalogRenderCallback; +use crate::world_state::executor_skills_world_state_section; +use crate::world_state::host_skills_world_state_section; +use crate::world_state::orchestrator_skills_world_state_section; + +type CatalogWarningEmitter = Arc; + +#[derive(Clone, Copy, Eq, PartialEq)] +enum CatalogKind { + Executor, + Orchestrator, + Host, +} + +impl CatalogKind { + fn metrics_surface(self) -> CatalogSurface { + match self { + Self::Executor => CatalogSurface::ExecutorWorldState, + Self::Orchestrator => CatalogSurface::OrchestratorWorldState, + Self::Host => CatalogSurface::HostWorldState, + } + } +} + +#[derive(Clone, Copy, Eq, PartialEq)] +pub(crate) enum CatalogStatus { + Unavailable, + Disabled, + Enabled, +} + +struct CatalogContribution { + catalog: SkillCatalog, + status: CatalogStatus, +} + +impl CatalogContribution { + fn unavailable() -> Self { + Self { + catalog: SkillCatalog::default(), + status: CatalogStatus::Unavailable, + } + } +} + +pub(crate) struct CatalogContributions { + executor: CatalogContribution, + orchestrator: CatalogContribution, + host: CatalogContribution, +} + +pub(crate) struct RenderedCatalogContribution { + kind: CatalogKind, + pub(crate) status: CatalogStatus, + rendered: Option, +} + +pub(crate) struct CatalogContext<'a> { + providers: &'a SkillProviders, + input: WorldStateContributionInput<'a>, + thread_state: Arc, + config: SkillsExtensionConfig, + metadata_budget: SkillMetadataBudget, + include_usage: bool, + warning_emitter: CatalogWarningEmitter, +} + +impl<'a> CatalogContext<'a> { + pub(crate) fn new( + providers: &'a SkillProviders, + event_sink: Arc, + input: WorldStateContributionInput<'a>, + ) -> Option { + let thread_state = input.thread_store.get::()?; + let config = thread_state.config(); + let model_info = input.thread_store.get::(); + let include_usage = model_info + .as_deref() + .is_some_and(|model_info| model_info.include_skills_usage_instructions); + let context_window = model_info + .as_deref() + .and_then(ModelInfo::resolved_context_window); + let emitted_warnings = input + .turn_store + .get_or_init(EmittedCatalogBudgetWarnings::default); + let thread_id = input.thread_store.level_id().to_string(); + let turn_id = input.turn_id.to_string(); + let warning_emitter: CatalogWarningEmitter = Arc::new(move |message| { + if emitted_warnings.insert(&message) { + event_sink.emit_warning(ExtensionWarning { + thread_id: thread_id.clone(), + turn_id: Some(turn_id.clone()), + message, + }); + } + }); + + Some(Self { + providers, + input, + thread_state, + config, + metadata_budget: skill_metadata_budget(context_window), + include_usage, + warning_emitter, + }) + } + + pub(crate) async fn discover_catalogs(&self) -> CatalogContributions { + let orchestrator_enabled = self.thread_state.orchestrator_skills_enabled() + && self.providers.has_orchestrator_provider(); + let query = SkillListQuery { + turn_id: self.input.turn_id.to_string(), + executor_roots: self.input.ready_selected_capability_roots.to_vec(), + resolved_executor_roots: Vec::new(), + host_snapshot: None, + include_host_skills: false, + include_bundled_skills: self.config.bundled_skills_enabled, + include_orchestrator_skills: orchestrator_enabled, + mcp_resources: self + .input + .session_store + .get::() + .and_then(|state| state.mcp_resources.clone()), + executor_capability_discovery: self.input.executor_capability_discovery.cloned(), + }; + + let (executor, orchestrator, host) = futures::join!( + self.discover_executor_catalog(query.clone()), + self.discover_orchestrator_catalog(query), + self.discover_host_catalog(), + ); + + CatalogContributions { + executor, + orchestrator, + host, + } + } + + async fn discover_executor_catalog(&self, query: SkillListQuery) -> CatalogContribution { + let catalog = self + .thread_state + .executor_catalog_snapshot(self.providers, query) + .await; + self.input + .turn_store + .insert(ExecutorSkillsStepState(catalog.clone())); + + CatalogContribution { + catalog, + status: CatalogStatus::Enabled, + } + } + + async fn discover_orchestrator_catalog(&self, query: SkillListQuery) -> CatalogContribution { + if !self.providers.has_orchestrator_provider() { + return CatalogContribution::unavailable(); + } + + if !query.include_orchestrator_skills { + return CatalogContribution { + catalog: SkillCatalog::default(), + status: CatalogStatus::Disabled, + }; + } + + CatalogContribution { + catalog: self + .thread_state + .orchestrator_catalog_snapshot(self.providers, query) + .await, + status: CatalogStatus::Enabled, + } + } + + async fn discover_host_catalog(&self) -> CatalogContribution { + let Some(host_snapshot) = self + .input + .turn_store + .get::() + .filter(|_| self.providers.has_host_provider()) + else { + return CatalogContribution::unavailable(); + }; + + let needs_catalog = + self.config.include_instructions || self.config.shadow_selection_enabled; + let catalog = if needs_catalog { + let catalog = self + .providers + .list_host_for_turn(SkillListQuery { + turn_id: self.input.turn_id.to_string(), + executor_roots: Vec::new(), + resolved_executor_roots: Vec::new(), + host_snapshot: Some(host_snapshot), + include_host_skills: true, + include_bundled_skills: false, + include_orchestrator_skills: false, + mcp_resources: None, + executor_capability_discovery: None, + }) + .await; + self.input + .turn_store + .insert(HostSkillsStepState(catalog.clone())); + catalog + } else { + SkillCatalog::default() + }; + + CatalogContribution { + catalog, + status: CatalogStatus::Enabled, + } + } + + pub(crate) fn render_catalogs( + &self, + catalogs: CatalogContributions, + ) -> [RenderedCatalogContribution; 3] { + let rendered = if self.config.include_instructions { + render_combined_available_skills( + &catalogs.executor.catalog, + &catalogs.orchestrator.catalog, + &catalogs.host.catalog, + self.metadata_budget, + ) + } else { + RenderedSkillCatalogs::default() + }; + + [ + (CatalogKind::Executor, catalogs.executor, rendered.executor), + ( + CatalogKind::Orchestrator, + catalogs.orchestrator, + rendered.orchestrator, + ), + (CatalogKind::Host, catalogs.host, rendered.host), + ] + .map(|(kind, catalog, rendered)| RenderedCatalogContribution { + kind, + status: catalog.status, + rendered, + }) + } + + pub(crate) fn build_world_state_section( + &self, + catalog: RenderedCatalogContribution, + ) -> WorldStateSectionContribution { + let RenderedCatalogContribution { + kind, + status, + rendered, + } = catalog; + let report = rendered + .as_ref() + .map(|rendered| rendered.report.clone()) + .unwrap_or_default(); + let body = rendered + .and_then(|rendered| rendered.into_fragment(self.include_usage)) + .map(|fragment| fragment.body()); + let include_instructions = self.config.include_instructions; + let metrics = self.input.extension_metrics.clone(); + let warning_emitter = Arc::clone(&self.warning_emitter); + let metadata_budget = self.metadata_budget; + let render_report = report.clone(); + let on_render: CatalogRenderCallback = Box::new(move || { + if !include_instructions || status != CatalogStatus::Enabled { + return; + } + + record_catalog_render( + metrics.as_deref(), + kind.metrics_surface(), + metadata_budget, + &render_report, + ); + if let Some(message) = render_report.warning_message() { + warning_emitter(message); + } + }); + + match kind { + CatalogKind::Executor => { + executor_skills_world_state_section(body, include_instructions, on_render) + } + CatalogKind::Orchestrator => orchestrator_skills_world_state_section( + body, + include_instructions, + status == CatalogStatus::Enabled, + on_render, + ), + CatalogKind::Host => { + self.input.turn_store.insert(HostSkillsCatalogInWorldState); + host_skills_world_state_section(body, include_instructions, &report, on_render) + } + } + } +} diff --git a/codex-rs/ext/skills/tests/skills_extension.rs b/codex-rs/ext/skills/tests/skills_extension.rs index cc36f5aa81..db720ecebb 100644 --- a/codex-rs/ext/skills/tests/skills_extension.rs +++ b/codex-rs/ext/skills/tests/skills_extension.rs @@ -19,11 +19,13 @@ use codex_extension_api::ExtensionRegistryBuilder; use codex_extension_api::ExtensionWarning; use codex_extension_api::NoopTurnItemEmitter; use codex_extension_api::PreviousWorldStateSection; +use codex_extension_api::RenderedWorldStateFragment; use codex_extension_api::ThreadStartInput; use codex_extension_api::ToolCall; use codex_extension_api::ToolPayload; use codex_extension_api::TurnInputContext; use codex_extension_api::WorldStateContributionInput; +use codex_extension_api::WorldStateSectionContribution; use codex_models_manager::model_info::model_info_from_slug; use codex_otel::MetricsClient; use codex_otel::MetricsConfig; @@ -77,6 +79,53 @@ const SKILLS_INTRO_WITH_ABSOLUTE_PATHS: &str = "A skill is a set of instructions const DEMO_SKILL_CONTENTS: &str = "---\nname: demo\ndescription: Demo skill.\n---\n# Demo\n\nUse the demo skill.\n"; +fn world_state_section<'a>( + sections: &'a [WorldStateSectionContribution], + id: &str, +) -> &'a WorldStateSectionContribution { + sections + .iter() + .find(|section| section.id() == id) + .unwrap_or_else(|| panic!("world-state section {id} should exist")) +} + +async fn skill_world_state_fragments( + registry: &codex_extension_api::ExtensionRegistry, + session_store: &ExtensionData, + thread_store: &ExtensionData, + turn_id: &str, +) -> Result<(RenderedWorldStateFragment, RenderedWorldStateFragment), Box> { + let turn_store = ExtensionData::new(turn_id); + let selected_roots = [SelectedCapabilityRoot { + id: "skills".to_string(), + location: CapabilityRootLocation::Environment { + environment_id: "env-1".to_string(), + path: PathUri::parse("file:///skills")?, + }, + }]; + let sections = registry.context_contributors()[0] + .contribute_world_state(WorldStateContributionInput { + thread_id: codex_protocol::ThreadId::new(), + turn_id, + environments: &[], + ready_selected_capability_roots: &selected_roots, + executor_capability_discovery: None, + extension_metrics: None, + session_store, + thread_store, + turn_store: &turn_store, + }) + .await; + + let executor = world_state_section(§ions, "skills") + .render_diff(PreviousWorldStateSection::Absent) + .ok_or("executor skills should render through world state")?; + let orchestrator = world_state_section(§ions, "orchestrator_skills") + .render_diff(PreviousWorldStateSection::Absent) + .ok_or("orchestrator skills should render through world state")?; + Ok((executor, orchestrator)) +} + #[tokio::test] async fn installed_extension_uses_host_service_snapshot() -> TestResult { let codex_home = test_codex_home(); @@ -223,18 +272,14 @@ async fn host_world_state_records_catalog_metrics_on_publish_and_change() -> Tes .await; assert_eq!(sections.len(), 2); - let host_section = §ions[1]; + let host_section = world_state_section(§ions, "host_skills"); let published_snapshot = host_section.snapshot().clone(); assert!( host_section .render_diff(PreviousWorldStateSection::Absent) .is_some() ); - let mut expected = expected_catalog_metric_samples("executor_world_state", /*count*/ 0); - expected.extend(expected_catalog_metric_samples( - "host_world_state", - /*count*/ 1, - )); + let mut expected = expected_catalog_metric_samples("host_world_state", /*count*/ 1); assert!(startup_metrics.samples().is_empty()); assert_eq!(turn_metrics.samples(), expected); @@ -252,14 +297,10 @@ async fn host_world_state_records_catalog_metrics_on_publish_and_change() -> Tes }) .await; assert!( - sections[1] + world_state_section(§ions, "host_skills") .render_diff(PreviousWorldStateSection::Known(&published_snapshot)) .is_none() ); - expected.extend(expected_catalog_metric_samples( - "executor_world_state", - /*count*/ 0, - )); assert!(startup_metrics.samples().is_empty()); assert_eq!(turn_metrics.samples(), expected); @@ -292,14 +333,10 @@ async fn host_world_state_records_catalog_metrics_on_publish_and_change() -> Tes }) .await; assert!( - sections[1] + world_state_section(§ions, "host_skills") .render_diff(PreviousWorldStateSection::Known(&published_snapshot)) .is_some() ); - expected.extend(expected_catalog_metric_samples( - "executor_world_state", - /*count*/ 0, - )); expected.extend(expected_catalog_metric_samples( "host_world_state", /*count*/ 2, @@ -366,9 +403,10 @@ async fn persisted_host_snapshot_deduplicates_warning_after_reinitialization() - turn_store: &turn_store, }) .await; - let published_snapshot = sections[1].snapshot().clone(); + let host_section = world_state_section(§ions, "host_skills"); + let published_snapshot = host_section.snapshot().clone(); assert!( - sections[1] + host_section .render_diff(PreviousWorldStateSection::Absent) .is_some() ); @@ -407,7 +445,7 @@ async fn persisted_host_snapshot_deduplicates_warning_after_reinitialization() - }) .await; assert!( - sections[1] + world_state_section(§ions, "host_skills") .render_diff(PreviousWorldStateSection::Known(&published_snapshot)) .is_none() ); @@ -416,6 +454,127 @@ async fn persisted_host_snapshot_deduplicates_warning_after_reinitialization() - Ok(()) } +#[tokio::test] +async fn executor_orchestrator_and_host_share_catalog_world_state_flow() -> TestResult { + let provider = |entry: SkillCatalogEntry| { + Arc::new(StaticSkillProvider { + catalog: SkillCatalog { + entries: vec![entry], + 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(test_entry( + SkillSourceKind::Executor, + "env-1", + "executor/executor-skill", + "skill://executor/executor-skill/SKILL.md", + ))) + .with_orchestrator_provider(provider(test_entry( + SkillSourceKind::Orchestrator, + "codex_apps", + "orchestrator/orchestrator-skill", + "skill://orchestrator/orchestrator-skill/SKILL.md", + ))) + .with_host_provider(provider( + test_entry( + SkillSourceKind::Host, + "host", + "host/host-skill", + "/skills/host-skill/SKILL.md", + ) + .with_display_path("/skills/host-skill/SKILL.md"), + )); + 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"); + registry.thread_lifecycle_contributors()[0] + .on_thread_start(ThreadStartInput { + config: &default_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 selected_roots = [SelectedCapabilityRoot { + id: "skills".to_string(), + location: CapabilityRootLocation::Environment { + environment_id: "env-1".to_string(), + path: PathUri::parse("file:///skills")?, + }, + }]; + let turn_store = ExtensionData::new("turn-1"); + turn_store.insert(HostSkillsSnapshot::new(Arc::new( + SkillLoadOutcome::default(), + ))); + let metrics = Arc::new(RecordingMetrics::default()); + let sections = registry.context_contributors()[0] + .contribute_world_state(WorldStateContributionInput { + thread_id: codex_protocol::ThreadId::new(), + turn_id: "turn-1", + environments: &[], + ready_selected_capability_roots: &selected_roots, + executor_capability_discovery: None, + extension_metrics: Some(metrics.clone()), + session_store: &session_store, + thread_store: &thread_store, + turn_store: &turn_store, + }) + .await; + + assert_eq!( + sections + .iter() + .map(WorldStateSectionContribution::id) + .collect::>(), + vec!["skills", "orchestrator_skills", "host_skills"] + ); + assert!(metrics.samples().is_empty()); + + for (section_id, expected_line) in [ + ( + "skills", + "- executor-skill: Fix lint errors. (environment resource: skill://executor/executor-skill/SKILL.md)", + ), + ( + "orchestrator_skills", + "- orchestrator-skill: Fix lint errors. (orchestrator resource: skill://orchestrator/orchestrator-skill/SKILL.md)", + ), + ( + "host_skills", + "- host-skill: Fix lint errors. (file: /skills/host-skill/SKILL.md)", + ), + ] { + let fragment = world_state_section(§ions, section_id) + .render_diff(PreviousWorldStateSection::Absent) + .ok_or("skill catalog should render through world state")?; + assert!(fragment.body().contains(expected_line)); + } + + let expected_metrics = [ + "executor_world_state", + "orchestrator_world_state", + "host_world_state", + ] + .into_iter() + .flat_map(|surface| expected_catalog_metric_samples(surface, /*count*/ 1)) + .collect::>(); + assert_eq!(metrics.samples(), expected_metrics); + + Ok(()) +} + #[tokio::test] async fn nonempty_executor_empty_host_records_catalog_metrics() -> TestResult { let executor_provider = Arc::new(StaticSkillProvider { @@ -488,15 +647,11 @@ async fn nonempty_executor_empty_host_records_catalog_metrics() -> TestResult { .is_some() ); assert!( - sections[1] + world_state_section(§ions, "host_skills") .render_diff(PreviousWorldStateSection::Absent) .is_none() ); - let mut expected = expected_catalog_metric_samples("executor_world_state", /*count*/ 1); - expected.extend(expected_catalog_metric_samples( - "host_world_state", - /*count*/ 0, - )); + let expected = expected_catalog_metric_samples("executor_world_state", /*count*/ 1); assert_eq!(metrics.samples(), expected); Ok(()) } @@ -561,7 +716,7 @@ async fn host_world_state_uses_provider_catalog_with_core_compatible_rendering() turn_store: &turn_store, }) .await; - let host_fragment = sections[1] + let host_fragment = world_state_section(§ions, "host_skills") .render_diff(PreviousWorldStateSection::Absent) .ok_or("host provider catalog should render")?; @@ -658,7 +813,7 @@ async fn shadow_selection_uses_host_catalog_when_instructions_are_disabled() -> .await; assert!( - sections[1] + world_state_section(§ions, "host_skills") .render_diff(PreviousWorldStateSection::Absent) .is_none() ); @@ -890,17 +1045,33 @@ async fn selected_executor_catalog_follows_step_availability_and_reuses_its_cach #[tokio::test] async fn default_context_truncates_catalog_descriptions() -> TestResult { let description = "x".repeat(1_025); - let mut entry = test_entry( + let mut executor_entry = test_entry( + SkillSourceKind::Executor, + "env-1", + "executor/executor-long-description", + "skill://executor/executor-long-description/SKILL.md", + ); + executor_entry.description = description.clone(); + let mut orchestrator_entry = test_entry( SkillSourceKind::Orchestrator, "codex_apps", - "orchestrator/long-description", - "skill://orchestrator/long-description/SKILL.md", + "orchestrator/orchestrator-long-description", + "skill://orchestrator/orchestrator-long-description/SKILL.md", ); - entry.description = description.clone(); - let providers = - SkillProviders::new().with_orchestrator_provider(Arc::new(StaticSkillProvider { + orchestrator_entry.description = description.clone(); + let providers = SkillProviders::new() + .with_executor_provider(Arc::new(StaticSkillProvider { catalog: SkillCatalog { - entries: vec![entry], + entries: vec![executor_entry], + warnings: Vec::new(), + }, + read_requests: Arc::new(Mutex::new(Vec::new())), + list_calls: None, + fail_first_list: false, + })) + .with_orchestrator_provider(Arc::new(StaticSkillProvider { + catalog: SkillCatalog { + entries: vec![orchestrator_entry], warnings: Vec::new(), }, read_requests: Arc::new(Mutex::new(Vec::new())), @@ -927,14 +1098,19 @@ async fn default_context_truncates_catalog_descriptions() -> TestResult { }) .await; - let fragments = registry.context_contributors()[0] - .contribute_thread_context(&session_store, &thread_store) - .await; - assert_eq!(1, fragments.len()); - let rendered = fragments[0].text(); - assert!(rendered.contains(&("x".repeat(1_021) + "..."))); - assert!(!rendered.contains(&"x".repeat(1_024))); - assert!(!rendered.contains(&description)); + let (executor, orchestrator) = + skill_world_state_fragments(®istry, &session_store, &thread_store, "turn-1").await?; + assert!(executor.body().contains("- executor-long-description:")); + assert!( + orchestrator + .body() + .contains("- orchestrator-long-description:") + ); + for rendered in [executor.body(), orchestrator.body()] { + assert!(rendered.contains(&("x".repeat(1_021) + "..."))); + assert!(!rendered.contains(&"x".repeat(1_024))); + assert!(!rendered.contains(&description)); + } Ok(()) } @@ -942,9 +1118,22 @@ async fn default_context_truncates_catalog_descriptions() -> TestResult { #[tokio::test] async fn moderate_budget_pressure_keeps_every_catalog_entry() -> TestResult { let description = "x".repeat(1_025); - let entries = (0..10) + let executor_entries = (0..5) .map(|index| { - let package_id = format!("orchestrator/skill-{index:02}"); + let package_id = format!("executor/executor-skill-{index:02}"); + let mut entry = test_entry( + SkillSourceKind::Executor, + "env-1", + &package_id, + &format!("skill://{package_id}/SKILL.md"), + ); + entry.description = description.clone(); + entry + }) + .collect(); + let orchestrator_entries = (0..5) + .map(|index| { + let package_id = format!("orchestrator/orchestrator-skill-{index:02}"); let mut entry = test_entry( SkillSourceKind::Orchestrator, "codex_apps", @@ -955,10 +1144,19 @@ async fn moderate_budget_pressure_keeps_every_catalog_entry() -> TestResult { entry }) .collect(); - let providers = - SkillProviders::new().with_orchestrator_provider(Arc::new(StaticSkillProvider { + let providers = SkillProviders::new() + .with_executor_provider(Arc::new(StaticSkillProvider { catalog: SkillCatalog { - entries, + entries: executor_entries, + warnings: Vec::new(), + }, + read_requests: Arc::new(Mutex::new(Vec::new())), + list_calls: None, + fail_first_list: false, + })) + .with_orchestrator_provider(Arc::new(StaticSkillProvider { + catalog: SkillCatalog { + entries: orchestrator_entries, warnings: Vec::new(), }, read_requests: Arc::new(Mutex::new(Vec::new())), @@ -985,27 +1183,32 @@ async fn moderate_budget_pressure_keeps_every_catalog_entry() -> TestResult { }) .await; - let fragments = registry.context_contributors()[0] - .contribute_thread_context(&session_store, &thread_store) - .await; - assert_eq!(1, fragments.len()); - let rendered = fragments[0].text(); - let description_lengths = (0..10) - .map(|index| { - let package_id = format!("orchestrator/skill-{index:02}"); - let line_prefix = format!("- skill-{index:02}: "); - let line_suffix = format!(" (orchestrator resource: skill://{package_id}/SKILL.md)"); - rendered - .lines() - .find_map(|line| { - line.strip_prefix(&line_prefix) - .and_then(|line| line.strip_suffix(&line_suffix)) - }) - .unwrap_or_else(|| panic!("rendered catalog should include skill-{index:02}")) - .chars() - .count() - }) - .collect::>(); + let (executor, orchestrator) = + skill_world_state_fragments(®istry, &session_store, &thread_store, "turn-1").await?; + let description_lengths = [ + ("executor", "environment", executor.body()), + ("orchestrator", "orchestrator", orchestrator.body()), + ] + .into_iter() + .flat_map(|(source, resource_kind, rendered)| { + (0..5).map(move |index| (source, resource_kind, rendered, index)) + }) + .map(|(source, resource_kind, rendered, index)| { + let name = format!("{source}-skill-{index:02}"); + let package_id = format!("{source}/{name}"); + let line_prefix = format!("- {name}: "); + let line_suffix = format!(" ({resource_kind} resource: skill://{package_id}/SKILL.md)"); + rendered + .lines() + .find_map(|line| { + line.strip_prefix(&line_prefix) + .and_then(|line| line.strip_suffix(&line_suffix)) + }) + .unwrap_or_else(|| panic!("rendered catalog should include {name}")) + .chars() + .count() + }) + .collect::>(); let shortest_description = *description_lengths .iter() .min() @@ -1017,31 +1220,55 @@ async fn moderate_budget_pressure_keeps_every_catalog_entry() -> TestResult { assert!(shortest_description > 0); assert!(longest_description < 1_024); assert!(longest_description.abs_diff(shortest_description) <= 1); - assert!(!rendered.contains("additional skills omitted from this bounded skills list")); - assert!(!rendered.contains(&"x".repeat(1_021))); + for rendered in [executor.body(), orchestrator.body()] { + assert!(!rendered.contains("additional skills omitted from this bounded skills list")); + assert!(!rendered.contains(&"x".repeat(1_021))); + } Ok(()) } #[tokio::test] async fn extreme_budget_pressure_removes_descriptions_before_omitting_entries() -> TestResult { - let entries = (0..200) + let executor_entries = (0..40) .map(|index| { - let package_id = format!("orchestrator/skill-{index:03}"); + let package_id = format!("executor/executor-skill-{index:03}"); + let mut entry = test_entry( + SkillSourceKind::Executor, + "env-1", + &package_id, + &format!("skill://{package_id}/SKILL.md"), + ); + entry.description = format!("executor-description-{index:03}"); + entry + }) + .collect(); + let orchestrator_entries = (0..160) + .map(|index| { + let package_id = format!("orchestrator/orchestrator-skill-{index:03}"); let mut entry = test_entry( SkillSourceKind::Orchestrator, "codex_apps", &package_id, &format!("skill://{package_id}/SKILL.md"), ); - entry.description = format!("description-{index:03}"); + entry.description = format!("orchestrator-description-{index:03}"); entry }) .collect(); - let providers = - SkillProviders::new().with_orchestrator_provider(Arc::new(StaticSkillProvider { + let providers = SkillProviders::new() + .with_executor_provider(Arc::new(StaticSkillProvider { catalog: SkillCatalog { - entries, + entries: executor_entries, + warnings: Vec::new(), + }, + read_requests: Arc::new(Mutex::new(Vec::new())), + list_calls: None, + fail_first_list: false, + })) + .with_orchestrator_provider(Arc::new(StaticSkillProvider { + catalog: SkillCatalog { + entries: orchestrator_entries, warnings: Vec::new(), }, read_requests: Arc::new(Mutex::new(Vec::new())), @@ -1070,25 +1297,44 @@ async fn extreme_budget_pressure_removes_descriptions_before_omitting_entries() }) .await; - let fragments = registry.context_contributors()[0] - .contribute_thread_context(&session_store, &thread_store) - .await; - assert_eq!(1, fragments.len()); - let rendered = fragments[0].text(); - let included_count = rendered + let (executor, orchestrator) = + skill_world_state_fragments(®istry, &session_store, &thread_store, "turn-1").await?; + let included_executor_count = executor + .body() .lines() - .filter(|line| line.starts_with("- skill-")) + .filter(|line| line.starts_with("- executor-skill-")) .count(); - assert!(included_count > 0); - assert!(included_count < 200); - assert!(rendered.contains("- skill-000: (orchestrator resource:")); - assert!(!rendered.contains("- skill-199:")); - assert!(!rendered.contains("description-")); - assert!(rendered.contains("additional skills omitted from this bounded skills list")); - let omitted_count = 200 - included_count; + let included_orchestrator_count = orchestrator + .body() + .lines() + .filter(|line| line.starts_with("- orchestrator-skill-")) + .count(); + assert_eq!(40, included_executor_count); + assert!(included_orchestrator_count > 0); + assert!(included_orchestrator_count < 160); + assert!( + executor + .body() + .contains("- executor-skill-039: (environment resource:") + ); + assert!( + orchestrator + .body() + .contains("- orchestrator-skill-000: (orchestrator resource:") + ); + assert!(!orchestrator.body().contains("- orchestrator-skill-159:")); + for rendered in [executor.body(), orchestrator.body()] { + assert!(!rendered.contains("description-")); + } + assert!( + orchestrator + .body() + .contains("additional skills omitted from this bounded skills list") + ); + let omitted_count = 200 - included_executor_count - included_orchestrator_count; let warning = event_rx.try_recv()?.into_warning(); assert_eq!(warning.thread_id, "thread"); - assert_eq!(warning.turn_id, None); + assert_eq!(warning.turn_id.as_deref(), Some("turn-1")); assert_eq!( warning.message, format!( @@ -1232,13 +1478,7 @@ async fn orchestrator_catalog_snapshot_caches_failure() -> TestResult { .contribute_thread_context(&session_store, &thread_store) .await; assert!(initial_fragments.is_empty()); - let warning = event_rx.try_recv()?.into_warning(); - assert_eq!(warning.thread_id, thread_store.level_id()); - assert_eq!(warning.turn_id, None); - assert_eq!( - warning.message, - "orchestrator skills unavailable: temporary orchestrator failure" - ); + assert!(event_rx.try_recv().is_err()); for turn_id in ["turn-1", "turn-2"] { let fragments = registry.turn_input_contributors()[0] @@ -1378,7 +1618,7 @@ async fn root_qualified_locator_selects_only_the_matching_executor_skill() -> Te } #[tokio::test] -async fn model_context_window_scales_executor_catalog_but_not_thread_catalog() -> TestResult { +async fn model_context_window_scales_executor_and_orchestrator_catalogs() -> TestResult { let orchestrator_entries = (0..40) .map(|index| { test_entry( @@ -1446,13 +1686,7 @@ async fn model_context_window_scales_executor_catalog_but_not_thread_catalog() - let thread_fragments = registry.context_contributors()[0] .contribute_thread_context(&session_store, &thread_store) .await; - assert_eq!(1, thread_fragments.len()); - assert!(thread_fragments[0].text().contains("skill-39")); - assert!( - !thread_fragments[0] - .text() - .contains("additional skills omitted") - ); + assert!(thread_fragments.is_empty()); let selected_roots = vec![SelectedCapabilityRoot { id: "skills".to_string(), @@ -1489,27 +1723,42 @@ async fn model_context_window_scales_executor_catalog_but_not_thread_catalog() - turn_store: &turn_store, }) .await; - let warning = event_rx.try_recv()?.into_warning(); - assert_eq!(warning.thread_id, thread_store.level_id()); - assert_eq!(warning.turn_id.as_deref(), Some("turn-1")); - assert!( - warning - .message - .starts_with("Exceeded skills context budget.") - ); - assert!( - warning - .message - .ends_with("additional skills were not included in the model-visible skills list.") - ); - let snapshot = sections[0].snapshot().clone(); - let fragment = sections[0] + assert!(event_rx.try_recv().is_err()); + let executor_section = world_state_section(§ions, "skills"); + let snapshot = executor_section.snapshot().clone(); + let executor_fragment = executor_section .render_diff(PreviousWorldStateSection::Absent) .ok_or("bounded executor catalog should render")?; - assert!(fragment.body().contains("additional skills omitted")); - assert!(!fragment.body().contains("skill-39")); + assert!(!executor_fragment.body().contains("skill-39")); + let orchestrator_fragment = world_state_section(§ions, "orchestrator_skills") + .render_diff(PreviousWorldStateSection::Absent) + .ok_or("bounded orchestrator catalog should render")?; assert!( - sections[0] + orchestrator_fragment + .body() + .contains("additional skills omitted") + ); + let warnings = event_rx + .try_iter() + .map(CapturedExtensionEvent::into_warning) + .collect::>(); + assert_eq!(warnings.len(), 2); + for warning in warnings { + assert_eq!(warning.thread_id, thread_store.level_id()); + assert_eq!(warning.turn_id.as_deref(), Some("turn-1")); + assert!( + warning + .message + .starts_with("Exceeded skills context budget.") + ); + assert!( + warning + .message + .ends_with("additional skills were not included in the model-visible skills list.") + ); + } + assert!( + executor_section .render_diff(PreviousWorldStateSection::Known(&snapshot)) .is_none() );