Add Guardian review routing and cached score metrics (#55237)

## What changed

- Count adaptive reviews and record synchronous review routes as `async_fallback` or `required_sync`, with the review reason in snake case.
- Record cached score readiness and score age in milliseconds, measured from input sampling time and including inference time.

## Testing

Extend approval tests to assert fallback routing, adaptive review counts, score readiness, and sampled score age.

GitOrigin-RevId: 1200b3297add6c4467e460de87b4a6d8d31e1fc4
This commit is contained in:
ioanbudea-OAI
2026-10-19 10:22:38 +00:00
committed by copyberry
parent 9dc3a1ce7e
commit eb354a928b
4 changed files with 88 additions and 6 deletions
@@ -100,7 +100,11 @@ impl GuardianApprovalReviewer {
if mode == GuardianReviewMode::Disabled && !input.require_guardian {
return ApprovalDecision::AskUser;
}
let reason = if mode == GuardianReviewMode::Adaptive && !input.require_fresh_review {
let adaptive = mode == GuardianReviewMode::Adaptive && !input.require_fresh_review;
if adaptive && let Some(metrics) = input.metrics.as_deref() {
metrics.counter("codex.guardian_v2.adaptive_review", /*inc*/ 1, &[]);
}
let reason = if adaptive {
match guardian_config.as_ref() {
Some(config) => {
match cached_evidence(thread, input, config, &policy, input.metrics.as_deref())
@@ -146,6 +150,23 @@ impl GuardianApprovalReviewer {
}
std::future::pending::<()>().await;
};
if let Some(metrics) = input.metrics.as_deref() {
metrics.counter(
"codex.guardian.review.route",
/*inc*/ 1,
&[
(
"route",
if adaptive {
"async_fallback"
} else {
"required_sync"
},
),
("reason", reason.into()),
],
);
}
match input
.synchronous_reviewer
.review(reason, Some(Box::pin(async_approval)))
@@ -167,6 +188,13 @@ async fn cached_evidence(
) -> Result<bool, GuardianReviewReason> {
let store = input.thread_store;
let Some(progress) = store.get::<GuardianV2ScoreProgress>() else {
if let Some(metrics) = metrics {
metrics.counter(
"codex.guardian_v2.score_readiness",
/*inc*/ 1,
&[("ready", "false")],
);
}
record_fast_decision(metrics, "deferred", "missing_score");
return Err(GuardianReviewReason::MissingScore);
};
@@ -177,6 +205,20 @@ async fn cached_evidence(
.is_ok_and(|action| action.len().saturating_add(1) <= max_action_bytes);
let history = thread.conversation_history_snapshot().await;
let cached = progress.inspect(input.tool_call_id);
if let Some(metrics) = metrics {
metrics.counter(
"codex.guardian_v2.score_readiness",
/*inc*/ 1,
&[(
"ready",
if cached.action_risk.is_some() {
"true"
} else {
"false"
},
)],
);
}
if !action_fits || cached.oversized {
record_fast_decision(metrics, "deferred", "scoring_failure");
return Err(GuardianReviewReason::ScoringFailure);
@@ -227,6 +269,19 @@ async fn cached_evidence(
let current = ScoreAuthorization::current(thread, permissions).await;
// Classification may publish or fail while authorization is collected.
let cached = progress.inspect(input.tool_call_id);
if let Some(metrics) = metrics
&& let Some(sampled_at) = cached.sampled_at
{
// Age of the sampled input, including inference time; not time since publication.
let age = std::time::SystemTime::now()
.duration_since(sampled_at)
.unwrap_or_default();
metrics.histogram(
"codex.guardian_v2.score_age_ms",
i64::try_from(age.as_millis()).unwrap_or(i64::MAX),
&[],
);
}
if cached.oversized {
record_fast_decision(metrics, "deferred", "scoring_failure");
return Err(GuardianReviewReason::ScoringFailure);
@@ -432,8 +432,8 @@ struct RecordingMetrics(Mutex<Vec<RecordedMetric>>);
impl RecordingMetrics {
fn classification_samples(&self) -> Vec<RecordedMetric> {
self.0.lock().unwrap().iter().filter(|sample| {
!matches!(sample, RecordedMetric::Histogram(name, _, _) if name == codex_guardian_context::SECTION_COST_METRIC || name == codex_guardian_context::REQUEST_TOKENS_METRIC || name.starts_with("codex.guardian_v2.connection.") || name.starts_with("codex.guardian_v2.sampling."))
&& !matches!(sample, RecordedMetric::Counter(name, _, _) if name.starts_with("codex.guardian_v2.sampling."))
!matches!(sample, RecordedMetric::Histogram(name, _, _) if name == codex_guardian_context::SECTION_COST_METRIC || name == codex_guardian_context::REQUEST_TOKENS_METRIC || name.starts_with("codex.guardian_v2.connection.") || name.starts_with("codex.guardian_v2.sampling.") || name == "codex.guardian_v2.score_age_ms")
&& !matches!(sample, RecordedMetric::Counter(name, _, _) if name.starts_with("codex.guardian_v2.sampling.") || name == "codex.guardian_v2.adaptive_review" || name == "codex.guardian_v2.score_readiness" || name == "codex.guardian.review.route")
}).cloned().collect()
}
}
@@ -1144,6 +1144,7 @@ async fn unscored_tools_invalidate_cached_scores() -> Result<()> {
.expect("score progress");
let mut expected = progress.inspect(/*call_id*/ None);
expected.action_risk = Some(0.25);
expected.sampled_at = None;
fixture.score_tool(ToolName::plain("wait")).await;
assert_eq!(progress.inspect(/*call_id*/ None), expected);
// A dynamic function named exec is not a Code Mode wrapper. An MCP tool
@@ -1593,7 +1594,7 @@ max_recent_non_user_entries = 8
scores: BTreeMap::from([("action_risk".to_owned(), 0.65)]),
call_id: None,
action: None,
sampled_at: None,
sampled_at: Some((std::time::SystemTime::now() - Duration::from_secs(1)).into()),
},
);
assert_eq!(
@@ -1769,8 +1770,15 @@ max_recent_non_user_entries = 8
.collect::<Vec<_>>()
);
let metrics = thread_store.get::<RecordingMetrics>().unwrap();
assert_eq!(metrics.0.lock().unwrap().iter().filter(|sample| matches!(sample, RecordedMetric::Counter(name, _, _) if name == "codex.guardian.review.route")).cloned().collect::<Vec<_>>(), vec![RecordedMetric::Counter(
"codex.guardian.review.route".to_owned(), 1,
vec![("route".to_owned(), "async_fallback".to_owned()), ("reason".to_owned(), "stale_score".to_owned())],
)]);
assert_eq!(metrics.0.lock().unwrap().iter().filter(|sample| matches!(sample, RecordedMetric::Counter(name, _, _) if name == "codex.guardian_v2.adaptive_review")).count(), 2);
assert_eq!(metrics.0.lock().unwrap().iter().filter(|sample| matches!(sample, RecordedMetric::Counter(name, _, tags) if name == "codex.guardian_v2.score_readiness" && tags == &vec![("ready".to_owned(), "true".to_owned())])).count(), 2);
assert!(initial_metrics.0.lock().unwrap().iter().any(|sample| matches!(sample, RecordedMetric::Histogram(name, age, _) if name == "codex.guardian_v2.score_age_ms" && *age >= 1000)));
assert_eq!(
*metrics.0.lock().unwrap(),
metrics.classification_samples(),
vec![
RecordedMetric::Histogram(TOOL_CALL_LAG_METRIC.to_owned(), 3, vec![]),
RecordedMetric::Counter(
@@ -3797,6 +3805,7 @@ async fn cached_score_publication_rejects_delayed_results_without_changing_cover
let mut expected = cached;
expected.lag = 0;
expected.action_risk = Some(0.0);
expected.sampled_at = delayed_score.sampled_at.map(std::time::SystemTime::from);
expected.authorization = Some(authorization);
assert_eq!(progress.inspect(/*call_id*/ None), expected);
assert!(progress.inspect(Some("active-overflow")).oversized);
@@ -73,6 +73,7 @@ struct ScoreState {
#[derive(Debug, PartialEq)]
pub(super) struct CachedScore {
pub(super) lag: usize,
pub(super) sampled_at: Option<SystemTime>,
pub(super) score_at_or_before_action: bool,
pub(super) has_unscored_failure: bool,
pub(super) js_executions: usize,
@@ -196,6 +197,11 @@ impl GuardianV2ScoreProgress {
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
CachedScore {
sampled_at: state
.score
.as_ref()
.and_then(|score| score.sampled_at)
.map(SystemTime::from),
// Use the accepted score's source, not the coverage high-water mark.
score_at_or_before_action: state
.wrapper_lag
+13 -1
View File
@@ -186,8 +186,20 @@ pub struct NetworkPolicyAmendment {
}
/// Why this approval needs a fresh Guardian assessment.
#[derive(Debug, Clone, Copy, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)]
#[derive(
Debug,
Clone,
Copy,
Deserialize,
Serialize,
PartialEq,
Eq,
JsonSchema,
TS,
strum_macros::IntoStaticStr,
)]
#[serde(rename_all = "snake_case")]
#[strum(serialize_all = "snake_case")]
pub enum GuardianReviewReason {
Policy,
FreshRequired,