openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-20 17:38:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add Guardian retry metrics and preserve turn attribution (#55347)

Browse Source 
## What changed

- Record review attempt counts and final outcomes with a `retried` tag.
- Measure retry waits by failure reason and whether the wait ends in a retry, cancellation, or deadline; count retries only when another attempt will run.
- Count required synchronous reviews with the `required_sync` route.
- Use the originating turn's telemetry for Guardian reviews so metrics retain that turn's model attribution.

## Testing

Add metric assertions for attempt counts, retry outcomes, and waits interrupted by cancellation or deadlines. Extend the delayed cached-score integration test to verify route and retry-wait metrics use the originating turn's model after a model override.

GitOrigin-RevId: 681a719ad5dd9e4436cb37b6f5f972cb11a5c890
This commit is contained in:
ioanbudea-OAI
2026-10-19 18:49:48 +00:00
committed by copyberry
parent 982695cdbc
commit 941b32bf57
7 changed files with 227 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/guardian/decision.rs
+2 -1
View File
@@ -42,6 +42,7 @@ pub(crate) fn spawn_approval_decision(
}
/// `None` requests the existing user flow. No contributor is never an implicit allow.
/// Both metric sinks use the originating turn retained by the review context.
pub(crate) async fn decide_approval(
session: Arc<Session>,
context: impl Into<GuardianReviewContext>,
@@ -102,7 +103,7 @@ pub(crate) async fn decide_approval(
cancellation: cancellation.clone(),
// Keep the existing turn-level reporting policy during this ownership move.
model: turn.model_info(),
telemetry: &session.services.session_telemetry,
telemetry: &turn.session_telemetry,
log_assessments: turn.config.otel.guardian_assessment_logging_enabled(),
analytics: &session.services.analytics_events_client,
metrics: Some(crate::session::extension_metrics::from_session_telemetry(
codex-rs/core/src/guardian/review.rs
+10 -4
View File
@@ -14,6 +14,8 @@ use codex_guardian_reviewer::GuardianReviewOutcome;
#[cfg(test)]
use codex_guardian_reviewer::GuardianReviewSessionLimits;
use codex_guardian_reviewer::ReviewModel;
#[cfg(test)]
use codex_guardian_reviewer::run_with_retry;
use codex_prompts::ResolvedModelMessages;
use codex_protocol::openai_models::GuardianScope;
use codex_protocol::protocol::EventMsg;
@@ -263,8 +265,11 @@ async fn run_guardian_review_session_with_retry_before_deadline(
limits: GuardianReviewSessionLimits,
) -> (GuardianReviewOutcome, GuardianReviewAnalyticsResult) {
let context = context.into();
let (outcome, analytics, _) =
codex_guardian_reviewer::run_with_retry(limits, external_cancel.as_ref(), |deadline| {
let (outcome, analytics, _) = run_with_retry(
limits,
external_cancel.as_ref(),
/*telemetry*/ None,
|deadline| {
let attempt = run_guardian_review_session_before_deadline(
Arc::clone(&session),
context.clone(),
@@ -281,7 +286,8 @@ async fn run_guardian_review_session_with_retry_before_deadline(
let (outcome, analytics) = attempt.await;
(outcome, analytics, None::<()>)
}
})
.await;
},
)
.await;
(outcome, analytics)
}
codex-rs/core/tests/suite/guardian_cached_score_tests.rs
+73 -5
View File
@@ -1,5 +1,6 @@
//! Exercises asynchronous score publication through real tool calls and approval results.
use std::collections::BTreeMap;
use std::time::Duration;
use anyhow::Result;
@@ -7,6 +8,8 @@ use codex_core::TurnInputRequest;
use codex_core::config::Constrained;
use codex_features::Feature;
use codex_history::RolloutItem;
use codex_otel::MetricsClient;
use codex_otel::MetricsConfig;
use codex_protocol::approvals::GuardianAssessmentStatus;
use codex_protocol::approvals::GuardianReviewReason;
use codex_protocol::config_types::ApprovalsReviewer;
@@ -14,6 +17,7 @@ use codex_protocol::models::NetworkPermissions;
use codex_protocol::models::PermissionProfile;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::ThreadSettingsOverrides;
use codex_protocol::request_permissions::PermissionGrantScope;
use codex_protocol::request_permissions::RequestPermissionProfile;
use codex_protocol::request_permissions::RequestPermissionsResponse;
@@ -29,6 +33,8 @@ use core_test_support::streaming_sse::StreamingSseChunk;
use core_test_support::streaming_sse::start_streaming_sse_server;
use core_test_support::test_codex::test_codex;
use core_test_support::wait_for_event;
use opentelemetry_sdk::metrics::data::AggregatedMetrics;
use opentelemetry_sdk::metrics::data::MetricData;
use pretty_assertions::assert_eq;
use serde_json::json;
use tokio::sync::oneshot;
@@ -123,6 +129,17 @@ async fn delayed_score_only_covers_the_tool_call_it_classified() -> Result<()> {
]),
)
.await;
Mock::given(method("POST"))
.and(body_partial_json(json!({ "model": "gpt-5.5" })))
.respond_with(responses::sse_response(sse(vec![
ev_assistant_message("invalid", "invalid assessment"),
ev_completed("invalid"),
])))
.with_priority(1)
.up_to_n_times(1)
.expect(1)
.mount(&server)
.await;
Mock::given(method("POST"))
.and(path("/v1/responses"))
.and(body_partial_json(json!({
@@ -148,7 +165,8 @@ async fn delayed_score_only_covers_the_tool_call_it_classified() -> Result<()> {
.mount(&server)
.await;
let test = test_codex()
let mut test = test_codex()
.with_model("guardian-initial-model")
.with_pre_build_hook(|home| {
std::fs::write(
home.join("config.toml"),
@@ -175,12 +193,26 @@ async fn delayed_score_only_covers_the_tool_call_it_classified() -> Result<()> {
})
.build_with_auto_env(&server)
.await?;
let metrics = MetricsClient::new(
MetricsConfig::in_memory("test", "test", "1", Default::default()).with_runtime_reader(),
)?;
let mut options = test.start_thread_options().await;
options.history_mode = Some(codex_protocol::protocol::ThreadHistoryMode::Legacy);
options.thread_extension_init.insert(metrics.clone());
test.codex.shutdown_and_wait().await?;
test.codex = test.thread_manager.start_thread(options).await?.thread;
test.codex.ensure_rollout_materialized().await;
test.codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "Inspect the available permissions.".to_owned(),
text_elements: Vec::new(),
}]))
.start_or_steer_turn(
TurnInputRequest::user_input(vec![UserInput::Text {
text: "Inspect the available permissions.".to_owned(),
text_elements: Vec::new(),
}])
.with_thread_settings(ThreadSettingsOverrides {
model: Some("guardian-publication-parent".into()),
..Default::default()
}),
)
.await?;
let mut parent_gates = parent_gates.into_iter();
@@ -283,6 +315,42 @@ async fn delayed_score_only_covers_the_tool_call_it_classified() -> Result<()> {
}
test.codex.shutdown_and_wait().await?;
let snapshot = metrics.snapshot()?;
let route = snapshot
.scope_metrics()
.flat_map(opentelemetry_sdk::metrics::data::ScopeMetrics::metrics)
.find(|metric| metric.name() == "codex.guardian.review.route")
.unwrap();
let AggregatedMetrics::U64(MetricData::Sum(routes)) = route.data() else {
panic!("expected route counter");
};
for point in routes.data_points() {
let model = point
.attributes()
.find(|tag| tag.key.as_str() == "model")
.unwrap();
assert_eq!(model.value.as_str(), "guardian-publication-parent");
}
let metric = snapshot
.scope_metrics()
.flat_map(opentelemetry_sdk::metrics::data::ScopeMetrics::metrics)
.find(|metric| metric.name() == "codex.guardian.review.retry_wait_ms")
.unwrap();
let AggregatedMetrics::F64(MetricData::Histogram(histogram)) = metric.data() else {
panic!("expected retry-wait histogram");
};
let points = histogram.data_points().collect::<Vec<_>>();
assert_eq!(points.len(), 1);
let point = points[0];
let tags = point
.attributes()
.map(|tag| (tag.key.as_str(), tag.value.as_str().to_string()))
.collect::<BTreeMap<_, _>>();
assert_eq!(tags["model"], "guardian-publication-parent");
assert_eq!(tags["failure_reason"], "parse_error");
assert_eq!(tags["outcome"], "retry");
assert_eq!(point.count(), 1);
assert!(point.sum() > 0.0);
parent_server.shutdown().await;
classifier_server.shutdown().await;
Ok(())
codex-rs/ext/guardian-reviewer/src/metrics.rs
+102 -1
View File
@@ -58,6 +58,22 @@ pub(crate) fn emit_guardian_review_metrics(
&tag_refs,
);
// Cached approvals may cancel an in-flight review: retain its attempt count,
// but use the final review status to distinguish recovery from async release.
session_telemetry.histogram_with_boundaries(
"codex.guardian.review.attempt_count",
result.attempt_count,
&[0.0, 1.0, 2.0, 3.0],
&tag_refs,
);
let mut retry_tags = tag_refs.clone();
retry_tags.push(("retried", bool_tag(result.attempt_count > 1)));
session_telemetry.counter(
"codex.guardian.review.retry_outcome",
/*inc*/ 1,
&retry_tags,
);
if let Some(time_to_first_token_ms) = result.time_to_first_token_ms {
session_telemetry.record_duration(
GUARDIAN_REVIEW_TTFT_DURATION_METRIC,
@@ -183,7 +199,7 @@ fn terminal_status_tag(status: GuardianReviewTerminalStatus) -> &'static str {
}
}
fn failure_reason_tag(reason: Option<GuardianReviewFailureReason>) -> &'static str {
pub(crate) fn failure_reason_tag(reason: Option<GuardianReviewFailureReason>) -> &'static str {
match reason {
Some(GuardianReviewFailureReason::Timeout) => "timeout",
Some(GuardianReviewFailureReason::Cancelled) => "cancelled",
@@ -407,6 +423,7 @@ mod tests {
) {
let session_telemetry = test_session_telemetry();
let result = GuardianReviewAnalyticsResult {
attempt_count: 2,
decision: GuardianReviewDecision::Approved,
terminal_status: GuardianReviewTerminalStatus::Approved,
risk_level: Some(GuardianRiskLevel::Low),
@@ -467,6 +484,29 @@ mod tests {
expected_counter_tags.insert("context_changed".to_string(), "false".to_string());
assert_eq!((attrs, value), (expected_counter_tags, 1));
let AggregatedMetrics::F64(MetricData::Histogram(histogram)) =
find_metric(&snapshot, "codex.guardian.review.attempt_count").data()
else {
panic!("expected attempt count histogram");
};
let point = histogram
.data_points()
.next()
.expect("attempt count sample");
assert_eq!(
(
point.bounds().collect::<Vec<_>>(),
point.bucket_counts().collect::<Vec<_>>()
),
(vec![0.0, 1.0, 2.0, 3.0], vec![0, 0, 1, 0, 0])
);
let mut retry_tags = expected_base_tags.clone();
retry_tags.insert("retried".to_string(), "true".to_string());
assert_eq!(
counter_point(&snapshot, "codex.guardian.review.retry_outcome"),
(retry_tags, 1)
);
assert_eq!(
histogram_sums(
&snapshot,
@@ -500,4 +540,65 @@ mod tests {
BTreeMap::from([("sample".to_string(), 123)])
);
}
#[test_case("retry", 2, 2000)]
#[test_case("cancelled", 1, 0)]
#[test_case("deadline", 1, 1000)]
#[tokio::test(start_paused = true)]
async fn retry_telemetry_records_waits_and_only_counts_resubmissions(
outcome: &str,
attempts: i64,
wait_ms: u64,
) {
let telemetry = test_session_telemetry();
let cancellation = tokio_util::sync::CancellationToken::new();
if outcome == "cancelled" {
cancellation.cancel();
}
let start = tokio::time::Instant::now();
let (_, result, _) = crate::run_with_retry(
crate::GuardianReviewSessionLimits {
max_attempts: 2,
deadline: start + Duration::from_secs(if outcome == "deadline" { 1 } else { 10 }),
},
Some(&cancellation),
Some(&telemetry),
|_| async {
(
crate::GuardianReviewOutcome::Error(crate::GuardianReviewError::Session {
message: "overloaded".into(),
error_info: Some(
codex_protocol::protocol::CodexErrorInfo::ServerOverloaded,
),
retry_at: Some(start + Duration::from_secs(2)),
}),
GuardianReviewAnalyticsResult::without_session(),
None::<()>,
)
},
)
.await;
assert_eq!(result.attempt_count, attempts);
let snapshot = telemetry.snapshot_metrics().unwrap();
let tags = BTreeMap::from([
("failure_reason".to_owned(), "session_error".to_owned()),
("outcome".to_owned(), outcome.to_owned()),
]);
assert_eq!(
histogram_sums(&snapshot, "codex.guardian.review.retry_wait_ms", &tags),
BTreeMap::from([("sample".to_owned(), wait_ms)]),
);
if outcome == "retry" {
assert_eq!(
counter_point(&snapshot, "codex.guardian.review.retry"),
(tags, 1)
);
} else {
assert!(
!snapshot
.scope_metrics()
.flat_map(opentelemetry_sdk::metrics::data::ScopeMetrics::metrics)
.any(|metric| metric.name() == "codex.guardian.review.retry")
);
}
}
}
codex-rs/ext/guardian-reviewer/src/retry.rs
+29 -3
View File
@@ -23,6 +23,7 @@ pub struct GuardianReviewSessionLimits {
pub async fn run_with_retry<Evidence, F, Attempt>(
limits: GuardianReviewSessionLimits,
external_cancel: Option<&CancellationToken>,
telemetry: Option<&codex_otel::SessionTelemetry>,
mut run_attempt: F,
) -> (
GuardianReviewOutcome,
@@ -60,7 +61,8 @@ where
}
_ => None,
};
if let Some(error) = wait_before_guardian_retry(
let wait_started = Instant::now();
let wait_error = wait_before_guardian_retry(
attempt_count,
max_attempts,
&outcome,
@@ -68,8 +70,32 @@ where
deadline,
external_cancel,
)
.await
{
.await;
if let Some(telemetry) = telemetry {
let reason = match &outcome {
GuardianReviewOutcome::Error(error) => {
crate::metrics::failure_reason_tag(Some(error.failure_reason()))
}
GuardianReviewOutcome::CachedApproval | GuardianReviewOutcome::Completed(_) => {
"none"
}
};
let wait_outcome = match &wait_error {
None => "retry",
Some(GuardianReviewError::Cancelled) => "cancelled",
Some(_) => "deadline",
};
let tags = [("failure_reason", reason), ("outcome", wait_outcome)];
telemetry.record_duration(
"codex.guardian.review.retry_wait_ms",
wait_started.elapsed(),
&tags,
);
if wait_error.is_none() {
telemetry.counter("codex.guardian.review.retry", /*inc*/ 1, &tags);
}
}
if let Some(error) = wait_error {
return (GuardianReviewOutcome::Error(error), analytics_result, None);
}
attempt_count += 1;
codex-rs/ext/guardian-reviewer/src/retry_tests.rs
+2
View File
@@ -176,6 +176,7 @@ async fn stale_authorization_uses_the_shared_attempt_budget(success_on_attempt:
deadline,
},
/*external_cancel*/ None,
/*telemetry*/ None,
|attempt_deadline| {
assert_eq!(attempt_deadline, deadline);
attempts += 1;
@@ -228,6 +229,7 @@ async fn stale_authorization_does_not_extend_the_deadline() {
deadline,
},
/*external_cancel*/ None,
/*telemetry*/ None,
|attempt_deadline| {
attempts += 1;
assert_eq!(attempt_deadline, deadline);
codex-rs/ext/guardian-reviewer/src/review.rs
+9
View File
@@ -99,6 +99,13 @@ impl<H: ReviewHost> SynchronousApprovalReviewer for ReviewRequest<'_, H> {
async_approval: Option<ExtensionFuture<'a, ()>>,
) -> ExtensionFuture<'a, Option<ReviewDecision>> {
Box::pin(async move {
if async_approval.is_none() {
self.telemetry.counter(
"codex.guardian.review.route",
/*inc*/ 1,
&[("route", "required_sync"), ("reason", reason.into())],
);
}
let deadline = Instant::now() + crate::REVIEW_TIMEOUT;
let (prepared, report) = match self
.host
@@ -119,6 +126,7 @@ impl<H: ReviewHost> SynchronousApprovalReviewer for ReviewRequest<'_, H> {
let prepared = Arc::clone(&prepared);
let cancellation = cancellation.clone();
let attempts = Arc::clone(&attempts);
let telemetry = self.telemetry.clone();
async move {
crate::run_with_retry(
GuardianReviewSessionLimits {
@@ -126,6 +134,7 @@ impl<H: ReviewHost> SynchronousApprovalReviewer for ReviewRequest<'_, H> {
deadline,
},
Some(&cancellation),
Some(&telemetry),
|deadline| {
attempts.fetch_add(/*val*/ 1, Ordering::Relaxed);
host.attempt(&prepared, deadline, &cancellation)