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,