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")), + ], + ); + } +}