From c517cc6d8436f1e8f0b7879278d88aeb514a3bdc Mon Sep 17 00:00:00 2001 From: felixxia-oai Date: Fri, 21 Aug 2026 17:27:49 +0000 Subject: [PATCH] Bypass risk scoring for models that require automatic review (#39981) ## Why Models listed in `auto_review.required_on_models` must always use the full automatic review path, regardless of any cached Guardian v2 risk score. ## What changed - Skip Guardian v2 risk classification for models that require automatic review and clear any cached `SecurityRiskScore` before review routing. - Count thread lookup failures as failed scoring attempts so stale scores cannot continue approving later tool calls. ## Testing - Verify required-review models do not start a classifier and always run full reviews. - Verify failed thread lookups advance score lag and fall back to strict review. GitOrigin-RevId: 048d9a80ac2a282e05437a3abb0c46ec21391be8 --- .../app-server/tests/suite/v2/guardian_v2.rs | 111 +++++----- .../guardian-v2/src/async_scorer/extension.rs | 91 ++++---- .../src/async_scorer/extension_tests.rs | 204 ++++++++++++++++++ 3 files changed, 310 insertions(+), 96 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 68573662b2..a0ad8d01df 100644 --- a/codex-rs/app-server/tests/suite/v2/guardian_v2.rs +++ b/codex-rs/app-server/tests/suite/v2/guardian_v2.rs @@ -175,10 +175,11 @@ async fn guardian_v2_routes_tool_approvals( lifecycle: ThreadLifecycle, requirement: ModelReviewRequirement, ) -> Result<()> { - let (luna_score, expected_guardian_reviews) = match risk { - GuardianRisk::Low => (0.25, 1), - GuardianRisk::Threshold => (0.5, 2), - GuardianRisk::High => (0.95, 2), + let (luna_score, expected_guardian_reviews) = match (requirement, risk) { + (ModelReviewRequirement::Required, _) => (0.25, 2), + (ModelReviewRequirement::Optional, GuardianRisk::Low) => (0.25, 1), + (ModelReviewRequirement::Optional, GuardianRisk::Threshold) => (0.5, 2), + (ModelReviewRequirement::Optional, GuardianRisk::High) => (0.95, 2), }; let responses_state = Arc::new(MockResponsesState { luna_score, @@ -316,44 +317,46 @@ async fn guardian_v2_routes_tool_approvals( .await??; assert_eq!(review_started.thread_id, thread_id); - let luna_request = timeout(TIMEOUT, async { - loop { - if let Some(request) = responses_state - .luna_requests - .lock() - .expect("Luna request lock should not be poisoned") - .first() - .cloned() - { - return request; + if matches!(requirement, ModelReviewRequirement::Optional) { + let luna_request = timeout(TIMEOUT, async { + loop { + if let Some(request) = responses_state + .luna_requests + .lock() + .expect("Luna request lock should not be poisoned") + .first() + .cloned() + { + return request; + } + tokio::task::yield_now().await; } - tokio::task::yield_now().await; - } - }) - .await?; - assert_eq!( - luna_request["prompt_cache_key"], - format!("guardian-v2:{thread_id}") - ); - assert!( - luna_request["input"] - .as_array() - .expect("Luna input should be an array") - .iter() - .any(|item| { - item["content"].as_array().is_some_and(|content| { - content.iter().any(|entry| { - entry["text"] - .as_str() - .is_some_and(|text| text.contains(USER_CONTEXT)) + }) + .await?; + assert_eq!( + luna_request["prompt_cache_key"], + format!("guardian-v2:{thread_id}") + ); + assert!( + luna_request["input"] + .as_array() + .expect("Luna input should be an array") + .iter() + .any(|item| { + item["content"].as_array().is_some_and(|content| { + content.iter().any(|entry| { + entry["text"] + .as_str() + .is_some_and(|text| text.contains(USER_CONTEXT)) + }) }) }) - }) - ); - responses_state.allow_luna.notify_one(); - timeout(TIMEOUT, responses_state.classification_completed.notified()).await?; + ); + responses_state.allow_luna.notify_one(); + timeout(TIMEOUT, responses_state.classification_completed.notified()).await?; + responses_state.allow_luna.notify_one(); + } responses_state.allow_guardian_review.notify_one(); - responses_state.allow_luna.notify_one(); timeout( TIMEOUT, app_server.read_stream_until_notification_message("turn/completed"), @@ -363,16 +366,25 @@ async fn guardian_v2_routes_tool_approvals( responses_state.guardian_reviews.load(Ordering::SeqCst), expected_guardian_reviews ); + if matches!(requirement, ModelReviewRequirement::Required) { + assert!( + responses_state + .luna_requests + .lock() + .expect("Luna request lock should not be poisoned") + .is_empty(), + "protected models must not receive Guardian v2 risk scoring" + ); + } + let requires_strict_review = matches!(requirement, ModelReviewRequirement::Optional) + && matches!(risk, GuardianRisk::Threshold | GuardianRisk::High); let strict_review_count = app_server .pending_notification_methods() .into_iter() .filter(|method| method == "autoApprovalReview/strictReviewRequired") .count(); - assert_eq!( - strict_review_count, - usize::from(matches!(risk, GuardianRisk::Threshold | GuardianRisk::High)) - ); - if matches!(risk, GuardianRisk::Threshold | GuardianRisk::High) { + assert_eq!(strict_review_count, usize::from(requires_strict_review)); + if requires_strict_review { let review_started: ItemGuardianApprovalReviewStartedNotification = timeout( TIMEOUT, app_server.read_notification("item/autoApprovalReview/started"), @@ -432,7 +444,7 @@ async fn guardian_v2_threshold_score_requires_full_reviews() -> Result<()> { } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn guardian_v2_required_model_honors_low_risk_assessment() -> Result<()> { +async fn guardian_v2_required_model_bypasses_scoring_and_runs_full_reviews() -> Result<()> { skip_if_no_network!(Ok(())); guardian_v2_routes_tool_approvals( GuardianRisk::Low, @@ -442,17 +454,6 @@ async fn guardian_v2_required_model_honors_low_risk_assessment() -> Result<()> { .await } -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn guardian_v2_required_model_high_risk_requires_full_review() -> Result<()> { - skip_if_no_network!(Ok(())); - guardian_v2_routes_tool_approvals( - GuardianRisk::High, - ThreadLifecycle::New, - ModelReviewRequirement::Required, - ) - .await -} - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn resumed_thread_starts_without_guardian_score() -> Result<()> { skip_if_no_network!(Ok(())); 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 66bb6be3b4..2a8992fe72 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/extension.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/extension.rs @@ -440,7 +440,57 @@ impl GuardianV2Extension { } let metrics = score_progress.metrics.clone(); let sampled_at = SystemTime::now(); + let tool_call_index = score_progress + .latest_tool_call + .fetch_add(/*val*/ 1, Ordering::Relaxed) + .saturating_add(1); + let event_sink = Arc::clone(&self.event_sink); + let thread_id = input.thread_store.level_id().to_owned(); + let turn_id = input.turn_id.to_owned(); + let thread_context: Result<_, String> = async { + let parsed_thread_id = + ThreadId::from_string(&thread_id).map_err(|error| error.to_string())?; + let manager = self + .thread_manager + .upgrade() + .ok_or_else(|| "thread manager is unavailable".to_string())?; + let thread = manager + .get_thread(parsed_thread_id) + .await + .map_err(|error| error.to_string())?; + let config = thread.config().await; + Ok((manager, thread, config)) + } + .await; + let (manager, thread, config) = match thread_context { + Ok(context) => context, + Err(error) => { + score_progress + .latest_failed_tool_call + .fetch_max(tool_call_index, Ordering::Release); + record_classification( + metrics.as_deref(), + classification_started_at.elapsed(), + "failure", + ); + event_sink.emit_warning(ExtensionWarning { + thread_id, + turn_id: Some(turn_id), + message: format!("Guardian V2 risk scoring failed: {error}"), + }); + return; + } + }; let parent_model = input.thread_store.get::(); + if parent_model.as_ref().is_some_and(|model| { + config + .config_layer_stack + .requirements() + .auto_review_required_for_model(&model.slug) + }) { + input.thread_store.remove::(); + return; + } let model_defaults = parent_model .as_ref() .and_then(|model| model.model_messages.as_ref()) @@ -469,10 +519,6 @@ impl GuardianV2Extension { .enable_image_capture(); } input.thread_store.insert(guardian_config.clone()); - let tool_call_index = score_progress - .latest_tool_call - .fetch_add(/*val*/ 1, Ordering::Relaxed) - .saturating_add(1); let latest_parent_compaction = if guardian_config.reuse_parent_compaction { input .conversation_history @@ -516,10 +562,6 @@ impl GuardianV2Extension { ); return; } - let event_sink = Arc::clone(&self.event_sink); - let thread_manager = self.thread_manager.clone(); - let thread_id = input.thread_store.level_id().to_owned(); - let turn_id = input.turn_id.to_owned(); let action = GuardianAction { tool_name: input.tool_name.clone(), payload: input.payload.clone(), @@ -542,38 +584,6 @@ impl GuardianV2Extension { }; tokio::spawn(async move { - let thread_context: Result<_, String> = async { - let parsed_thread_id = - ThreadId::from_string(&thread_id).map_err(|error| error.to_string())?; - let manager = thread_manager - .upgrade() - .ok_or_else(|| "thread manager is unavailable".to_string())?; - let thread = manager - .get_thread(parsed_thread_id) - .await - .map_err(|error| error.to_string())?; - Ok((manager, thread)) - } - .await; - let (manager, thread) = match thread_context { - Ok(context) => context, - Err(error) => { - score_progress - .latest_failed_tool_call - .fetch_max(tool_call_index, Ordering::Release); - record_classification( - metrics.as_deref(), - classification_started_at.elapsed(), - "failure", - ); - event_sink.emit_warning(ExtensionWarning { - thread_id, - turn_id: Some(turn_id), - message: format!("Guardian V2 risk scoring failed: {error}"), - }); - return; - } - }; let root_conversation = thread.guardian_root_conversation().await; let transcript = guardian_config .transcript @@ -627,7 +637,6 @@ impl GuardianV2Extension { ]); let mut classification_finished_at = None; let result: Result<&str, String> = async { - let config = thread.config().await; let review_model_messages = if config.guardian_policy_config.is_none() { let review_model_id = review_model_override.as_deref().unwrap_or_else(|| { create_model_provider( 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 ca3140bb25..1ad7831307 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 @@ -7,6 +7,8 @@ use std::time::SystemTime; use anyhow::Result; use codex_core::config::Config; +use codex_core::config::ConfigBuilder; +use codex_core::config::LoaderOverrides; use codex_core::context::NodeReplReviewEvidence; use codex_extension_api::ConversationHistorySnapshot; use codex_extension_api::ExtensionData; @@ -1565,6 +1567,208 @@ async fn contributor_samples_tool_calls_with_the_existing_luna_pool() -> Result< Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn contributor_skips_models_requiring_managed_guardian_review() -> Result<()> { + skip_if_no_network!(Ok(())); + + let thread_server = responses::start_mock_server().await; + let initial = test_codex().build_with_auto_env(&thread_server).await?; + std::fs::write( + initial.home.path().join("requirements.toml"), + "[auto_review]\nrequired_on_models = [\"protected-model\"]\n", + )?; + let config_layer_stack = ConfigBuilder::default() + .codex_home(initial.home.path().to_path_buf()) + .loader_overrides(LoaderOverrides::with_managed_config_path_for_tests( + initial.home.path().join("managed_config.toml"), + )) + .build() + .await? + .config_layer_stack; + let test = test_codex() + .with_home(Arc::clone(&initial.home)) + .with_config(move |config| { + config.config_layer_stack = config_layer_stack; + config + .features + .enable(Feature::GuardianV2) + .expect("Guardian v2 should remain globally enabled"); + }) + .build_with_auto_env(&thread_server) + .await?; + + let server = responses::start_websocket_server(vec![Vec::new(), Vec::new()]).await; + let provider_info = ModelProviderInfo::create_openai_provider(Some(format!( + "http://{}/v1", + server.uri().trim_start_matches("ws://") + ))); + let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::from_api_key("test-api-key")); + let mut config = test.config.clone(); + config.model_provider = provider_info; + let mut builder = ExtensionRegistryBuilder::new(); + super::install( + &mut builder, + auth_manager, + Arc::downgrade(&test.thread_manager), + ); + let registry = builder.build(); + let session_store = ExtensionData::new("session-1"); + let thread_store = test.codex.thread_extension_data(); + registry.thread_lifecycle_contributors()[0] + .on_thread_start(ThreadStartInput { + config: &config, + session_source: &SessionSource::Exec, + persistent_thread_state_available: false, + environments: &[], + mcp_resource_client: None, + extension_metrics: None, + session_store: &session_store, + thread_store, + }) + .await; + + let mut model_info = test + .thread_manager + .get_models_manager() + .get_model_info("gpt-5.5", &config.to_models_manager_config()) + .await; + model_info.slug = "protected-model".to_owned(); + thread_store.insert(model_info); + thread_store.insert(SecurityRiskScore { + scores: BTreeMap::from([("action_risk".to_owned(), 0.25)]), + sampled_at: None, + }); + assert_eq!( + registry + .approval_review( + &session_store, + thread_store, + "review action", + /*extension_metrics*/ None, + ) + .await, + Some(ReviewDecision::Approved) + ); + + let turn_store = ExtensionData::new("turn-1"); + let tool_name = ToolName::plain("read_file"); + let payload = ToolPayload::Function { + arguments: json!({ "path": "protected.md" }).to_string(), + }; + let oversized_compaction = ResponseItem::Compaction { + id: Some(ResponseItemId::from_server("cmp_oversized".to_owned())), + encrypted_content: "a" + .repeat(TruncationPolicy::Tokens(DEFAULT_PARENT_COMPACTION_TOKENS).byte_budget()), + internal_chat_message_metadata_passthrough: None, + }; + registry.tool_lifecycle_contributors()[0] + .on_tool_start(ToolStartInput { + session_store: &session_store, + thread_store, + turn_store: &turn_store, + turn_id: "turn-1", + call_id: "protected.md", + tool_name: &tool_name, + payload: &payload, + conversation_history: Arc::new(TestConversationHistory(vec![oversized_compaction])), + source: ToolCallSource::Direct, + }) + .await; + + assert!( + thread_store.get::().is_none(), + "protected models must not receive Guardian v2 fail-closed scores" + ); + assert!( + server.connections().iter().all(Vec::is_empty), + "protected models must not spawn Guardian v2 classifiers" + ); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn contributor_counts_failed_thread_lookups_toward_score_lag() -> Result<()> { + skip_if_no_network!(Ok(())); + + let (_, test, registry) = sample_configured_conversation_history( + Vec::new(), + r#"{"path":"README.md"}"#, + Some(TEST_GUARDIAN_POLICY), + "[features.guardianv2]\nenabled = true\nmax_tool_call_lag = 0\n", + /*model_defaults*/ None, + ) + .await?; + let session_store = ExtensionData::new("session-1"); + let thread_store = test.codex.thread_extension_data(); + let score_progress = thread_store + .get::() + .expect("Guardian v2 should track score progress per thread"); + tokio::time::timeout(Duration::from_secs(5), async { + while score_progress + .latest_scored_tool_call + .load(Ordering::Acquire) + == 0 + { + tokio::task::yield_now().await; + } + }) + .await?; + thread_store.insert(SecurityRiskScore { + scores: BTreeMap::from([("action_risk".to_owned(), 0.25)]), + sampled_at: None, + }); + assert_eq!( + registry + .approval_review( + &session_store, + thread_store, + "review action", + /*extension_metrics*/ None, + ) + .await, + Some(ReviewDecision::Approved) + ); + + test.thread_manager + .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(), + }; + registry.tool_lifecycle_contributors()[0] + .on_tool_start(ToolStartInput { + session_store: &session_store, + thread_store, + turn_store: &turn_store, + turn_id: "turn-1", + call_id: "missing.md", + tool_name: &tool_name, + payload: &payload, + conversation_history: Arc::new(TestConversationHistory(Vec::new())), + source: ToolCallSource::Direct, + }) + .await; + + assert_eq!(score_progress.latest_tool_call.load(Ordering::Acquire), 2); + assert_eq!( + registry + .approval_review( + &session_store, + thread_store, + "review action", + /*extension_metrics*/ None, + ) + .await, + None + ); + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn contributor_uses_catalog_policy_without_a_configured_override() -> Result<()> { skip_if_no_network!(Ok(()));