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

Preserve turn configuration in Guardian scoring and approvals (#59642)

Browse Source 
## Why

Guardian scoring and cached approvals need to use the requesting turn's captured policy even when the live thread configuration changes. Newly tightened mandatory review requirements must still prevent cached approvals.

## What changed

- Pass captured turn configuration through tool lifecycle and approval review inputs, and use it to resolve Guardian policy and classifier instructions.
- Apply current reviewer and model requirements before scoring and cached approval, including rechecking restrictions after collecting score authorization.
- Count computer-use JavaScript execution attempts even when background scoring is disabled, so a synchronous call consumes the initial-call allowance.
- Retain runtime configuration snapshots when applying agent role overrides, and share Guardian policy text resolution between configuration loading and runtime publication.

## Testing

Add and extend regression coverage for captured policy instructions, reviewer restrictions, stale cached evidence, and a synchronous computer-use call followed by an adaptive turn.

GitOrigin-RevId: 24ed4a1b6ce1733ad775790f30e440da614f9cd6
This commit is contained in:
ioanbudea-OAI
2026-11-05 11:41:09 +00:00
committed by copyberry
parent 2be3dd3ff3
commit 86ac091659
23 changed files with 532 additions and 174 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/tests/suite/v2/guardian_v2.rs
+92
View File
@@ -2559,6 +2559,98 @@ async fn first_cua_review_does_not_wait_for_initial_score(
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn synchronous_cua_call_consumes_the_initial_call_allowance() -> Result<()> {
skip_if_no_network!(Ok(()));
const SYNCHRONOUS_MODEL: &str = "synchronous-cua";
let state = Arc::new(MockResponsesState {
mcp_server_name: Some("cua_repl"),
mcp_tool_sequence: Some(&["js"]),
..Default::default()
});
let listener = TcpListener::bind("127.0.0.1:0").await?;
let responses_url = format!("http://{}", listener.local_addr()?);
let router = Router::new()
.route("/v1/responses", get(luna_websocket).post(parent_response))
.with_state(Arc::clone(&state));
let responses_server = tokio::spawn(async move {
let _ = axum::serve(listener, router).await;
});
let (mcp_url, mcp_server) =
start_mcp_server_with_tools(&["js"], /*sensitive_action*/ None).await?;
let codex_home = TempDir::new()?;
MockResponsesConfig::new(&responses_url)
.with_model(SYNCHRONOUS_MODEL)
.with_provider_config("supports_websockets = false")
.with_approval_policy("on-request")
.with_root_config("approvals_reviewer = \"auto_review\"")
.with_extra_config(&format!(
"[mcp_servers.cua_repl]\nurl = \"{mcp_url}/mcp\"\ndefault_tools_approval_mode = \"auto\""
))
.enable_feature(Feature::GuardianApproval)
.enable_feature(Feature::GuardianV2)
.write(codex_home.path())?;
let config = load_default_config_for_test(&codex_home).await;
let models = [
(
SYNCHRONOUS_MODEL,
codex_protocol::openai_models::GuardianReviewMode::Synchronous,
),
(
MODEL,
codex_protocol::openai_models::GuardianReviewMode::Adaptive,
),
]
.map(|(slug, mode)| {
let mut model = codex_core::test_support::construct_model_info_offline(slug, &config);
model.node_repl_auto_review_required = true;
model.guardian = Some(codex_protocol::openai_models::GuardianModelPolicy {
computer_use: Some(mode),
..Default::default()
});
model
});
write_models_cache_with_models(codex_home.path(), models.into()).await?;
let mut app_server = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized_with_timeout(TIMEOUT)
.await?;
let thread = app_server
.start_thread(ThreadStartParams::default())
.await?
.thread;
// Both calls need fresh review: the synchronous first turn must consume the
// allowance before the second turn enables adaptive scoring. Scores stay pending.
for model in [SYNCHRONOUS_MODEL, MODEL] {
state.parent_requests.store(0, Ordering::SeqCst);
let expected_reviews = state.guardian_reviews.load(Ordering::SeqCst) + 1;
state.allow_guardian_review.notify_one();
let request_id = app_server
.send_turn_start_request(TurnStartParams {
thread_id: thread.id.clone(),
model: Some(model.to_owned()),
input: vec![UserInput::Text {
text: USER_CONTEXT.to_owned(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
let _: TurnStartResponse = timeout(TIMEOUT, app_server.read_response(request_id)).await??;
let completed: TurnCompletedNotification =
timeout(TIMEOUT, app_server.read_notification("turn/completed")).await??;
assert_eq!(completed.thread_id, thread.id);
assert_eq!(
state.guardian_reviews.load(Ordering::SeqCst),
expected_reviews
);
}
app_server.shutdown_gracefully().await?;
mcp_server.abort();
responses_server.abort();
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn user_approval_skips_async_guardian_without_changing_other_modes() -> Result<()> {
skip_if_no_network!(Ok(()));
codex-rs/config/src/guardian.rs
+18 -15
View File
@@ -1,4 +1,4 @@
//! Translates legacy Guardian settings and applies live managed constraints.
//! Translates legacy Guardian settings and applies captured managed constraints.
//! Catalog policies take precedence over legacy scope and feature settings.
use codex_features::FeatureToml;
@@ -15,15 +15,11 @@ use crate::ConfigRequirements;
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct GuardianPolicyLoader {
default: GuardianModelPolicy,
force_synchronous_review: bool,
scoring_disabled: bool,
}
impl GuardianPolicyLoader {
pub fn new(
legacy: Option<&FeatureToml<GuardianV2ConfigToml>>,
requirements: &ConfigRequirements,
) -> Self {
pub fn new(legacy: Option<&FeatureToml<GuardianV2ConfigToml>>) -> Self {
let scope = match legacy {
Some(FeatureToml::Config(config)) => config.review_scope.as_ref(),
Some(FeatureToml::Enabled(_)) | None => None,
@@ -61,14 +57,14 @@ impl GuardianPolicyLoader {
Self {
default,
scoring_disabled: !enabled.unwrap_or(/*default*/ false),
force_synchronous_review: requirements
.approvals_reviewer
.can_set(&ApprovalsReviewer::User)
.is_err(),
}
}
pub fn resolve(&self, model: Option<&ModelInfo>) -> GuardianModelPolicy {
pub fn resolve(
&self,
model: Option<&ModelInfo>,
requirements: &ConfigRequirements,
) -> GuardianModelPolicy {
let catalog = model.and_then(|model| model.guardian.as_ref());
let mut policy = catalog.cloned().unwrap_or_else(|| self.default.clone());
if catalog.is_none()
@@ -80,17 +76,24 @@ impl GuardianPolicyLoader {
policy.unscored_action = GuardianUnscoredAction::InvalidateScore;
}
// Preserve the legacy feature gate: explicit catalog policies bypass it.
if self.force_synchronous_review || self.scoring_disabled && catalog.is_none() {
if self.scoring_disabled && catalog.is_none() {
policy.disable_scoring();
}
requirements.constrain_guardian_policy(&mut policy, model.map(|model| model.slug.as_str()));
policy
}
}
impl ConfigRequirements {
/// Applies live administrator requirements to an already resolved model policy.
pub fn constrain_guardian_policy(&self, policy: &mut GuardianModelPolicy, model: &str) {
if self.auto_review_required_for_model(model) {
/// Applies the supplied administrator requirements to an already resolved model policy.
/// The reviewer allowlist applies even when model metadata is unavailable.
pub fn constrain_guardian_policy(&self, policy: &mut GuardianModelPolicy, model: Option<&str>) {
if [ApprovalsReviewer::User, ApprovalsReviewer::AutoReview]
.iter()
.any(|reviewer| self.approvals_reviewer.can_set(reviewer).is_err())
{
policy.disable_scoring();
} else if model.is_some_and(|model| self.auto_review_required_for_model(model)) {
let computer_use = policy.computer_use;
let initial_cua_call = policy.allows_initial_cua_call();
policy.disable_scoring();
codex-rs/config/src/guardian_tests.rs
+35 -9
View File
@@ -29,7 +29,7 @@ fn explicit_catalog_policy_preserves_precedence_over_the_legacy_switch() {
let mut model = test_model();
model.guardian = Some(catalog.clone());
assert_eq!(
GuardianPolicyLoader::new(Some(&legacy), &requirements).resolve(Some(&model)),
GuardianPolicyLoader::new(Some(&legacy)).resolve(Some(&model), &requirements),
catalog
);
}
@@ -43,7 +43,7 @@ fn legacy_scope_applies_only_to_models_without_catalog_policy() {
"review_scope": {"computer_use_only": false, "sandboxed_exec_commands": true}
}))
.unwrap();
let loader = GuardianPolicyLoader::new(Some(&legacy), &ConfigRequirements::default());
let loader = GuardianPolicyLoader::new(Some(&legacy));
let expected = GuardianModelPolicy {
computer_use: Some(GuardianReviewMode::Adaptive),
shell: Some(GuardianReviewMode::Adaptive),
@@ -56,12 +56,38 @@ fn legacy_scope_applies_only_to_models_without_catalog_policy() {
initial_cua_call: Some(false),
sandboxed_exec_commands: Some(true),
};
assert_eq!(loader.resolve(/*model*/ None), expected);
assert_eq!(
loader.resolve(/*model*/ None, &ConfigRequirements::default()),
expected
);
let catalog: GuardianModelPolicy =
serde_json::from_value(json!({"computer_use": "adaptive"})).unwrap();
let mut model = test_model();
model.guardian = Some(catalog.clone());
assert_eq!(loader.resolve(Some(&model)), catalog);
assert_eq!(
loader.resolve(Some(&model), &ConfigRequirements::default()),
catalog
);
for reviewer in [ApprovalsReviewer::AutoReview, ApprovalsReviewer::User] {
let restricted: ConfigRequirements = crate::ConfigRequirementsWithSources {
allowed_approvals_reviewers: Some(Sourced::new(
vec![reviewer],
RequirementSource::Unknown,
)),
..Default::default()
}
.try_into()
.unwrap();
// Reuse the same cached base across tightening and relaxation, including
// decisions without model metadata.
for model in [None, Some(&model)] {
let base = loader.resolve(model, &ConfigRequirements::default());
let mut synchronous = base.clone();
synchronous.disable_scoring();
assert_eq!(loader.resolve(model, &restricted), synchronous);
assert_eq!(loader.resolve(model, &ConfigRequirements::default()), base);
}
}
}
#[test]
@@ -79,15 +105,15 @@ fn required_models_preserve_cua_allowance_and_disable_all_tools_scoring() {
.unwrap();
let cua_only = FeatureToml::Enabled(true);
for legacy in [Some(&cua_only), Some(&legacy)] {
let loader = GuardianPolicyLoader::new(legacy, &requirements);
let ordinary = loader.resolve(/*model*/ None);
let loader = GuardianPolicyLoader::new(legacy);
let ordinary = loader.resolve(/*model*/ None, &requirements);
let mut expected = ordinary.clone();
expected.disable_scoring();
if ordinary.allows_initial_cua_call() {
expected.computer_use = ordinary.computer_use;
}
let mut actual = ordinary;
requirements.constrain_guardian_policy(&mut actual, "provider/protected-model");
requirements.constrain_guardian_policy(&mut actual, Some("provider/protected-model"));
assert_eq!(actual, expected);
}
}
@@ -101,11 +127,11 @@ fn legacy_cua_opt_in_and_disabled_feature_preserve_scoring_behavior() {
] {
let enabled = legacy == json!(true);
let legacy = serde_json::from_value(legacy).unwrap();
let loader = GuardianPolicyLoader::new(Some(&legacy), &ConfigRequirements::default());
let loader = GuardianPolicyLoader::new(Some(&legacy));
for required in [false, true] {
let mut model = test_model();
model.node_repl_auto_review_required = required;
let policy = loader.resolve(Some(&model));
let policy = loader.resolve(Some(&model), &ConfigRequirements::default());
assert_eq!(policy.scoring_enabled(), enabled && required);
assert_eq!(model.computer_use_review_required(), required);
}
codex-rs/config/src/state.rs
+17 -10
View File
@@ -553,6 +553,20 @@ impl ConfigLayerStack {
})
}
/// Adds local session configuration while retaining independently owned runtime snapshots.
pub fn with_session_config(mut self, config: TomlValue) -> std::io::Result<Self> {
let index = self.layers.partition_point(|layer| {
layer.name.precedence() <= ConfigLayerSource::SessionFlags.precedence()
});
self.layers.insert(
index,
ConfigLayerEntry::new(ConfigLayerSource::SessionFlags, config),
);
validate_enabled_config_layers(&self.layers)?;
self.validate_project_mitm_profiles()?;
Ok(self)
}
pub fn with_session_plugins(mut self, plugins: TomlValue) -> std::io::Result<Self> {
let session_index = self.layers.iter().rposition(|layer| {
layer.name == ConfigLayerSource::SessionFlags && !layer.is_disabled()
@@ -577,16 +591,9 @@ impl ConfigLayerStack {
*layer = ConfigLayerEntry::new(ConfigLayerSource::SessionFlags, config);
}
} else {
let index = self.layers.partition_point(|layer| {
layer.name.precedence() <= ConfigLayerSource::SessionFlags.precedence()
});
self.layers.insert(
index,
ConfigLayerEntry::new(
ConfigLayerSource::SessionFlags,
TomlValue::Table([("plugins".to_owned(), plugins)].into_iter().collect()),
),
);
return self.with_session_config(TomlValue::Table(
[("plugins".to_owned(), plugins)].into_iter().collect(),
));
}
Ok(self)
}
codex-rs/core/src/agent/role.rs
+3 -32
View File
@@ -8,9 +8,6 @@ use crate::config::Config;
use crate::config::deserialize_config_toml_with_base;
use anyhow::anyhow;
use codex_agent_roles::parse_agent_role_file_contents;
use codex_config::ConfigLayerEntry;
use codex_config::ConfigLayerSource;
use codex_config::ConfigLayerStack;
use codex_config::SkillsConfig;
use codex_config::loader::resolve_relative_paths_in_config_toml;
use codex_exec_server::read_sensitive_file_to_string;
@@ -180,7 +177,9 @@ mod role_overrides {
overrides: &AgentRoleOverrides,
) -> anyhow::Result<Config> {
let mut next_config = config.clone();
next_config.config_layer_stack = build_config_layer_stack(config, &role_layer_toml)?;
next_config.config_layer_stack = next_config
.config_layer_stack
.with_session_config(role_layer_toml)?;
if let Some(model) = &overrides.model {
next_config.model = Some(model.clone());
}
@@ -238,34 +237,6 @@ mod role_overrides {
}
Ok(next_config)
}
fn build_config_layer_stack(
config: &Config,
role_layer_toml: &TomlValue,
) -> anyhow::Result<ConfigLayerStack> {
let mut layers: Vec<_> = config
.config_layer_stack
.all_layers_low_to_high()
.cloned()
.collect();
let role_layer =
ConfigLayerEntry::new(ConfigLayerSource::SessionFlags, role_layer_toml.clone());
let insertion_index = layers.partition_point(|layer| layer.name <= role_layer.name);
layers.insert(insertion_index, role_layer);
Ok(ConfigLayerStack::new(
layers,
config.config_layer_stack.requirements().clone(),
config.config_layer_stack.requirements_toml().clone(),
)?
.with_cloud_config_binding(config.config_layer_stack.cloud_config_binding().cloned())
.with_mcp_ema_cancellation(config.config_layer_stack.mcp_ema_cancellation().clone())
.with_applied_config_server(config.config_layer_stack.applied_config_server().cloned())
.with_user_and_project_exec_policy_rules_ignored(
config
.config_layer_stack
.ignore_user_and_project_exec_policy_rules(),
))
}
}
pub(crate) mod spawn_tool_spec {
codex-rs/core/src/agent/role_tests.rs
+1
View File
@@ -5,6 +5,7 @@ use crate::skills_load_input_from_config;
use codex_config::CloudConfigBundleBindingStatus;
use codex_config::CloudConfigBundlePolicy;
use codex_config::CloudConfigBundleSnapshot;
use codex_config::ConfigLayerSource;
use codex_config::test_support::CloudConfigBundleFixture;
use codex_login::test_support::auth_manager_from_optional_auth;
use codex_protocol::config_types::ServiceTier;
codex-rs/core/src/config/guardian_policy.rs
+22 -1
View File
@@ -1,8 +1,29 @@
//! Resolves the effective Guardian reviewer while retaining the requested choice.
//! Resolves Guardian requirements and retained configuration fallbacks identically
//! during startup and runtime publication.
use super::guardian_policy_config_from_requirements;
use super::normalize_guardian_policy_config;
use codex_config::ConfigRequirements;
use codex_config::ConfigRequirementsToml;
use codex_config::config_toml::AutoReviewToml;
use codex_protocol::config_types::ApprovalsReviewer;
pub(crate) fn resolve_guardian_policy_text(
requirements: &ConfigRequirementsToml,
fallback: Option<&AutoReviewToml>,
) -> (Option<String>, Option<String>) {
let policy = guardian_policy_config_from_requirements(requirements).or_else(|| {
normalize_guardian_policy_config(fallback.and_then(|fallback| fallback.policy.as_deref()))
});
let extra = normalize_guardian_policy_config(requirements.guardian_extra_policy.as_deref())
.or_else(|| {
normalize_guardian_policy_config(
fallback.and_then(|fallback| fallback.extra_policy.as_deref()),
)
});
(policy, extra)
}
/// Resolves the effective reviewer without replacing the retained request.
/// Trusted private reviewers omit the model to bypass model-required review.
pub(crate) fn resolve_guardian_reviewer(
codex-rs/core/src/config/mod.rs
+7 -20
View File
@@ -181,6 +181,7 @@ mod permission_profile_selection;
mod permissions;
mod requirements;
mod resolved_permission_profile;
pub(crate) use guardian_policy::resolve_guardian_policy_text;
pub(crate) use guardian_policy::resolve_guardian_reviewer;
mod runtime_refresh;
pub(crate) use runtime_refresh::RuntimeConfigRefresh;
@@ -729,6 +730,8 @@ pub struct Config {
/// `# Policy Configuration` section rather than replacing the whole
/// guardian developer prompt.
pub guardian_policy_config: Option<String>,
/// Guardian publication observed before startup loading or last policy refresh.
pub guardian_policy_revision: u64,
/// Additional Guardian policy from requirements.toml or config.toml.
/// Rendered into `{{ extra_policy }}` alongside the resolved tenant policy.
@@ -4304,26 +4307,9 @@ impl Config {
.and_then(|skills| skills.max_context_tokens);
let include_environment_context = cfg.include_environment_context.unwrap_or(true);
let include_environment_context_time = cfg.include_environment_context_time.unwrap_or(true);
let guardian_policy_config =
guardian_policy_config_from_requirements(config_layer_stack.requirements_toml())
.or_else(|| {
cfg.auto_review
.as_ref()
.and_then(|auto_review| normalize_guardian_policy_config(
auto_review.policy.as_deref(),
))
});
let guardian_extra_policy = normalize_guardian_policy_config(
config_layer_stack
.requirements_toml()
.guardian_extra_policy
.as_deref(),
)
.or_else(|| {
cfg.auto_review.as_ref().and_then(|auto_review| {
normalize_guardian_policy_config(auto_review.extra_policy.as_deref())
})
});
let (guardian_policy_config, guardian_extra_policy) = resolve_guardian_policy_text(
config_layer_stack.requirements_toml(), cfg.auto_review.as_ref(),
);
let guardian_transcript_mode = cfg
.features
.as_ref()
@@ -4760,6 +4746,7 @@ impl Config {
.or(show_raw_agent_reasoning)
.unwrap_or(false),
guardian_policy_config,
guardian_policy_revision: 0,
guardian_extra_policy,
guardian_policy_template,
guardian_transcript_mode,
codex-rs/core/src/guardian/decision.rs
+2
View File
@@ -102,6 +102,7 @@ pub(crate) fn decide_approval(
};
let _cancel_on_drop = cancellation.clone().drop_guard();
let turn = context.turn();
turn.extension_data.insert_arc(Arc::clone(&turn.config));
let live_config = session.get_config().await;
let requirements = live_config.config_layer_stack.requirements();
let model_requires_review =
@@ -159,6 +160,7 @@ pub(crate) fn decide_approval(
action: request.action.as_ref().ok(),
thread_id: session.thread_id,
thread_store: &session.services.thread_extension_data,
turn_store: &turn.extension_data,
category: request.category,
approval_policy: context.approval_policy,
approvals_reviewer: context.approvals_reviewer,
codex-rs/core/src/tools/lifecycle.rs
+4
View File
@@ -30,6 +30,10 @@ pub(crate) async fn notify_tool_start(
if contributors.is_empty() {
return;
}
invocation
.turn
.extension_data
.insert_arc(Arc::clone(&invocation.turn.config));
let thread_store = &invocation.session.services.thread_extension_data;
let conversation_history = invocation.session.conversation_history_snapshot().await;
let root_turn_id = invocation.turn.turn_metadata_state.root_turn_id();
codex-rs/ext/extension-api/src/contributors/approval_review.rs
+2
View File
@@ -42,6 +42,8 @@ pub struct ApprovalDecisionInput<'a> {
pub action: &'a serde_json::Value,
pub thread_id: ThreadId,
pub thread_store: &'a ExtensionData,
/// Captured configuration and evidence owned by the requesting turn.
pub turn_store: &'a ExtensionData,
pub category: codex_protocol::openai_models::GuardianScope,
pub approval_policy: codex_protocol::protocol::AskForApproval,
pub approvals_reviewer: codex_protocol::config_types::ApprovalsReviewer,
codex-rs/ext/extension-api/tests/registry.rs
+1
View File
@@ -256,6 +256,7 @@ async fn build_round_trips_every_contributor_category() {
action: &serde_json::Value::Null,
thread_id: codex_protocol::ThreadId::new(),
thread_store: &thread_store,
turn_store: &thread_store,
category: codex_protocol::openai_models::GuardianScope::Shell,
approval_policy: codex_protocol::protocol::AskForApproval::OnRequest,
approvals_reviewer: codex_protocol::config_types::ApprovalsReviewer::AutoReview,
codex-rs/ext/guardian-reviewer/src/review_tests.rs
+1
View File
@@ -167,6 +167,7 @@ async fn async_approval_revalidates_without_waiting_for_tracked_cleanup(state: A
action: Some(&action),
thread_id,
thread_store: &store,
turn_store: &store,
category: GuardianScope::Permissions,
approval_policy: AskForApproval::OnRequest,
approvals_reviewer: ApprovalsReviewer::AutoReview,
codex-rs/ext/guardian-reviewer/src/routing.rs
+2
View File
@@ -34,6 +34,7 @@ pub struct ReviewRequest<'a, H> {
pub action: Option<&'a serde_json::Value>,
pub thread_id: ThreadId,
pub thread_store: &'a ExtensionData,
pub turn_store: &'a ExtensionData,
pub category: GuardianScope,
pub approval_policy: AskForApproval,
pub approvals_reviewer: ApprovalsReviewer,
@@ -111,6 +112,7 @@ impl<H: ReviewHost> ReviewRequest<'_, H> {
action,
thread_id: self.thread_id,
thread_store: self.thread_store,
turn_store: self.turn_store,
category: self.category,
approval_policy: self.approval_policy,
approvals_reviewer: self.approvals_reviewer,
codex-rs/ext/guardian-v2/src/async_scorer/approval.rs
+54 -16
View File
@@ -15,6 +15,7 @@ use codex_core::context::GuardianContextMode;
use codex_extension_api::ApprovalDecision;
use codex_extension_api::ApprovalDecisionInput;
use codex_extension_api::ApprovalReviewContributor;
use codex_extension_api::ExtensionData;
use codex_extension_api::ExtensionFuture;
use codex_extension_api::ExtensionMetrics;
use codex_protocol::approvals::GuardianReviewReason;
@@ -43,7 +44,8 @@ impl ApprovalReviewContributor for GuardianApprovalReviewer {
record_fast_decision(input.metrics.as_deref(), "deferred", "scoring_failure");
return None;
};
Some(self.decide_request(&thread, input).await)
let config = input.turn_store.get::<codex_core::config::Config>()?;
Some(self.decide_request(&thread, input, &config).await)
})
}
}
@@ -54,6 +56,7 @@ impl GuardianApprovalReviewer {
&self,
thread: &CodexThread,
input: &ApprovalDecisionInput<'_>,
config: &codex_core::config::Config,
) -> ApprovalDecision {
if input.full_access {
return ApprovalDecision::Allow;
@@ -67,7 +70,6 @@ impl GuardianApprovalReviewer {
{
return ApprovalDecision::AskUser;
}
let config = thread.config().await;
let model = input
.thread_store
.get::<codex_protocol::openai_models::ModelInfo>();
@@ -75,24 +77,28 @@ impl GuardianApprovalReviewer {
.thread_store
.get::<GuardianV2Config>()
.map(|config| (*config).clone())
.map_or_else(|| GuardianV2Config::resolve(&config), Ok)
.map_or_else(|| GuardianV2Config::resolve(config), Ok)
.ok();
let requirements = config.config_layer_stack.requirements();
let mut policy = guardian_config.as_ref().map_or_else(
|| {
codex_config::GuardianPolicyLoader::new(
Some(&codex_features::FeatureToml::Enabled(true)),
&codex_config::ConfigRequirements::default(),
)
.resolve(model.as_deref())
codex_config::GuardianPolicyLoader::new(Some(
&codex_features::FeatureToml::Enabled(true),
))
.resolve(model.as_deref(), requirements)
},
|config| config.policy_for_model(model.as_deref()),
|config| config.policy_for_model(model.as_deref(), requirements),
);
if let Some(model) = model.as_ref() {
config
.config_layer_stack
.requirements()
.constrain_guardian_policy(&mut policy, &model.slug);
}
// Live mandatory restrictions may harden the turn's captured policy.
thread
.config()
.await
.config_layer_stack
.requirements()
.constrain_guardian_policy(
&mut policy,
model.as_ref().map(|model| model.slug.as_str()),
);
let mode = policy.review_mode(input.category);
if mode != GuardianReviewMode::Adaptive {
record_fast_decision(input.metrics.as_deref(), "deferred", "out_of_scope");
@@ -218,6 +224,7 @@ async fn cached_evidence(
let action_fits = serde_json::to_string_pretty(&action)
.is_ok_and(|action| action.len().saturating_add(1) <= max_action_bytes);
let history = thread.conversation_history_snapshot().await;
let permitted = constrained_live_policy(thread, store, policy).await;
let cached = progress.inspect(input.tool_call_id);
if let Some(metrics) = metrics {
metrics.counter(
@@ -273,6 +280,11 @@ async fn cached_evidence(
)
&& cached.js_executions == 1
{
if !permitted.allows_initial_cua_call()
|| permitted.review_mode(input.category) != GuardianReviewMode::Adaptive
{
return Err(GuardianReviewReason::Policy);
}
record_fast_decision(metrics, "approved", "initial_cua_call");
return Ok(false);
}
@@ -281,7 +293,8 @@ async fn cached_evidence(
return Err(GuardianReviewReason::AuthorizationChanged);
};
let current = ScoreAuthorization::current(thread, permissions).await;
// Classification may publish or fail while authorization is collected.
// Classification may publish or fail while authorization and live policy are collected.
let permitted = constrained_live_policy(thread, store, policy).await;
let cached = progress.inspect(input.tool_call_id);
if let Some(metrics) = metrics
&& let Some(sampled_at) = cached.sampled_at
@@ -341,6 +354,10 @@ async fn cached_evidence(
// actions without a tool-call ID retain their existing thread-level checks.
(GuardianReviewReason::MissingScore, "missing_score")
} else {
// A pending review can receive a late score after requirements tighten.
if permitted.review_mode(input.category) != GuardianReviewMode::Adaptive {
return Err(GuardianReviewReason::Policy);
}
record_fast_decision(metrics, "approved", "low_risk");
return Ok(cached.score_at_or_before_action);
}
@@ -358,3 +375,24 @@ async fn cached_evidence(
record_fast_decision(metrics, "deferred", label);
Err(reason)
}
async fn constrained_live_policy(
thread: &CodexThread,
store: &ExtensionData,
policy: &GuardianModelPolicy,
) -> GuardianModelPolicy {
let mut permitted = policy.clone();
thread
.config()
.await
.config_layer_stack
.requirements()
.constrain_guardian_policy(
&mut permitted,
store
.get::<codex_protocol::openai_models::ModelInfo>()
.as_ref()
.map(|model| model.slug.as_str()),
);
permitted
}
codex-rs/ext/guardian-v2/src/async_scorer/budget_tests.rs
+7 -1
View File
@@ -42,6 +42,9 @@ async fn catalog_budget_fixture(base_url: String, window: i64) -> Result<Guardia
);
let registry = builder.build();
let session_store = ExtensionData::new("session-1");
test.codex
.thread_extension_data()
.insert(test.config.clone());
registry.thread_lifecycle_contributors()[0]
.on_thread_start(ThreadStartInput {
config: &test.config,
@@ -195,6 +198,8 @@ async fn assert_catalog_budget(evidence: BudgetEvidence) -> Result<()> {
}
};
let thread_store = fixture.test.codex.thread_extension_data();
let turn_store = ExtensionData::new("turn-1");
turn_store.insert_arc(thread_store.get::<Config>().expect("captured test config"));
if matches!(evidence, BudgetEvidence::UserInstructions) {
set_cached_score(
thread_store,
@@ -213,6 +218,7 @@ async fn assert_catalog_budget(evidence: BudgetEvidence) -> Result<()> {
cached_approval(
&fixture.registry,
thread_store,
&turn_store,
"review action",
/*metrics*/ None
)
@@ -220,7 +226,6 @@ async fn assert_catalog_budget(evidence: BudgetEvidence) -> Result<()> {
Some(ReviewDecision::Approved)
);
}
let turn_store = ExtensionData::new("turn-1");
let tool_name = ToolName::plain("read_file");
let payload = ToolPayload::Function {
arguments: r#"{"path":"README.md"}"#.to_owned(),
@@ -290,6 +295,7 @@ async fn assert_catalog_budget(evidence: BudgetEvidence) -> Result<()> {
cached_approval(
&fixture.registry,
thread_store,
&turn_store,
"review action",
/*metrics*/ None
)
codex-rs/ext/guardian-v2/src/async_scorer/config.rs
+7 -10
View File
@@ -79,8 +79,7 @@ impl GuardianV2Config {
enabled: Some(config.features.enabled(codex_features::Feature::GuardianV2)),
..configured
});
resolved.policy =
GuardianPolicyLoader::new(Some(&legacy), config.config_layer_stack.requirements());
resolved.policy = GuardianPolicyLoader::new(Some(&legacy));
resolved.review_multi_agent_calls = config
.features
.enabled(codex_features::Feature::GuardianNativeMultiAgentHistory);
@@ -91,8 +90,9 @@ impl GuardianV2Config {
pub(super) fn policy_for_model(
&self,
model: Option<&codex_protocol::openai_models::ModelInfo>,
requirements: &codex_config::ConfigRequirements,
) -> GuardianModelPolicy {
self.policy.resolve(model)
self.policy.resolve(model, requirements)
}
pub(crate) fn with_model_defaults(
@@ -267,13 +267,10 @@ impl GuardianV2Config {
);
}
let policy = GuardianPolicyLoader::new(
Some(&FeatureToml::Config(GuardianV2ConfigToml {
enabled: Some(true),
..configured.clone()
})),
&codex_config::ConfigRequirements::default(),
);
let policy = GuardianPolicyLoader::new(Some(&FeatureToml::Config(GuardianV2ConfigToml {
enabled: Some(true),
..configured.clone()
})));
Ok(Self {
local_overrides: configured.clone(),
review_multi_agent_calls: false,
codex-rs/ext/guardian-v2/src/async_scorer/conversation_tests.rs
+8
View File
@@ -301,6 +301,8 @@ async fn delayed_preparation_does_not_block_newer_requests_and_capacity_is_bound
.get::<ConversationBackend>()
.unwrap();
let store = fixture.test.codex.thread_extension_data();
let turn_store = ExtensionData::new("approval-turn");
turn_store.insert_arc(store.get::<Config>().expect("captured test config"));
let progress = store.get::<GuardianV2ScoreProgress>().unwrap();
let mut reservations = (0..16)
.map(|index| backend.reserve(|| index).1.unwrap())
@@ -323,6 +325,7 @@ async fn delayed_preparation_does_not_block_newer_requests_and_capacity_is_bound
cached_approval(
&fixture.registry,
store,
&turn_store,
"review action",
/*metrics*/ None
)
@@ -532,6 +535,8 @@ async fn preflight_skips_only_after_score_and_history_are_superseded() -> Result
)
.await?;
let store = fixture.test.codex.thread_extension_data();
let turn_store = ExtensionData::new("approval-turn");
turn_store.insert_arc(store.get::<Config>().expect("captured test config"));
let mut cached = cached_score(store).unwrap();
cached.scores.insert("action_risk".to_owned(), risk);
set_cached_score(store, cached);
@@ -578,6 +583,7 @@ async fn preflight_skips_only_after_score_and_history_are_superseded() -> Result
let approval_before = cached_approval(
&fixture.registry,
store,
&turn_store,
r#"{"tool":"mcp_tool_call","server":"node_repl"}"#,
/*metrics*/ None,
)
@@ -609,6 +615,7 @@ async fn preflight_skips_only_after_score_and_history_are_superseded() -> Result
cached_approval(
&fixture.registry,
store,
&turn_store,
r#"{"tool":"mcp_tool_call","server":"node_repl"}"#,
/*metrics*/ None
)
@@ -652,6 +659,7 @@ async fn delayed_classification_skips_after_newer_score_and_history_complete() -
let mut updates = progress.updates.subscribe();
let metrics = store.get::<RecordingMetrics>().unwrap();
let turn_store = ExtensionData::new("turn-1");
turn_store.insert_arc(store.get::<Config>().expect("captured test config"));
let tool_name = ToolName::plain("read_file");
let payload = ToolPayload::Function {
arguments: r#"{"path":"README.md"}"#.to_owned(),
codex-rs/ext/guardian-v2/src/async_scorer/extension.rs
+4 -8
View File
@@ -63,14 +63,10 @@ impl ThreadLifecycleContributor<Config> for GuardianV2Extension {
return;
}
};
let mut policy = guardian_config.policy_for_model(model.as_deref());
if let Some(model) = model.as_ref() {
input
.config
.config_layer_stack
.requirements()
.constrain_guardian_policy(&mut policy, &model.slug);
}
let policy = guardian_config.policy_for_model(
model.as_deref(),
input.config.config_layer_stack.requirements(),
);
let scoring_enabled = policy.scoring_enabled();
let sampler_config =
super::startup::sampler_config(&input, self.thread_manager.upgrade()).await;
codex-rs/ext/guardian-v2/src/async_scorer/extension_cached_delivery_tests.rs
+26 -3
View File
@@ -205,6 +205,7 @@ async fn confirmed_root_delivery_invalidates_root_and_worker_cached_approvals()
let mut config = test.config.clone();
config.features.enable(Feature::GuardianV2)?;
let session_store = ExtensionData::new("worker-cache-session");
worker.thread_extension_data().insert(config.clone());
registry.thread_lifecycle_contributors()[0]
.on_thread_start(ThreadStartInput {
config: &config,
@@ -229,10 +230,19 @@ async fn confirmed_root_delivery_invalidates_root_and_worker_cached_approvals()
// Hold tool lag at zero so only the delivery can invalidate either approval.
for (thread, before) in [(&test.codex, &root_before), (&worker, &worker_before)] {
let store = thread.thread_extension_data();
let turn_store = ExtensionData::new("approval-turn");
turn_store.insert_arc(store.get::<Config>().expect("captured test config"));
let progress = store.get::<GuardianV2ScoreProgress>().unwrap();
seed_cached_score(&progress, store, /*index*/ 1_000, before.clone());
assert_eq!(
cached_approval(&registry, store, "review action", /*metrics*/ None).await,
cached_approval(
&registry,
store,
&turn_store,
"review action",
/*metrics*/ None
)
.await,
Some(ReviewDecision::Approved)
);
}
@@ -266,10 +276,14 @@ async fn confirmed_root_delivery_invalidates_root_and_worker_cached_approvals()
}
);
for thread in [&test.codex, &worker] {
let store = thread.thread_extension_data();
let turn_store = ExtensionData::new("approval-turn");
turn_store.insert_arc(store.get::<Config>().expect("captured test config"));
assert_eq!(
cached_approval(
&registry,
thread.thread_extension_data(),
store,
&turn_store,
"review action",
/*metrics*/ None
)
@@ -298,12 +312,21 @@ async fn confirmed_root_delivery_invalidates_root_and_worker_cached_approvals()
.contains(&codex_core::GuardianRootMessage::IncompleteRootInstructions)
);
let store = worker.thread_extension_data();
let turn_store = ExtensionData::new("approval-turn");
turn_store.insert_arc(store.get::<Config>().expect("captured test config"));
let progress = store.get::<GuardianV2ScoreProgress>().unwrap();
let authorization = ScoreAuthorization::current(&worker, &Default::default()).await;
assert!(authorization.local.retained_context_complete);
seed_cached_score(&progress, store, /*index*/ 1_000, authorization);
assert_eq!(
cached_approval(&registry, store, "review action", /*metrics*/ None).await,
cached_approval(
&registry,
store,
&turn_store,
"review action",
/*metrics*/ None
)
.await,
Some(ReviewDecision::Approved)
);
let shutdown = test
codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs
+185 -28
View File
@@ -132,20 +132,20 @@ fn should_classify_tool(
fn legacy_loader(
scope: Option<&GuardianV2ReviewScopeConfigToml>,
) -> codex_config::GuardianPolicyLoader {
codex_config::GuardianPolicyLoader::new(
Some(&codex_features::FeatureToml::Config(
codex_features::GuardianV2ConfigToml {
enabled: Some(true),
review_scope: scope.cloned(),
..Default::default()
},
)),
&codex_config::ConfigRequirements::default(),
)
codex_config::GuardianPolicyLoader::new(Some(&codex_features::FeatureToml::Config(
codex_features::GuardianV2ConfigToml {
enabled: Some(true),
review_scope: scope.cloned(),
..Default::default()
},
)))
}
fn legacy_policy(scope: Option<&GuardianV2ReviewScopeConfigToml>) -> GuardianModelPolicy {
legacy_loader(scope).resolve(/*model*/ None)
legacy_loader(scope).resolve(
/*model*/ None,
&codex_config::ConfigRequirements::default(),
)
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
@@ -239,6 +239,7 @@ async fn installed_extension_warms_connections_without_blocking_thread_start() -
model.node_repl_auto_review_required = true;
thread_store.insert(model);
thread_store.insert(config.clone());
registry.thread_lifecycle_contributors()[0]
.on_thread_start(ThreadStartInput {
config: &config,
@@ -313,6 +314,7 @@ async fn installed_extension_uses_http_after_warm_socket_auth_expires() -> Resul
.await;
model.node_repl_auto_review_required = true;
thread_store.insert_arc(model.model);
thread_store.insert(config.clone());
registry.thread_lifecycle_contributors()[0]
.on_thread_start(ThreadStartInput {
config: &config,
@@ -334,6 +336,7 @@ async fn installed_extension_uses_http_after_warm_socket_auth_expires() -> Resul
.get::<GuardianV2ScoreProgress>()
.expect("Guardian v2 should initialize");
let turn_store = ExtensionData::new("turn-1");
turn_store.insert_arc(thread_store.get::<Config>().expect("captured test config"));
let tool_name = ToolName::namespaced("mcp__node_repl__", "js");
let payload = ToolPayload::Function {
arguments: r#"{"path":"README.md"}"#.to_owned(),
@@ -371,6 +374,7 @@ async fn installed_extension_uses_http_after_warm_socket_auth_expires() -> Resul
cached_approval(
&registry,
thread_store,
&turn_store,
r#"{"tool":"mcp_tool_call","server":"node_repl"}"#,
/*metrics*/ None,
)
@@ -618,6 +622,7 @@ async fn sandboxed_shell_classification_respects_review_scope() -> Result<()> {
.expect("Guardian v2 should track score progress per thread");
let mut expected = score_progress.inspect(/*call_id*/ None);
let turn_store = ExtensionData::new("turn-1");
turn_store.insert_arc(thread_store.get::<Config>().expect("captured test config"));
let tool_name = ToolName::plain("exec_command");
for (call_id, payload, expected_decision) in [
("call-2", sandboxed, Some(ReviewDecision::Approved)),
@@ -646,6 +651,7 @@ async fn sandboxed_shell_classification_respects_review_scope() -> Result<()> {
cached_approval(
&fixture.registry,
thread_store,
&turn_store,
r#"{"tool":"exec_command","cmd":"pwd"}"#,
/*metrics*/ None,
)
@@ -719,6 +725,7 @@ async fn computer_use_only_scores_cannot_approve_other_actions() -> Result<()> {
seed_cached_score(&progress, thread_store, /*index*/ 1, authorization);
let cached = progress.inspect(/*call_id*/ None);
let turn_store = ExtensionData::new("turn-1");
turn_store.insert_arc(thread_store.get::<Config>().expect("captured test config"));
let ordinary_tool = ToolName::namespaced("mcp__ordinary__", "write_record");
let payload = ToolPayload::Function {
arguments: r#"{"record":"sensitive"}"#.to_owned(),
@@ -770,6 +777,7 @@ async fn computer_use_only_scores_cannot_approve_other_actions() -> Result<()> {
cached_approval(
&fixture.registry,
thread_store,
&turn_store,
&prompt,
thread_store
.get::<RecordingMetrics>()
@@ -784,6 +792,7 @@ async fn computer_use_only_scores_cannot_approve_other_actions() -> Result<()> {
cached_approval(
&fixture.registry,
thread_store,
&turn_store,
"not valid JSON",
thread_store
.get::<RecordingMetrics>()
@@ -808,6 +817,7 @@ async fn computer_use_only_scores_cannot_approve_other_actions() -> Result<()> {
cached_approval(
&fixture.registry,
thread_store,
&turn_store,
&json!({"tool": "mcp_tool_call", "server": "node_repl", "tool_name": "js"}).to_string(),
/*metrics*/ None,
)
@@ -866,6 +876,7 @@ async fn computer_use_only_scores_cannot_approve_other_actions() -> Result<()> {
cached_approval(
&fixture.registry,
thread_store,
&turn_store,
r#"{"tool":"mcp_tool_call","server":"node_repl","tool_name":"js"}"#,
/*metrics*/ None,
)
@@ -979,7 +990,7 @@ async fn sample_configured_conversation_history_with_delivery(
.with_model("gpt-5.5")
.with_config(move |config| {
config.approvals_reviewer = ApprovalsReviewer::AutoReview;
config.guardian_policy_config = guardian_policy;
config.guardian_policy_config = Some("Policy from the live thread.".to_owned());
if code_mode {
for feature in [
Feature::Apps,
@@ -1031,7 +1042,10 @@ async fn sample_configured_conversation_history_with_delivery(
server.uri().trim_start_matches("ws://")
)));
let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::from_api_key("test-api-key"));
// Capture policy independently of the newer live thread configuration.
let mut config = test.config.clone();
config.guardian_policy_config = guardian_policy;
config.guardian_policy_revision = 0;
config.model_provider = provider_info;
config.features.enable(Feature::GuardianV2)?;
let mut builder = ExtensionRegistryBuilder::new();
@@ -1043,6 +1057,7 @@ async fn sample_configured_conversation_history_with_delivery(
let registry = builder.build();
let session_store = ExtensionData::new("session-1");
let thread_store = test.codex.thread_extension_data();
let turn_store = ExtensionData::new("turn-1");
thread_store.insert(RecordingMetrics::default());
let metrics = thread_store.get::<RecordingMetrics>().unwrap();
if has_model_defaults {
@@ -1057,12 +1072,14 @@ async fn sample_configured_conversation_history_with_delivery(
cached_approval(
&registry,
thread_store,
&turn_store,
"review action",
/*metrics*/ None,
)
.await,
None
);
thread_store.insert(config.clone());
registry.thread_lifecycle_contributors()[0]
.on_thread_start(ThreadStartInput {
config: &config,
@@ -1080,7 +1097,7 @@ async fn sample_configured_conversation_history_with_delivery(
.expect("Guardian v2 should initialize")
.wait_for_prewarm(PREWARM_TIMEOUT)
.await?;
let turn_store = ExtensionData::new("turn-1");
turn_store.insert_arc(thread_store.get::<Config>().expect("captured test config"));
let tool_name = ToolName::plain("read_file");
let tool_payload = ToolPayload::Function {
arguments: arguments.to_owned(),
@@ -1144,7 +1161,18 @@ async fn unscored_tools_invalidate_cached_scores() -> Result<()> {
computer_use: Some(codex_protocol::openai_models::GuardianReviewMode::Adaptive),
..Default::default()
});
let mut config = thread_store.get::<Config>().unwrap().as_ref().clone();
let mut requirements = config.config_layer_stack.requirements().clone();
requirements.auto_review_required_models = Some(codex_config::Sourced::new(
[model.slug.clone()].into_iter().collect(),
codex_config::RequirementSource::Unknown,
));
config.config_layer_stack =
codex_config::ConfigLayerStack::new(Vec::new(), requirements, Default::default())?;
thread_store.insert(config);
thread_store.insert(model);
let turn_store = ExtensionData::new("approval-turn");
turn_store.insert_arc(thread_store.get::<Config>().expect("captured test config"));
let progress = thread_store
.get::<GuardianV2ScoreProgress>()
.expect("score progress");
@@ -1164,6 +1192,21 @@ async fn unscored_tools_invalidate_cached_scores() -> Result<()> {
expected.has_unscored_failure = true;
assert_eq!(progress.inspect(/*call_id*/ None), expected);
}
// A classifier started before these actions may still publish, but cannot approve later CUA.
let authorization = ScoreAuthorization::current(&fixture.test.codex, &Default::default()).await;
seed_cached_score(&progress, thread_store, /*index*/ 0, authorization);
assert!(progress.inspect(/*call_id*/ None).has_unscored_failure);
assert_eq!(
cached_approval(
&fixture.registry,
thread_store,
&turn_store,
r#"{"tool":"mcp_tool_call","server":"cua_repl","tool_name":"js"}"#,
/*metrics*/ None,
)
.await,
None
);
Ok(())
}
@@ -1194,10 +1237,12 @@ impl GuardianFailureFixture {
})
.await?;
let session_store = ExtensionData::new("session-1");
session_store.insert_arc(thread_store.get::<Config>().expect("captured test config"));
Ok(Self {
test,
registry,
session_store: ExtensionData::new("session-1"),
session_store,
})
}
@@ -1213,6 +1258,7 @@ impl GuardianFailureFixture {
},
);
let turn_store = ExtensionData::new("turn-1");
turn_store.insert_arc(thread_store.get::<Config>().expect("captured test config"));
let payload = ToolPayload::Function {
arguments: r#"{"path":"README.md"}"#.to_owned(),
};
@@ -1238,6 +1284,8 @@ impl GuardianFailureFixture {
async fn assert_fails_closed(&self, expected_reason: &str) -> Result<()> {
let thread_store = self.test.codex.thread_extension_data();
let turn_store = ExtensionData::new("approval-turn");
turn_store.insert_arc(thread_store.get::<Config>().expect("captured test config"));
let score_progress = thread_store
.get::<GuardianV2ScoreProgress>()
.expect("Guardian v2 should track score progress per thread");
@@ -1257,6 +1305,7 @@ impl GuardianFailureFixture {
cached_approval(
&self.registry,
thread_store,
&turn_store,
"review action",
thread_store
.get::<RecordingMetrics>()
@@ -1379,7 +1428,7 @@ extra_policy = "Allow proactive drafting. Keep {{ tenant_policy_config }} and {{
enabled = true
classifier_instructions = "Predict future violations.\n# Security Policy\n{{ tenant_policy_config }}\nReturn action_risk."
"#;
let (request, _test, _registry) = sample_configured_conversation_history(
let (request, test, registry) = sample_configured_conversation_history(
Vec::new(),
r#"{"path":"README.md"}"#,
Some(TEST_GUARDIAN_POLICY),
@@ -1408,6 +1457,32 @@ classifier_instructions = "Predict future violations.\n# Security Policy\n{{ ten
}],
})
);
let store = test.codex.thread_extension_data();
let turn_store = ExtensionData::new("approval-turn");
turn_store.insert_arc(store.get::<Config>().expect("captured test config"));
let progress = store.get::<GuardianV2ScoreProgress>().unwrap();
let mut updates = progress.updates.subscribe();
tokio::time::timeout(ASYNC_TEST_TIMEOUT, async {
while progress.inspect(/*call_id*/ None).authorization.is_none() {
updates.changed().await?;
}
Ok::<(), anyhow::Error>(())
})
.await??;
let mut score = cached_score(store).expect("captured policy score");
score.scores.insert("action_risk".to_owned(), 0.0);
set_cached_score(store, score);
assert_eq!(
cached_approval(
&registry,
store,
&turn_store,
"review action",
/*metrics*/ None
)
.await,
Some(ReviewDecision::Approved)
);
Ok(())
}
@@ -1570,6 +1645,8 @@ max_recent_non_user_entries = 8
);
let thread_store = test.codex.thread_extension_data();
let turn_store = ExtensionData::new("approval-turn");
turn_store.insert_arc(thread_store.get::<Config>().expect("captured test config"));
let score_progress = thread_store
.get::<GuardianV2ScoreProgress>()
.expect("Guardian v2 should track score progress per thread");
@@ -1608,6 +1685,7 @@ max_recent_non_user_entries = 8
cached_approval(
&registry,
thread_store,
&turn_store,
"review action",
thread_store
.get::<RecordingMetrics>()
@@ -1629,6 +1707,7 @@ max_recent_non_user_entries = 8
cached_approval(
&registry,
thread_store,
&turn_store,
"review action",
thread_store
.get::<RecordingMetrics>()
@@ -1656,6 +1735,7 @@ max_recent_non_user_entries = 8
cached_approval(
&registry,
thread_store,
&turn_store,
&action.to_string(),
/*metrics*/ None
)
@@ -1671,6 +1751,7 @@ max_recent_non_user_entries = 8
cached_approval(
&registry,
thread_store,
&turn_store,
"review action",
thread_store
.get::<RecordingMetrics>()
@@ -1687,6 +1768,7 @@ max_recent_non_user_entries = 8
cached_approval(
&registry,
thread_store,
&turn_store,
"review action",
thread_store
.get::<RecordingMetrics>()
@@ -1706,6 +1788,7 @@ max_recent_non_user_entries = 8
cached_approval(
&registry,
thread_store,
&turn_store,
"review action",
thread_store
.get::<RecordingMetrics>()
@@ -2007,6 +2090,8 @@ async fn contributor_uses_model_defaults_and_preserves_local_overrides() -> Resu
assert!(action.len() <= TruncationPolicy::Tokens(/*limit*/ 128).byte_budget());
let thread_store = test.codex.thread_extension_data();
let turn_store = ExtensionData::new("approval-turn");
turn_store.insert_arc(thread_store.get::<Config>().expect("captured test config"));
let guardian_config = thread_store
.get::<crate::async_scorer::config::GuardianV2Config>()
.expect("Guardian v2 configuration should be installed");
@@ -2046,6 +2131,7 @@ async fn contributor_uses_model_defaults_and_preserves_local_overrides() -> Resu
cached_approval(
&registry,
thread_store,
&turn_store,
"review action",
/*metrics*/ None,
)
@@ -2127,6 +2213,8 @@ async fn contributor_samples_tool_calls_with_the_existing_luna_pool() -> Result<
.await?;
let thread_id = test.session_configured.thread_id;
let thread_store = test.codex.thread_extension_data();
let turn_store = ExtensionData::new("approval-turn");
turn_store.insert_arc(thread_store.get::<Config>().expect("captured test config"));
assert_eq!(request["model"], "gpt-5.6-luna");
let classifier_thread_id = request["client_metadata"]["thread_id"]
.as_str()
@@ -2245,6 +2333,7 @@ async fn contributor_samples_tool_calls_with_the_existing_luna_pool() -> Result<
cached_approval(
&registry,
thread_store,
&turn_store,
"review action",
/*metrics*/ None,
)
@@ -2264,6 +2353,7 @@ async fn contributor_samples_tool_calls_with_the_existing_luna_pool() -> Result<
cached_approval(
&registry,
thread_store,
&turn_store,
"review action",
/*metrics*/ None,
)
@@ -2284,6 +2374,7 @@ async fn contributor_samples_tool_calls_with_the_existing_luna_pool() -> Result<
cached_approval(
&registry,
thread_store,
&turn_store,
"review action",
/*metrics*/ None,
)
@@ -2312,6 +2403,7 @@ async fn contributor_samples_tool_calls_with_the_existing_luna_pool() -> Result<
cached_approval(
&registry,
thread_store,
&turn_store,
"review action",
/*metrics*/ None
)
@@ -2334,6 +2426,7 @@ async fn contributor_samples_tool_calls_with_the_existing_luna_pool() -> Result<
cached_approval(
&registry,
&disabled_thread_store,
&turn_store,
"review action",
/*metrics*/ None
)
@@ -2444,6 +2537,7 @@ async fn contributor_skips_required_models_in_standard_scope() -> Result<()> {
let registry = builder.build();
let session_store = ExtensionData::new("session-1");
let thread_store = test.codex.thread_extension_data();
thread_store.insert(config.clone());
registry.thread_lifecycle_contributors()[0]
.on_thread_start(ThreadStartInput {
config: &config,
@@ -2491,10 +2585,13 @@ async fn contributor_skips_required_models_in_standard_scope() -> Result<()> {
sampled_at: None,
},
);
let turn_store = ExtensionData::new("turn-1");
turn_store.insert_arc(thread_store.get::<Config>().expect("captured test config"));
assert_eq!(
cached_approval(
&registry,
thread_store,
&turn_store,
r#"{"tool":"mcp_tool_call","server":"node_repl"}"#,
/*metrics*/ None,
)
@@ -2502,7 +2599,6 @@ async fn contributor_skips_required_models_in_standard_scope() -> Result<()> {
None
);
let turn_store = ExtensionData::new("turn-1");
let tool_name = ToolName::namespaced("mcp__node_repl__", "js");
let payload = ToolPayload::Function {
arguments: json!({ "path": "protected.md" }).to_string(),
@@ -2513,6 +2609,7 @@ async fn contributor_skips_required_models_in_standard_scope() -> Result<()> {
.repeat(3 * TruncationPolicy::Tokens(DEFAULT_PARENT_COMPACTION_TOKENS).byte_budget()),
internal_chat_message_metadata_passthrough: None,
};
let previous_score = cached_score(thread_store);
registry.tool_lifecycle_contributors()[0]
.on_tool_start(ToolStartInput {
serving_assessment: None,
@@ -2532,15 +2629,35 @@ async fn contributor_skips_required_models_in_standard_scope() -> Result<()> {
})
.await;
assert!(
cached_score(thread_store).is_none(),
"protected models must not receive Guardian v2 fail-closed scores"
);
assert_eq!(cached_score(thread_store), previous_score);
assert_eq!(progress.inspect(/*call_id*/ None).lag, 1);
assert!(
server.connections().iter().all(Vec::is_empty),
"protected models must not spawn Guardian v2 classifiers"
);
// An old turn can still request adaptive scoring after live requirements tighten.
let mut captured = config;
captured.config_layer_stack = initial.config.config_layer_stack.clone();
thread_store.insert(captured);
let mut model = thread_store.get::<ModelInfo>().unwrap().as_ref().clone();
model.guardian = Some(GuardianModelPolicy {
computer_use: Some(codex_protocol::openai_models::GuardianReviewMode::Adaptive),
other_tools: codex_protocol::openai_models::GuardianReviewMode::Adaptive,
..Default::default()
});
thread_store.insert(model);
let fixture = GuardianFailureFixture {
test,
registry,
session_store,
};
fixture.score_tool(ToolName::plain("read_file")).await;
let store = fixture.test.codex.thread_extension_data();
let authorization = ScoreAuthorization::current(&fixture.test.codex, &Default::default()).await;
seed_cached_score(&progress, store, /*index*/ 0, authorization);
assert!(progress.inspect(/*call_id*/ None).has_unscored_failure);
Ok(())
}
@@ -2557,6 +2674,8 @@ async fn cached_score_survives_compaction_and_internal_context_but_not_user_inpu
)
.await?;
let thread_store = test.codex.thread_extension_data();
let turn_store = ExtensionData::new("approval-turn");
turn_store.insert_arc(thread_store.get::<Config>().expect("captured test config"));
let progress = thread_store.get::<GuardianV2ScoreProgress>().unwrap();
tokio::time::timeout(ASYNC_TEST_TIMEOUT, async {
while progress.inspect(/*call_id*/ None).authorization.is_none() {
@@ -2593,6 +2712,7 @@ async fn cached_score_survives_compaction_and_internal_context_but_not_user_inpu
cached_approval(
&registry,
thread_store,
&turn_store,
"review action",
/*metrics*/ None,
)
@@ -2617,6 +2737,7 @@ async fn cached_score_survives_compaction_and_internal_context_but_not_user_inpu
cached_approval(
&registry,
thread_store,
&turn_store,
"review action",
/*metrics*/ None
)
@@ -2633,6 +2754,8 @@ async fn stale_review_context_blocks_cached_approval_without_changing_authorizat
let fixture = GuardianFailureFixture::new().await?;
let store = fixture.test.codex.thread_extension_data();
let turn_store = ExtensionData::new("captured-turn");
turn_store.insert_arc(store.get::<Config>().expect("captured turn config"));
let progress = store.get::<GuardianV2ScoreProgress>().unwrap();
let mut score = cached_score(store).unwrap();
score.scores.insert("action_risk".to_owned(), 0.25);
@@ -2650,13 +2773,13 @@ async fn stale_review_context_blocks_cached_approval_without_changing_authorizat
(current.clone(), Some(ReviewDecision::Approved)),
(stale_local, None),
(stale_root, None),
(current, Some(ReviewDecision::Approved)),
] {
seed_cached_score(&progress, store, /*index*/ 0, authorization);
assert_eq!(
cached_approval(
&fixture.registry,
store,
&turn_store,
"review action",
/*metrics*/ None
)
@@ -2664,6 +2787,20 @@ async fn stale_review_context_blocks_cached_approval_without_changing_authorizat
expected,
);
}
seed_cached_score(&progress, store, /*index*/ 0, current);
// The recovered approval must read its config only from the captured turn.
store.remove::<Config>().expect("captured turn config");
assert_eq!(
cached_approval(
&fixture.registry,
store,
&turn_store,
"review action",
/*metrics*/ None,
)
.await,
Some(ReviewDecision::Approved),
);
Ok(())
}
@@ -2676,6 +2813,8 @@ async fn incompatible_compaction_blocks_cached_score_and_initial_cua_allowance()
let fixture = GuardianFailureFixture::with_config("").await?;
let thread_store = fixture.test.codex.thread_extension_data();
let turn_store = ExtensionData::new("approval-turn");
turn_store.insert_arc(thread_store.get::<Config>().expect("captured test config"));
set_cached_score(
thread_store,
SecurityRiskScore {
@@ -2689,6 +2828,7 @@ async fn incompatible_compaction_blocks_cached_score_and_initial_cua_allowance()
cached_approval(
&fixture.registry,
thread_store,
&turn_store,
"review action",
/*metrics*/ None
)
@@ -2753,6 +2893,7 @@ async fn incompatible_compaction_blocks_cached_score_and_initial_cua_allowance()
cached_approval(
&fixture.registry,
thread_store,
&turn_store,
prompt,
/*metrics*/ None
)
@@ -2777,6 +2918,8 @@ async fn contributor_counts_failed_thread_lookups_toward_score_lag() -> Result<(
.await?;
let session_store = ExtensionData::new("session-1");
let thread_store = test.codex.thread_extension_data();
let turn_store = ExtensionData::new("turn-1");
turn_store.insert_arc(thread_store.get::<Config>().expect("captured test config"));
let score_progress = thread_store
.get::<GuardianV2ScoreProgress>()
.expect("Guardian v2 should track score progress per thread");
@@ -2803,6 +2946,7 @@ async fn contributor_counts_failed_thread_lookups_toward_score_lag() -> Result<(
cached_approval(
&registry,
thread_store,
&turn_store,
"review action",
/*metrics*/ None,
)
@@ -2814,7 +2958,6 @@ async fn contributor_counts_failed_thread_lookups_toward_score_lag() -> Result<(
.remove_thread(&test.session_configured.thread_id)
.await
.expect("the test thread should exist before simulating a failed lookup");
let turn_store = ExtensionData::new("turn-1");
let tool_name = ToolName::plain("read_file");
let payload = ToolPayload::Function {
arguments: r#"{"path":"missing.md"}"#.to_owned(),
@@ -2843,6 +2986,7 @@ async fn contributor_counts_failed_thread_lookups_toward_score_lag() -> Result<(
cached_approval(
&registry,
thread_store,
&turn_store,
"review action",
/*metrics*/ None,
)
@@ -3291,6 +3435,7 @@ async fn assert_parent_compaction_reuse(parent_context_for_review: bool) -> Resu
let session_store = ExtensionData::new("session-1");
let thread_store = test.codex.thread_extension_data();
let metrics = Arc::new(RecordingMetrics::default());
thread_store.insert(config.clone());
registry.thread_lifecycle_contributors()[0]
.on_thread_start(ThreadStartInput {
config: &config,
@@ -3309,6 +3454,7 @@ async fn assert_parent_compaction_reuse(parent_context_for_review: bool) -> Resu
.wait_for_prewarm(PREWARM_TIMEOUT)
.await?;
let turn_store = ExtensionData::new("turn-1");
turn_store.insert_arc(thread_store.get::<Config>().expect("captured test config"));
let tool_name = ToolName::plain("read_file");
let tool_payload = ToolPayload::Function {
arguments: r#"{"path":"README.md"}"#.to_owned(),
@@ -3419,6 +3565,7 @@ async fn assert_parent_compaction_reuse(parent_context_for_review: bool) -> Resu
cached_approval(
&registry,
thread_store,
&turn_store,
"review action",
/*metrics*/ None,
)
@@ -3493,6 +3640,7 @@ async fn assert_parent_compaction_reuse(parent_context_for_review: bool) -> Resu
cached_approval(
&registry,
thread_store,
&turn_store,
"review action",
/*metrics*/ None,
)
@@ -3537,6 +3685,8 @@ async fn cached_approval_respects_action_order_and_wrapper_lag() -> Result<()> {
skip_if_no_network!(Ok(()));
let fixture = GuardianFailureFixture::new().await?;
let store = fixture.test.codex.thread_extension_data();
let turn_store = ExtensionData::new("approval-turn");
turn_store.insert_arc(store.get::<Config>().expect("captured test config"));
let progress = store.get::<GuardianV2ScoreProgress>().unwrap();
let mut score = cached_score(store).unwrap();
score.scores.insert("action_risk".to_owned(), 0.0);
@@ -3576,6 +3726,7 @@ async fn cached_approval_respects_action_order_and_wrapper_lag() -> Result<()> {
cached_approval(
&fixture.registry,
store,
&turn_store,
&json!({"tool": "mcp_tool_call", "server": "example", "id": call_id}).to_string(),
/*metrics*/ None,
)
@@ -3676,19 +3827,24 @@ async fn cached_approval_respects_action_order_and_wrapper_lag() -> Result<()> {
Ok(())
}
/// Exercises decision routing; a fresh review is observed as a cache miss.
/// Routes an approval using the thread and captured turn supplied by the test.
async fn cached_approval(
registry: &codex_extension_api::ExtensionRegistry<Config>,
store: &ExtensionData,
thread_store: &ExtensionData,
turn_store: &ExtensionData,
action: &str,
metrics: Option<Arc<dyn ExtensionMetrics>>,
) -> Option<ReviewDecision> {
let action = serde_json::from_str(action).unwrap_or(serde_json::Value::Null);
let config = turn_store.get::<Config>()?;
let category = match review_scope(&action) {
Some(category) => category,
None if store
None if thread_store
.get::<super::GuardianV2Config>()?
.policy_for_model(store.get::<ModelInfo>().as_deref())
.policy_for_model(
thread_store.get::<ModelInfo>().as_deref(),
config.config_layer_stack.requirements(),
)
.other_tools
== codex_protocol::openai_models::GuardianReviewMode::Adaptive =>
{
@@ -3709,8 +3865,9 @@ async fn cached_approval(
approval_id: "cache-probe",
tool_call_id: action.get("id").and_then(serde_json::Value::as_str),
action: &action,
thread_id: codex_protocol::ThreadId::from_string(store.level_id()).unwrap(),
thread_store: store,
thread_id: codex_protocol::ThreadId::from_string(thread_store.level_id()).unwrap(),
thread_store,
turn_store,
category,
approval_policy: codex_protocol::protocol::AskForApproval::OnRequest,
approvals_reviewer: ApprovalsReviewer::AutoReview,
codex-rs/ext/guardian-v2/src/async_scorer/observation.rs
+33 -21
View File
@@ -81,16 +81,24 @@ impl GuardianV2Extension {
info.tool.name == "js"
}
});
// Count execution attempts even when policy disables background scoring.
if is_js_execution {
score_progress.observe_js_execution();
}
let Some(config) = input.turn_store.get::<codex_core::config::Config>() else {
score_progress.clear_score();
return;
};
if input.serving_assessment.is_some() {
if is_js_execution {
score_progress.observe_js_execution();
}
let index = score_progress.observe(&input);
score_progress.invalidate(index);
return;
}
let parent_model = input.thread_store.get::<ModelInfo>();
let policy = guardian_config.policy_for_model(parent_model.as_deref());
let policy = guardian_config.policy_for_model(
parent_model.as_deref(),
config.config_layer_stack.requirements(),
);
if !policy.scoring_enabled() {
input.thread_store.remove::<GuardianV2Enabled>();
}
@@ -111,14 +119,18 @@ impl GuardianV2Extension {
}
})
.or_else(|| GuardianScope::for_tool(input.tool_name));
// Model policies review nested actions; the Code Mode wrapper leaves their scores alone.
// Mandatory review constraints must not turn transparent catalog wrappers into actions.
// The legacy all-tools policy still scores wrappers through `other_tools`.
let wrapper_mode = parent_model
.as_ref()
.and_then(|model| model.guardian.as_ref())
.map_or(policy.other_tools, |catalog| catalog.other_tools);
if scope.is_none()
&& input.tool_name.is_default_namespace()
&& input.tool_name.name == "exec"
&& matches!(input.payload, ToolPayload::Custom { .. })
&& matches!(input.source, ToolCallSource::Direct)
&& policy.other_tools == GuardianReviewMode::Disabled
&& wrapper_mode == GuardianReviewMode::Disabled
{
return;
}
@@ -147,9 +159,6 @@ impl GuardianV2Extension {
}
return;
}
if is_js_execution {
score_progress.observe_js_execution();
}
let metrics = score_progress.metrics.clone();
let context_mode = GuardianContextMode::from_history(input.conversation_history.as_ref());
let analytics = input.session_store.get::<AnalyticsEventsClient>();
@@ -201,11 +210,10 @@ impl GuardianV2Extension {
.get_thread(parsed_thread_id)
.await
.map_err(|error| error.to_string())?;
let config = thread.config().await;
Ok((manager, thread, config))
Ok((manager, thread))
}
.await;
let (manager, thread, config) = match thread_context {
let (manager, thread) = match thread_context {
Ok(context) => context,
Err(error) => {
score_progress.invalidate(tool_call_index);
@@ -233,15 +241,19 @@ impl GuardianV2Extension {
score_progress.invalidate(tool_call_index);
return;
}
// A required model keeps synchronous review outside its CUA allowance.
if !(scope == Some(GuardianScope::ComputerUse) && policy.allows_initial_cua_call())
&& parent_model.as_ref().is_some_and(|model| {
config
.config_layer_stack
.requirements()
.auto_review_required_for_model(&model.slug)
})
{
// Keep live mandatory review gates without replacing the captured instructions.
let mut permitted = policy.clone();
thread
.config()
.await
.config_layer_stack
.requirements()
.constrain_guardian_policy(
&mut permitted,
parent_model.as_ref().map(|model| model.slug.as_str()),
);
if !scores_tool(&permitted, input.tool_name, input.payload, scope) {
score_progress.invalidate(tool_call_index);
score_progress.clear_score();
return;
}
codex-rs/thread-manager-sample/src/main.rs
+1
View File
@@ -272,6 +272,7 @@ async fn new_config(
base_instructions: None,
base_instructions_provenance: None,
developer_instructions: None,
guardian_policy_revision: 0,
guardian_policy_config: None,
guardian_extra_policy: None,
guardian_policy_template: None,