From ca466061d64f0b44f416135c7fd06aa7af850bbc Mon Sep 17 00:00:00 2001 From: swastik-oai Date: Fri, 2 Oct 2026 10:36:09 +0000 Subject: [PATCH] Record Guardian V2 Decisions agreement and latency metrics (#50273) ## What changed Record `agree`, `disagree`, or `unavailable` comparisons between Decisions risk classifications and accepted Responses scores. Emit paired latency histograms only when both backends succeed, and include request preparation and queue time in Decisions duration. Report unavailable comparisons when the comparison feature is enabled but the Decisions sampler is not initialized. ## Testing Extend lifecycle coverage for an uninitialized Decisions sampler, checking skipped outcome and unavailable comparison metrics while preserving the baseline score. GitOrigin-RevId: f11789ab5d5fd6957b7913d849bbb2ae7006d3ed --- .../src/async_scorer/classification.rs | 33 ++++++++- .../guardian-v2/src/async_scorer/decisions.rs | 40 ++--------- .../async_scorer/decisions_lifecycle_tests.rs | 47 ++++++++---- .../src/async_scorer/decisions_tests.rs | 15 ++-- .../guardian-v2/src/async_scorer/metrics.rs | 72 +++++++++++++++++++ 5 files changed, 148 insertions(+), 59 deletions(-) diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/classification.rs b/codex-rs/ext/guardian-v2/src/async_scorer/classification.rs index ecabb3dd23..abfa27f22d 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/classification.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/classification.rs @@ -266,6 +266,7 @@ impl Classification { super::metrics::record_section_costs(metrics.as_deref(), context.section_costs()); } let mut decisions_task = None; + let mut responses_duration = None; let mut failure_reason = "invalid_output"; let mut classification_risk = None; let mut classification_finished_at = None; @@ -307,6 +308,7 @@ impl Classification { parent_turn_id: turn_id.clone(), root_turn_id, }; + let mut sampling_started = Instant::now(); let result = match transcript { ClassificationContext::Snapshot(context) => { sampling.input = context.into_messages(); @@ -323,6 +325,7 @@ impl Classification { "not_initialized", ); } + sampling_started = Instant::now(); sampler.sample(sampling).await } @@ -355,6 +358,7 @@ impl Classification { score.await.unwrap_or(Err(LunaSamplerError::Superseded)) } }; + responses_duration = Some(sampling_started.elapsed()); let output = match result { Ok(output) => output, Err(LunaSamplerError::Superseded) => { @@ -468,6 +472,12 @@ impl Classification { "superseded", ); } + // Only compare against a score accepted by the authoritative publication path. + let baseline_risk = if matches!(result, Ok(ClassificationOutcome::Scored)) { + classification_risk + } else { + None + }; if let Err(error) = result { event_sink.emit_warning(ExtensionWarning { thread_id, @@ -476,9 +486,26 @@ impl Classification { }); } if let Some(decisions_task) = decisions_task { - decisions_task - .finish_and_record_outcome(metrics.as_deref()) - .await; + super::metrics::record_decisions_comparison( + decisions_task.finish().await, + baseline_risk.zip(responses_duration), + metrics.as_deref(), + ); + } else if config + .features + .enabled(codex_features::Feature::GuardianV2DecisionsComparison) + && let Some(risk) = baseline_risk + && let Some(metrics) = metrics.as_deref() + { + metrics.counter( + "codex.guardian_v2.decisions_comparison.comparison", + /*inc*/ 1, + &[ + ("comparison", "unavailable"), + ("responses", risk), + ("decisions", "unavailable"), + ], + ); } } } diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/decisions.rs b/codex-rs/ext/guardian-v2/src/async_scorer/decisions.rs index 579150e148..d62680a8fa 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/decisions.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/decisions.rs @@ -3,12 +3,9 @@ //! Unsupported evidence is rejected intact; errors never contain credentials or wire bodies. //! The caller records measurements after baseline publication; dropping its task aborts Decisions work. -use super::metrics::decisions_failure_reason; -use super::metrics::record_decisions_comparison_outcome; use super::sampler::LunaSampler; use super::sampler::LunaSamplingRequest; use codex_context_fragments::RenderedFragment; -use codex_extension_api::ExtensionMetrics; use codex_history::ResponseItemEnvelope; use codex_http_client::HttpClient; use codex_protocol::models::ContentItem; @@ -89,35 +86,10 @@ impl Drop for DecisionsTask { } impl DecisionsTask { - pub(super) async fn finish_and_record_outcome( + pub(super) async fn finish( mut self, - metrics: Option<&dyn ExtensionMetrics>, - ) { - let (result, _duration) = match (&mut self.handle).await { - Ok(result) => result, - Err(error) => { - let (outcome, reason) = if error.is_cancelled() { - ("skipped", "superseded") - } else { - ("failure", "task_error") - }; - record_decisions_comparison_outcome(metrics, outcome, reason); - return; - } - }; - let outcome = match &result { - Ok(_) => "success", - Err(DecisionsError::UnsupportedEvidence | DecisionsError::InputTooLarge) => "skipped", - Err(_) => "failure", - }; - record_decisions_comparison_outcome( - metrics, - outcome, - result - .as_ref() - .err() - .map_or("none", |error| decisions_failure_reason(*error)), - ); + ) -> Result<(Result<&'static str, DecisionsError>, Duration), tokio::task::JoinError> { + (&mut self.handle).await } } @@ -146,6 +118,7 @@ impl DecisionsSampler { request: &LunaSamplingRequest, max_input_tokens: usize, ) -> DecisionsTask { + let started = Instant::now(); let body = request_body( &request.instructions, &request.input, @@ -173,7 +146,7 @@ impl DecisionsSampler { Ok(body) => body, Err(error) => { return DecisionsTask { - handle: tokio::spawn(async move { (Err(error), Duration::ZERO) }), + handle: tokio::spawn(async move { (Err(error), started.elapsed()) }), sampler: Weak::new(), }; } @@ -191,9 +164,8 @@ impl DecisionsSampler { let sampler = Arc::clone(self); let handle = tokio::spawn(async move { let Ok(_permit) = sampler.slots.acquire().await else { - return (Err(DecisionsError::Transport), Duration::ZERO); + return (Err(DecisionsError::Transport), started.elapsed()); }; - let started = Instant::now(); let result = tokio::time::timeout(DEADLINE, sampler.request(body)) .await .unwrap_or(Err(DecisionsError::Timeout)); diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/decisions_lifecycle_tests.rs b/codex-rs/ext/guardian-v2/src/async_scorer/decisions_lifecycle_tests.rs index 518420c08c..6c7bb39074 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/decisions_lifecycle_tests.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/decisions_lifecycle_tests.rs @@ -9,13 +9,17 @@ async fn decisions_disagreement_failures_and_disabled_flag_preserve_baseline() - skip_if_no_network!(Ok(())); for (case, enabled, status, choice, baseline, expected_risk) in [ ("disabled", false, 200, "high", "low", 0.0), + ("not_initialized", true, 200, "high", "low", 0.0), ("cannot_escalate", true, 200, "high", "low", 0.0), ("cannot_lower", true, 200, "low", "high", 1.0), ("candidate_failure", true, 403, "low", "low", 0.0), ("baseline_failure", true, 200, "low", "invalid", 1.0), ] { let server = responses::start_mock_server().await; - let fixture = GuardianFailureFixture::new().await?; + let fixture = GuardianFailureFixture::with_config(&format!( + "[features]\nguardianv2_decisions_comparison = {enabled}\n" + )) + .await?; let test = &fixture.test; let mut config = test.config.clone(); config.features.enable(Feature::GuardianApproval)?; @@ -65,11 +69,13 @@ async fn decisions_disagreement_failures_and_disabled_flag_preserve_baseline() - "choice":choice}] })), ) - .expect(/*r*/ u64::from(enabled)) + .expect(/*r*/ u64::from(enabled && case != "not_initialized")) .mount(&decisions_server) .await; // Inject only the transport, using the existing mock HTTP server instead of credentials. - if enabled { + if case == "not_initialized" { + store.remove::(); + } else if enabled { store.insert(DecisionsSampler::new( HttpClientBuilder::new().build_direct()?, "synthetic-key".into(), @@ -103,7 +109,7 @@ async fn decisions_disagreement_failures_and_disabled_flag_preserve_baseline() - tokio::time::timeout(ASYNC_TEST_TIMEOUT, async { loop { if metrics.0.lock().unwrap().iter().any(|sample| matches!(sample, - RecordedMetric::Counter(name, _, _) if name == "codex.guardian_v2.decisions_comparison")) { break; } + RecordedMetric::Counter(name, _, _) if name == "codex.guardian_v2.decisions_comparison.comparison")) { break; } tokio::task::yield_now().await; } }).await?; @@ -127,7 +133,11 @@ async fn decisions_disagreement_failures_and_disabled_flag_preserve_baseline() - .collect::>(); assert_eq!(body["input"], json!(evidence)); } - let reason = if status == 403 { "http_403" } else { "none" }; + let (outcome, reason) = match case { + "not_initialized" => ("skipped", "not_initialized"), + "candidate_failure" => ("failure", "http_403"), + _ => ("success", "none"), + }; let outcomes = metrics.0.lock().unwrap().iter().filter(|sample| matches!(sample, RecordedMetric::Counter(name, _, _) if name == "codex.guardian_v2.decisions_comparison")) .cloned().collect::>(); @@ -137,20 +147,29 @@ async fn decisions_disagreement_failures_and_disabled_flag_preserve_baseline() - "codex.guardian_v2.decisions_comparison".into(), 1, vec![ - ( - "outcome".into(), - if reason == "none" { - "success" - } else { - "failure" - } - .into() - ), + ("outcome".into(), outcome.into()), ("reason".into(), reason.into()) ] )], "{case}" ); + if case == "not_initialized" { + let comparisons = metrics.0.lock().unwrap().iter().filter(|sample| matches!(sample, + RecordedMetric::Counter(name, _, _) if name == "codex.guardian_v2.decisions_comparison.comparison")) + .cloned().collect::>(); + assert_eq!( + comparisons, + vec![RecordedMetric::Counter( + "codex.guardian_v2.decisions_comparison.comparison".into(), + 1, + vec![ + ("comparison".into(), "unavailable".into()), + ("responses".into(), "low".into()), + ("decisions".into(), "unavailable".into()), + ], + )] + ); + } } assert_eq!(cached_score(store), Some(published), "{case}"); decisions_server.verify().await; diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/decisions_tests.rs b/codex-rs/ext/guardian-v2/src/async_scorer/decisions_tests.rs index dc6976fd75..cdcd588956 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/decisions_tests.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/decisions_tests.rs @@ -76,9 +76,8 @@ async fn http_contract_and_untrusted_response_validation() { ) .unwrap(); let mut sampler = Arc::new(sampler); - let mut task = sampler.spawn(&request, /*max_input_tokens*/ 128_000); - assert_eq!((&mut task.handle).await.unwrap().0, Ok("low")); - drop(task); + let task = sampler.spawn(&request, /*max_input_tokens*/ 128_000); + assert_eq!(task.finish().await.unwrap().0, Ok("low")); let mut invalid = response; invalid["answers"][0]["choice"] = json!("untrusted server text"); assert_eq!(parse_answer(&invalid), Err(DecisionsError::InvalidResponse)); @@ -114,17 +113,17 @@ async fn decisions_admits_newest_request_by_cancelling_oldest() { .acquire_many(MAX_CONCURRENT_REQUESTS as u32) .await .unwrap(); - let mut oldest = sampler.spawn(&request, /*max_input_tokens*/ 128_000); + let oldest = sampler.spawn(&request, /*max_input_tokens*/ 128_000); let mut remaining = (1..MAX_CONCURRENT_REQUESTS) .map(|_| sampler.spawn(&request, /*max_input_tokens*/ 128_000)) .collect::>(); remaining.push(sampler.spawn(&request, /*max_input_tokens*/ 128_000)); - assert!((&mut oldest.handle).await.unwrap_err().is_cancelled()); + assert!(oldest.finish().await.unwrap_err().is_cancelled()); assert!(server.received_requests().await.unwrap().is_empty()); drop(permits); - for mut task in remaining { + for task in remaining { assert_eq!( - (&mut task.handle).await.unwrap().0, + task.finish().await.unwrap().0, Err(DecisionsError::Http(403)) ); } @@ -199,7 +198,7 @@ async fn cancelling_a_decisions_request_releases_thread_capacity() { .await .unwrap(); { - let recording = task.finish_and_record_outcome(/*metrics*/ None); + let recording = task.finish(); tokio::pin!(recording); std::future::poll_fn(|cx| { assert!(std::future::Future::poll(recording.as_mut(), cx).is_pending()); diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/metrics.rs b/codex-rs/ext/guardian-v2/src/async_scorer/metrics.rs index 92ad375621..25b50494c7 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/metrics.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/metrics.rs @@ -194,3 +194,75 @@ pub(super) fn decisions_failure_reason(error: DecisionsError) -> &'static str { DecisionsError::InvalidResponse => "invalid_output", } } + +// Called after the authoritative baseline has been published. +pub(super) fn record_decisions_comparison( + completed: Result<(Result<&'static str, DecisionsError>, Duration), tokio::task::JoinError>, + responses_sample: Option<(&str, Duration)>, + metrics: Option<&dyn ExtensionMetrics>, +) { + let decisions_sample = match completed { + Ok((result, duration)) => { + let outcome = match &result { + Ok(_) => "success", + Err(DecisionsError::UnsupportedEvidence | DecisionsError::InputTooLarge) => { + "skipped" + } + Err(_) => "failure", + }; + record_decisions_comparison_outcome( + metrics, + outcome, + result + .as_ref() + .err() + .map_or("none", |error| decisions_failure_reason(*error)), + ); + result.ok().map(|risk| (risk, duration)) + } + Err(error) => { + let (outcome, reason) = if error.is_cancelled() { + ("skipped", "superseded") + } else { + ("failure", "task_error") + }; + record_decisions_comparison_outcome(metrics, outcome, reason); + None + } + }; + if let Some(metrics) = metrics { + if let (Some((_, responses_duration)), Some((_, decisions_duration))) = + (responses_sample, decisions_sample) + { + // Compare latency only for the same successfully classified requests. + for (backend, duration) in [ + ("responses", responses_duration), + ("decisions", decisions_duration), + ] { + metrics.histogram( + "codex.guardian_v2.decisions_comparison.duration_ms", + i64::try_from(duration.as_millis()).unwrap_or(i64::MAX), + &[("backend", backend), ("outcome", "success")], + ); + } + } + let responses_risk = responses_sample.map(|(risk, _)| risk); + let decisions_risk = decisions_sample.map(|(risk, _)| risk); + let comparison = match (responses_risk, decisions_risk) { + (Some(responses_risk), Some(decisions_risk)) if responses_risk == decisions_risk => { + "agree" + } + (Some(_), Some(_)) => "disagree", + _ => "unavailable", + }; + metrics.counter( + "codex.guardian_v2.decisions_comparison.comparison", + /*inc*/ 1, + &[ + ("comparison", comparison), + ("responses", responses_risk.unwrap_or("unavailable")), + ("decisions", decisions_risk.unwrap_or("unavailable")), + ], + ); + } +}