openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-20 17:38:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Prepare skill catalogs before building step tools (#55357)

Browse Source 
## Why

Skill tools depended on World State rendering to initialize their catalogs and could return `skill catalog is not ready for this step` before rendering completed.

## What changed

- Add an asynchronous `ToolContributor::prepare_for_step` hook and await all contributors before constructing tools.
- Prepare immutable skill catalogs, aliases, plugin policy, and configuration for each step. Tools, invocation detection, and World State rendering consume the prepared state without relying on rendering to perform discovery.
- Apply configuration reloads to subsequent steps while preserving live authorization checks for skill reads.

## Testing

Add coverage for preparation completing before tool construction and World State rendering, skill tools working before rendering, catalogs remaining stable across steps, and configuration reloads during discovery.

GitOrigin-RevId: 1fb95c58c412488353f3bd4403824d6d1476d3d0
This commit is contained in:
Anthony Tafoya
2026-10-19 19:33:11 +00:00
committed by copyberry
parent 49df716ba9
commit 9b865ab31e
33 changed files with 1522 additions and 997 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/session/mod.rs
+21 -19
View File
@@ -3932,6 +3932,8 @@ impl Session {
.executor_capability_discovery_for_step_v2(&turn_context.config, &environments)
.await;
let v2_snapshots = executor_capability_discovery_v2.as_ref();
// V1 discovery is still retained on StepContext for executor Stop/Interrupt hooks.
// TODO: Prepare hook sources at capture time, including V2, instead.
let (executor_capability_discovery, discovery_recovered) = if v2_snapshots.is_some() {
(None, false)
} else {
@@ -3947,9 +3949,6 @@ impl Session {
}
let extension_data =
codex_extension_api::ExtensionData::new(turn_context.sub_id.clone());
if let Some(snapshots) = &executor_capability_discovery_v2 {
extension_data.insert(snapshots.clone());
}
if let Some(messages) = settings
.model_info
.model_messages
@@ -3959,10 +3958,6 @@ impl Session {
{
extension_data.insert(messages.clone());
}
extension_data.insert(selected_capability_roots.clone());
if let Some(discovery) = &executor_capability_discovery {
extension_data.insert(discovery.as_ref().clone());
}
let (mcp, prepared_recommendations) = tokio::join!(
// MCP refresh can be large; keep it off the sampling request's stack.
Box::pin(self.mcp_runtime_for_step(
@@ -3973,6 +3968,7 @@ impl Session {
)),
turn::prepare_tool_recommendations(self.as_ref(), turn_context.as_ref()),
);
extension_data.insert(selected_capability_roots);
// A step keeps the plugins from the environments it captured, even if shared MCP
// moves on to another environment. Its skill tools and the model use this same copy.
// Policy follows this step's MCP binding, which may have refreshed mid-turn.
@@ -3995,6 +3991,23 @@ impl Session {
&turn_context.disabled_plugin_ids,
)
.await;
for contributor in self.services.extensions.tool_contributors() {
contributor
.prepare_for_step(codex_extension_api::StepPreparationInput {
turn_id: &turn_context.sub_id,
ready_selected_capability_roots: &ready_selected_capability_roots,
executor_capability_discovery: executor_capability_discovery.as_deref(),
executor_capability_discovery_v2: executor_capability_discovery_v2
.as_deref(),
selected_plugins: &selected_plugins,
session_store: &self.services.session_extension_data,
thread_store: &self.services.thread_extension_data,
turn_store: turn_context.extension_data.as_ref(),
step_store: &extension_data,
})
.or_cancel(cancellation_token)
.await?;
}
extension_data.insert(selected_plugins);
let tool_router = turn::built_tools(
self.as_ref(),
@@ -4007,9 +4020,7 @@ impl Session {
)
.await?;
Ok::<_, CodexErr>((
selected_capability_roots,
executor_capability_discovery,
executor_capability_discovery_v2,
mcp,
tool_router,
extension_data,
@@ -4022,14 +4033,7 @@ impl Session {
if cancellation_token.is_cancelled() {
return Err(CodexErr::TurnAborted);
}
let (
selected_capability_roots,
executor_capability_discovery,
executor_capability_discovery_v2,
mcp,
tool_router,
extension_data,
) = prepared_tools??;
let (executor_capability_discovery, mcp, tool_router, extension_data) = prepared_tools??;
Ok(Arc::new(StepContext {
preempt: turn_context
.config
@@ -4043,9 +4047,7 @@ impl Session {
turn: turn_context,
extension_data,
environments,
selected_capability_roots,
executor_capability_discovery,
executor_capability_discovery_v2,
mcp,
tool_router,
loaded_agents_md,
codex-rs/core/src/session/step_context.rs
+1 -6
View File
@@ -11,8 +11,6 @@ use crate::session::turn_context::TurnContext;
use crate::session::turn_context::TurnEnvironment;
use crate::tools::router::ToolRouter;
use codex_exec_server::ExecutorCapabilityDiscoverySnapshot;
use codex_exec_server::ExecutorCapabilityDiscoveryV2Snapshot;
use codex_exec_server::ResolvedSelectedCapabilityRoot;
use codex_extension_api::ExtensionData;
use codex_file_system::EnvironmentAccess;
use codex_mcp::McpBinding;
@@ -35,13 +33,10 @@ pub(crate) struct StepContext {
/// Telemetry context tagged with this sampling request's model.
pub(crate) session_telemetry: SessionTelemetry,
pub(crate) environments: TurnEnvironmentSnapshot,
/// Capability roots bound to ready environments in this exact step.
pub(crate) selected_capability_roots: Vec<ResolvedSelectedCapabilityRoot>,
/// Executor-materialized capability files shared by MCP and skills in this exact step.
/// Legacy discovery retained for executor hook materialization in this exact step.
pub(crate) executor_capability_discovery: Option<Arc<ExecutorCapabilityDiscoverySnapshot>>,
/// Keeps the extension inputs used to build this step's tools for its World State as well.
pub(crate) extension_data: ExtensionData,
pub(crate) executor_capability_discovery_v2: Option<Vec<ExecutorCapabilityDiscoveryV2Snapshot>>,
/// The exact MCP connections, configuration, and catalog captured for this step.
pub(crate) mcp: Arc<McpBinding>,
/// The finalized tool plan advertised and executed for this exact sampling request.
codex-rs/core/src/session/tests.rs
-2
View File
@@ -290,10 +290,8 @@ impl StepContext {
},
turn: Arc::clone(&turn),
environments,
selected_capability_roots: Vec::new(),
executor_capability_discovery: None,
extension_data: codex_extension_api::ExtensionData::new(turn.sub_id.clone()),
executor_capability_discovery_v2: None,
mcp: Arc::new(codex_mcp::McpBinding::empty(mcp_config_for_test(
&turn.config,
))),
codex-rs/core/src/session/world_state.rs
+4 -13
View File
@@ -72,7 +72,10 @@ impl Session {
let model_info = settings.model_info.as_ref();
let model_messages = ResolvedModelMessages::from_model(model_info);
tracing::trace!(
selected_capability_root_count = step_context.selected_capability_roots.len(),
selected_capability_root_count = step_context
.extension_data
.get::<Vec<codex_exec_server::ResolvedSelectedCapabilityRoot>>()
.map_or(0, |roots| roots.len()),
"building step world state"
);
let render_current_instructions = || {
@@ -339,26 +342,14 @@ impl Session {
));
}
let environments = step_context.environments.to_selections();
let ready_selected_capability_roots = step_context
.selected_capability_roots
.iter()
.map(|root| root.selected_root().clone())
.collect::<Vec<_>>();
let previous_world_state = self.state.lock().await.history.world_state_checkpoint();
for contributor in self.services.extensions.context_contributors() {
for section in contributor
.contribute_world_state(WorldStateContributionInput {
executor_capability_discovery_v2: step_context
.executor_capability_discovery_v2
.as_deref(),
thread_id: self.thread_id(),
turn_id: turn_context.sub_id.as_str(),
model_info: &step_context.settings.model_info,
environments: &environments,
ready_selected_capability_roots: &ready_selected_capability_roots,
executor_capability_discovery: step_context
.executor_capability_discovery
.as_deref(),
extension_metrics: Some(Arc::clone(&extension_metrics)),
session_store: &self.services.session_extension_data,
thread_store: &self.services.thread_extension_data,
codex-rs/core/tests/suite/executor_discovery_v2.rs
+22 -12
View File
@@ -127,10 +127,21 @@ impl McpServerContributor<Config> for DiscoveryRecorder {
}
}
impl codex_extension_api::ToolContributor for DiscoveryRecorder {
fn prepare_for_step<'a>(
&'a self,
input: codex_extension_api::StepPreparationInput<'a>,
) -> ExtensionFuture<'a, ()> {
Box::pin(async move {
*self.selected_plugins.lock().await = Some(input.selected_plugins.clone());
})
}
}
impl ContextContributor for DiscoveryRecorder {
fn contribute_world_state<'a>(
&'a self,
input: WorldStateContributionInput<'a>,
_input: WorldStateContributionInput<'a>,
) -> ExtensionFuture<'a, Vec<WorldStateSectionContribution>> {
Box::pin(async move {
let snapshot_count = self.snapshots.lock().await.len();
@@ -138,9 +149,6 @@ impl ContextContributor for DiscoveryRecorder {
.lock()
.await
.push(snapshot_count);
if let Some(snapshot) = input.step_store.get::<SelectedPluginSnapshot>() {
*self.selected_plugins.lock().await = Some(snapshot.as_ref().clone());
}
Vec::new()
})
}
@@ -315,15 +323,17 @@ async fn v2_changed_inventory_refreshes_mcp_without_selection_changes() -> Resul
}
}
impl ContextContributor for InventoryRecorder {
fn contribute_world_state<'a>(
impl codex_extension_api::ToolContributor for InventoryRecorder {
fn prepare_for_step<'a>(
&'a self,
input: WorldStateContributionInput<'a>,
) -> ExtensionFuture<'a, Vec<WorldStateSectionContribution>> {
input: codex_extension_api::StepPreparationInput<'a>,
) -> ExtensionFuture<'a, ()> {
Box::pin(async move {
let inventory = input
let snapshots = input
.executor_capability_discovery_v2
.and_then(|snapshots| snapshots.first())
.expect("captured inventories");
let inventory = snapshots
.first()
.and_then(|snapshot| snapshot.result.as_ref().ok())
.expect("step V2 inventory")
.clone();
@@ -334,7 +344,6 @@ async fn v2_changed_inventory_refreshes_mcp_without_selection_changes() -> Resul
input
.thread_store
.insert(ExecutorCapabilityDiscoveryV2Cache::default());
Vec::new()
})
}
}
@@ -343,7 +352,7 @@ async fn v2_changed_inventory_refreshes_mcp_without_selection_changes() -> Resul
let model = responses::start_mock_server().await;
let recorder = Arc::new(InventoryRecorder::default());
let mut extensions = ExtensionRegistryBuilder::new();
extensions.prompt_contributor(recorder.clone());
extensions.tool_contributor(recorder.clone());
extensions.mcp_server_contributor(recorder.clone());
let test = test_codex()
.with_extensions(Arc::new(extensions.build()))
@@ -498,6 +507,7 @@ async fn check_acquisition(enabled: bool, supported: bool) -> Result<()> {
mcp_snapshots_per_step: Arc::clone(&mcp_snapshots_per_step),
});
extensions.prompt_contributor(recorder.clone());
extensions.tool_contributor(recorder.clone());
extensions.mcp_server_contributor(recorder);
codex_skills_extension::install_with_providers(
&mut extensions,
codex-rs/core/tests/suite/remote_env.rs
+10 -3
View File
@@ -189,14 +189,21 @@ impl ContextContributor for ReadyCapabilityRootsTestExtension {
&'a self,
input: WorldStateContributionInput<'a>,
) -> ExtensionFuture<'a, Vec<WorldStateSectionContribution>> {
let roots = input
.step_store
.get::<Vec<codex_exec_server::ResolvedSelectedCapabilityRoot>>()
.expect("captured capability roots should be present in step state");
let selected_roots = roots
.iter()
.map(|root| root.selected_root().clone())
.collect::<Vec<_>>();
if let Some(observed_roots) = &self.observed_roots {
observed_roots
.lock()
.expect("observed capability roots should not be poisoned")
.push(input.ready_selected_capability_roots.to_vec());
.push(selected_roots.clone());
}
let root_ids = input
.ready_selected_capability_roots
let root_ids = selected_roots
.iter()
.map(|root| root.id.clone())
.collect::<Vec<_>>();
codex-rs/core/tests/suite/startup_cancellation.rs
+118 -1
View File
@@ -1,4 +1,4 @@
//! Cancelling partially initialized sessions must release their persistent writers.
//! Startup and step preparation establish readiness and release state on cancellation.
use codex_protocol::protocol::TurnEnvironmentSelection;
use std::sync::Arc;
@@ -179,3 +179,120 @@ async fn failed_resume_releases_writer_before_returning() -> Result<()> {
resumed.thread.shutdown_and_wait().await?;
Ok(())
}
#[tokio::test]
async fn step_preparation_finishes_before_tools_and_world_state() -> Result<()> {
use codex_extension_api::ContextContributor;
use codex_extension_api::ExtensionData;
use codex_extension_api::ExtensionFuture;
use codex_extension_api::SelectedPluginSnapshot;
use codex_extension_api::StepPreparationInput;
use codex_extension_api::ToolCall;
use codex_extension_api::ToolContributor;
use codex_extension_api::ToolExecutor;
use codex_extension_api::WorldStateContributionInput;
use codex_extension_api::WorldStateSectionContribution;
use std::sync::Mutex;
use std::sync::atomic::AtomicUsize;
skip_if_no_network!(Ok(()));
struct Prepared(Mutex<[bool; 2]>);
struct Contributor {
index: usize,
constructed: AtomicUsize,
}
impl ToolContributor for Contributor {
fn prepare_for_step<'a>(
&'a self,
input: StepPreparationInput<'a>,
) -> ExtensionFuture<'a, ()> {
Box::pin(async move {
tokio::task::yield_now().await;
let prepared = input
.step_store
.get_or_init(|| Prepared(Mutex::new([false; 2])));
let mut flags = prepared.0.lock().unwrap();
assert!(!flags[self.index], "a step must be prepared only once");
flags[self.index] = true;
})
}
fn tools_for_step(
&self,
_session_store: &ExtensionData,
_thread_store: &ExtensionData,
step_store: &ExtensionData,
) -> Vec<Arc<dyn for<'call> ToolExecutor<ToolCall<'call>>>> {
assert!(step_store.get::<SelectedPluginSnapshot>().is_some());
assert_eq!(
*step_store
.get::<Prepared>()
.expect("prepared step")
.0
.lock()
.unwrap(),
[true; 2]
);
self.constructed.fetch_add(1, Ordering::SeqCst);
Vec::new()
}
}
impl ContextContributor for Contributor {
fn contribute_world_state<'a>(
&'a self,
input: WorldStateContributionInput<'a>,
) -> ExtensionFuture<'a, Vec<WorldStateSectionContribution>> {
Box::pin(async move {
assert_eq!(
*input
.step_store
.get::<Prepared>()
.expect("prepared step")
.0
.lock()
.unwrap(),
[true; 2]
);
Vec::new()
})
}
}
let first = Arc::new(Contributor {
index: 0,
constructed: AtomicUsize::new(0),
});
let second = Arc::new(Contributor {
index: 1,
constructed: AtomicUsize::new(0),
});
let mut extensions = ExtensionRegistryBuilder::<Config>::new();
for contributor in [&first, &second] {
extensions.tool_contributor(contributor.clone());
extensions.prompt_contributor(contributor.clone());
}
let server = start_mock_server().await;
let mock = core_test_support::responses::mount_sse_sequence(
&server,
vec![
core_test_support::responses::sse(vec![core_test_support::responses::ev_completed(
"first",
)]),
core_test_support::responses::sse(vec![core_test_support::responses::ev_completed(
"second",
)]),
],
)
.await;
let test = test_codex()
.with_extensions(Arc::new(extensions.build()))
.build_with_auto_env(&server)
.await?;
test.submit_turn("First turn.").await?;
test.submit_turn("Second turn.").await?;
assert_eq!(mock.requests().len(), 2);
assert!(first.constructed.load(Ordering::SeqCst) >= 2);
assert_eq!(
first.constructed.load(Ordering::SeqCst),
second.constructed.load(Ordering::SeqCst)
);
Ok(())
}
codex-rs/ext/extension-api/src/contributors.rs
+25
View File
@@ -4,6 +4,9 @@ use std::sync::Arc;
use codex_config::types::PluginsConfigToml;
use codex_context_fragments::ContextualUserFragment;
use codex_exec_server_protocol::ExecutorCapabilityDiscoverySnapshot;
use codex_exec_server_protocol::ExecutorCapabilityDiscoveryV2Snapshot;
use codex_protocol::capabilities::SelectedCapabilityRoot;
use codex_protocol::items::TurnItem;
use codex_protocol::protocol::TokenUsageInfo;
use codex_tools::ToolCall;
@@ -349,8 +352,30 @@ pub trait SkillInvocationContributor: Send + Sync {
}
}
/// Runtime inputs shared by preparation, tool construction, and later context consumers.
#[derive(Clone, Copy)]
pub struct StepPreparationInput<'a> {
pub turn_id: &'a str,
pub ready_selected_capability_roots: &'a [SelectedCapabilityRoot],
pub executor_capability_discovery: Option<&'a ExecutorCapabilityDiscoverySnapshot>,
pub executor_capability_discovery_v2: Option<&'a [ExecutorCapabilityDiscoveryV2Snapshot]>,
/// Captured plugin policy; an empty snapshot represents no selected plugins.
pub selected_plugins: &'a SelectedPluginSnapshot,
pub session_store: &'a ExtensionData,
pub thread_store: &'a ExtensionData,
pub turn_store: &'a ExtensionData,
/// Contains captured environment handles and permissions.
pub step_store: &'a ExtensionData,
}
/// Extension contribution that exposes native tools owned by a feature.
pub trait ToolContributor: Send + Sync {
/// Prepares step state once, before tools are built and without rendering.
/// Never publish partial state; discovery failures may report warnings.
fn prepare_for_step<'a>(&'a self, _input: StepPreparationInput<'a>) -> ExtensionFuture<'a, ()> {
Box::pin(async {})
}
/// Returns native tools bound to the supplied extension state.
/// Step-only contributors can leave this empty default unchanged.
fn tools(
codex-rs/ext/extension-api/src/contributors/world_state.rs
-9
View File
@@ -1,9 +1,6 @@
use std::sync::Arc;
use codex_exec_server_protocol::ExecutorCapabilityDiscoverySnapshot;
use codex_exec_server_protocol::ExecutorCapabilityDiscoveryV2Snapshot;
use codex_protocol::ThreadId;
use codex_protocol::capabilities::SelectedCapabilityRoot;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ContentItemMetadata;
use codex_protocol::openai_models::ModelInfo;
@@ -20,12 +17,6 @@ pub struct WorldStateContributionInput<'a> {
/// Resolved model metadata captured for this sampling step, retained across discovery.
pub model_info: &'a ModelInfo,
pub environments: &'a [TurnEnvironmentSelection],
/// Selected roots whose stable environments are ready in this sampling step.
pub ready_selected_capability_roots: &'a [SelectedCapabilityRoot],
/// Executor-materialized capability files shared by all consumers in this exact step.
pub executor_capability_discovery: Option<&'a ExecutorCapabilityDiscoverySnapshot>,
/// Opted-in executor inventories captured for this exact step.
pub executor_capability_discovery_v2: Option<&'a [ExecutorCapabilityDiscoveryV2Snapshot]>,
/// Metrics bound to the captured model for this sampling step.
pub extension_metrics: Option<Arc<dyn ExtensionMetrics>>,
pub session_store: &'a ExtensionData,
codex-rs/ext/extension-api/src/lib.rs
+1
View File
@@ -71,6 +71,7 @@ pub use contributors::SelectedPluginSnapshot;
pub use contributors::SkillInvocationContributor;
pub use contributors::SkillInvocationInput;
pub use contributors::SkillInvocationKind;
pub use contributors::StepPreparationInput;
pub use contributors::SynchronousApprovalReviewer;
pub use contributors::ThreadIdleCause;
pub use contributors::ThreadIdleInput;
codex-rs/ext/skills/src/aliases.rs
+1 -1
View File
@@ -9,7 +9,7 @@ use serde::Deserialize;
use serde::Serialize;
use crate::catalog::SkillSourceKind;
use crate::skills_extension_state::SkillToolAuthoritySelector;
use crate::step_state::SkillToolAuthoritySelector;
fn alias_prefix(source: &SkillSourceKind) -> Option<&'static str> {
match source {
codex-rs/ext/skills/src/skills_extension_state.rs → codex-rs/ext/skills/src/catalog_state.rs
+3 -116
View File
@@ -1,4 +1,5 @@
//! Stores published skill catalogs, executor discovery projections, and cached cloud resources.
//! Reusable provider catalog caches and Cloud resource contents.
//! Provider catalogs remain unfiltered until step preparation applies policy.
use std::collections::HashMap;
use std::sync::Arc;
@@ -11,26 +12,19 @@ use codex_mcp::McpResourceClientAuthKey;
use codex_mcp::McpResourceServerCacheKey;
use codex_protocol::capabilities::SelectedCapabilityRoot;
use crate::aliases::SkillCatalogAliases;
use crate::catalog::SkillAuthority;
use crate::catalog::SkillCatalog;
use crate::catalog::SkillPackageId;
use crate::catalog::SkillReadResult;
use crate::catalog::SkillResourceId;
use crate::catalog::SkillSourceKind;
use crate::catalog::merge_catalogs;
use crate::provider::SkillReadRequest;
use schemars::JsonSchema;
use serde::Deserialize;
use serde::Serialize;
use tokio::sync::OnceCell;
const MAX_CACHED_CLOUD_RESOURCES: usize = 100;
const MAX_CACHED_CLOUD_CONTENT_BYTES: usize = 8 * 1024 * 1024;
/// Reusable provider catalogs, before step-specific plugin policy and merging.
#[derive(Clone, Debug, Default)]
pub(crate) struct SkillsExtensionState {
pub(crate) struct SkillCatalogState {
// Roots-only discovery, reused within a turn.
pub(crate) executor_cache: Option<CachedExecutorCatalog>,
// DiscoveryV1.
@@ -226,110 +220,3 @@ impl std::fmt::Debug for CloudResourceCache {
.finish_non_exhaustive()
}
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub(crate) enum SkillToolAuthoritySelector {
Cloud,
Executor,
}
impl SkillToolAuthoritySelector {
pub(crate) fn matches(self, authority: &SkillAuthority) -> bool {
match self {
Self::Cloud => authority.kind == SkillSourceKind::Cloud,
Self::Executor => authority.kind == SkillSourceKind::Executor,
}
}
}
/// Filled once by World State preparation; retained by this step's list/read tools.
pub(crate) type SkillsStepState = OnceCell<Arc<SkillStepCatalog>>;
/// Policy-checked, enabled skill routes for one step. Hidden skills remain explicitly readable.
#[derive(Debug)]
pub(crate) struct SkillStepCatalog {
pub(super) catalog: SkillCatalog,
pub(crate) aliases: SkillCatalogAliases,
/// None means no Cloud catalog was captured for this step.
pub(super) cloud_warnings: Option<Vec<String>>,
pub(super) executor_warnings: Vec<String>,
}
impl SkillStepCatalog {
pub(crate) fn new(
cloud: Option<SkillCatalog>,
executor: SkillCatalog,
discovery_v2: bool,
aliases: SkillCatalogAliases,
) -> Self {
let cloud_skill_count = cloud.as_ref().map(|catalog| catalog.entries.len());
let executor_skill_count = executor.entries.len();
let cloud_warnings = cloud.as_ref().map(|catalog| catalog.warnings.clone());
let executor_warnings = executor.warnings.clone();
let mut catalog = if discovery_v2 {
merge_catalogs(cloud, executor)
} else {
let mut catalog = cloud.unwrap_or_default();
catalog.is_partial |= executor.is_partial;
catalog.entries.extend(executor.entries);
catalog.warnings.extend(executor.warnings);
catalog
};
catalog.entries.retain_mut(|entry| {
if !entry.enabled {
return false;
}
entry.sources.retain(|source| source.enabled);
true
});
tracing::debug!(
discovery_v2,
?cloud_skill_count,
executor_skill_count,
enabled_skill_count = catalog.entries.len(),
warning_count = catalog.warnings.len(),
"prepared step skill catalog"
);
Self {
catalog,
aliases,
cloud_warnings,
executor_warnings,
}
}
/// Selects authorities from the prepared catalog without consulting thread state.
pub(crate) fn select_from_catalog(
&self,
authorities: &[SkillToolAuthoritySelector],
) -> SkillCatalog {
let entries = self
.catalog
.entries
.iter()
.filter(|entry| {
authorities
.iter()
.any(|authority| authority.matches(&entry.authority))
})
.cloned()
.collect();
let mut catalog = SkillCatalog {
is_partial: self.catalog.is_partial,
entries,
warnings: Vec::new(),
};
if authorities.contains(&SkillToolAuthoritySelector::Cloud)
&& let Some(warnings) = &self.cloud_warnings
{
catalog.warnings.extend(warnings.iter().cloned());
}
if authorities.contains(&SkillToolAuthoritySelector::Executor) {
catalog
.warnings
.extend(self.executor_warnings.iter().cloned());
}
catalog
}
}
codex-rs/ext/skills/src/cloud_cache_tests.rs
+1 -1
View File
@@ -163,7 +163,7 @@ async fn late_discovery_cannot_publish_into_a_new_generation() {
assert!(late.is_err());
assert!(
state
.skills_extension_state
.catalog_state
.lock()
.unwrap()
.cloud_cache
codex-rs/ext/skills/src/extension.rs
+29 -41
View File
@@ -3,7 +3,6 @@ use std::sync::Arc;
use crate::HostSkillsSnapshot;
use crate::InjectedHostSkillPrompts;
use codex_analytics::InvocationType;
use codex_exec_server::ExecutorCapabilityDiscoveryV2Snapshot;
use codex_exec_server::LOCAL_ENVIRONMENT_ID;
use codex_extension_api::ConfigContributor;
use codex_extension_api::ContextContributor;
@@ -40,6 +39,7 @@ use crate::catalog::SkillSourceKind;
use crate::fragments::AvailableSkillsInstructions;
use crate::fragments::SkillInstructions;
use crate::fragments::SkillResourceAccess;
use crate::preparation::prepare_catalogs;
use crate::provider::HostSkillProvider;
use crate::provider::SkillListQuery;
use crate::provider::SkillReadContext;
@@ -58,20 +58,17 @@ use crate::render_observability::CatalogSurface;
use crate::render_observability::record_catalog_render;
use crate::selection::collect_explicit_skill_mentions;
use crate::shadow_selection_experiment::ShadowSelectionExperiment;
use crate::skills_extension_state::SkillToolAuthoritySelector;
use crate::skills_extension_state::SkillsStepState;
use crate::sources::SkillProviders;
use crate::state::HostSkillsCatalogInWorldState;
use crate::state::HostSkillsStepState;
use crate::state::SkillsSessionState;
use crate::state::SkillsThreadState;
use crate::step_state::SkillsStepState;
use crate::telemetry::SkillTelemetry;
use crate::tools::SkillAnalytics;
use crate::tools::SkillToolAuthority;
use crate::tools::skill_tools;
use crate::warnings::bounded_warnings;
use crate::world_state_catalogs::CatalogContext;
use crate::world_state_catalogs::CatalogStatus;
use crate::world_state_catalogs::render_world_state_sections;
#[path = "cloud_skill.rs"]
mod cloud_skill;
@@ -253,21 +250,7 @@ where
&'a self,
input: WorldStateContributionInput<'a>,
) -> ExtensionFuture<'a, Vec<WorldStateSectionContribution>> {
Box::pin(async move {
let Some(context) =
CatalogContext::new(&self.providers, Arc::clone(&self.event_sink), input)
else {
return Vec::new();
};
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()
})
Box::pin(async move { render_world_state_sections(Arc::clone(&self.event_sink), input) })
}
}
@@ -275,6 +258,21 @@ impl<C> ToolContributor for SkillsExtension<C>
where
C: Send + Sync + 'static,
{
fn prepare_for_step<'a>(
&'a self,
input: codex_extension_api::StepPreparationInput<'a>,
) -> ExtensionFuture<'a, ()> {
Box::pin(async move {
if input.step_store.get::<SkillsStepState>().is_some() {
return;
}
if let Some(thread_state) = input.thread_store.get::<SkillsThreadState>() {
let state = prepare_catalogs(&thread_state, &self.providers, &input).await;
input.step_store.insert(state);
}
})
}
fn tools_for_step(
&self,
session_store: &ExtensionData,
@@ -336,14 +334,17 @@ where
return Vec::new();
};
let config = thread_state.config();
let mcp_resources = session_store
.get::<SkillsSessionState>()
.and_then(|state| state.mcp_resources.clone());
let host_snapshot = turn_store.get::<HostSkillsSnapshot>();
let host_catalog_in_world_state =
turn_store.get::<HostSkillsCatalogInWorldState>().is_some();
let host_skills = turn_store.get::<HostSkillsStepState>();
let Some(step_state) = input.step_store.get::<SkillsStepState>() else {
return Vec::new();
};
let config = &step_state.config;
let host_skills = step_state.host_catalog.as_ref();
let query = SkillListQuery {
turn_id: input.turn_id.clone(),
executor_roots: Vec::new(),
@@ -357,23 +358,10 @@ where
executor_capability_discovery: None,
executor_capability_discovery_v2: None,
};
let step_state = input.step_store.get::<SkillsStepState>();
let step_catalog = step_state.as_ref().and_then(|state| state.get());
let mut catalog = match &step_catalog {
Some(step_catalog) => step_catalog.select_from_catalog(&[
SkillToolAuthoritySelector::Cloud,
SkillToolAuthoritySelector::Executor,
]),
// Preserve legacy callers that select Cloud skills before World State preparation.
None if input
.step_store
.get::<Vec<ExecutorCapabilityDiscoveryV2Snapshot>>()
.is_none() =>
{
thread_state.available_cloud_catalog().unwrap_or_default()
}
None => SkillCatalog::default(),
};
let mut catalog = step_state.catalog.merged.clone();
if !host_catalog_in_world_state && let Some(host_skills) = host_skills {
catalog.extend(host_skills.clone());
}
catalog.extend(self.providers.list_for_turn(query).await);
for warning in bounded_warnings(&catalog.warnings) {
self.emit_warning(thread_store.level_id(), Some(&input.turn_id), warning);
@@ -383,7 +371,7 @@ where
let shadow_selection_turn = if config.shadow_selection_enabled {
let mut shadow_catalog = catalog.clone();
if let Some(host_skills) = host_skills {
shadow_catalog.extend(host_skills.0.clone());
shadow_catalog.extend(host_skills.clone());
}
let shadow_selected_entries =
collect_explicit_skill_mentions(&input.user_input, &shadow_catalog);
codex-rs/ext/skills/src/invocation.rs
+3 -3
View File
@@ -71,8 +71,8 @@ pub fn detect_implicit_skill_invocation(
if accesses.is_empty() {
return None;
}
let step_state = step_store.get::<crate::skills_extension_state::SkillsStepState>()?;
let step_catalog = step_state.get()?;
let step_state = step_store.get::<crate::step_state::SkillsStepState>()?;
let step_catalog = &step_state.catalog;
for access in accesses {
let matches_route = |authority: &SkillAuthority, main_prompt: &SkillResourceId| {
if authority.kind != SkillSourceKind::Executor
@@ -91,7 +91,7 @@ pub fn detect_implicit_skill_invocation(
.is_some_and(|scripts_dir| path.starts_with(&scripts_dir)),
}
};
for entry in step_catalog.catalog.entries.iter().filter(|entry| {
for entry in step_catalog.merged.entries.iter().filter(|entry| {
matches!(
entry.authority.kind,
SkillSourceKind::Cloud | SkillSourceKind::Executor
codex-rs/ext/skills/src/invocation_tests.rs
+16 -6
View File
@@ -1,5 +1,4 @@
use pretty_assertions::assert_eq;
use std::sync::Arc;
use super::*;
use crate::catalog::SkillAuthority;
@@ -31,16 +30,27 @@ fn detect_executor_skill(command: &str, environments: &[&str]) -> Option<SkillIn
warnings: Vec::new(),
};
let turn_store = ExtensionData::new("test-turn");
let step_catalog = Arc::new(crate::skills_extension_state::SkillStepCatalog::new(
let step_catalog = crate::step_state::SkillStepCatalog::new(
/*cloud*/ None,
catalog,
/*discovery_v2*/ false,
Default::default(),
));
);
let step_store = ExtensionData::new("step");
step_store.insert(crate::skills_extension_state::SkillsStepState::new_with(
Some(step_catalog),
));
step_store.insert(crate::step_state::SkillsStepState {
discovery_v2: false,
tools_enabled: true,
catalog: step_catalog,
host_catalog: None,
cloud_status: crate::step_state::CatalogStatus::Unavailable,
config: crate::SkillsExtensionConfig {
include_instructions: true,
max_context_tokens: None,
bundled_skills_enabled: false,
cloud_skill_enabled: false,
shadow_selection_enabled: false,
},
});
detect_implicit_skill_invocation(
&turn_store,
codex-rs/ext/skills/src/lib.rs
+3 -2
View File
@@ -1,6 +1,7 @@
mod aliases;
pub mod catalog;
mod catalog_prompt;
mod catalog_state;
mod config;
mod dynamic_skill_selector;
mod extension;
@@ -13,15 +14,16 @@ mod host_service;
mod host_snapshot;
mod invocation;
mod loader;
mod preparation;
pub mod provider;
mod render;
mod render_observability;
mod required;
mod selection;
mod shadow_selection_experiment;
mod skills_extension_state;
mod sources;
mod state;
mod step_state;
mod telemetry;
mod tools;
mod warnings;
@@ -46,7 +48,6 @@ pub use provider::HostSkillProvider;
pub use provider::SkillProvider;
pub use required::EnvironmentSkillRequirements;
pub use required::validate_required_skills;
pub(crate) use skills_extension_state::SkillsExtensionState;
pub use sources::SkillProviderSource;
pub use sources::SkillProviders;
pub use state::SkillsThreadState;
codex-rs/ext/skills/src/preparation.rs
+137
View File
@@ -0,0 +1,137 @@
//! Resolves step policy once and prepares immutable catalogs before tool construction.
//! Cache operations and live read-time authorization remain on the thread state.
use crate::HostSkillsSnapshot;
use crate::SkillsExtensionConfig;
use crate::catalog::SkillCatalog;
use crate::catalog::apply_plugin_denials;
use crate::provider::SkillListQuery;
use crate::provider::attribute_executor_plugins;
use crate::sources::SkillProviders;
use crate::state::SkillsSessionState;
use crate::state::SkillsThreadState;
use crate::step_state::CatalogStatus;
use crate::step_state::SkillsStepState;
use codex_exec_server::ResolvedSelectedCapabilityRoot;
use codex_extension_api::StepPreparationInput;
pub(crate) async fn prepare_catalogs(
thread_state: &SkillsThreadState,
providers: &SkillProviders,
input: &StepPreparationInput<'_>,
) -> SkillsStepState {
let config = thread_state.config();
let cloud_enabled = config.cloud_skill_enabled && thread_state.cloud_skills_available();
let discovery_v2 = input.executor_capability_discovery_v2.is_some();
let mut cloud = cloud_enabled
.then(|| thread_state.available_cloud_catalog())
.flatten();
let (mut executor, host_catalog) = futures::join!(
discover_executor_catalog(thread_state, providers, input, &config, cloud_enabled),
discover_host_catalog(providers, input, &config),
);
if discovery_v2 {
apply_plugin_denials(&mut executor, input.selected_plugins);
if let Some(cloud) = &mut cloud {
apply_plugin_denials(cloud, input.selected_plugins);
}
} else {
attribute_executor_plugins(&mut executor, input.selected_plugins);
}
let cloud_status = if !providers.has_cloud_provider() {
CatalogStatus::Unavailable
} else if cloud_enabled {
CatalogStatus::Enabled
} else {
CatalogStatus::Disabled
};
SkillsStepState {
discovery_v2,
tools_enabled: cloud_enabled
|| input
.step_store
.get::<Vec<ResolvedSelectedCapabilityRoot>>()
.map_or_else(
|| !input.ready_selected_capability_roots.is_empty(),
|roots| !roots.is_empty(),
)
|| discovery_v2,
catalog: thread_state.prepare_step_catalog(cloud, executor, discovery_v2),
host_catalog,
cloud_status,
config,
}
}
async fn discover_executor_catalog(
thread_state: &SkillsThreadState,
providers: &SkillProviders,
input: &StepPreparationInput<'_>,
config: &SkillsExtensionConfig,
cloud_enabled: bool,
) -> SkillCatalog {
// Core supplies captured roots even when none are ready. Descriptor-only callers
// retain their unbound discovery path.
let captured_roots = input
.step_store
.get::<Vec<ResolvedSelectedCapabilityRoot>>();
let executor_roots = captured_roots
.as_ref()
.map(|roots| {
roots
.iter()
.map(|root| root.selected_root().clone())
.collect()
})
.unwrap_or_else(|| input.ready_selected_capability_roots.to_vec());
let query = SkillListQuery {
turn_id: input.turn_id.to_string(),
executor_roots,
resolved_executor_roots: captured_roots
.map(|roots| roots.as_ref().clone())
.unwrap_or_default(),
host_snapshot: None,
include_host_skills: false,
include_bundled_skills: config.bundled_skills_enabled,
include_cloud_skills: cloud_enabled,
mcp_resources: input
.session_store
.get::<SkillsSessionState>()
.and_then(|state| state.mcp_resources.clone()),
executor_capability_discovery: input.executor_capability_discovery.cloned(),
executor_capability_discovery_v2: input.executor_capability_discovery_v2.map(Vec::from),
};
thread_state
.refresh_executor_catalog(providers, query)
.await
}
async fn discover_host_catalog(
providers: &SkillProviders,
input: &StepPreparationInput<'_>,
config: &SkillsExtensionConfig,
) -> Option<SkillCatalog> {
let host_snapshot = input
.turn_store
.get::<HostSkillsSnapshot>()
.filter(|_| providers.has_host_provider())?;
if !config.include_instructions && !config.shadow_selection_enabled {
return Some(SkillCatalog::default());
}
Some(
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(host_snapshot),
include_host_skills: true,
include_bundled_skills: false,
include_cloud_skills: false,
mcp_resources: None,
executor_capability_discovery: None,
executor_capability_discovery_v2: None,
})
.await,
)
}
codex-rs/ext/skills/src/required.rs
+2 -6
View File
@@ -2,7 +2,7 @@
use codex_extension_api::ExtensionData;
use crate::skills_extension_state::SkillsStepState;
use crate::step_state::SkillsStepState;
/// Skill names that must be supplied by one selected environment before inference.
pub struct EnvironmentSkillRequirements<'a> {
@@ -23,17 +23,13 @@ pub fn validate_required_skills<'a>(
"Required skills cannot be validated because the skills extension is unavailable"
.to_string()
})?;
let step_catalog = step_state.get().ok_or_else(|| {
"Required skills cannot be validated because the step skill catalog is unavailable"
.to_string()
})?;
for EnvironmentSkillRequirements {
environment_id,
skill_names,
} in environments
{
for name in skill_names {
let available = step_catalog.catalog.entries.iter().any(|entry| {
let available = step_state.catalog.merged.entries.iter().any(|entry| {
entry.enabled
&& entry.name == name.as_str()
&& (entry.main_prompt.environment_id() == Some(environment_id)
codex-rs/ext/skills/src/shadow_selection_experiment/experiment_tests.rs
+19 -3
View File
@@ -54,7 +54,8 @@ async fn turn_input_and_history_advance_while_ranking_is_blocked()
use crate::catalog::SkillPackageId;
use crate::catalog::SkillResourceId;
use crate::install_with_providers_and_metrics;
use crate::state::HostSkillsStepState;
use crate::step_state::CatalogStatus;
use crate::step_state::SkillsStepState;
let _serial = RANKING_TESTS.acquire().await.unwrap();
let slots = RANKING_SLOTS.acquire_many(2).await?;
@@ -99,7 +100,7 @@ async fn turn_input_and_history_advance_while_ranking_is_blocked()
for turn_id in ["first", "second"] {
let turn_store = ExtensionData::new(turn_id);
turn_store.insert(HostSkillsStepState(SkillCatalog {
let host_catalog = SkillCatalog {
is_partial: false,
entries: ["x", "hidden", "selected"]
.into_iter()
@@ -116,7 +117,22 @@ async fn turn_input_and_history_advance_while_ranking_is_blocked()
})
.collect(),
warnings: Vec::new(),
}));
};
turn_store.insert(SkillsStepState {
discovery_v2: false,
tools_enabled: false,
catalog: thread_store
.get::<SkillsThreadState>()
.unwrap()
.prepare_step_catalog(
/*cloud*/ None,
SkillCatalog::default(),
/*discovery_v2*/ false,
),
host_catalog: Some(host_catalog),
cloud_status: CatalogStatus::Unavailable,
config: config.clone(),
});
let fragments = tokio::time::timeout(
Duration::from_secs(10),
registry.turn_input_contributors()[0].contribute(
codex-rs/ext/skills/src/state.rs
+61 -64
View File
@@ -1,3 +1,6 @@
//! Shared skill lifecycle state, including session resources and thread-owned caches.
//! Keeps discovery caches and live authorization checks separate from step preparation.
use std::collections::HashSet;
use std::sync::Arc;
use std::sync::Mutex;
@@ -7,7 +10,6 @@ use codex_extension_api::ExtensionMetrics;
use codex_mcp::McpResourceClient;
use crate::SkillsExtensionConfig;
use crate::SkillsExtensionState;
use crate::aliases::SkillAliasState;
use crate::catalog::SkillAuthority;
use crate::catalog::SkillCatalog;
@@ -16,21 +18,22 @@ use crate::catalog::SkillProviderError;
use crate::catalog::SkillProviderResult;
use crate::catalog::SkillReadResult;
use crate::catalog::SkillSourceKind;
use crate::catalog_state::CachedExecutorCatalog;
use crate::catalog_state::CachedExecutorDiscoveryCatalog;
use crate::catalog_state::CachedExecutorV2Catalog;
use crate::catalog_state::CloudResourceCache;
use crate::catalog_state::CloudSkillGeneration;
use crate::catalog_state::ExecutorCatalogCacheKey;
use crate::catalog_state::SkillCatalogState;
use crate::catalog_state::SkillReadCacheKey;
use crate::provider::SkillListQuery;
use crate::provider::SkillReadContext;
use crate::provider::SkillReadRequest;
use crate::shadow_selection_experiment::RecentSkillInvocations;
use crate::shadow_selection_experiment::ShadowSelectionTurnState;
use crate::shadow_selection_experiment::ShadowTaskContext;
use crate::skills_extension_state::CachedExecutorCatalog;
use crate::skills_extension_state::CachedExecutorDiscoveryCatalog;
use crate::skills_extension_state::CachedExecutorV2Catalog;
use crate::skills_extension_state::CloudResourceCache;
use crate::skills_extension_state::CloudSkillGeneration;
use crate::skills_extension_state::ExecutorCatalogCacheKey;
use crate::skills_extension_state::SkillReadCacheKey;
use crate::skills_extension_state::SkillStepCatalog;
use crate::sources::SkillProviders;
use crate::step_state::SkillStepCatalog;
#[cfg(test)]
#[path = "cloud_cache_tests.rs"]
@@ -45,7 +48,7 @@ pub(crate) struct SkillsSessionState {
pub struct SkillsThreadState {
config: Mutex<SkillsExtensionConfig>,
cloud_skills_available: bool,
skills_extension_state: Mutex<SkillsExtensionState>,
catalog_state: Mutex<SkillCatalogState>,
aliases: SkillAliasState,
shadow_selection_turn: Mutex<Option<ShadowSelectionTurn>>,
pub(crate) executor_read_snapshot: Mutex<Option<ExecutorReadSnapshot>>,
@@ -58,7 +61,7 @@ impl SkillsThreadState {
Self {
config: Mutex::new(config),
cloud_skills_available,
skills_extension_state: Mutex::new(SkillsExtensionState::default()),
catalog_state: Mutex::new(SkillCatalogState::default()),
aliases: SkillAliasState::default(),
shadow_selection_turn: Mutex::new(None),
executor_read_snapshot: Mutex::new(None),
@@ -81,7 +84,7 @@ impl SkillsThreadState {
.unwrap_or_else(std::sync::PoisonError::into_inner);
if previous.cloud_skill_enabled != config.cloud_skill_enabled {
let mut catalogs = self
.skills_extension_state
.catalog_state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
catalogs.cloud_cache = None;
@@ -89,37 +92,14 @@ impl SkillsThreadState {
*previous = config;
}
pub(crate) fn cloud_skills_available(&self) -> bool {
self.cloud_skills_available
}
pub(crate) fn cloud_skill_enabled(&self) -> bool {
self.cloud_skills_available && self.config().cloud_skill_enabled
}
pub(crate) fn replace_shadow_selection_turn(
&self,
turn_id: String,
state: Option<ShadowSelectionTurnState>,
) {
*self
.shadow_selection_turn
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner) =
state.map(|state| ShadowSelectionTurn {
turn_id,
state: Arc::new(state),
});
}
pub(crate) fn shadow_selection_turn(
&self,
turn_id: &str,
) -> Option<Arc<ShadowSelectionTurnState>> {
self.shadow_selection_turn
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.as_ref()
.filter(|turn| turn.turn_id == turn_id)
.map(|turn| Arc::clone(&turn.state))
}
pub(crate) fn prepare_step_catalog(
&self,
cloud: Option<SkillCatalog>,
@@ -128,7 +108,7 @@ impl SkillsThreadState {
) -> SkillStepCatalog {
let mut step = SkillStepCatalog::new(cloud, executor, discovery_v2, Default::default());
step.aliases = self.aliases.capture(
step.catalog
step.merged
.entries
.iter()
.flat_map(super::catalog::SkillCatalogEntry::source_locations)
@@ -148,13 +128,12 @@ impl SkillsThreadState {
pub(crate) async fn refresh_executor_catalog(
&self,
providers: &SkillProviders,
mut query: SkillListQuery,
query: SkillListQuery,
) -> SkillCatalog {
query.include_bundled_skills = self.config().bundled_skills_enabled;
if let Some(snapshots) = &query.executor_capability_discovery_v2 {
let snapshots = snapshots.clone();
let cached = self
.skills_extension_state
.catalog_state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.executor_v2_cache
@@ -165,7 +144,7 @@ impl SkillsThreadState {
Some(catalog) => catalog,
None => {
let catalog = providers.list_executor_for_turn(query).await;
self.skills_extension_state
self.catalog_state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.executor_v2_cache = Some(CachedExecutorV2Catalog {
@@ -180,7 +159,7 @@ impl SkillsThreadState {
.await
} else {
if let Some(cached) = self
.skills_extension_state
.catalog_state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.executor_cache
@@ -200,7 +179,7 @@ impl SkillsThreadState {
} else {
providers.list_executor_for_turn(query).await
};
self.skills_extension_state
self.catalog_state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.executor_cache = Some(CachedExecutorCatalog {
@@ -228,7 +207,7 @@ impl SkillsThreadState {
});
if !discovery_failed
&& let Some(cached) = self
.skills_extension_state
.catalog_state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.executor_discovery_cache
@@ -248,7 +227,7 @@ impl SkillsThreadState {
let include_bundled_skills = query.include_bundled_skills;
let discovered = providers.list_executor_for_turn(query).await;
let mut state = self
.skills_extension_state
.catalog_state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
// Failed discovery is returned to this caller but is never reused by another request.
@@ -262,16 +241,10 @@ impl SkillsThreadState {
discovered
}
/// Returns the cached Cloud catalog when enabled and authorized, without refreshing it.
/// Returns the currently authorized cached Cloud catalog without refreshing it.
/// Preparation decides whether Cloud is enabled using its captured config.
pub(crate) fn available_cloud_catalog(&self) -> Option<SkillCatalog> {
let config = self
.config
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if !config.cloud_skill_enabled || !self.cloud_skills_available {
return None;
}
self.skills_extension_state
self.catalog_state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.cloud_cache
@@ -282,7 +255,7 @@ impl SkillsThreadState {
fn cloud_cache(&self, mcp_resources: Option<&McpResourceClient>) -> Arc<CloudSkillGeneration> {
let mut state = self
.skills_extension_state
.catalog_state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let cache_key = mcp_resources
@@ -333,7 +306,7 @@ impl SkillsThreadState {
let mut catalog = providers.list_cloud_for_turn(query).await?;
catalog.entries.sort_by(|a, b| a.id.0.cmp(&b.id.0));
let mut catalogs = self
.skills_extension_state
.catalog_state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if !cache.is_current()
@@ -398,7 +371,7 @@ impl SkillsThreadState {
return Err(SkillProviderError::new("cloud skills are disabled"));
}
let catalogs = self
.skills_extension_state
.catalog_state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if !cache.is_current()
@@ -423,7 +396,7 @@ impl SkillsThreadState {
let result = providers.read(request).await?;
let catalogs = self
.skills_extension_state
.catalog_state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if !cache.is_current()
@@ -445,6 +418,33 @@ impl SkillsThreadState {
.unwrap_or_else(std::sync::PoisonError::into_inner)
.insert(cache_key, result))
}
pub(crate) fn replace_shadow_selection_turn(
&self,
turn_id: String,
state: Option<ShadowSelectionTurnState>,
) {
*self
.shadow_selection_turn
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner) =
state.map(|state| ShadowSelectionTurn {
turn_id,
state: Arc::new(state),
});
}
pub(crate) fn shadow_selection_turn(
&self,
turn_id: &str,
) -> Option<Arc<ShadowSelectionTurnState>> {
self.shadow_selection_turn
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.as_ref()
.filter(|turn| turn.turn_id == turn_id)
.map(|turn| Arc::clone(&turn.state))
}
}
/// One bounded executor resource, retained for continuations until replacement or thread drop.
@@ -476,6 +476,3 @@ impl EmittedCatalogBudgetWarnings {
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(crate) struct HostSkillsCatalogInWorldState;
#[derive(Clone, Debug, Default)]
pub(crate) struct HostSkillsStepState(pub(crate) SkillCatalog);
codex-rs/ext/skills/src/step_state.rs
+121
View File
@@ -0,0 +1,121 @@
//! Immutable, policy-checked skill catalogs captured for one step.
use crate::aliases::SkillCatalogAliases;
use crate::catalog::SkillAuthority;
use crate::catalog::SkillCatalog;
use crate::catalog::SkillSourceKind;
use crate::catalog::merge_catalogs;
use schemars::JsonSchema;
use serde::Deserialize;
use serde::Serialize;
/// Fully prepared before tool construction; retained unchanged by all step consumers.
#[derive(Debug)]
pub(crate) struct SkillsStepState {
pub(crate) discovery_v2: bool,
pub(crate) tools_enabled: bool,
pub(crate) catalog: SkillStepCatalog,
pub(crate) host_catalog: Option<SkillCatalog>,
pub(crate) cloud_status: CatalogStatus,
pub(crate) config: crate::SkillsExtensionConfig,
}
/// Policy-checked, enabled skill routes for one step. Hidden skills remain explicitly readable.
#[derive(Debug)]
pub(crate) struct SkillStepCatalog {
pub(super) merged: SkillCatalog,
pub(crate) aliases: SkillCatalogAliases,
/// None means no Cloud catalog was captured for this step.
cloud_warnings: Option<Vec<String>>,
executor_warnings: Vec<String>,
}
impl SkillStepCatalog {
pub(crate) fn new(
cloud: Option<SkillCatalog>,
executor: SkillCatalog,
discovery_v2: bool,
aliases: SkillCatalogAliases,
) -> Self {
let cloud_skill_count = cloud.as_ref().map(|catalog| catalog.entries.len());
let executor_skill_count = executor.entries.len();
let cloud_warnings = cloud.as_ref().map(|catalog| catalog.warnings.clone());
let executor_warnings = executor.warnings.clone();
let mut catalog = if discovery_v2 {
merge_catalogs(cloud, executor)
} else {
let mut catalog = cloud.unwrap_or_default();
catalog.is_partial |= executor.is_partial;
catalog.entries.extend(executor.entries);
catalog.warnings.extend(executor.warnings);
catalog
};
catalog.entries.retain_mut(|entry| {
if !entry.enabled {
return false;
}
entry.sources.retain(|source| source.enabled);
true
});
tracing::debug!(
discovery_v2,
?cloud_skill_count,
executor_skill_count,
enabled_skill_count = catalog.entries.len(),
warning_count = catalog.warnings.len(),
"prepared step skill catalog"
);
Self {
merged: catalog,
aliases,
cloud_warnings,
executor_warnings,
}
}
/// Selects one authority from the prepared catalog without consulting thread state.
pub(crate) fn select_from_catalog(
&self,
authority: SkillToolAuthoritySelector,
) -> SkillCatalog {
let entries = self
.merged
.entries
.iter()
.filter(|entry| authority.matches(&entry.authority))
.cloned()
.collect();
let warnings = match authority {
SkillToolAuthoritySelector::Cloud => self.cloud_warnings.clone().unwrap_or_default(),
SkillToolAuthoritySelector::Executor => self.executor_warnings.clone(),
};
SkillCatalog {
is_partial: self.merged.is_partial,
entries,
warnings,
}
}
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub(crate) enum SkillToolAuthoritySelector {
Cloud,
Executor,
}
impl SkillToolAuthoritySelector {
pub(crate) fn matches(self, authority: &SkillAuthority) -> bool {
match self {
Self::Cloud => authority.kind == SkillSourceKind::Cloud,
Self::Executor => authority.kind == SkillSourceKind::Executor,
}
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum CatalogStatus {
Unavailable,
Disabled,
Enabled,
}
codex-rs/ext/skills/src/tools/installations.rs
+1 -1
View File
@@ -77,7 +77,7 @@ impl<'call> ToolExecutor<ToolCall<'call>> for InstallationsTool {
if let Some(source) = &args.source {
source.validate()?;
}
let matches = self.context.matching_packages(&args.package)?;
let matches = self.context.matching_packages(&args.package);
let (entry, _) = matches.first().ok_or_else(|| {
FunctionCallError::RespondToModel("skill package is not available".to_string())
})?;
codex-rs/ext/skills/src/tools/list.rs
+6 -4
View File
@@ -86,14 +86,16 @@ impl<'call> ToolExecutor<ToolCall<'call>> for ListTool {
Box::pin(async move {
let args: ListArgs = parse_args(&call)?;
let response_byte_budget = call.response_byte_budget(MAX_SKILL_RESPONSE_BYTES);
let catalog = self.context.catalog(&[args.authority])?;
let catalog = self
.context
.step_state
.catalog
.select_from_catalog(args.authority);
let mut omitted_oversized_entry = false;
let canonical_skills = catalog
.entries
.into_iter()
.filter(|entry| {
entry.is_model_visible() && args.authority.matches(&entry.authority)
})
.filter(SkillCatalogEntry::is_model_visible)
.filter_map(|entry| {
let listed = listed_skill(entry);
omitted_oversized_entry |= listed.is_none();
codex-rs/ext/skills/src/tools/mod.rs
+15 -41
View File
@@ -6,8 +6,6 @@ use std::sync::Arc;
use codex_analytics::AnalyticsEventsClient;
use codex_analytics::InvocationType;
use codex_analytics::build_track_events_context;
use codex_exec_server::ExecutorCapabilityDiscoveryV2Snapshot;
use codex_exec_server::ResolvedSelectedCapabilityRoot;
use codex_extension_api::ExtensionData;
use codex_extension_api::ExtensionMetrics;
use codex_extension_api::FunctionCallError;
@@ -36,15 +34,14 @@ use serde_json::Value;
use crate::aliases::SkillLocator;
use crate::catalog::SkillAuthority;
use crate::catalog::SkillCatalog;
use crate::catalog::SkillCatalogEntry;
use crate::catalog::SkillSourceKind;
use crate::catalog::SkillSourceLocation;
use crate::skills_extension_state::SkillToolAuthoritySelector;
use crate::skills_extension_state::SkillsStepState;
use crate::sources::SkillProviders;
use crate::state::SkillsSessionState;
use crate::state::SkillsThreadState;
use crate::step_state::SkillToolAuthoritySelector;
use crate::step_state::SkillsStepState;
use crate::telemetry::ActiveSkillTurnMetrics;
use crate::telemetry::SkillTurnMetrics;
@@ -66,14 +63,10 @@ pub(crate) fn skill_tools(
let Some(thread_state) = thread_store.get::<SkillsThreadState>() else {
return Vec::new();
};
let discovery_v2 = step_store
.get::<Vec<ExecutorCapabilityDiscoveryV2Snapshot>>()
.is_some();
let has_executor = step_store
.get::<Vec<ResolvedSelectedCapabilityRoot>>()
.is_some_and(|roots| !roots.is_empty())
|| discovery_v2;
if !thread_state.cloud_skill_enabled() && !has_executor {
let Some(step_state) = step_store.get::<SkillsStepState>() else {
return Vec::new();
};
if !step_state.tools_enabled {
return Vec::new();
}
let mcp_resources = session_store
@@ -81,12 +74,11 @@ pub(crate) fn skill_tools(
.and_then(|state| state.mcp_resources.clone());
let analytics = SkillAnalytics::from_stores(session_store, thread_store);
let context = SkillToolContext {
discovery_v2,
providers,
mcp_resources,
thread_state,
analytics,
step_catalog: step_store.get_or_init(SkillsStepState::new),
step_state,
};
let mut tools: Vec<Arc<dyn for<'call> ToolExecutor<ToolCall<'call>>>> = vec![
Arc::new(list::ListTool {
@@ -96,7 +88,7 @@ pub(crate) fn skill_tools(
context: context.clone(),
}),
];
if context.discovery_v2 {
if context.step_state.discovery_v2 {
tools.push(Arc::new(installations::InstallationsTool { context }));
}
tools
@@ -212,25 +204,19 @@ impl SkillAnalytics {
#[derive(Clone)]
struct SkillToolContext {
discovery_v2: bool,
providers: SkillProviders,
mcp_resources: Option<Arc<McpResourceClient>>,
thread_state: Arc<SkillsThreadState>,
analytics: Option<SkillAnalytics>,
step_catalog: Arc<SkillsStepState>,
step_state: Arc<SkillsStepState>,
}
impl SkillToolContext {
fn matching_packages<'a>(
&'a self,
package: &SkillLocator,
) -> Result<Vec<(&'a SkillCatalogEntry, SkillSourceLocation)>, FunctionCallError> {
let snapshot = self.step_catalog.get().ok_or_else(|| {
FunctionCallError::RespondToModel(
"skill catalog is not ready for this step".to_string(),
)
})?;
let catalog = &snapshot.catalog;
) -> Vec<(&'a SkillCatalogEntry, SkillSourceLocation)> {
let catalog = &self.step_state.catalog.merged;
let matching_entries = |resolved_package: &str, alias_source: Option<&SkillSourceKind>| {
catalog
.entries
@@ -250,24 +236,12 @@ impl SkillToolContext {
};
let exact = matching_entries(package.as_str(), /*alias_source*/ None);
if !exact.is_empty() || package.alias_source().is_none() {
return Ok(exact);
return exact;
}
let resolved_package = snapshot.aliases.resolve(package);
Ok(resolved_package.map_or_else(Vec::new, |resolved| {
let resolved_package = self.step_state.catalog.aliases.resolve(package);
resolved_package.map_or_else(Vec::new, |resolved| {
matching_entries(&resolved, package.alias_source())
}))
}
fn catalog(
&self,
authorities: &[SkillToolAuthoritySelector],
) -> Result<SkillCatalog, FunctionCallError> {
let snapshot = self.step_catalog.get().ok_or_else(|| {
FunctionCallError::RespondToModel(
"skill catalog is not ready for this step".to_string(),
)
})?;
Ok(snapshot.select_from_catalog(authorities))
})
}
}
codex-rs/ext/skills/src/tools/read.rs
+6 -6
View File
@@ -98,7 +98,7 @@ impl<'call> ToolExecutor<ToolCall<'call>> for ReadTool {
}
fn spec(&self) -> ToolSpec {
if !self.context.discovery_v2 {
if !self.context.step_state.discovery_v2 {
return skill_function_tool::<ReadArgs, ReadResponse<LegacyReadMetadata>>(
TOOL_NAME,
"Read one page from a skill. Pass its provided package directly; root aliases are resolved automatically. Omit resource to read SKILL.md; to read another file, use the same package and pass the file's complete skill:// identifier as resource. For executor-backed skills, skill_root is the skill's absolute directory in the executor filesystem and can be used to locate bundled scripts. If the package is not provided, use skills.list to find it. Pass next_cursor back as cursor to continue the same snapshot while it is cached; omit cursor to read again.",
@@ -115,7 +115,7 @@ impl<'call> ToolExecutor<ToolCall<'call>> for ReadTool {
'call: 'a,
{
Box::pin(async move {
let (args, source) = if self.context.discovery_v2 {
let (args, source) = if self.context.step_state.discovery_v2 {
let args: ReadInstallationArgs = parse_args(&call)?;
(args.read, args.source)
} else {
@@ -130,7 +130,7 @@ impl<'call> ToolExecutor<ToolCall<'call>> for ReadTool {
validate_handle("resource", resource, MAX_HANDLE_BYTES)?;
}
let matches = self.context.matching_packages(&args.package)?;
let matches = self.context.matching_packages(&args.package);
let (skill_entry, selected_source, output_authority) = matches
.iter()
.find_map(|(entry, matched_source)| {
@@ -186,7 +186,7 @@ impl<'call> ToolExecutor<ToolCall<'call>> for ReadTool {
}
};
let cursor_start = |contents: &str| {
if self.context.discovery_v2 {
if self.context.step_state.discovery_v2 {
parse_pagination_cursor(
args.cursor.as_deref(),
&(&source, &package, contents),
@@ -267,7 +267,7 @@ impl<'call> ToolExecutor<ToolCall<'call>> for ReadTool {
{
state.record_invocation(skill_entry.main_prompt.as_str());
}
let metadata = if self.context.discovery_v2 {
let metadata = if self.context.step_state.discovery_v2 {
let mut installations = Vec::new();
let mut installations_truncated = false;
let mut installation_bytes = 2; // JSON array brackets.
@@ -320,7 +320,7 @@ impl<'call> ToolExecutor<ToolCall<'call>> for ReadTool {
start,
response_byte_budget,
|offset| {
if self.context.discovery_v2 {
if self.context.step_state.discovery_v2 {
pagination_cursor(&(&source, &package, result.contents.as_str()), offset)
} else {
pagination_cursor(result.contents.as_str(), offset)
codex-rs/ext/skills/src/world_state_catalogs.rs
+72 -194
View File
@@ -1,21 +1,16 @@
//! Renders prepared step catalogs without performing discovery or changing tool readiness.
use std::sync::Arc;
use codex_exec_server::ResolvedSelectedCapabilityRoot;
use codex_extension_api::ContextualUserFragment;
use codex_extension_api::ExtensionEventSink;
use codex_extension_api::ExtensionWarning;
use codex_extension_api::SelectedPluginSnapshot;
use codex_extension_api::WorldStateContributionInput;
use codex_extension_api::WorldStateSectionContribution;
use serde::Deserialize;
use serde::Serialize;
use crate::HostSkillsSnapshot;
use crate::SkillsExtensionConfig;
use crate::catalog::SkillCatalog;
use crate::catalog::apply_plugin_denials;
use crate::provider::SkillListQuery;
use crate::provider::attribute_executor_plugins;
use crate::render::AvailableSkillsRender;
use crate::render::PreparedSkillCatalog;
use crate::render::RenderedSkillCatalogs;
@@ -26,14 +21,11 @@ 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;
use crate::skills_extension_state::SkillToolAuthoritySelector;
use crate::skills_extension_state::SkillsStepState;
use crate::sources::SkillProviders;
use crate::state::EmittedCatalogBudgetWarnings;
use crate::state::HostSkillsCatalogInWorldState;
use crate::state::HostSkillsStepState;
use crate::state::SkillsSessionState;
use crate::state::SkillsThreadState;
use crate::step_state::CatalogStatus;
use crate::step_state::SkillToolAuthoritySelector;
use crate::step_state::SkillsStepState;
use crate::world_state::CatalogRenderCallback;
use crate::world_state::cloud_skills_world_state_section;
use crate::world_state::executor_skills_world_state_section;
@@ -72,13 +64,6 @@ impl CatalogKind {
}
}
#[derive(Clone, Copy, Eq, PartialEq)]
pub(crate) enum CatalogStatus {
Unavailable,
Disabled,
Enabled,
}
struct CatalogContribution {
catalog: SkillCatalog,
status: CatalogStatus,
@@ -93,42 +78,39 @@ impl CatalogContribution {
}
}
pub(crate) struct CatalogContributions {
struct CatalogContributions {
executor: CatalogContribution,
cloud: CatalogContribution,
host: CatalogContribution,
}
pub(crate) struct RenderedCatalogContribution {
struct RenderedCatalogContribution {
kind: CatalogKind,
pub(crate) status: CatalogStatus,
status: CatalogStatus,
budget: SkillMetadataBudget,
allocation: CatalogBudgetAllocation,
rendered: Option<AvailableSkillsRender>,
}
pub(crate) struct CatalogContext<'a> {
providers: &'a SkillProviders,
struct CatalogRenderingContext<'a> {
input: WorldStateContributionInput<'a>,
thread_state: Arc<SkillsThreadState>,
config: SkillsExtensionConfig,
step_state: Arc<SkillsStepState>,
metadata_budget: SkillMetadataBudget,
render_policy: SkillCatalogRenderPolicy,
include_usage: bool,
warning_emitter: CatalogWarningEmitter,
}
impl<'a> CatalogContext<'a> {
pub(crate) fn new(
providers: &'a SkillProviders,
impl<'a> CatalogRenderingContext<'a> {
fn new(
event_sink: Arc<dyn ExtensionEventSink>,
input: WorldStateContributionInput<'a>,
) -> Option<Self> {
let thread_state = input.thread_store.get::<SkillsThreadState>()?;
let config = thread_state.config();
let step_state = input.step_store.get::<SkillsStepState>()?;
let include_usage = input.model_info.include_skills_usage_instructions;
let context_window = input.model_info.resolved_context_window();
let metadata_budget = skill_metadata_budget(context_window, config.max_context_tokens);
let metadata_budget =
skill_metadata_budget(context_window, step_state.config.max_context_tokens);
let emitted_warnings = input
.turn_store
.get_or_init(EmittedCatalogBudgetWarnings::default);
@@ -144,16 +126,14 @@ impl<'a> CatalogContext<'a> {
}
});
let render_policy = if input.executor_capability_discovery_v2.is_some() {
let render_policy = if step_state.discovery_v2 {
SkillCatalogRenderPolicy::DiscoveryV2
} else {
SkillCatalogRenderPolicy::ExtensionCompatible
};
Some(Self {
providers,
input,
thread_state,
config,
step_state,
metadata_budget,
render_policy,
include_usage,
@@ -161,150 +141,38 @@ impl<'a> CatalogContext<'a> {
})
}
pub(crate) async fn discover_catalogs(&self) -> CatalogContributions {
let cloud_enabled = self.thread_state.cloud_skill_enabled();
// Core always supplies this vector, including when no captured roots are ready.
// Descriptor-only callers retain their unbound discovery path.
let captured_roots = self
.input
.step_store
.get::<Vec<ResolvedSelectedCapabilityRoot>>();
let executor_roots = captured_roots
.as_ref()
.map(|roots| {
roots
.iter()
.map(|root| root.selected_root().clone())
.collect()
})
.unwrap_or_else(|| self.input.ready_selected_capability_roots.to_vec());
let query = SkillListQuery {
turn_id: self.input.turn_id.to_string(),
executor_roots,
resolved_executor_roots: captured_roots
.map(|roots| roots.as_ref().clone())
.unwrap_or_default(),
host_snapshot: None,
include_host_skills: false,
include_bundled_skills: self.config.bundled_skills_enabled,
include_cloud_skills: cloud_enabled,
mcp_resources: self
.input
.session_store
.get::<SkillsSessionState>()
.and_then(|state| state.mcp_resources.clone()),
executor_capability_discovery: self.input.executor_capability_discovery.cloned(),
executor_capability_discovery_v2: self
.input
.executor_capability_discovery_v2
.map(Vec::from),
};
let selected_plugins = self.input.step_store.get::<SelectedPluginSnapshot>();
let step_state = self.input.step_store.get_or_init(SkillsStepState::new);
let (step_catalog, host) = futures::join!(
step_state.get_or_init(|| async {
let discovery_v2_enabled = query.executor_capability_discovery_v2.is_some();
let mut executor = self
.thread_state
.refresh_executor_catalog(self.providers, query)
.await;
let mut cloud = self.thread_state.available_cloud_catalog();
if let Some(selected_plugins) = selected_plugins.as_deref() {
if discovery_v2_enabled {
apply_plugin_denials(&mut executor, selected_plugins);
if let Some(cloud) = &mut cloud {
apply_plugin_denials(cloud, selected_plugins);
}
} else {
attribute_executor_plugins(&mut executor, selected_plugins);
}
}
Arc::new(self.thread_state.prepare_step_catalog(
cloud,
executor,
discovery_v2_enabled,
))
}),
self.discover_host_catalog(),
);
let cloud_status = if !self.providers.has_cloud_provider() {
CatalogStatus::Unavailable
} else if cloud_enabled || step_catalog.cloud_warnings.is_some() {
CatalogStatus::Enabled
} else {
CatalogStatus::Disabled
};
CatalogContributions {
fn render_world_state_catalogs(&self) -> [RenderedCatalogContribution; 3] {
let catalogs = CatalogContributions {
executor: CatalogContribution {
catalog: step_catalog.select_from_catalog(&[SkillToolAuthoritySelector::Executor]),
catalog: self
.step_state
.catalog
.select_from_catalog(SkillToolAuthoritySelector::Executor),
status: CatalogStatus::Enabled,
},
cloud: CatalogContribution {
catalog: step_catalog.select_from_catalog(&[SkillToolAuthoritySelector::Cloud]),
status: cloud_status,
catalog: self
.step_state
.catalog
.select_from_catalog(SkillToolAuthoritySelector::Cloud),
status: self.step_state.cloud_status,
},
host,
}
}
async fn discover_host_catalog(&self) -> CatalogContribution {
let Some(host_snapshot) = self
.input
.turn_store
.get::<HostSkillsSnapshot>()
.filter(|_| self.providers.has_host_provider())
else {
return CatalogContribution::unavailable();
host: self.step_state.host_catalog.as_ref().map_or_else(
CatalogContribution::unavailable,
|catalog| CatalogContribution {
catalog: catalog.clone(),
status: CatalogStatus::Enabled,
},
),
};
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_cloud_skills: false,
mcp_resources: None,
executor_capability_discovery: None,
executor_capability_discovery_v2: 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 total_limit = self.metadata_budget.limit();
let default_cloud_limit = total_limit - total_limit / FILESYSTEM_BUDGET_DIVISOR;
let step_state = self.input.step_store.get::<SkillsStepState>();
let executor_aliases = step_state
.as_ref()
.and_then(|state| state.get())
.map(|catalog| {
catalog
.aliases
.for_authority(SkillToolAuthoritySelector::Executor)
});
let executor_aliases = Some(
self.step_state
.catalog
.aliases
.for_authority(SkillToolAuthoritySelector::Executor),
);
let mut executor = PreparedSkillCatalog::with_alias_plan(
&catalogs.executor.catalog,
self.render_policy,
@@ -402,15 +270,12 @@ impl<'a> CatalogContext<'a> {
return None;
}
let step_state = self.input.step_store.get::<SkillsStepState>();
let cloud_aliases = step_state
.as_ref()
.and_then(|state| state.get())
.map(|catalog| {
catalog
.aliases
.for_authority(SkillToolAuthoritySelector::Cloud)
});
let cloud_aliases = Some(
self.step_state
.catalog
.aliases
.for_authority(SkillToolAuthoritySelector::Cloud),
);
let shared = render_prepared_skill_catalogs(
executor,
&PreparedSkillCatalog::with_alias_plan(
@@ -447,19 +312,16 @@ impl<'a> CatalogContext<'a> {
executor: &PreparedSkillCatalog<'_>,
cloud_limit: usize,
) -> (RenderedSkillCatalogs, SkillMetadataBudget) {
if !self.config.include_instructions {
if !self.step_state.config.include_instructions {
return (RenderedSkillCatalogs::default(), self.metadata_budget);
}
let step_state = self.input.step_store.get::<SkillsStepState>();
let cloud_aliases = step_state
.as_ref()
.and_then(|state| state.get())
.map(|catalog| {
catalog
.aliases
.for_authority(SkillToolAuthoritySelector::Cloud)
});
let cloud_aliases = Some(
self.step_state
.catalog
.aliases
.for_authority(SkillToolAuthoritySelector::Cloud),
);
let cloud_budget = self.metadata_budget.with_limit(cloud_limit);
let cloud = render_available_skills(
&catalogs.cloud.catalog,
@@ -490,7 +352,7 @@ impl<'a> CatalogContext<'a> {
(rendered, filesystem_budget)
}
pub(crate) fn build_world_state_section(
fn build_world_state_section(
&self,
catalog: RenderedCatalogContribution,
) -> WorldStateSectionContribution {
@@ -508,7 +370,7 @@ impl<'a> CatalogContext<'a> {
let body = rendered
.and_then(|rendered| rendered.into_fragment(self.include_usage))
.map(|fragment| fragment.body().into_text().unwrap_or_default());
let include_instructions = self.config.include_instructions;
let include_instructions = self.step_state.config.include_instructions;
let metrics = self.input.extension_metrics.clone();
let warning_emitter = Arc::clone(&self.warning_emitter);
let render_report = report.clone();
@@ -551,3 +413,19 @@ impl<'a> CatalogContext<'a> {
}
}
}
/// Returns complete sections, omitting catalogs unavailable in this step.
pub(crate) fn render_world_state_sections(
event_sink: Arc<dyn ExtensionEventSink>,
input: WorldStateContributionInput<'_>,
) -> Vec<WorldStateSectionContribution> {
let Some(context) = CatalogRenderingContext::new(event_sink, input) else {
return Vec::new();
};
context
.render_world_state_catalogs()
.into_iter()
.filter(|catalog| catalog.status != CatalogStatus::Unavailable)
.map(|catalog| context.build_world_state_section(catalog))
.collect()
}
codex-rs/ext/skills/tests/skills_extension.rs
+460 -353
View File
File diff suppressed because it is too large Load Diff
codex-rs/ext/skills/tests/skills_extension/discovery_v2_tests.rs
+25 -8
View File
@@ -207,23 +207,40 @@ async fn v2_steps_refresh_metadata_and_preserve_separate_environments() -> TestR
} else {
snapshots.clear();
}
let sections = registry.context_contributors()[0]
.contribute_world_state(WorldStateContributionInput {
let step_store = ExtensionData::new("step");
step_store.insert(snapshots.clone());
step_store.insert(SelectedPluginSnapshot {
plugins: snapshots
.iter()
.map(|snapshot| codex_extension_api::SelectedPluginIdentity {
selected_root_id: Some(format!(
"{}:local-name@marketplace",
snapshot.environment_id
)),
plugin_id: codex_core_plugin_common::PluginIdentity::Remote {
remote_plugin_id: "Plugin_123".into(),
},
})
.collect(),
..Default::default()
});
let sections = prepared_world_state(
&registry,
&[],
WorldStateContributionInput {
previous_world_state: None,
step_store: &ExtensionData::new("step"),
executor_capability_discovery_v2: Some(&snapshots),
step_store: &step_store,
model_info: &catalog_model_info(),
thread_id: codex_protocol::ThreadId::new(),
turn_id: "turn-1",
environments: &[],
ready_selected_capability_roots: &[],
executor_capability_discovery: None,
extension_metrics: None,
session_store: &session_store,
thread_store: &thread_store,
turn_store: &turn_store,
})
.await;
},
)
.await;
let rendered = sections
.iter()
.filter_map(|section| section.render_diff(PreviousWorldStateSection::Absent).1)
codex-rs/ext/skills/tests/skills_extension/merge_catalog_tests.rs
+159 -53
View File
@@ -42,7 +42,8 @@ fn installed_skill(
),
)
.with_display_path(format!("{package}/SKILL.md"))
.with_remote_plugin_id(Some("Plugin_123".to_string())))
.with_remote_plugin_id(Some("Plugin_123".to_string()))
.with_plugin_id(Some(root.to_string())))
}
fn provider(entries: Vec<SkillCatalogEntry>) -> StaticSkillProvider {
@@ -135,7 +136,7 @@ impl MergeHarness {
Ok(harness)
}
fn tools(
async fn tools(
&self,
step: &ExtensionData,
) -> Vec<Arc<dyn for<'call> ToolExecutor<ToolCall<'call>>>> {
@@ -145,6 +146,25 @@ impl MergeHarness {
if let Some(discovery) = &self.discovery {
step.insert(discovery.clone());
}
if let Some(plugins) = self.turn_store.get::<SelectedPluginSnapshot>() {
step.insert(plugins.as_ref().clone());
}
let selected_plugins = step.get::<SelectedPluginSnapshot>().unwrap_or_default();
prepare_step(
&self.registry,
codex_extension_api::StepPreparationInput {
selected_plugins: &selected_plugins,
turn_id: self.turn_store.level_id(),
ready_selected_capability_roots: &self.roots,
executor_capability_discovery: self.discovery.as_ref(),
executor_capability_discovery_v2: self.snapshots.as_deref(),
session_store: &self.session_store,
thread_store: &self.thread_store,
turn_store: &self.turn_store,
step_store: step,
},
)
.await;
self.registry.tool_contributors()[0].tools_for_step(
&self.session_store,
&self.thread_store,
@@ -164,23 +184,29 @@ impl MergeHarness {
if let Some(plugins) = turn_store.get::<codex_extension_api::SelectedPluginSnapshot>() {
step_store.insert(plugins.as_ref().clone());
}
let sections = self.registry.context_contributors()[0]
.contribute_world_state(WorldStateContributionInput {
if let Some(snapshots) = &self.snapshots {
step_store.insert(snapshots.clone());
}
if let Some(discovery) = &self.discovery {
step_store.insert(discovery.clone());
}
let sections = prepared_world_state(
&self.registry,
&self.roots,
WorldStateContributionInput {
previous_world_state: None,
step_store,
model_info: &catalog_model_info(),
thread_id: codex_protocol::ThreadId::new(),
turn_id: turn_store.level_id(),
environments: &[],
ready_selected_capability_roots: &self.roots,
executor_capability_discovery: self.discovery.as_ref(),
executor_capability_discovery_v2: self.snapshots.as_deref(),
extension_metrics: None,
session_store: &self.session_store,
thread_store: &self.thread_store,
turn_store,
})
.await;
},
)
.await;
Ok(sections
.iter()
.filter_map(|section| section.render_diff(PreviousWorldStateSection::Absent).1)
@@ -190,7 +216,7 @@ impl MergeHarness {
}
}
async fn call_tool(
pub(super) async fn call_tool(
tools: &[Arc<dyn for<'call> ToolExecutor<ToolCall<'call>>>],
name: &str,
arguments: Value,
@@ -293,7 +319,8 @@ async fn v2_merged_skills_preserve_cloud_metadata_and_accepted_plugin_routes() -
let standalone_cloud =
cloud_skill("cloud").with_remote_plugin_id(/*remote_plugin_id*/ None);
let standalone_executor = installed_skill("standalone", "executor")?
.with_remote_plugin_id(/*remote_plugin_id*/ None);
.with_remote_plugin_id(/*remote_plugin_id*/ None)
.with_plugin_id(/*plugin_id*/ None);
installed.push(standalone_executor.clone());
let mut executor_provider = provider(installed.clone());
executor_provider.catalog.warnings =
@@ -318,7 +345,7 @@ async fn v2_merged_skills_preserve_cloud_metadata_and_accepted_plugin_routes() -
},
});
harness.turn_store.insert(selection);
let tools = harness.tools(&harness.step_store);
let tools = harness.tools(&harness.step_store).await;
let rendered = harness.render().await?;
assert_eq!(rendered.matches("- deploy:work:").count(), 1);
assert!(rendered.contains("Cloud work description"));
@@ -389,7 +416,9 @@ async fn executor_aliases_survive_cloud_promotion_without_reassigning_remaining_
for name in names {
let mut entry = installed_skill(root, name)?.with_alias_root(format!("skill://{root}"));
if root == &local_root {
entry = entry.with_remote_plugin_id(/*remote_plugin_id*/ None);
entry = entry
.with_remote_plugin_id(/*remote_plugin_id*/ None)
.with_plugin_id(/*plugin_id*/ None);
entry.name = format!("local:{name}");
}
installed.push(entry);
@@ -426,7 +455,11 @@ async fn executor_aliases_survive_cloud_promotion_without_reassigning_remaining_
)
.await;
}
let tools = harness.tools(&harness.step_store);
harness.turn_store.insert(selected_plugins(
&[None, Some(&format!("env-1:{promoted_root}"))],
&[],
));
let tools = harness.tools(&harness.step_store).await;
let rendered = harness.render().await?;
assert!(rendered.contains("e1/work"), "{rendered}");
assert!(
@@ -485,14 +518,24 @@ async fn executor_aliases_survive_cloud_promotion_without_reassigning_remaining_
#[tokio::test]
async fn rejected_roots_do_not_starve_accepted_alternate_and_standalone_aliases() -> TestResult {
let mut entries = Vec::new();
let mut selection = selected_plugins(&[None], &[]);
// Together these individually allocatable roots exceed the shared 16 KiB allowance.
for index in 0..180 {
let root = format!("a-rejected-{index:03}-{}", "x".repeat(120));
let entry = installed_skill(&root, "work")?
.with_alias_root(format!("skill://{root}"))
.with_remote_plugin_id(Some(format!("Rejected_{index}")));
selection.plugins.push(SelectedPluginIdentity {
selected_root_id: Some(format!("env-1:{root}")),
plugin_id: PluginIdentity::Remote {
remote_plugin_id: format!("Rejected_{index}"),
},
});
entries.push(entry);
}
selection
.plugins
.extend(selected_plugins(&[Some("env-1:b-promoted")], &[]).plugins);
let promoted = installed_skill("b-promoted", "work")?.with_alias_root("skill://b-promoted");
let mut standalone = installed_skill("c-local", "work")?
.with_alias_root("skill://c-local")
@@ -506,7 +549,8 @@ async fn rejected_roots_do_not_starve_accepted_alternate_and_standalone_aliases(
default_config(),
)
.await?;
let tools = harness.tools(&harness.step_store);
harness.turn_store.insert(selection);
let tools = harness.tools(&harness.step_store).await;
harness.render().await?;
for (alias, entry) in [("e0/work", promoted), ("e1/work", standalone)] {
let output = call_tool(&tools, "read", json!({"package": alias})).await?;
@@ -541,6 +585,18 @@ async fn cloud_rendering_and_reads_share_aliases_after_a_plugin_is_disabled() ->
.collect();
let harness =
MergeHarness::new(provider(Vec::new()), provider(entries), default_config()).await?;
harness.step_store.insert(SelectedPluginSnapshot {
plugins: roots
.iter()
.map(|root| SelectedPluginIdentity {
selected_root_id: None,
plugin_id: PluginIdentity::Remote {
remote_plugin_id: root.clone(),
},
})
.collect(),
..Default::default()
});
let rendered = harness.render().await?;
assert!(rendered.contains(
"For each alias listed below, this mapping replaces any earlier mapping for that alias."
@@ -564,7 +620,7 @@ async fn cloud_rendering_and_reads_share_aliases_after_a_plugin_is_disabled() ->
let rendered = harness.render_step(&step).await?;
assert!(rendered.contains("c1/work"), "{rendered}");
assert!(!rendered.contains(&roots[0]), "{rendered}");
let tools = harness.tools(&step);
let tools = harness.tools(&step).await;
let aliased = call_tool(&tools, "read", json!({"package": "c1/work"})).await?;
let full = call_tool(
&tools,
@@ -604,7 +660,8 @@ async fn v2_only_complete_cloud_catalog_excludes_unmatched_remote_skills() -> Te
provider(vec![
installed_skill("matching", "work")?.with_plugin_id(Some("matching".to_string())),
installed_skill("standalone", "local")?
.with_remote_plugin_id(/*remote_plugin_id*/ None),
.with_remote_plugin_id(/*remote_plugin_id*/ None)
.with_plugin_id(/*plugin_id*/ None),
]),
cloud,
config,
@@ -636,7 +693,7 @@ async fn v2_disabled_remote_plugin_blocks_both_read_routes() -> TestResult {
&[None, Some("env-1:matching")],
&["Plugin_123"],
));
let tools = harness.tools(&harness.step_store);
let tools = harness.tools(&harness.step_store).await;
assert!(!harness.render().await?.contains("- deploy:work:"));
for package in [&cloud.id, &installed.id] {
let result = call_tool(&tools, "read", json!({"package": package.0})).await;
@@ -659,13 +716,14 @@ async fn v1_reads_preserve_response_shape_without_installations_tool() -> TestRe
assert_eq!(
disabled
.tools(&disabled.step_store)
.await
.iter()
.map(|tool| tool.tool_name().name)
.collect::<Vec<_>>(),
["list", "read"],
);
disabled.render().await?;
let tools = disabled.tools(&disabled.step_store);
let tools = disabled.tools(&disabled.step_store).await;
for entry in [&cloud, &executor] {
let mut expected = json!({
"resource": entry.main_prompt.as_str(),
@@ -707,12 +765,26 @@ async fn v2_reads_preserve_content_routes_with_bounded_installations() -> TestRe
"environment_id": "env-1",
"skill_root": PathUri::parse("file:///skills/selected")?.inferred_native_path_string(),
}));
let mut selection = selected_plugins(&[None], &[]);
selection
.plugins
.extend(entries.iter().map(|entry| SelectedPluginIdentity {
selected_root_id: Some(format!(
"{}:{}",
entry.main_prompt.environment_id().unwrap(),
entry.plugin_id.as_ref().unwrap()
)),
plugin_id: PluginIdentity::Remote {
remote_plugin_id: "Plugin_123".into(),
},
}));
let executor = provider(entries);
let executor_reads = executor.read_requests.clone();
let cloud_provider = provider(vec![cloud.clone()]);
let cloud_reads = cloud_provider.read_requests.clone();
let harness = MergeHarness::new(executor, cloud_provider, default_config()).await?;
let tools = harness.tools(&harness.step_store);
harness.turn_store.insert(selection);
let tools = harness.tools(&harness.step_store).await;
harness.render().await?;
for (package, source, requested, limit) in [
("c0/work", Value::Null, &cloud, 10_000),
@@ -770,6 +842,7 @@ async fn v2_reads_preserve_content_routes_with_bounded_installations() -> TestRe
}
let first = installed_skill("shared", "work")?
.with_plugin_id(/*plugin_id*/ None)
.with_remote_plugin_id(/*remote_plugin_id*/ None)
.with_alias_root("skill://shared");
let mut second = first.clone();
@@ -781,7 +854,7 @@ async fn v2_reads_preserve_content_routes_with_bounded_installations() -> TestRe
let executor = provider(vec![first.clone(), second.clone()]);
let reads = executor.read_requests.clone();
let harness = MergeHarness::new(executor, provider(Vec::new()), default_config()).await?;
let tools = harness.tools(&harness.step_store);
let tools = harness.tools(&harness.step_store).await;
harness.render().await?;
let locations = ["env-1", "env-2"].map(|environment_id| {
json!({
@@ -913,12 +986,26 @@ async fn v2_installations_filter_all_matching_locations_and_paginate_within_budg
}));
}
entries.push(entries[2].clone());
let mut selection = selected_plugins(&[None], &[]);
selection
.plugins
.extend(entries.iter().map(|entry| SelectedPluginIdentity {
selected_root_id: Some(format!(
"{}:{}",
entry.main_prompt.environment_id().unwrap(),
entry.plugin_id.as_ref().unwrap()
)),
plugin_id: PluginIdentity::Remote {
remote_plugin_id: "Plugin_123".into(),
},
}));
let executor = provider(entries);
let executor_reads = executor.read_requests.clone();
let cloud_provider = provider(vec![cloud]);
let cloud_reads = cloud_provider.read_requests.clone();
let harness = MergeHarness::new(executor, cloud_provider, default_config()).await?;
let tools = harness.tools(&harness.step_store);
harness.turn_store.insert(selection);
let tools = harness.tools(&harness.step_store).await;
harness.render().await?;
assert_eq!(
call_tool(
@@ -1013,23 +1100,24 @@ async fn v2_rendering_and_tools_retain_their_step_catalog() -> TestResult {
harness
.turn_store
.insert(selected_plugins(&[None, Some("env-1:matching")], &[]));
// Core constructs tools before World State prepares their shared step store.
let original_tools = harness.tools(&harness.step_store);
// Preparation makes tools usable before any World State rendering.
let original_tools = harness.tools(&harness.step_store).await;
let early_list = call_tool(
&original_tools,
"list",
json!({"authority": {"kind": "cloud"}}),
)
.await;
assert_matches!(early_list, Err(FunctionCallError::RespondToModel(message)) if message == "skill catalog is not ready for this step");
assert_eq!(executor_calls.load(Ordering::SeqCst), 0);
.await?;
assert_eq!(early_list["skills"].as_array().unwrap().len(), 1);
call_tool(&original_tools, "read", json!({"package": installed.id.0})).await?;
assert_eq!(executor_calls.load(Ordering::SeqCst), 1);
let original_rendering = harness.render().await?;
assert!(original_rendering.contains("- deploy:work:"));
let next_step = ExtensionData::new("step-2");
let next_tools = harness.tools(&next_step);
harness
.turn_store
.insert(selected_plugins(&[], &["Plugin_123"]));
let next_tools = harness.tools(&next_step).await;
assert!(
!harness
.render_step(&next_step)
@@ -1056,7 +1144,7 @@ async fn v2_rendering_and_tools_retain_their_step_catalog() -> TestResult {
harness
.turn_store
.insert(selected_plugins(&[None, Some("env-1:matching")], &[]));
let enabled_tools = harness.tools(&enabled_step);
let enabled_tools = harness.tools(&enabled_step).await;
assert_eq!(
harness.render_step(&enabled_step).await?,
original_rendering
@@ -1117,23 +1205,23 @@ async fn v2_implicit_and_explicit_invocations_retain_their_step_catalog() -> Tes
);
assert!(detect(&next_step).is_none());
for (step_store, allowed) in [(&harness.step_store, true), (&next_step, false)] {
let fragments = harness.registry.turn_input_contributors()[0]
.contribute(
TurnInputContext {
step_store,
turn_id: "turn-1".to_string(),
user_input: vec![UserInput::Mention {
name: installed.name.clone(),
path: installed.main_prompt.as_str().to_string(),
}],
environments: Vec::new(),
},
/*extension_metrics*/ None,
&harness.session_store,
&harness.thread_store,
&harness.turn_store,
)
.await;
let fragments = prepared_turn_input(
&harness.registry,
TurnInputContext {
step_store,
turn_id: "turn-1".to_string(),
user_input: vec![UserInput::Mention {
name: installed.name.clone(),
path: installed.main_prompt.as_str().to_string(),
}],
environments: Vec::new(),
},
/*extension_metrics*/ None,
&harness.session_store,
&harness.thread_store,
&harness.turn_store,
)
.await;
assert_eq!(fragments.len(), usize::from(allowed));
if allowed {
assert!(
@@ -1182,7 +1270,7 @@ async fn v1_retains_legacy_plugin_disablement() -> TestResult {
},
..Default::default()
});
let tools = harness.tools(&harness.step_store);
let tools = harness.tools(&harness.step_store).await;
let rendered = harness.render().await?;
assert!(rendered.contains("Cloud work description"));
assert!(!rendered.contains("Installed legacy description"));
@@ -1225,14 +1313,19 @@ async fn v2_shadow_observation_tracks_merged_skills_independently_of_read_route(
let cloud_provider = provider(vec![cloud.clone()]);
let executor_reads = executor.read_requests.clone();
let cloud_reads = cloud_provider.read_requests.clone();
let config = TestConfig {
include_instructions: false,
shadow_selection_enabled: true,
..default_config()
};
let updated_config = TestConfig {
shadow_selection_enabled: false,
..config.clone()
};
let mut harness = MergeHarness::with_metrics(
executor,
cloud_provider,
TestConfig {
include_instructions: false,
shadow_selection_enabled: true,
..default_config()
},
config.clone(),
Some(metrics.clone()),
)
.await?;
@@ -1249,8 +1342,15 @@ async fn v2_shadow_observation_tracks_merged_skills_independently_of_read_route(
harness
.turn_store
.insert(selected_plugins(&[None, Some("env-1:matching")], &[]));
let tools = harness.tools(&harness.step_store);
let tools = harness.tools(&harness.step_store).await;
harness.render().await?;
// Turn input must retain the config captured before this update.
harness.registry.config_contributors()[0].on_config_changed(
&harness.session_store,
&harness.thread_store,
&config,
&updated_config,
);
let fragments = harness.registry.turn_input_contributors()[0]
.contribute(
TurnInputContext {
@@ -1269,6 +1369,12 @@ async fn v2_shadow_observation_tracks_merged_skills_independently_of_read_route(
)
.await;
assert!(fragments.is_empty());
harness.registry.config_contributors()[0].on_config_changed(
&harness.session_store,
&harness.thread_store,
&updated_config,
&config,
);
if turn_id == "turn-1" {
let entry = if executor_route { &installed } else { &cloud };
// Repeated reads should yield only one shadow observation for this turn.
codex-rs/ext/skills/tests/skills_extension/preparation_tests.rs
+150
View File
@@ -0,0 +1,150 @@
//! Exercises config reloads while preparation is waiting on executor discovery.
use super::*;
use pretty_assertions::assert_eq;
use tokio::sync::Notify;
struct PausedExecutor {
entered: Notify,
resume: Notify,
queries: Mutex<Vec<(bool, bool)>>,
}
impl SkillProvider for PausedExecutor {
fn list(&self, query: SkillListQuery) -> SkillProviderFuture<'_, SkillCatalog> {
Box::pin(async move {
self.queries
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.push((query.include_bundled_skills, query.include_cloud_skills));
self.entered.notify_one();
self.resume.notified().await;
Ok(SkillCatalog::default())
})
}
fn read<'a>(
&'a self,
_request: SkillReadRequest<'a>,
) -> SkillProviderFuture<'a, SkillReadResult> {
Box::pin(async { panic!("executor reads are not expected") })
}
fn search(&self, _request: SkillSearchRequest) -> SkillProviderFuture<'_, SkillSearchResult> {
Box::pin(async { panic!("executor searches are not expected") })
}
}
#[tokio::test]
async fn config_reload_during_preparation_applies_to_next_step_but_reads_still_check_live_policy()
-> TestResult {
let executor = Arc::new(PausedExecutor {
entered: Notify::new(),
resume: Notify::new(),
queries: Mutex::default(),
});
let cloud = test_entry(
SkillSourceKind::Cloud,
"codex_apps",
"cloud/demo",
"skill://cloud/demo/SKILL.md",
);
let cloud_reads = Arc::default();
let providers = SkillProviders::new()
.with_executor_provider(executor.clone())
.with_cloud_provider(Arc::new(StaticSkillProvider {
catalog: SkillCatalog {
entries: vec![cloud],
..Default::default()
},
read_requests: Arc::clone(&cloud_reads),
list_calls: None,
fail_first_list: false,
}));
let mut builder = ExtensionRegistryBuilder::new();
install_with_providers(&mut builder, providers, skills_extension_config);
let registry = builder.build();
let session = ExtensionData::new("session");
let thread = ExtensionData::new("thread");
let turn = ExtensionData::new("turn");
let config = default_config();
let updated = TestConfig {
cloud_skill_enabled: false,
bundled_skills_enabled: false,
..config.clone()
};
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,
thread_store: &thread,
})
.await;
start_registered_turn(&registry, &session, &thread, "turn").await;
let roots = vec![SelectedCapabilityRoot {
id: "skills".into(),
location: CapabilityRootLocation::Environment {
environment_id: "env-1".into(),
path: PathUri::parse("file:///skills")?,
},
}];
let step = ExtensionData::new("step");
let next_step = ExtensionData::new("next-step");
let model_info = catalog_model_info();
for (store, expected_enabled) in [(&step, true), (&next_step, false)] {
let (sections, ()) = tokio::join!(
prepared_world_state(
&registry,
&roots,
WorldStateContributionInput {
previous_world_state: None,
step_store: store,
model_info: &model_info,
thread_id: codex_protocol::ThreadId::new(),
turn_id: "turn",
environments: &[],
extension_metrics: None,
session_store: &session,
thread_store: &thread,
turn_store: &turn,
}
),
async {
executor.entered.notified().await;
registry.config_contributors()[0]
.on_config_changed(&session, &thread, &config, &updated);
executor.resume.notify_one();
}
);
let rendered = sections
.iter()
.filter_map(|section| section.render_diff(PreviousWorldStateSection::Absent).1)
.map(|fragment| fragment.body().clone().expect_text())
.collect::<Vec<_>>()
.join("\n");
assert_eq!(
rendered.contains("cloud/demo"),
expected_enabled,
"{rendered}"
);
}
assert_eq!(
*executor.queries.lock().unwrap(),
vec![(true, true), (false, false)]
);
let tools = registry.tool_contributors()[0].tools_for_step(&session, &thread, &step);
let result = merge_catalog_tests::call_tool(
&tools,
"read",
serde_json::json!({"package": "cloud/demo"}),
)
.await;
assert_matches!(result, Err(FunctionCallError::RespondToModel(message)) if message == "failed to read skill resource");
assert!(read_request_keys(&cloud_reads).is_empty());
Ok(())
}
codex-rs/ext/skills/tests/skills_extension/presentation_dedup_tests.rs
+8 -7
View File
@@ -111,23 +111,24 @@ async fn cloud_preference_preserves_aliases_reads_and_executor_fallback() -> Tes
] {
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 {
let sections = prepared_world_state(
&registry,
if ready { &roots } else { &[] },
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,
executor_capability_discovery_v2: None,
extension_metrics: None,
session_store: &session_store,
thread_store: &thread_store,
turn_store: &turn_store,
step_store: &turn_store,
})
.await;
},
)
.await;
let snapshots: serde_json::Map<String, serde_json::Value> = sections
.iter()
.map(|section| {
codex-rs/ext/skills/tests/skills_extension/shadow_task_context_tests.rs
+22 -22
View File
@@ -165,28 +165,28 @@ async fn task_context_recovers_prior_requests_and_explicit_intent_without_changi
.await;
let observed_turn = turns.last().ok_or("test needs a turn")?.0;
for (turn_id, user_input) in turns {
let fragments = registry.turn_input_contributors()[0]
.contribute(
TurnInputContext {
step_store: &ExtensionData::new("step"),
turn_id: turn_id.to_string(),
user_input,
environments: vec![codex_extension_api::TurnInputEnvironment {
environment_id: "test".to_string(),
cwd: PathUri::from_host_native_path(
std::env::current_dir().expect("test cwd"),
)
.expect("absolute cwd"),
is_primary: true,
fs: &FileSystemEnvironmentAccessor::unrestricted(&LOCAL_FS),
}],
},
/*extension_metrics*/ None,
&session_store,
&thread_store,
&ExtensionData::new(turn_id),
)
.await;
let fragments = prepared_turn_input(
&registry,
TurnInputContext {
step_store: &ExtensionData::new("step"),
turn_id: turn_id.to_string(),
user_input,
environments: vec![codex_extension_api::TurnInputEnvironment {
environment_id: "test".to_string(),
cwd: PathUri::from_host_native_path(
std::env::current_dir().expect("test cwd"),
)
.expect("absolute cwd"),
is_primary: true,
fs: &FileSystemEnvironmentAccessor::unrestricted(&LOCAL_FS),
}],
},
/*extension_metrics*/ None,
&session_store,
&thread_store,
&ExtensionData::new(turn_id),
)
.await;
assert!(fragments.is_empty());
}
registry.skill_invocation_contributors()[0]