github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-19 16:17:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Enforce required environment skills before model inference (#51157)

Browse Source 
## What changed

- Accept per-environment `skills.required` lists through `environment/add` and `environments.toml`.
- Before inference, require each named skill to be enabled and supplied by the selected environment. Missing skills fail the turn before a model request, including when environment setup fails.
- Stop applying an environment's requirements after it is deselected. Exempt isolated Guardian reviewers, which have no skill catalog.

## Testing

Add tests for available, missing, disabled, and wrong-environment skills, failed environment setup, recovery after deselection, and propagation of TOML requirements to the correct environment.

GitOrigin-RevId: 302ced4774697d2f414f745fe0535e71f747eda2
This commit is contained in:
Adam Perry @ OpenAI
2026-10-05 18:54:17 +00:00
committed by copyberry
parent c9253c4977
commit 42312d4ff4
33 changed files with 615 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/schema/json/ClientRequest.json
Generated
+15
View File
@@ -867,6 +867,21 @@
}
]
},
"EnvironmentSkillsParams": {
"properties": {
"required": {
"description": "Exact catalog names that must be available from this environment.",
"items": {
"type": "string"
},
"type": [
"array",
"null"
]
}
},
"type": "object"
},
"ExperimentalFeatureEnablementSetParams": {
"properties": {
"enablement": {
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json
Generated
+15
View File
@@ -10983,6 +10983,21 @@
"title": "EnvironmentConnectionNotification",
"type": "object"
},
"EnvironmentSkillsParams": {
"properties": {
"required": {
"description": "Exact catalog names that must be available from this environment.",
"items": {
"type": "string"
},
"type": [
"array",
"null"
]
}
},
"type": "object"
},
"ErrorNotification": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json
Generated
+15
View File
@@ -6819,6 +6819,21 @@
"title": "EnvironmentConnectionNotification",
"type": "object"
},
"EnvironmentSkillsParams": {
"properties": {
"required": {
"description": "Exact catalog names that must be available from this environment.",
"items": {
"type": "string"
},
"type": [
"array",
"null"
]
}
},
"type": "object"
},
"ErrorNotification": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst
Generated
BIN
View File
Binary file not shown.
codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst
Generated
BIN
View File
Binary file not shown.
codex-rs/app-server-protocol/schema/typescript/v2/EnvironmentSkillsParams.ts
Generated
+9
View File
@@ -0,0 +1,9 @@
// GENERATED CODE! DO NOT MODIFY BY HAND!
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
export type EnvironmentSkillsParams = {
/**
* Exact catalog names that must be available from this environment.
*/
required?: Array<string> | null, };
codex-rs/app-server-protocol/schema/typescript/v2/index.ts
Generated
+1
View File
@@ -131,6 +131,7 @@ export type { DynamicToolNamespaceSpec } from "./DynamicToolNamespaceSpec";
export type { DynamicToolNamespaceTool } from "./DynamicToolNamespaceTool";
export type { DynamicToolSpec } from "./DynamicToolSpec";
export type { EnvironmentConnectionNotification } from "./EnvironmentConnectionNotification";
export type { EnvironmentSkillsParams } from "./EnvironmentSkillsParams";
export type { ErrorNotification } from "./ErrorNotification";
export type { ExecPolicyAmendment } from "./ExecPolicyAmendment";
export type { ExperimentalFeature } from "./ExperimentalFeature";
codex-rs/app-server-protocol/src/protocol/common.rs
+5 -1
View File
@@ -2587,6 +2587,7 @@ mod tests {
let environment_add = ClientRequest::EnvironmentAdd {
request_id: request_id(),
params: v2::EnvironmentAddParams {
skills: None,
auth_bearer_token: None,
environment_id: "remote-a".to_string(),
exec_server_url: "ws://127.0.0.1:8765".to_string(),
@@ -3841,6 +3842,7 @@ mod tests {
let request = ClientRequest::EnvironmentAdd {
request_id: RequestId::Integer(9),
params: v2::EnvironmentAddParams {
skills: None,
auth_bearer_token: Some("private-executor-token".into()),
environment_id: "remote-a".to_string(),
exec_server_url: "ws://127.0.0.1:8765".to_string(),
@@ -3856,7 +3858,8 @@ mod tests {
"environmentId": "remote-a",
"execServerUrl": "ws://127.0.0.1:8765",
"connectTimeoutMs": 300000,
"authBearerToken": "private-executor-token"
"authBearerToken": "private-executor-token",
"skills": null
}
}),
serde_json::to_value(&request)?,
@@ -4381,6 +4384,7 @@ mod tests {
let request = ClientRequest::EnvironmentAdd {
request_id: RequestId::Integer(1),
params: v2::EnvironmentAddParams {
skills: None,
auth_bearer_token: None,
environment_id: "remote-a".to_string(),
exec_server_url: "ws://127.0.0.1:8765".to_string(),
codex-rs/app-server-protocol/src/protocol/v2/environment.rs
+12
View File
@@ -47,6 +47,9 @@ pub struct EnvironmentAddParams {
#[ts(type = "number | null")]
#[ts(optional = nullable)]
pub connect_timeout_ms: Option<u64>,
/// Required skills supplied by this environment, checked before model inference.
#[ts(optional = nullable)]
pub skills: Option<EnvironmentSkillsParams>,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
@@ -54,6 +57,15 @@ pub struct EnvironmentAddParams {
#[ts(export_to = "v2/")]
pub struct EnvironmentAddResponse {}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct EnvironmentSkillsParams {
/// Exact catalog names that must be available from this environment.
#[ts(optional = nullable)]
pub required: Option<Vec<String>>,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(rename_all = "camelCase", export_to = "v2/")]
codex-rs/app-server/src/request_processors/environment_processor.rs
+8 -1
View File
@@ -1,4 +1,5 @@
use super::*;
use codex_config::ScopedSkillsConfig;
use codex_exec_server::RemoteEnvironmentOptions;
use std::time::Duration;
@@ -18,6 +19,12 @@ impl EnvironmentRequestProcessor {
&self,
params: EnvironmentAddParams,
) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
let skills = ScopedSkillsConfig {
required: params
.skills
.and_then(|skills| skills.required)
.unwrap_or_default(),
};
let options = RemoteEnvironmentOptions {
exec_server_url: params.exec_server_url,
connect_timeout: params.connect_timeout_ms.map(Duration::from_millis),
@@ -33,7 +40,7 @@ impl EnvironmentRequestProcessor {
.collect(),
};
self.environment_manager
.upsert_environment_with_options(params.environment_id, options)
.upsert_environment_with_options(params.environment_id, options, skills)
.map_err(|err| invalid_request(err.to_string()))?;
Ok(Some(EnvironmentAddResponse {}.into()))
}
codex-rs/app-server/tests/suite/v2/environment_status.rs
+1
View File
@@ -164,6 +164,7 @@ async fn add_environment(
exec_server_url: &str,
) -> Result<()> {
let params = EnvironmentAddParams {
skills: None,
auth_bearer_token: None,
environment_id: environment_id.to_string(),
exec_server_url: exec_server_url.to_string(),
codex-rs/config/src/lib.rs
+1
View File
@@ -211,6 +211,7 @@ pub use requirements_layers::compose_requirements;
pub use requirements_layers::compose_requirements_for_hostname;
pub use shell_environment_policy::validate_shell_environment_policy_filter_config;
pub use skills_config::BundledSkillsConfig;
pub use skills_config::ScopedSkillsConfig;
pub use skills_config::SkillConfig;
pub use skills_config::SkillConfigRule;
pub use skills_config::SkillConfigRuleSelector;
codex-rs/config/src/skills_config.rs
+8
View File
@@ -27,6 +27,14 @@ pub struct SkillConfig {
pub enabled: bool,
}
/// Skills required from one environment or plugin before inference.
#[derive(Serialize, Deserialize, Debug, Clone, Default, PartialEq, Eq, JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct ScopedSkillsConfig {
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub required: Vec<String>,
}
#[derive(Serialize, Deserialize, Debug, Clone, Default, PartialEq, Eq, JsonSchema)]
#[schemars(deny_unknown_fields)]
pub struct SkillsConfig {
codex-rs/core/src/context/world_state/environment.rs
+4 -1
View File
@@ -494,7 +494,10 @@ fn environment_states(snapshot: &TurnEnvironmentSnapshot) -> BTreeMap<String, En
const MAX_TOTAL_ERROR_BYTES: usize = 512;
let mut remaining_error_bytes = MAX_TOTAL_ERROR_BYTES;
for environment in &snapshot.environments {
if let TurnEnvironmentState::Failed { selection, error } = environment {
if let TurnEnvironmentState::Failed {
selection, error, ..
} = environment
{
let detail = error
[..error.floor_char_boundary(remaining_error_bytes.min(MAX_ERROR_BYTES))]
.to_string();
codex-rs/core/src/context/world_state/environment_tests.rs
+1
View File
@@ -424,6 +424,7 @@ fn failure_context_limits_total_detail_bytes_at_utf8_boundaries() {
let snapshot = TurnEnvironmentSnapshot {
environments: (0..4)
.map(|index| TurnEnvironmentState::Failed {
required_skills: Vec::new(),
selection: TurnEnvironmentSelection {
environment_id: format!("remote-{index}"),
cwd: PathUri::parse("file:///workspace").unwrap(),
codex-rs/core/src/environment_selection.rs
+35
View File
@@ -27,6 +27,7 @@ use codex_protocol::protocol::EnvironmentConnectionEvent;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::TurnEnvironmentSelection;
use codex_skills_extension::EnvironmentSkillRequirements;
use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_path_uri::PathUri;
use futures::FutureExt;
@@ -982,6 +983,7 @@ pub(crate) enum TurnEnvironmentState {
Failed {
// Keep the input form so connection failure doesn't hide whether config comes from the thread.
selection: TurnEnvironmentSelection,
required_skills: Vec<String>,
error: String,
},
}
@@ -991,9 +993,11 @@ impl TurnEnvironmentState {
starting: StartingTurnEnvironment,
resolved: Option<TurnEnvironmentResult>,
) -> Self {
let required_skills = starting.environment.required_skills().to_vec();
if let EnvironmentConfigState::Failed(error) = &starting.selection.config {
let error = error.clone();
return Self::Failed {
required_skills,
selection: starting
.config_origin
.into_input_selection(starting.selection),
@@ -1006,6 +1010,7 @@ impl TurnEnvironmentState {
if matches!(selection.config, EnvironmentConfigState::Pending) {
let Some(config) = environment.installed_config else {
return Self::Failed {
required_skills,
selection: starting.config_origin.into_input_selection(selection),
error: "Environment configuration was not supplied.".to_string(),
};
@@ -1035,6 +1040,7 @@ impl TurnEnvironmentState {
"skipping failed turn environment: {err}"
);
Self::Failed {
required_skills,
selection: starting
.config_origin
.into_input_selection(starting.selection),
@@ -1211,6 +1217,35 @@ impl TurnEnvironmentSnapshot {
.collect()
}
/// Captures requirements from the selected registration even when connection setup fails.
pub(crate) fn required_skills(&self) -> impl Iterator<Item = EnvironmentSkillRequirements<'_>> {
self.environments.iter().filter_map(|environment| {
let (selection, required) = match environment {
TurnEnvironmentState::Ready(environment) => (
&environment.selection,
environment.environment.required_skills(),
),
TurnEnvironmentState::Starting(environment) => (
&environment.selection,
environment.environment.required_skills(),
),
TurnEnvironmentState::Failed {
selection,
required_skills,
..
} => (selection, required_skills.as_slice()),
};
if required.is_empty() {
None
} else {
Some(EnvironmentSkillRequirements {
environment_id: &selection.environment_id,
skill_names: required,
})
}
})
}
/// Native children retain ready and starting attachments captured by their spawning step.
pub(crate) fn inheritable_selections(&self) -> Vec<TurnEnvironmentSelection> {
self.environments
codex-rs/core/src/guardian/tests.rs
+1
View File
@@ -747,6 +747,7 @@ async fn guardian_mcp_uses_thread_permissions_for_an_unavailable_captured_enviro
selection.config = codex_protocol::protocol::EnvironmentConfigState::Failed("offline".into());
let captured = crate::environment_selection::TurnEnvironmentSnapshot {
environments: vec![TurnEnvironmentState::Failed {
required_skills: Vec::new(),
selection,
error: "offline".into(),
}],
codex-rs/core/src/session/turn.rs
+11
View File
@@ -115,6 +115,7 @@ use codex_skills::collect_explicit_skill_mentions;
use codex_skills::tool_kind_for_path;
use codex_skills_extension::HostSkillPrompts;
use codex_skills_extension::InjectedHostSkillPrompts;
use codex_skills_extension::validate_required_skills;
use codex_thread_store::PersistContext;
use codex_tools::DiscoverableTool;
use codex_tools::ToolName;
@@ -508,6 +509,16 @@ pub(crate) async fn run_turn(
.record_step_world_state_if_changed(step_context.as_ref())
.await?;
// Isolated Guardian reviewers deliberately have no skill catalog.
if !crate::guardian::is_basic_session_source(&turn_context.session_source) {
validate_required_skills(
&sess.services.thread_extension_data,
&turn_context.extension_data,
step_context.environments.required_skills(),
)
.map_err(CodexErr::Fatal)?;
}
// Keep the override after accepted input so history truncation removes them together.
sess.record_reasoning_effort_override(step_context.as_ref())
.await;
codex-rs/core/src/tools/handlers/wait_for_environment.rs
+3 -5
View File
@@ -135,11 +135,9 @@ impl ToolExecutor<ToolInvocation> for WaitForEnvironmentHandler {
.environments
.iter()
.find_map(|state| match state {
TurnEnvironmentState::Failed { selection, error }
if selection.environment_id == environment_id =>
{
Some(error)
}
TurnEnvironmentState::Failed {
selection, error, ..
} if selection.environment_id == environment_id => Some(error),
_ => None,
})
{
codex-rs/core/tests/suite/remote_env.rs
+2
View File
@@ -2,6 +2,8 @@
mod capability_roots;
#[path = "guardian_environments_tests.rs"]
mod guardian_environments;
#[path = "remote_env_required_skills_tests.rs"]
mod required_skills;
#[path = "remote_env_spawn_tests.rs"]
pub(super) mod spawn_tests;
codex-rs/core/tests/suite/remote_env_required_skills_tests.rs
+165
View File
@@ -0,0 +1,165 @@
//! Required environment skills gate model requests and stop applying after deselection.
use super::super::skills_extension::CatalogSkillProvider;
use super::*;
use codex_config::ScopedSkillsConfig;
use codex_exec_server::RemoteEnvironmentOptions;
use codex_features::Feature;
use codex_skills_extension::SkillProviders;
use codex_skills_extension::SkillsExtensionConfig;
use codex_skills_extension::catalog::SkillAuthority;
use codex_skills_extension::catalog::SkillCatalog;
use codex_skills_extension::catalog::SkillCatalogEntry;
use codex_skills_extension::catalog::SkillPackageId;
use codex_skills_extension::catalog::SkillResourceId;
use codex_skills_extension::catalog::SkillSourceKind;
use codex_skills_extension::install_with_providers;
use core_test_support::responses::mount_sse_once;
use core_test_support::test_codex::test_env;
use pretty_assertions::assert_eq;
use test_case::test_case;
use tokio_util::task::AbortOnDropHandle;
#[derive(Clone, Copy)]
enum Availability {
Available,
Missing,
Disabled,
OtherEnvironment,
FailedEnvironment,
}
/// Only an enabled skill from the required environment permits inference; deselection recovers.
#[test_case(Availability::Available; "available")]
#[test_case(Availability::Missing; "missing")]
#[test_case(Availability::Disabled; "disabled")]
#[test_case(Availability::OtherEnvironment; "other environment")]
#[test_case(Availability::FailedEnvironment; "failed environment")]
#[tokio::test]
async fn required_environment_skill_gates_inference(availability: Availability) -> Result<()> {
skip_if_no_network!(Ok(()));
let environment = test_env().await?;
let primary = environment.selection().clone();
let source = if matches!(availability, Availability::OtherEnvironment) {
primary.environment_id.as_str()
} else {
"required"
};
let mut entry = SkillCatalogEntry::new(
SkillPackageId("test/review".to_string()),
SkillAuthority::new(SkillSourceKind::Executor, source),
"review",
"Review the workspace.",
SkillResourceId::environment(
"skill://test/review/SKILL.md",
source,
primary.cwd.join("SKILL.md")?,
),
);
entry.enabled = !matches!(availability, Availability::Disabled);
let catalog = SkillCatalog {
entries: if matches!(
availability,
Availability::Missing | Availability::FailedEnvironment
) {
Vec::new()
} else {
vec![entry]
},
warnings: Vec::new(),
};
let mut extensions = ExtensionRegistryBuilder::<Config>::new();
install_with_providers(
&mut extensions,
SkillProviders::new().with_executor_provider(Arc::new(CatalogSkillProvider { catalog })),
|_: &Config| SkillsExtensionConfig {
include_instructions: false,
max_context_tokens: None,
bundled_skills_enabled: false,
cloud_skill_enabled: false,
shadow_selection_enabled: false,
},
);
let server = start_mock_server().await;
let response = mount_sse_once(&server, sse(vec![ev_completed("done")])).await;
let test = test_codex()
.with_extensions(Arc::new(extensions.build()))
.with_config(|config| {
config.project_doc_max_bytes = 0;
config
.features
.enable(Feature::ExecutorCapabilityDiscovery)
.expect("enable executor capability discovery");
})
.build_with_environment(&server, environment)
.await?;
let listener = TcpListener::bind("127.0.0.1:0").await?;
let executor_url = format!("ws://{}", listener.local_addr()?);
let (attach, connection) = tokio::sync::oneshot::channel();
let (shutdown, stop) = tokio::sync::oneshot::channel();
let executor = AbortOnDropHandle::new(tokio::spawn(serve_environment_with_agents_md(
listener, "", connection, stop,
)));
attach.send(()).expect("attach required environment");
let manager = test.thread_manager.environment_manager();
manager.upsert_environment_with_options(
"required".to_string(),
RemoteEnvironmentOptions {
exec_server_url: executor_url,
connect_timeout: None,
http_headers: HashMap::new(),
},
ScopedSkillsConfig {
required: vec!["review".to_string()],
},
)?;
manager
.get_environment("required")
.context("required environment")?
.info()
.await?;
let mut required = TurnEnvironmentSelection {
environment_id: "required".to_string(),
..primary.clone()
};
if matches!(availability, Availability::FailedEnvironment) {
required.config = EnvironmentConfigState::Failed("configuration unavailable".to_string());
}
submit_turn_with_approval_and_environments(
&test,
"Review the workspace.",
vec![primary.clone(), required],
AskForApproval::Never,
)
.await?;
let mut errors = Vec::new();
wait_for_event(&test.codex, |event| {
if let EventMsg::Error(error) = event {
errors.push(error.message.clone());
}
matches!(event, EventMsg::TurnComplete(_))
})
.await;
if matches!(availability, Availability::Available) {
assert_eq!(errors, Vec::<String>::new());
assert_eq!(response.requests().len(), 1);
} else {
assert!(response.requests().is_empty());
assert_eq!(
errors,
vec![
"Fatal error: Required skill \"review\" from environment \"required\" is unavailable"
]
);
test.submit_turn_with_environments(
"Continue without that environment.",
Some(vec![primary]),
)
.await?;
assert_eq!(response.requests().len(), 1);
}
shutdown.send(()).expect("stop required environment");
executor.await?;
Ok(())
}
codex-rs/exec-server/src/client_refresh_tests.rs
+21 -8
View File
@@ -7,6 +7,7 @@ use std::time::Duration;
use anyhow::Result;
use bytes::Bytes;
use codex_config::ScopedSkillsConfig;
use futures::future::BoxFuture;
use pretty_assertions::assert_eq;
use tokio::net::TcpListener;
@@ -321,8 +322,11 @@ async fn registration_snapshot_changes_only_after_same_key_replacement_is_instal
};
let new = Executor::start_with_identity(validator.clone(), old.target.identity.clone()).await?;
let (client, registry) = old.client()?;
let environment =
crate::Environment::remote_with_client(client.clone(), /*local_runtime_paths*/ None);
let environment = crate::Environment::remote_with_client(
client.clone(),
/*local_runtime_paths*/ None,
ScopedSkillsConfig::default(),
);
assert_eq!(environment.cached_executor_registration_id(), None);
let original = client.get().await?;
assert_eq!(
@@ -364,8 +368,11 @@ async fn registration_renewal_recovers_the_session_and_running_process() -> Resu
)
.await?;
let (client, registry) = old.client()?;
let environment =
crate::Environment::remote_with_client(client.clone(), /*local_runtime_paths*/ None);
let environment = crate::Environment::remote_with_client(
client.clone(),
/*local_runtime_paths*/ None,
ScopedSkillsConfig::default(),
);
let original = client.get().await?;
let session_id = original.session_id();
let process =
@@ -450,8 +457,11 @@ async fn same_key_replacement_cannot_resume_a_missing_session() -> Result<()> {
let replacement =
Executor::start_with_identity(Validator::default(), old.target.identity.clone()).await?;
let (client, registry) = old.client()?;
let environment =
crate::Environment::remote_with_client(client.clone(), /*local_runtime_paths*/ None);
let environment = crate::Environment::remote_with_client(
client.clone(),
/*local_runtime_paths*/ None,
ScopedSkillsConfig::default(),
);
let original = client.get().await?;
let process = original
.register_session(&ProcessId::from("old-process"))
@@ -488,8 +498,11 @@ async fn stale_refresh_lookup_does_not_retire_a_renewed_session() -> Result<()>
)
.await?;
let (client, registry) = old.client()?;
let environment =
crate::Environment::remote_with_client(client.clone(), /*local_runtime_paths*/ None);
let environment = crate::Environment::remote_with_client(
client.clone(),
/*local_runtime_paths*/ None,
ScopedSkillsConfig::default(),
);
let original = client.get().await?;
let release = registry.block_next_lookup();
let refreshing = client.refresh_connection();
codex-rs/exec-server/src/environment.rs
+73 -19
View File
@@ -7,6 +7,7 @@ use std::sync::Mutex;
use std::sync::RwLock;
use arc_swap::ArcSwapOption;
use codex_config::ScopedSkillsConfig;
use codex_http_client::HttpClientFactory;
use codex_http_client::OutboundProxyPolicy;
use codex_protocol::capabilities::CapabilityRootLocation;
@@ -33,6 +34,7 @@ use crate::environment_config::read_environment_config;
use crate::environment_provider::DefaultEnvironmentProvider;
use crate::environment_provider::EnvironmentDefault;
use crate::environment_provider::EnvironmentProvider;
use crate::environment_provider::EnvironmentProviderEntry;
use crate::environment_provider::EnvironmentProviderSnapshot;
use crate::environment_provider::normalize_exec_server_url;
use crate::environment_toml::environment_provider_from_codex_home;
@@ -245,6 +247,7 @@ impl EnvironmentManager {
},
manager.local_runtime_paths.clone(),
manager.http_client_factory.clone(),
ScopedSkillsConfig::default(),
));
manager.insert_environment(REMOTE_ENVIRONMENT_ID.to_string(), environment)?;
Ok(manager)
@@ -299,7 +302,12 @@ impl EnvironmentManager {
} else {
None
};
for (id, transport) in environments {
for EnvironmentProviderEntry {
id,
transport,
skills,
} in environments
{
if id.is_empty() {
return Err(ExecServerError::Protocol(
"environment id cannot be empty".to_string(),
@@ -314,6 +322,7 @@ impl EnvironmentManager {
transport,
/*local_runtime_paths*/ None,
http_client_factory.clone(),
skills,
);
if environment_map
.insert(id.clone(), Arc::new(environment))
@@ -477,6 +486,7 @@ impl EnvironmentManager {
connect_timeout,
http_headers: HashMap::new(),
},
ScopedSkillsConfig::default(),
)
}
@@ -489,15 +499,17 @@ impl EnvironmentManager {
&self,
environment_id: String,
options: RemoteEnvironmentOptions,
skills: ScopedSkillsConfig,
) -> Result<(), ExecServerError> {
validate_environment_id(&environment_id)?;
let transport = options.into_transport_params()?;
let environment = Arc::new(Environment::remote_with_transport(
let environment = Environment::remote_with_transport(
transport,
self.local_runtime_paths.clone(),
self.http_client_factory.clone(),
));
self.insert_environment(environment_id, environment)
skills,
);
self.insert_environment(environment_id, Arc::new(environment))
}
/// Returns the stable environment for an ID, creating it as pending when absent.
@@ -538,6 +550,7 @@ impl EnvironmentManager {
})),
self.local_runtime_paths.clone(),
self.http_client_factory.clone(),
ScopedSkillsConfig::default(),
);
environment.provisioning_status_tx = Some(provisioning_status_tx);
Ok(environment)
@@ -665,6 +678,7 @@ fn optional_environment_value(name: &str) -> Option<String> {
/// paths used by filesystem helpers.
#[derive(Clone)]
pub struct Environment {
skills: ScopedSkillsConfig,
remote_client: Option<LazyRemoteExecServerClient>,
ready_info: Arc<ArcSwapOption<EnvironmentReadyInfo>>,
// No sender means an ordinary environment. A provisioned environment retains a sender whose
@@ -683,6 +697,7 @@ impl Environment {
pub fn default_for_tests() -> Self {
Self {
remote_client: None,
skills: ScopedSkillsConfig::default(),
ready_info: Arc::new(ArcSwapOption::empty()),
provisioning_status_tx: None,
startup_task: Arc::new(Mutex::new(None)),
@@ -748,6 +763,7 @@ impl Environment {
),
local_runtime_paths,
http_client_factory,
ScopedSkillsConfig::default(),
),
None => match local_runtime_paths {
Some(local_runtime_paths) => Self::local(local_runtime_paths, http_client_factory),
@@ -762,6 +778,7 @@ impl Environment {
) -> Self {
Self {
remote_client: None,
skills: ScopedSkillsConfig::default(),
ready_info: Arc::new(ArcSwapOption::empty()),
provisioning_status_tx: None,
startup_task: Arc::new(Mutex::new(None)),
@@ -780,14 +797,16 @@ impl Environment {
remote_transport: ExecServerTransportParams,
local_runtime_paths: Option<ExecServerRuntimeOptions>,
http_client_factory: HttpClientFactory,
skills: ScopedSkillsConfig,
) -> Self {
let client = LazyRemoteExecServerClient::new(remote_transport, http_client_factory);
Self::remote_with_client(client, local_runtime_paths)
Self::remote_with_client(client, local_runtime_paths, skills)
}
pub(crate) fn remote_with_client(
client: LazyRemoteExecServerClient,
local_runtime_paths: Option<ExecServerRuntimeOptions>,
skills: ScopedSkillsConfig,
) -> Self {
let exec_backend: Arc<dyn ExecBackend> = Arc::new(RemoteProcess::new(client.clone()));
let filesystem: Arc<dyn ExecutorFileSystem> =
@@ -795,6 +814,7 @@ impl Environment {
Self {
remote_client: Some(client.clone()),
skills,
ready_info: Arc::new(ArcSwapOption::empty()),
provisioning_status_tx: None,
startup_task: Arc::new(Mutex::new(None)),
@@ -805,6 +825,11 @@ impl Environment {
}
}
/// Skill names required whenever this registration is selected by a turn.
pub fn required_skills(&self) -> &[String] {
&self.skills.required
}
pub fn is_remote(&self) -> bool {
self.remote_client.is_some()
}
@@ -1177,7 +1202,9 @@ mod tests {
use crate::client_api::ExecServerTransportParams;
use crate::client_api::StdioExecServerCommand;
use crate::environment_provider::EnvironmentDefault;
use crate::environment_provider::EnvironmentProviderEntry;
use crate::environment_provider::EnvironmentProviderSnapshot;
use codex_config::ScopedSkillsConfig;
use codex_http_client::HttpClientFactory;
use codex_http_client::OutboundProxyPolicy;
use codex_utils_path_uri::PathUri;
@@ -1354,10 +1381,11 @@ mod tests {
#[tokio::test]
async fn environment_manager_builds_from_snapshot() {
let snapshot = EnvironmentProviderSnapshot {
environments: vec![(
REMOTE_ENVIRONMENT_ID.to_string(),
prepared_websocket_environment(),
)],
environments: vec![EnvironmentProviderEntry {
id: REMOTE_ENVIRONMENT_ID.to_string(),
transport: prepared_websocket_environment(),
skills: Default::default(),
}],
default: EnvironmentDefault::EnvironmentId(REMOTE_ENVIRONMENT_ID.to_string()),
include_local: false,
};
@@ -1385,7 +1413,11 @@ mod tests {
#[tokio::test]
async fn environment_manager_rejects_empty_environment_id() {
let snapshot = EnvironmentProviderSnapshot {
environments: vec![("".to_string(), prepared_websocket_environment())],
environments: vec![EnvironmentProviderEntry {
id: "".to_string(),
transport: prepared_websocket_environment(),
skills: Default::default(),
}],
default: EnvironmentDefault::Disabled,
include_local: false,
};
@@ -1405,10 +1437,11 @@ mod tests {
#[tokio::test]
async fn environment_manager_rejects_provider_supplied_local_environment() {
let snapshot = EnvironmentProviderSnapshot {
environments: vec![(
LOCAL_ENVIRONMENT_ID.to_string(),
prepared_websocket_environment(),
)],
environments: vec![EnvironmentProviderEntry {
id: LOCAL_ENVIRONMENT_ID.to_string(),
transport: prepared_websocket_environment(),
skills: Default::default(),
}],
default: EnvironmentDefault::Disabled,
include_local: false,
};
@@ -1428,7 +1461,11 @@ mod tests {
#[tokio::test]
async fn environment_manager_uses_explicit_provider_default() {
let snapshot = EnvironmentProviderSnapshot {
environments: vec![("devbox".to_string(), prepared_websocket_environment())],
environments: vec![EnvironmentProviderEntry {
id: "devbox".to_string(),
transport: prepared_websocket_environment(),
skills: Default::default(),
}],
default: EnvironmentDefault::EnvironmentId("devbox".to_string()),
include_local: true,
};
@@ -1450,7 +1487,11 @@ mod tests {
#[tokio::test]
async fn environment_manager_disables_provider_default() {
let snapshot = EnvironmentProviderSnapshot {
environments: vec![("devbox".to_string(), prepared_websocket_environment())],
environments: vec![EnvironmentProviderEntry {
id: "devbox".to_string(),
transport: prepared_websocket_environment(),
skills: Default::default(),
}],
default: EnvironmentDefault::Disabled,
include_local: true,
};
@@ -1474,7 +1515,11 @@ mod tests {
#[tokio::test]
async fn environment_manager_rejects_unknown_provider_default() {
let snapshot = EnvironmentProviderSnapshot {
environments: vec![("devbox".to_string(), prepared_websocket_environment())],
environments: vec![EnvironmentProviderEntry {
id: "devbox".to_string(),
transport: prepared_websocket_environment(),
skills: Default::default(),
}],
default: EnvironmentDefault::EnvironmentId("missing".to_string()),
include_local: true,
};
@@ -1650,6 +1695,7 @@ mod tests {
},
/*local_runtime_paths*/ None,
legacy_http_client_factory(),
ScopedSkillsConfig::default(),
);
assert_eq!(
@@ -1672,7 +1718,11 @@ mod tests {
};
let manager = EnvironmentManager::from_snapshot(
EnvironmentProviderSnapshot {
environments: vec![("stdio".to_string(), transport)],
environments: vec![EnvironmentProviderEntry {
id: "stdio".to_string(),
transport,
skills: Default::default(),
}],
default: EnvironmentDefault::Disabled,
include_local: false,
},
@@ -1703,7 +1753,11 @@ mod tests {
};
let manager = EnvironmentManager::from_snapshot(
EnvironmentProviderSnapshot {
environments: vec![("stdio".to_string(), transport)],
environments: vec![EnvironmentProviderEntry {
id: "stdio".to_string(),
transport,
skills: Default::default(),
}],
default: EnvironmentDefault::Disabled,
include_local: false,
},
codex-rs/exec-server/src/environment/accepted.rs
+4 -1
View File
@@ -10,6 +10,7 @@ use crate::ExecServerClientConnectOptions;
use crate::ExecServerError;
use crate::client::LazyRemoteExecServerClient;
use axum::extract::ws::WebSocket;
use codex_config::ScopedSkillsConfig;
use codex_http_client::HttpClientFactory;
impl EnvironmentManager {
@@ -29,7 +30,9 @@ impl EnvironmentManager {
let client =
LazyRemoteExecServerClient::from_connected(client, http_client_factory.clone());
let environment = Arc::new(Environment::remote_with_client(
client, /*local_runtime_paths*/ None,
client,
/*local_runtime_paths*/ None,
ScopedSkillsConfig::default(),
));
Ok(Self {
default_environment: Some(environment_id.clone()),
codex-rs/exec-server/src/environment/connect_options_tests.rs
+7
View File
@@ -1,6 +1,7 @@
use std::collections::HashMap;
use std::time::Duration;
use codex_config::ScopedSkillsConfig;
use codex_http_client::HttpClientFactory;
use codex_http_client::OutboundProxyPolicy;
use futures::SinkExt;
@@ -59,6 +60,7 @@ fn trusted_headers_require_tls_for_non_loopback_destinations() {
"customer-session".to_string(),
)]),
},
ScopedSkillsConfig::default(),
)
.expect_err("trusted headers must not be sent over an insecure remote connection");
@@ -86,6 +88,7 @@ fn duplicate_case_insensitive_websocket_headers_fail_before_registration() {
("x-session-id".to_string(), "second-session".to_string()),
]),
},
ScopedSkillsConfig::default(),
)
.expect_err("duplicate header names must fail regardless of case");
@@ -110,6 +113,7 @@ fn invalid_websocket_header_names_fail_before_registration() {
connect_timeout: None,
http_headers: HashMap::from([("bad header".to_string(), "value".to_string())]),
},
ScopedSkillsConfig::default(),
)
.expect_err("invalid header name should fail");
@@ -137,6 +141,7 @@ fn invalid_websocket_header_values_fail_before_registration() {
"customer\nspoofed".to_string(),
)]),
},
ScopedSkillsConfig::default(),
)
.expect_err("invalid header value should fail");
@@ -162,6 +167,7 @@ fn websocket_controlled_headers_fail_before_registration() {
connect_timeout: None,
http_headers: HashMap::from([(header.to_string(), "overridden".to_string())]),
},
ScopedSkillsConfig::default(),
)
.expect_err("connection-controlled header should fail");
@@ -228,6 +234,7 @@ async fn trusted_headers_are_sent_on_initial_websocket_and_session_reconnect() {
),
]),
},
ScopedSkillsConfig::default(),
)
.expect("environment with routing headers should register");
codex-rs/exec-server/src/environment_bootstrap.rs
+1 -1
View File
@@ -33,7 +33,7 @@ impl PreparedEnvironmentManager {
EnvironmentDefault::EnvironmentId(default_id) => snapshot
.environments
.iter()
.any(|(environment_id, _)| environment_id == default_id),
.any(|entry| &entry.id == default_id),
},
}
}
codex-rs/exec-server/src/environment_bootstrap_tests.rs
+6 -4
View File
@@ -10,6 +10,7 @@ use crate::REMOTE_ENVIRONMENT_ID;
use crate::client_api::DEFAULT_REMOTE_EXEC_SERVER_CONNECT_TIMEOUT;
use crate::client_api::ExecServerTransportParams;
use crate::environment_provider::EnvironmentDefault;
use crate::environment_provider::EnvironmentProviderEntry;
use crate::environment_provider::EnvironmentProviderSnapshot;
use crate::remote::NoiseRendezvousEnvironmentConfig;
@@ -17,13 +18,14 @@ use crate::remote::NoiseRendezvousEnvironmentConfig;
fn prepared_remote_environment_is_detected_without_constructing_a_connection() {
let prepared = PreparedEnvironmentManager {
source: PreparedEnvironmentSource::Snapshot(EnvironmentProviderSnapshot {
environments: vec![(
REMOTE_ENVIRONMENT_ID.to_string(),
ExecServerTransportParams::websocket_url(
environments: vec![EnvironmentProviderEntry {
id: REMOTE_ENVIRONMENT_ID.to_string(),
transport: ExecServerTransportParams::websocket_url(
"ws://username:password@executor.example/private?token=secret".to_string(),
DEFAULT_REMOTE_EXEC_SERVER_CONNECT_TIMEOUT,
),
)],
skills: Default::default(),
}],
default: EnvironmentDefault::EnvironmentId(REMOTE_ENVIRONMENT_ID.to_string()),
include_local: false,
}),
codex-rs/exec-server/src/environment_provider.rs
+34 -11
View File
@@ -1,6 +1,8 @@
use std::future::Future;
use std::pin::Pin;
use codex_config::ScopedSkillsConfig;
use crate::ExecServerError;
use crate::client_api::DEFAULT_REMOTE_EXEC_SERVER_CONNECT_TIMEOUT;
use crate::client_api::ExecServerTransportParams;
@@ -25,14 +27,22 @@ pub type EnvironmentProviderFuture<'a> =
#[derive(Clone)]
pub struct EnvironmentProviderSnapshot {
pub(crate) environments: Vec<(String, ExecServerTransportParams)>,
pub(crate) environments: Vec<EnvironmentProviderEntry>,
pub default: EnvironmentDefault,
pub include_local: bool,
}
/// Keeps each configured environment's requirements with its transport.
#[derive(Clone, Debug)]
pub(crate) struct EnvironmentProviderEntry {
pub(crate) id: String,
pub(crate) transport: ExecServerTransportParams,
pub(crate) skills: ScopedSkillsConfig,
}
impl std::fmt::Debug for EnvironmentProviderSnapshot {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let environment_ids: Vec<_> = self.environments.iter().map(|(id, _)| id).collect();
let environment_ids: Vec<_> = self.environments.iter().map(|entry| &entry.id).collect();
f.debug_struct("EnvironmentProviderSnapshot")
.field("environments", &environment_ids)
.field("default", &self.default)
@@ -69,18 +79,19 @@ impl DefaultEnvironmentProvider {
let (exec_server_url, disabled) = normalize_exec_server_url(self.exec_server_url.clone());
if let Some(exec_server_url) = exec_server_url {
environments.push((
REMOTE_ENVIRONMENT_ID.to_string(),
ExecServerTransportParams::websocket_url(
environments.push(EnvironmentProviderEntry {
id: REMOTE_ENVIRONMENT_ID.to_string(),
transport: ExecServerTransportParams::websocket_url(
exec_server_url,
DEFAULT_REMOTE_EXEC_SERVER_CONNECT_TIMEOUT,
),
));
skills: ScopedSkillsConfig::default(),
});
}
let has_remote = environments
.iter()
.any(|(id, _environment)| id == REMOTE_ENVIRONMENT_ID);
.any(|entry| entry.id == REMOTE_ENVIRONMENT_ID);
let include_local = !disabled && !has_remote;
let default = if disabled {
EnvironmentDefault::Disabled
@@ -129,7 +140,10 @@ mod tests {
default,
include_local,
} = snapshot;
let environments: HashMap<_, _> = environments.into_iter().collect();
let environments: HashMap<_, _> = environments
.into_iter()
.map(|entry| (entry.id, entry.transport))
.collect();
assert!(include_local);
assert!(!environments.contains_key(LOCAL_ENVIRONMENT_ID));
@@ -149,7 +163,10 @@ mod tests {
default,
include_local,
} = snapshot;
let environments: HashMap<_, _> = environments.into_iter().collect();
let environments: HashMap<_, _> = environments
.into_iter()
.map(|entry| (entry.id, entry.transport))
.collect();
assert!(include_local);
assert!(!environments.contains_key(LOCAL_ENVIRONMENT_ID));
@@ -169,7 +186,10 @@ mod tests {
default,
include_local,
} = snapshot;
let environments: HashMap<_, _> = environments.into_iter().collect();
let environments: HashMap<_, _> = environments
.into_iter()
.map(|entry| (entry.id, entry.transport))
.collect();
assert!(!include_local);
assert!(!environments.contains_key(LOCAL_ENVIRONMENT_ID));
@@ -186,7 +206,10 @@ mod tests {
default,
include_local,
} = snapshot;
let environments: HashMap<_, _> = environments.into_iter().collect();
let environments: HashMap<_, _> = environments
.into_iter()
.map(|entry| (entry.id, entry.transport))
.collect();
assert!(!include_local);
assert!(!environments.contains_key(LOCAL_ENVIRONMENT_ID));
codex-rs/exec-server/src/environment_toml.rs
+71 -18
View File
@@ -5,6 +5,7 @@ use std::path::PathBuf;
use std::time::Duration;
use crate::RemoteEnvironmentOptions;
use codex_config::ScopedSkillsConfig;
use codex_utils_redacted_string::RedactedString;
use serde::Deserialize;
use tokio_tungstenite::tungstenite::client::IntoClientRequest;
@@ -19,6 +20,7 @@ use crate::client_api::ExecServerTransportParams;
use crate::client_api::StdioExecServerCommand;
use crate::environment::LOCAL_ENVIRONMENT_ID;
use crate::environment_provider::EnvironmentDefault;
use crate::environment_provider::EnvironmentProviderEntry;
use crate::environment_provider::EnvironmentProviderSnapshot;
const ENVIRONMENTS_TOML_FILE: &str = "environments.toml";
@@ -38,6 +40,10 @@ struct EnvironmentsToml {
#[serde(deny_unknown_fields)]
struct EnvironmentToml {
id: String,
// Requirements for this [[environments]] entry. environment/add sets the same
// per-environment value dynamically instead of loading it from this file.
#[serde(default)]
skills: ScopedSkillsConfig,
url: Option<String>,
auth_bearer_token: Option<RedactedString>,
program: Option<String>,
@@ -54,7 +60,7 @@ struct EnvironmentToml {
struct TomlEnvironmentProvider {
default: EnvironmentDefault,
include_local: bool,
environments: Vec<(String, ExecServerTransportParams)>,
environments: Vec<EnvironmentProviderEntry>,
}
impl TomlEnvironmentProvider {
@@ -79,13 +85,14 @@ impl TomlEnvironmentProvider {
}
let mut parsed_environments = Vec::with_capacity(environments.len());
for item in environments {
let (id, transport) = parse_environment_toml(item, config_dir)?;
if !ids.insert(id.clone()) {
let entry = parse_environment_toml(item, config_dir)?;
if !ids.insert(entry.id.clone()) {
return Err(ExecServerError::Protocol(format!(
"environment id `{id}` is duplicated"
"environment id `{}` is duplicated",
entry.id
)));
}
parsed_environments.push((id, transport));
parsed_environments.push(entry);
}
let default = normalize_default_environment_id(default.as_deref(), include_local, &ids)?;
Ok(Self {
@@ -113,9 +120,10 @@ impl EnvironmentProvider for TomlEnvironmentProvider {
fn parse_environment_toml(
item: EnvironmentToml,
config_dir: Option<&Path>,
) -> Result<(String, ExecServerTransportParams), ExecServerError> {
) -> Result<EnvironmentProviderEntry, ExecServerError> {
let EnvironmentToml {
id,
skills,
url,
auth_bearer_token,
program,
@@ -197,7 +205,11 @@ fn parse_environment_toml(
}
};
Ok((id, transport_params))
Ok(EnvironmentProviderEntry {
id,
transport: transport_params,
skills,
})
}
fn normalize_stdio_cwd(
@@ -357,6 +369,9 @@ mod option_duration_secs {
#[cfg(test)]
mod tests {
use crate::EnvironmentManager;
use codex_http_client::HttpClientFactory;
use codex_http_client::OutboundProxyPolicy;
use pretty_assertions::assert_eq;
use tempfile::tempdir;
@@ -380,7 +395,7 @@ mod tests {
http_headers,
initialize_timeout,
..
} = &provider.environments[0].1
} = &provider.environments[0].transport
else {
panic!("expected websocket")
};
@@ -437,12 +452,12 @@ mod tests {
default,
include_local,
} = snapshot;
let environment_ids: Vec<_> = environments
.iter()
.map(|(id, _environment)| id.as_str())
.collect();
let environment_ids: Vec<_> = environments.iter().map(|entry| entry.id.as_str()).collect();
assert_eq!(environment_ids, vec!["devbox", "ssh-dev"]);
let environments: HashMap<_, _> = environments.into_iter().collect();
let environments: HashMap<_, _> = environments
.into_iter()
.map(|entry| (entry.id, entry.transport))
.collect();
assert!(include_local);
assert!(!environments.contains_key(LOCAL_ENVIRONMENT_ID));
@@ -642,7 +657,7 @@ mod tests {
let ExecServerTransportParams::StdioCommand {
command,
initialize_timeout,
} = &provider.environments[0].1
} = &provider.environments[0].transport
else {
panic!("expected stdio transport");
};
@@ -689,7 +704,7 @@ mod tests {
connect_timeout,
initialize_timeout,
..
} = &provider.environments[0].1
} = &provider.environments[0].transport
else {
panic!("expected websocket transport");
};
@@ -700,7 +715,7 @@ mod tests {
let ExecServerTransportParams::StdioCommand {
command,
initialize_timeout,
} = &provider.environments[1].1
} = &provider.environments[1].transport
else {
panic!("expected stdio transport");
};
@@ -799,6 +814,44 @@ mod tests {
);
}
/// Requirements from a TOML entry reach that environment without affecting other entries.
#[tokio::test]
async fn required_skills_from_toml_reach_their_environment() {
let codex_home = tempdir().expect("tempdir");
std::fs::write(
codex_home.path().join(ENVIRONMENTS_TOML_FILE),
r#"
include_local = false
[[environments]]
id = "training"
url = "ws://127.0.0.1:4512"
[environments.skills]
required = ["computer-use"]
[[environments]]
id = "other"
url = "ws://127.0.0.1:4513"
"#,
)
.expect("write environments.toml");
let provider =
environment_provider_from_codex_home(codex_home.path()).expect("environment provider");
let manager = EnvironmentManager::from_snapshot(
provider.snapshot().await.expect("environment snapshot"),
/*local_runtime_paths*/ None,
HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault),
)
.expect("environment manager");
let requirements = ["training", "other"].map(|id| {
manager
.get_environment(id)
.expect("registered environment")
.required_skills()
.to_vec()
});
assert_eq!(requirements, [vec!["computer-use".to_string()], Vec::new()]);
}
#[test]
fn load_environments_toml_reads_root_environment_list() {
let codex_home = tempdir().expect("tempdir");
@@ -930,7 +983,7 @@ include_local = false
let environment_ids: Vec<_> = snapshot
.environments
.into_iter()
.map(|(id, _environment)| id)
.map(|entry| entry.id)
.collect();
assert!(!snapshot.include_local);
@@ -949,7 +1002,7 @@ include_local = false
let environment_ids: Vec<_> = snapshot
.environments
.into_iter()
.map(|(id, _environment)| id)
.map(|entry| entry.id)
.collect();
assert!(snapshot.include_local);
codex-rs/ext/skills/src/lib.rs
+3
View File
@@ -16,6 +16,7 @@ mod loader;
pub mod provider;
mod render;
mod render_observability;
mod required;
mod selection;
mod shadow_selection_experiment;
mod skills_extension_state;
@@ -42,6 +43,8 @@ pub use invocation::detect_implicit_skill_invocation;
pub use provider::ExecutorSkillProvider;
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;
codex-rs/ext/skills/src/required.rs
+57
View File
@@ -0,0 +1,57 @@
//! Checks source-scoped requirements against skills discovered before inference.
use codex_extension_api::ExtensionData;
use crate::state::ExecutorSkillsStepState;
use crate::state::SkillsThreadState;
/// Skill names that must be supplied by one selected environment before inference.
pub struct EnvironmentSkillRequirements<'a> {
pub environment_id: &'a str,
pub skill_names: &'a [String],
}
/// Validates requirements after discovery refreshes the turn's skill snapshots.
pub fn validate_required_skills<'a>(
thread_store: &ExtensionData,
turn_store: &ExtensionData,
environments: impl IntoIterator<Item = EnvironmentSkillRequirements<'a>>,
) -> Result<(), String> {
let mut environments = environments.into_iter().peekable();
if environments.peek().is_none() {
return Ok(());
}
let thread_state = thread_store.get::<SkillsThreadState>().ok_or_else(|| {
"Required skills cannot be validated because the skills extension is unavailable"
.to_string()
})?;
let executor = turn_store.get::<ExecutorSkillsStepState>();
thread_state.with_cloud_catalog(|cloud| {
for EnvironmentSkillRequirements {
environment_id,
skill_names,
} in environments
{
for name in skill_names {
let available = executor
.iter()
.flat_map(|state| &state.0.entries)
.chain(cloud.into_iter().flat_map(|catalog| &catalog.entries))
.any(|entry| {
entry.enabled
&& entry.name == name.as_str()
&& entry
.main_prompt
.environment_path()
.is_some_and(|(id, _)| id == environment_id)
});
if !available {
return Err(format!(
"Required skill {name:?} from environment {environment_id:?} is unavailable"
));
}
}
}
Ok(())
})
}
codex-rs/ext/skills/src/state.rs
+16 -8
View File
@@ -235,21 +235,29 @@ impl SkillsThreadState {
/// Reads the last authorized turn-start catalog without discovery or retries.
/// Same-auth failures may retain it; disabled or invalidated catalogs are empty.
pub fn cloud_catalog_snapshot(&self) -> SkillCatalog {
self.with_cloud_catalog(|catalog| catalog.cloned().unwrap_or_default())
}
/// Borrows the authorized catalog under the cache lock. The reader must not re-enter this state.
pub(crate) fn with_cloud_catalog<T>(&self, read: impl FnOnce(Option<&SkillCatalog>) -> T) -> T {
let config = self
.config
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if !config.cloud_skill_enabled || !self.cloud_skills_available {
return SkillCatalog::default();
return read(None);
}
self.skills_extension_state
let state = self
.skills_extension_state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.cloud_cache
.as_ref()
.filter(|cache| cache.is_current())
.and_then(|cache| cache.catalog.clone())
.unwrap_or_default()
.unwrap_or_else(std::sync::PoisonError::into_inner);
read(
state
.cloud_cache
.as_ref()
.filter(|cache| cache.is_current())
.and_then(|cache| cache.catalog.as_ref()),
)
}
fn cloud_cache(&self, mcp_resources: Option<&McpResourceClient>) -> Arc<CloudSkillGeneration> {
sdk/python/src/openai_codex/generated/v2_all.py
+10
View File
@@ -1302,6 +1302,16 @@ class EnvironmentConnectionNotification(BaseModel):
thread_id: Annotated[str, Field(alias="threadId")]
class EnvironmentSkillsParams(BaseModel):
model_config = ConfigDict(
populate_by_name=True,
)
required: Annotated[
list[str] | None,
Field(description="Exact catalog names that must be available from this environment."),
] = None
class ExperimentalFeatureEnablementSetParams(BaseModel):
model_config = ConfigDict(
populate_by_name=True,