From 86ac0916597244a2e67a197822a5483924660ff1 Mon Sep 17 00:00:00 2001 From: ioanbudea-OAI Date: Thu, 5 Nov 2026 11:30:05 +0000 Subject: [PATCH] Preserve turn configuration in Guardian scoring and approvals (#59642) ## 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 --- .../app-server/tests/suite/v2/guardian_v2.rs | 92 ++++++++ codex-rs/config/src/guardian.rs | 33 +-- codex-rs/config/src/guardian_tests.rs | 44 +++- codex-rs/config/src/state.rs | 27 ++- codex-rs/core/src/agent/role.rs | 35 +-- codex-rs/core/src/agent/role_tests.rs | 1 + codex-rs/core/src/config/guardian_policy.rs | 23 +- codex-rs/core/src/config/mod.rs | 27 +-- codex-rs/core/src/guardian/decision.rs | 2 + codex-rs/core/src/tools/lifecycle.rs | 4 + .../src/contributors/approval_review.rs | 2 + codex-rs/ext/extension-api/tests/registry.rs | 1 + .../ext/guardian-reviewer/src/review_tests.rs | 1 + codex-rs/ext/guardian-reviewer/src/routing.rs | 2 + .../guardian-v2/src/async_scorer/approval.rs | 70 ++++-- .../src/async_scorer/budget_tests.rs | 8 +- .../guardian-v2/src/async_scorer/config.rs | 17 +- .../src/async_scorer/conversation_tests.rs | 8 + .../guardian-v2/src/async_scorer/extension.rs | 12 +- .../extension_cached_delivery_tests.rs | 29 ++- .../src/async_scorer/extension_tests.rs | 213 +++++++++++++++--- .../src/async_scorer/observation.rs | 54 +++-- codex-rs/thread-manager-sample/src/main.rs | 1 + 23 files changed, 532 insertions(+), 174 deletions(-) diff --git a/codex-rs/app-server/tests/suite/v2/guardian_v2.rs b/codex-rs/app-server/tests/suite/v2/guardian_v2.rs index c15253a118..76dee71d43 100644 --- a/codex-rs/app-server/tests/suite/v2/guardian_v2.rs +++ b/codex-rs/app-server/tests/suite/v2/guardian_v2.rs @@ -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(())); diff --git a/codex-rs/config/src/guardian.rs b/codex-rs/config/src/guardian.rs index 8125c95894..189a3918cb 100644 --- a/codex-rs/config/src/guardian.rs +++ b/codex-rs/config/src/guardian.rs @@ -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>, - requirements: &ConfigRequirements, - ) -> Self { + pub fn new(legacy: Option<&FeatureToml>) -> 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(); diff --git a/codex-rs/config/src/guardian_tests.rs b/codex-rs/config/src/guardian_tests.rs index 097c510ff9..d21c42f1ac 100644 --- a/codex-rs/config/src/guardian_tests.rs +++ b/codex-rs/config/src/guardian_tests.rs @@ -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); } diff --git a/codex-rs/config/src/state.rs b/codex-rs/config/src/state.rs index 2d679410d7..9aea73e881 100644 --- a/codex-rs/config/src/state.rs +++ b/codex-rs/config/src/state.rs @@ -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 { + 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 { 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) } diff --git a/codex-rs/core/src/agent/role.rs b/codex-rs/core/src/agent/role.rs index b1642fe5c0..fd21d2229e 100644 --- a/codex-rs/core/src/agent/role.rs +++ b/codex-rs/core/src/agent/role.rs @@ -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 { 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 { - 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 { diff --git a/codex-rs/core/src/agent/role_tests.rs b/codex-rs/core/src/agent/role_tests.rs index 09c7447382..bc54948f38 100644 --- a/codex-rs/core/src/agent/role_tests.rs +++ b/codex-rs/core/src/agent/role_tests.rs @@ -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; diff --git a/codex-rs/core/src/config/guardian_policy.rs b/codex-rs/core/src/config/guardian_policy.rs index e1f21cde38..92f98587f5 100644 --- a/codex-rs/core/src/config/guardian_policy.rs +++ b/codex-rs/core/src/config/guardian_policy.rs @@ -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, Option) { + 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( diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index f11bb4ab10..44addff5e4 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/codex-rs/core/src/config/mod.rs @@ -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, + /// 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, diff --git a/codex-rs/core/src/guardian/decision.rs b/codex-rs/core/src/guardian/decision.rs index ab11c061f4..372ff1a8e9 100644 --- a/codex-rs/core/src/guardian/decision.rs +++ b/codex-rs/core/src/guardian/decision.rs @@ -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, diff --git a/codex-rs/core/src/tools/lifecycle.rs b/codex-rs/core/src/tools/lifecycle.rs index 7a787aa71f..1b6bec2b0d 100644 --- a/codex-rs/core/src/tools/lifecycle.rs +++ b/codex-rs/core/src/tools/lifecycle.rs @@ -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(); diff --git a/codex-rs/ext/extension-api/src/contributors/approval_review.rs b/codex-rs/ext/extension-api/src/contributors/approval_review.rs index 26ba18e0a5..5c88034981 100644 --- a/codex-rs/ext/extension-api/src/contributors/approval_review.rs +++ b/codex-rs/ext/extension-api/src/contributors/approval_review.rs @@ -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, diff --git a/codex-rs/ext/extension-api/tests/registry.rs b/codex-rs/ext/extension-api/tests/registry.rs index d8e495f45d..0f08651ea5 100644 --- a/codex-rs/ext/extension-api/tests/registry.rs +++ b/codex-rs/ext/extension-api/tests/registry.rs @@ -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, diff --git a/codex-rs/ext/guardian-reviewer/src/review_tests.rs b/codex-rs/ext/guardian-reviewer/src/review_tests.rs index 4c606953a9..78c2911e8f 100644 --- a/codex-rs/ext/guardian-reviewer/src/review_tests.rs +++ b/codex-rs/ext/guardian-reviewer/src/review_tests.rs @@ -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, diff --git a/codex-rs/ext/guardian-reviewer/src/routing.rs b/codex-rs/ext/guardian-reviewer/src/routing.rs index 03aa11a13c..c173bd376a 100644 --- a/codex-rs/ext/guardian-reviewer/src/routing.rs +++ b/codex-rs/ext/guardian-reviewer/src/routing.rs @@ -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 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, diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/approval.rs b/codex-rs/ext/guardian-v2/src/async_scorer/approval.rs index 12a62619fb..dedf89cd92 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/approval.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/approval.rs @@ -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::()?; + 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::(); @@ -75,24 +77,28 @@ impl GuardianApprovalReviewer { .thread_store .get::() .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::() + .as_ref() + .map(|model| model.slug.as_str()), + ); + permitted +} diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/budget_tests.rs b/codex-rs/ext/guardian-v2/src/async_scorer/budget_tests.rs index d2f7655e55..9b6d245810 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/budget_tests.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/budget_tests.rs @@ -42,6 +42,9 @@ async fn catalog_budget_fixture(base_url: String, window: i64) -> Result Result<()> { } }; let thread_store = fixture.test.codex.thread_extension_data(); + let turn_store = ExtensionData::new("turn-1"); + turn_store.insert_arc(thread_store.get::().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 ) diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/config.rs b/codex-rs/ext/guardian-v2/src/async_scorer/config.rs index adc4402260..6cc765638c 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/config.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/config.rs @@ -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, diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/conversation_tests.rs b/codex-rs/ext/guardian-v2/src/async_scorer/conversation_tests.rs index 1f1bd06965..033a4c0e70 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/conversation_tests.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/conversation_tests.rs @@ -301,6 +301,8 @@ async fn delayed_preparation_does_not_block_newer_requests_and_capacity_is_bound .get::() .unwrap(); let store = fixture.test.codex.thread_extension_data(); + let turn_store = ExtensionData::new("approval-turn"); + turn_store.insert_arc(store.get::().expect("captured test config")); let progress = store.get::().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::().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::().unwrap(); let turn_store = ExtensionData::new("turn-1"); + turn_store.insert_arc(store.get::().expect("captured test config")); let tool_name = ToolName::plain("read_file"); let payload = ToolPayload::Function { arguments: r#"{"path":"README.md"}"#.to_owned(), diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/extension.rs b/codex-rs/ext/guardian-v2/src/async_scorer/extension.rs index 034333defe..32911bb006 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/extension.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/extension.rs @@ -63,14 +63,10 @@ impl ThreadLifecycleContributor 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; diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/extension_cached_delivery_tests.rs b/codex-rs/ext/guardian-v2/src/async_scorer/extension_cached_delivery_tests.rs index f9de5e5003..97258a9c37 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/extension_cached_delivery_tests.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/extension_cached_delivery_tests.rs @@ -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::().expect("captured test config")); let progress = store.get::().unwrap(); seed_cached_score(&progress, store, /*index*/ 1_000, before.clone()); assert_eq!( - cached_approval(®istry, store, "review action", /*metrics*/ None).await, + cached_approval( + ®istry, + 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::().expect("captured test config")); assert_eq!( cached_approval( ®istry, - 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::().expect("captured test config")); let progress = store.get::().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(®istry, store, "review action", /*metrics*/ None).await, + cached_approval( + ®istry, + store, + &turn_store, + "review action", + /*metrics*/ None + ) + .await, Some(ReviewDecision::Approved) ); let shutdown = test diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs b/codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs index a6d20d3ab6..ff16ff40ca 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs @@ -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::() .expect("Guardian v2 should initialize"); let turn_store = ExtensionData::new("turn-1"); + turn_store.insert_arc(thread_store.get::().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( ®istry, 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::().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::().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::() @@ -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::() @@ -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::().unwrap(); if has_model_defaults { @@ -1057,12 +1072,14 @@ async fn sample_configured_conversation_history_with_delivery( cached_approval( ®istry, 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::().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::().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::().expect("captured test config")); let progress = thread_store .get::() .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::().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::().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::().expect("captured test config")); let score_progress = thread_store .get::() .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::() @@ -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::().expect("captured test config")); + let progress = store.get::().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( + ®istry, + 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::().expect("captured test config")); let score_progress = thread_store .get::() .expect("Guardian v2 should track score progress per thread"); @@ -1608,6 +1685,7 @@ max_recent_non_user_entries = 8 cached_approval( ®istry, thread_store, + &turn_store, "review action", thread_store .get::() @@ -1629,6 +1707,7 @@ max_recent_non_user_entries = 8 cached_approval( ®istry, thread_store, + &turn_store, "review action", thread_store .get::() @@ -1656,6 +1735,7 @@ max_recent_non_user_entries = 8 cached_approval( ®istry, thread_store, + &turn_store, &action.to_string(), /*metrics*/ None ) @@ -1671,6 +1751,7 @@ max_recent_non_user_entries = 8 cached_approval( ®istry, thread_store, + &turn_store, "review action", thread_store .get::() @@ -1687,6 +1768,7 @@ max_recent_non_user_entries = 8 cached_approval( ®istry, thread_store, + &turn_store, "review action", thread_store .get::() @@ -1706,6 +1788,7 @@ max_recent_non_user_entries = 8 cached_approval( ®istry, thread_store, + &turn_store, "review action", thread_store .get::() @@ -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::().expect("captured test config")); let guardian_config = thread_store .get::() .expect("Guardian v2 configuration should be installed"); @@ -2046,6 +2131,7 @@ async fn contributor_uses_model_defaults_and_preserves_local_overrides() -> Resu cached_approval( ®istry, 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::().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( ®istry, 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( ®istry, 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( ®istry, 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( ®istry, 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( ®istry, &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::().expect("captured test config")); assert_eq!( cached_approval( ®istry, 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::().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::().expect("captured test config")); let progress = thread_store.get::().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( ®istry, 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( ®istry, 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::().expect("captured turn config")); let progress = store.get::().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::().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::().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::().expect("captured test config")); let score_progress = thread_store .get::() .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( ®istry, 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( ®istry, 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::().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( ®istry, 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( ®istry, 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::().expect("captured test config")); let progress = store.get::().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, - store: &ExtensionData, + thread_store: &ExtensionData, + turn_store: &ExtensionData, action: &str, metrics: Option>, ) -> Option { let action = serde_json::from_str(action).unwrap_or(serde_json::Value::Null); + let config = turn_store.get::()?; let category = match review_scope(&action) { Some(category) => category, - None if store + None if thread_store .get::()? - .policy_for_model(store.get::().as_deref()) + .policy_for_model( + thread_store.get::().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, diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/observation.rs b/codex-rs/ext/guardian-v2/src/async_scorer/observation.rs index d71e2ad8b9..bd8f7212f6 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/observation.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/observation.rs @@ -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::() 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::(); - 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::(); } @@ -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::(); @@ -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; } diff --git a/codex-rs/thread-manager-sample/src/main.rs b/codex-rs/thread-manager-sample/src/main.rs index e7e8ed4c7b..d844abd2ab 100644 --- a/codex-rs/thread-manager-sample/src/main.rs +++ b/codex-rs/thread-manager-sample/src/main.rs @@ -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,