From e683c3118b25bc2fce65e9cbccded7a8343307f1 Mon Sep 17 00:00:00 2001 From: felixxia-oai Date: Tue, 18 Aug 2026 17:39:08 +0000 Subject: [PATCH] Record Guardian v2 classification metrics (#39241) ## What changed - Emit a `codex.guardian_v2.classification` counter for each classification. - Record classification latency in `codex.guardian_v2.classification.duration_ms`. - Tag both metrics with `success`, `failure`, or `superseded` outcomes. ## Testing - Verify metrics for successful classification and fail-closed handling. GitOrigin-RevId: 021702f3b26565d6da5708d2b8f1f8881942a441 --- codex-rs/ext/guardian-v2/src/extension.rs | 71 ++++++++++++++++--- .../ext/guardian-v2/src/extension_tests.rs | 36 +++++++++- 2 files changed, 95 insertions(+), 12 deletions(-) diff --git a/codex-rs/ext/guardian-v2/src/extension.rs b/codex-rs/ext/guardian-v2/src/extension.rs index eaa8fb6f6f..f11ee0fe47 100644 --- a/codex-rs/ext/guardian-v2/src/extension.rs +++ b/codex-rs/ext/guardian-v2/src/extension.rs @@ -3,6 +3,8 @@ use std::sync::Arc; use std::sync::Weak; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; +use std::time::Duration; +use std::time::Instant; use std::time::SystemTime; use codex_core::ThreadManager; @@ -195,11 +197,31 @@ struct GuardianV2Enabled; // Sampled when the approval contributor evaluates the cached Luna score. const TOOL_CALL_LAG_METRIC: &str = "codex.guardian_v2.tool_call_lag"; const REVIEW_FALLBACK_METRIC: &str = "codex.guardian_v2.review_fallback"; +const CLASSIFICATION_METRIC: &str = "codex.guardian_v2.classification"; +const CLASSIFICATION_DURATION_METRIC: &str = "codex.guardian_v2.classification.duration_ms"; #[derive(Default)] struct GuardianV2ScoreProgress { latest_tool_call: AtomicUsize, latest_scored_tool_call: AtomicUsize, + metrics: Option>, +} + +fn record_classification( + metrics: Option<&dyn ExtensionMetrics>, + duration: Duration, + outcome: &str, +) { + let Some(metrics) = metrics else { + return; + }; + let tags = [("outcome", outcome)]; + metrics.counter(CLASSIFICATION_METRIC, /*inc*/ 1, &tags); + metrics.histogram( + CLASSIFICATION_DURATION_METRIC, + i64::try_from(duration.as_millis()).unwrap_or(i64::MAX), + &tags, + ); } #[derive(Clone)] @@ -275,9 +297,10 @@ impl ThreadLifecycleContributor for GuardianV2Extension { } input.thread_store.insert(sampler); input.thread_store.insert(guardian_config); - input - .thread_store - .insert(GuardianV2ScoreProgress::default()); + input.thread_store.insert(GuardianV2ScoreProgress { + metrics: input.extension_metrics.clone(), + ..Default::default() + }); input.thread_store.insert(GuardianV2Enabled); } Err(error) => self.event_sink.emit_warning(ExtensionWarning { @@ -345,12 +368,17 @@ impl ApprovalReviewContributor for GuardianV2Extension { impl ToolLifecycleContributor for GuardianV2Extension { fn on_tool_start<'a>(&'a self, input: ToolStartInput<'a>) -> ToolLifecycleFuture<'a> { + let classification_started_at = Instant::now(); let Some(sampler) = input.thread_store.get::() else { return Box::pin(std::future::ready(())); }; let Some(guardian_config) = input.thread_store.get::() else { return Box::pin(std::future::ready(())); }; + let Some(score_progress) = input.thread_store.get::() else { + return Box::pin(std::future::ready(())); + }; + let metrics = score_progress.metrics.clone(); let parent_model = input.thread_store.get::(); let model_defaults = parent_model .as_ref() @@ -359,6 +387,11 @@ impl ToolLifecycleContributor for GuardianV2Extension { let guardian_config = match guardian_config.with_model_defaults(model_defaults) { Ok(config) => config, Err(error) => { + record_classification( + metrics.as_deref(), + classification_started_at.elapsed(), + "failure", + ); self.event_sink.emit_warning(ExtensionWarning { thread_id: input.thread_store.level_id().to_owned(), turn_id: Some(input.turn_id.to_owned()), @@ -368,9 +401,6 @@ impl ToolLifecycleContributor for GuardianV2Extension { } }; input.thread_store.insert(guardian_config.clone()); - let Some(score_progress) = input.thread_store.get::() else { - return Box::pin(std::future::ready(())); - }; let tool_call_index = score_progress .latest_tool_call .fetch_add(/*val*/ 1, Ordering::Relaxed) @@ -414,6 +444,11 @@ impl ToolLifecycleContributor for GuardianV2Extension { input.thread_store.insert_if(score.clone(), |previous| { previous.is_none_or(|previous| previous.sampled_at <= score.sampled_at) }); + record_classification( + metrics.as_deref(), + classification_started_at.elapsed(), + "failure", + ); return Box::pin(std::future::ready(())); } let event_sink = Arc::clone(&self.event_sink); @@ -452,6 +487,11 @@ impl ToolLifecycleContributor for GuardianV2Extension { let planned_action = match action.render(guardian_config.max_action_tokens) { Ok(planned_action) => planned_action, Err(error) => { + record_classification( + metrics.as_deref(), + classification_started_at.elapsed(), + "failure", + ); event_sink.emit_warning(ExtensionWarning { thread_id, turn_id: Some(turn_id), @@ -470,7 +510,8 @@ impl ToolLifecycleContributor for GuardianV2Extension { format!("{planned_action}\n"), ">>> APPROVAL REQUEST END\n".to_owned(), ]); - let result: Result<(), String> = async { + let mut classification_finished_at = None; + let result: Result<&str, String> = async { let parsed_thread_id = ThreadId::from_string(&thread_id).map_err(|error| error.to_string())?; let manager = thread_manager @@ -544,7 +585,7 @@ impl ToolLifecycleContributor for GuardianV2Extension { .await { Ok(output) => output, - Err(LunaSamplerError::Superseded) => return Ok(()), + Err(LunaSamplerError::Superseded) => return Ok("superseded"), Err(error) => return Err(error.to_string()), }; let output: serde_json::Value = @@ -573,20 +614,30 @@ impl ToolLifecycleContributor for GuardianV2Extension { previous.is_none_or(|previous| previous.sampled_at < score.sampled_at) }) { - return Ok(()); + return Ok("superseded"); } score_progress .latest_scored_tool_call .fetch_max(tool_call_index, Ordering::Release); + classification_finished_at = Some(Instant::now()); if !config.ephemeral { thread .append_rollout_items(&[RolloutItem::SecurityRiskScore(score)]) .await .map_err(|error| error.to_string())?; } - Ok(()) + Ok("success") } .await; + record_classification( + metrics.as_deref(), + classification_finished_at + .map(|finished_at: Instant| { + finished_at.duration_since(classification_started_at) + }) + .unwrap_or_else(|| classification_started_at.elapsed()), + result.as_deref().unwrap_or("failure"), + ); if let Err(error) = result { event_sink.emit_warning(ExtensionWarning { thread_id, diff --git a/codex-rs/ext/guardian-v2/src/extension_tests.rs b/codex-rs/ext/guardian-v2/src/extension_tests.rs index 30d1f62ed9..6389f62602 100644 --- a/codex-rs/ext/guardian-v2/src/extension_tests.rs +++ b/codex-rs/ext/guardian-v2/src/extension_tests.rs @@ -48,6 +48,8 @@ use core_test_support::test_codex::test_codex; use pretty_assertions::assert_eq; use serde_json::json; +use super::CLASSIFICATION_DURATION_METRIC; +use super::CLASSIFICATION_METRIC; use super::GuardianV2ScoreProgress; use super::REVIEW_FALLBACK_METRIC; use super::StrictReviewReason; @@ -632,11 +634,13 @@ max_recent_non_user_entries = 8 let score_progress = thread_store .get::() .expect("Guardian v2 should track score progress per thread"); + let metrics = thread_store.get::().unwrap(); tokio::time::timeout(Duration::from_secs(5), async { while score_progress .latest_scored_tool_call .load(Ordering::Acquire) == 0 + || metrics.0.lock().unwrap().len() < 2 { tokio::task::yield_now().await; } @@ -745,9 +749,28 @@ max_recent_non_user_entries = 8 Some(ReviewDecision::Approved) ); + let samples = initial_metrics.0.lock().unwrap(); + let classification_duration_ms = match &samples[1] { + RecordedMetric::Histogram(name, duration_ms, _) + if name == CLASSIFICATION_DURATION_METRIC => + { + *duration_ms + } + sample => panic!("expected classification duration metric, got {sample:?}"), + }; assert_eq!( - *initial_metrics.0.lock().unwrap(), + *samples, vec![ + RecordedMetric::Counter( + CLASSIFICATION_METRIC.to_owned(), + 1, + vec![("outcome".to_owned(), "success".to_owned())], + ), + RecordedMetric::Histogram( + CLASSIFICATION_DURATION_METRIC.to_owned(), + classification_duration_ms, + vec![("outcome".to_owned(), "success".to_owned())], + ), RecordedMetric::Histogram(TOOL_CALL_LAG_METRIC.to_owned(), 0, vec![]), RecordedMetric::Histogram(TOOL_CALL_LAG_METRIC.to_owned(), 0, vec![]), RecordedMetric::Histogram(TOOL_CALL_LAG_METRIC.to_owned(), 2, vec![]), @@ -1338,6 +1361,7 @@ async fn contributor_reuses_the_latest_compatible_parent_compaction() -> Result< let registry = builder.build(); let session_store = ExtensionData::new("session-1"); let thread_store = test.codex.thread_extension_data(); + let metrics = Arc::new(RecordingMetrics::default()); thread_store.insert(parent_model); registry.thread_lifecycle_contributors()[0] .on_thread_start(ThreadStartInput { @@ -1346,7 +1370,7 @@ async fn contributor_reuses_the_latest_compatible_parent_compaction() -> Result< persistent_thread_state_available: false, environments: &[], mcp_resource_client: None, - extension_metrics: None, + extension_metrics: Some(metrics.clone()), session_store: &session_store, thread_store, }) @@ -1488,6 +1512,14 @@ async fn contributor_reuses_the_latest_compatible_parent_compaction() -> Result< 1, "an oversized latest compaction must bypass Luna rather than reuse stale context" ); + assert!(metrics.0.lock().unwrap().iter().any(|sample| { + matches!( + sample, + RecordedMetric::Counter(name, 1, tags) + if name == CLASSIFICATION_METRIC + && tags == &[("outcome".to_owned(), "failure".to_owned())] + ) + })); Ok(()) }