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
This commit is contained in:
felixxia-oai
2026-08-21 17:27:49 +00:00
committed by copyberry
parent f580dd886f
commit c517cc6d84
3 changed files with 310 additions and 96 deletions

View File

@@ -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(()));

View File

@@ -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::<ModelInfo>();
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::<SecurityRiskScore>();
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(

View File

@@ -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::<SecurityRiskScore>().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::<GuardianV2ScoreProgress>()
.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(()));