From eb354a928b347856055b29fb55ecd740addf8f77 Mon Sep 17 00:00:00 2001 From: ioanbudea-OAI Date: Mon, 19 Oct 2026 10:14:47 +0000 Subject: [PATCH] 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 --- .../guardian-v2/src/async_scorer/approval.rs | 57 ++++++++++++++++++- .../src/async_scorer/extension_tests.rs | 17 ++++-- .../ext/guardian-v2/src/async_scorer/score.rs | 6 ++ codex-rs/protocol/src/approvals.rs | 14 ++++- 4 files changed, 88 insertions(+), 6 deletions(-) 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 ad3cf5638b..2b3d192cd8 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/approval.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/approval.rs @@ -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 { let store = input.thread_store; let Some(progress) = store.get::() 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); 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 05e6aedd85..d5a88815d9 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 @@ -432,8 +432,8 @@ struct RecordingMetrics(Mutex>); impl RecordingMetrics { fn classification_samples(&self) -> Vec { 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::>() ); let metrics = thread_store.get::().unwrap(); + assert_eq!(metrics.0.lock().unwrap().iter().filter(|sample| matches!(sample, RecordedMetric::Counter(name, _, _) if name == "codex.guardian.review.route")).cloned().collect::>(), 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); diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/score.rs b/codex-rs/ext/guardian-v2/src/async_scorer/score.rs index 8599be243b..555e06ddb7 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/score.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/score.rs @@ -73,6 +73,7 @@ struct ScoreState { #[derive(Debug, PartialEq)] pub(super) struct CachedScore { pub(super) lag: usize, + pub(super) sampled_at: Option, 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 diff --git a/codex-rs/protocol/src/approvals.rs b/codex-rs/protocol/src/approvals.rs index 09b7115824..f740befc93 100644 --- a/codex-rs/protocol/src/approvals.rs +++ b/codex-rs/protocol/src/approvals.rs @@ -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,