github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-19 16:17:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Record Guardian V2 Decisions agreement and latency metrics (#50273)

Browse Source 
## 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
This commit is contained in:
swastik-oai
2026-10-02 10:44:29 +00:00
committed by copyberry
parent 14a477ea89
commit ca466061d6
5 changed files with 148 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/ext/guardian-v2/src/async_scorer/classification.rs
+30 -3
View File
@@ -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"),
],
);
}
}
}
codex-rs/ext/guardian-v2/src/async_scorer/decisions.rs
+6 -34
View File
@@ -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));
codex-rs/ext/guardian-v2/src/async_scorer/decisions_lifecycle_tests.rs
+33 -14
View File
@@ -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::<DecisionsSampler>();
} 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::<Vec<_>>();
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::<Vec<_>>();
@@ -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::<Vec<_>>();
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;
codex-rs/ext/guardian-v2/src/async_scorer/decisions_tests.rs
+7 -8
View File
@@ -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::<Vec<_>>();
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());
codex-rs/ext/guardian-v2/src/async_scorer/metrics.rs
+72
View File
@@ -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")),
],
);
}
}