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

Honor server retry advice across Responses retries and fallback (#49441)

Browse Source 
## Why

Responses overload and exhausted-request errors could terminate despite server retry advice, and WebSocket-to-HTTP fallback could issue a request before the advised deadline.

## What changed

- Allow `ServerOverloaded` and `RetryLimit` errors to retry when server advice is present, within the configured retry budgets.
- Preserve `Retry-After` headers from rejected WebSocket upgrades and wait for the server deadline before falling back to HTTP.
- Keep quota, usage-limit, and policy failures terminal even when they include retry advice.

## Testing

Add coverage for HTTP overload and rate-limit recovery, WebSocket upgrade rejection and fallback timing, retry deadline preservation through error mapping, and retry limits for sampling and manual and automatic compaction.

GitOrigin-RevId: 72d30a9f078652664754dfa971fd7dc7a78f20e8
This commit is contained in:
Adam Perry @ OpenAI
2026-09-30 02:12:42 +00:00
committed by copyberry
parent fc81a7154b
commit 6ba4bf9e64
6 changed files with 434 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/codex-api/src/api_bridge_tests.rs
+50
View File
@@ -74,6 +74,56 @@ async fn map_api_error_preserves_retry_delay() {
}
}
/// Mapping a new error category preserves elapsed advice without making hard failures retryable.
#[tokio::test(start_paused = true)]
async fn http_retry_deadline_survives_mapping_and_respects_terminal_errors() {
use std::time::Duration;
let advice = RetryAfter::from_delay(Duration::from_secs(10)).expect("retry advice");
tokio::time::advance(Duration::from_secs(4)).await;
for (status, code, retryable) in [
(503, "server_is_overloaded", true),
(429, "rate_limit_exceeded", true),
(429, "insufficient_quota", false),
(429, "usage_limit_reached", false),
(400, "cyber_policy", false),
] {
let error = map_api_error(ApiError::Transport(TransportError::Http {
status: http::StatusCode::from_u16(status).unwrap(),
url: None,
headers: None,
body: Some(serde_json::json!({"error": {"type": code, "code": code, "message": "sensitive upstream detail"}}).to_string()),
retry_after: Some(advice),
}));
assert_eq!(
(error.retry_after(), error.server_retry_delay()),
(Some(advice), Some(Duration::from_secs(6))),
"{code}",
);
assert_eq!(
error.retry_delay(/*retry_count*/ 1),
retryable.then_some(Duration::from_secs(6)),
"{code}",
);
if status == 429 && retryable {
assert_eq!(
(
error.to_codex_protocol_error(),
error.http_status_code_value(),
error.to_string()
),
(
CodexErrorInfo::ResponseTooManyFailedAttempts {
http_status_code: Some(429),
},
Some(429),
"exceeded retry limit, last status: 429 Too Many Requests".to_string(),
),
);
}
}
}
#[test]
fn map_api_error_distinguishes_capacity_from_slow_down() {
for (code, expected, retryable) in [
codex-rs/codex-api/src/endpoint/responses_websocket.rs
+3 -1
View File
@@ -14,6 +14,7 @@ use crate::sse::process_responses_event;
use crate::telemetry::WebsocketTelemetry;
use codex_client::TransportError;
use codex_http_client::HttpClientFactory;
use codex_http_client::RetryAfter;
use codex_websocket_client::WebSocketConnection;
use codex_websocket_client::WebSocketConnector;
use futures::FutureExt;
@@ -570,6 +571,7 @@ fn map_ws_error(err: WsError, url: &Url) -> ApiError {
WsError::Http(response) => {
let status = response.status();
let headers = response.headers().clone();
let retry_after = RetryAfter::from_headers(&headers);
let body = response
.body()
.as_ref()
@@ -579,7 +581,7 @@ fn map_ws_error(err: WsError, url: &Url) -> ApiError {
url: Some(url.to_string()),
headers: Some(headers),
body,
retry_after: None,
retry_after,
})
}
WsError::ConnectionClosed | WsError::AlreadyClosed => {
codex-rs/core/src/responses_retry.rs
+5 -1
View File
@@ -1,5 +1,6 @@
//! Shared retry and transport fallback decisions for Responses requests.
//! Content-filter guidance is recorded for sampling requests before retry decisions.
//! Server advice controls timing without extending configured retry limits.
use std::time::Duration;
@@ -116,13 +117,16 @@ pub(crate) async fn handle_response_stream_error(
return Ok(());
}
// TODO(anp): Respect server retry advice before issuing the fallback HTTP request.
if retry_state.retries >= max_retries
&& client_session.try_switch_fallback_transport(
&turn_context.session_telemetry,
turn_context.model_info(),
)
{
// Changing transport must not bypass the server's retry deadline.
if let Some(retry_after) = retry_after {
tokio::time::sleep_until(retry_after.deadline()).await;
}
sess.send_event(
turn_context,
EventMsg::Warning(WarningEvent {
codex-rs/core/tests/suite/compact.rs
+188
View File
@@ -23,6 +23,7 @@ use codex_protocol::openai_models::ModelInfo;
use codex_protocol::openai_models::ModelServiceTier;
use codex_protocol::openai_models::ModelsResponse;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::CodexErrorInfo;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::HookEventName;
use codex_protocol::protocol::HookRunStatus;
@@ -3074,6 +3075,193 @@ async fn pre_sampling_compact_falls_back_after_previous_model_stream_retries_are
);
}
/// Both compaction models keep the configured retry budget when the server supplies advice,
/// including the selected model's final attempt.
#[test_case::test_case(2; "selected_model_recovers")]
#[test_case::test_case(3; "selected_model_exhausts_retries")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn pre_sampling_compact_advised_errors_fall_back_then_retry_on_selected_model(
selected_model_failures: usize,
) {
skip_if_no_network!();
let server = MockServer::start().await;
let previous_model = "gpt-5.4";
let selected_model = "gpt-5.2";
let _models_mock = mount_models_once(
&server,
ModelsResponse {
models: vec![
model_info_with_context_window(previous_model, /*context_window*/ 273_000),
model_info_with_context_window(selected_model, /*context_window*/ 125_000),
],
},
)
.await;
let overload = wiremock::ResponseTemplate::new(/*s*/ 503)
.insert_header("Retry-After", "0")
.set_body_json(json!({ "error": { "code": "server_is_overloaded" } }));
let mut responses = vec![
sse_response(sse(vec![
ev_assistant_message("m1", "before switch"),
ev_completed_with_tokens("r1", /*total_tokens*/ 120_000),
])),
overload.clone(), // Old model: initial attempt and two retries.
overload.clone(),
overload.clone(),
];
responses.extend(vec![overload; selected_model_failures]);
if selected_model_failures < 3 {
responses.extend([
sse_response(remote_v2_compaction_response()),
sse_response(sse(vec![
ev_assistant_message("m3", "after switch"),
ev_completed_with_tokens("r3", /*total_tokens*/ 100),
])),
]);
}
let request_log = mount_response_sequence(&server, responses).await;
let mut model_provider = openai_model_provider(&server);
model_provider.request_max_retries = Some(0);
model_provider.stream_max_retries = Some(2);
let test = test_codex()
.with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
.with_model(previous_model)
.with_config(move |config| {
config.model_provider = model_provider;
set_test_compact_prompt(config);
})
.build_with_auto_env(&server)
.await
.expect("build codex");
test.codex
.start_or_steer_turn(disabled_permission_user_turn(
"before switch",
test.cwd.path().to_path_buf(),
previous_model.to_string(),
))
.await
.expect("submit first turn");
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
test.codex
.start_or_steer_turn(disabled_permission_user_turn(
"after switch",
test.cwd.path().to_path_buf(),
selected_model.to_string(),
))
.await
.expect("submit selected-model turn");
if selected_model_failures < 3 {
assert_compaction_uses_turn_lifecycle_id(&test.codex).await;
} else {
let mut errors = Vec::new();
loop {
match wait_for_event(&test.codex, |_| true).await {
EventMsg::Error(error) => errors.push(error.codex_error_info),
EventMsg::TurnComplete(event) => {
assert_eq!(
event.error.and_then(|error| error.codex_error_info),
Some(CodexErrorInfo::ServerOverloaded)
);
break;
}
_ => {}
}
}
assert_eq!(errors, vec![Some(CodexErrorInfo::ServerOverloaded)]);
}
let actual_models = request_log
.requests()
.iter()
.map(|request| {
request.body_json()["model"]
.as_str()
.expect("request model")
.to_string()
})
.collect::<Vec<_>>();
let mut expected_models = vec![
previous_model, // First turn.
previous_model, // Old-model compaction attempt and retries.
previous_model,
previous_model,
selected_model, // Selected-model compaction attempt and retries.
selected_model,
selected_model,
];
if selected_model_failures < 3 {
expected_models.push(selected_model); // Sampling after compaction succeeds.
}
assert_eq!(actual_models, expected_models);
}
/// Manual compaction keeps the configured retry limit even when the server supplies advice.
#[test_case::test_case(0; "no_retries")]
#[test_case::test_case(2; "two_retries")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn manual_remote_compact_advice_respects_retry_limit(max_retries: u64) {
skip_if_no_network!();
let server = start_mock_server().await;
let overload = wiremock::ResponseTemplate::new(/*s*/ 503)
.insert_header("Retry-After", "0")
.set_body_json(json!({ "error": { "code": "server_is_overloaded" } }));
let request_log =
mount_response_sequence(&server, vec![overload; (max_retries + 1) as usize]).await;
let mut model_provider = openai_model_provider(&server);
model_provider.request_max_retries = Some(0);
model_provider.stream_max_retries = Some(max_retries);
let test = test_codex()
.with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
.with_model("gpt-5.4")
.with_config(move |config| {
config.model_provider = model_provider;
set_test_compact_prompt(config);
})
.build_with_auto_env(&server)
.await
.expect("build codex");
test.codex.submit(Op::Compact).await.expect("run /compact");
let mut errors = Vec::new();
let mut reconnect_messages = Vec::new();
loop {
match wait_for_event(&test.codex, |_| true).await {
EventMsg::Error(error) => errors.push(error.codex_error_info),
EventMsg::StreamError(error) => reconnect_messages.push(error.message),
EventMsg::TurnComplete(event) => {
assert_eq!(
event.error.and_then(|error| error.codex_error_info),
Some(CodexErrorInfo::ServerOverloaded)
);
break;
}
_ => {}
}
}
assert_eq!(errors, vec![Some(CodexErrorInfo::ServerOverloaded)]);
assert_eq!(
reconnect_messages,
(1..=max_retries)
.map(|retry| format!("Reconnecting... {retry}/{max_retries}"))
.collect::<Vec<_>>()
);
let models = request_log
.requests()
.iter()
.map(|request| {
request.body_json()["model"]
.as_str()
.expect("request model")
.to_string()
})
.collect::<Vec<_>>();
assert_eq!(models, vec!["gpt-5.4"; (max_retries + 1) as usize]);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn pre_sampling_compact_keeps_unknown_previous_model_for_api_key_auth_and_custom_provider() {
skip_if_no_network!();
codex-rs/core/tests/suite/retry_after.rs
+185 -22
View File
@@ -51,11 +51,11 @@ const SECOND_RETRY_MIN_DELAY: Duration = Duration::from_millis(360);
const SECOND_RETRY_MAX_DELAY: Duration = Duration::from_millis(440);
#[derive(Debug, PartialEq, Eq)]
struct RetryTelemetryEvent {
pub(super) struct RetryTelemetryEvent {
attempt: u64,
delay: Duration,
layer: String,
operation: String,
pub(super) delay: Duration,
pub(super) layer: String,
pub(super) operation: String,
}
#[derive(Default)]
@@ -193,7 +193,7 @@ where
}
}
struct RetryTelemetryCapture {
pub(super) struct RetryTelemetryCapture {
events: mpsc::UnboundedReceiver<RetryTelemetryEvent>,
resumptions: mpsc::UnboundedReceiver<Duration>,
_subscriber: DefaultGuard,
@@ -201,7 +201,7 @@ struct RetryTelemetryCapture {
}
impl RetryTelemetryCapture {
fn install() -> Self {
pub(super) fn install() -> Self {
let (sender, events) = mpsc::unbounded_channel();
let (resumptions_sender, resumptions) = mpsc::unbounded_channel();
// Avoid caching no interest when another test first reaches a shared callsite.
@@ -225,7 +225,7 @@ impl RetryTelemetryCapture {
}
}
async fn next_retry(&mut self) -> RetryTelemetryEvent {
pub(super) async fn next_retry(&mut self) -> RetryTelemetryEvent {
tokio::time::timeout(Duration::from_secs(10), self.events.recv())
.await
.expect("timed out waiting for retry telemetry")
@@ -275,7 +275,7 @@ async fn wait_for_turn_completion(test: &TestCodex) {
assert_eq!(completed.error, None, "turn should complete successfully");
}
/// HTTP overloads use the upstream header before another HTTP request.
/// HTTP overloads use server advice within the request and stream retry budgets.
#[tokio::test(flavor = "current_thread")]
async fn responses_http_uses_retry_after() -> Result<()> {
skip_if_no_network!(Ok(()));
@@ -285,6 +285,9 @@ async fn responses_http_uses_retry_after() -> Result<()> {
let response_mock = responses::mount_response_sequence(
&server,
vec![
ResponseTemplate::new(503)
.insert_header("Retry-After", "1")
.set_body_json(json!({ "error": { "code": "server_is_overloaded" } })),
ResponseTemplate::new(503)
.insert_header("Retry-After", "1")
.set_body_json(json!({ "error": { "code": "server_is_overloaded" } })),
@@ -298,7 +301,7 @@ async fn responses_http_uses_retry_after() -> Result<()> {
let test = test_codex()
.with_config(|config| {
config.model_provider.request_max_retries = Some(1);
config.model_provider.stream_max_retries = Some(0);
config.model_provider.stream_max_retries = Some(1);
})
.build_with_auto_env(&server)
.await?;
@@ -316,8 +319,79 @@ async fn responses_http_uses_retry_after() -> Result<()> {
}
);
assert!(wait_for_retry(&mut telemetry, &retry).await >= Duration::from_secs(1));
let retry = telemetry.next_retry().await;
assert!(retry.delay <= Duration::from_secs(1));
assert_eq!(
retry,
RetryTelemetryEvent {
attempt: 1,
delay: retry.delay,
layer: "stream".into(),
operation: "sampling".into(),
}
);
assert!(wait_for_retry(&mut telemetry, &retry).await >= Duration::from_secs(1));
wait_for_turn_completion(&test).await;
assert_eq!(response_mock.requests().len(), 3);
assert_eq!(
telemetry.events.try_recv(),
Err(mpsc::error::TryRecvError::Empty)
);
Ok(())
}
/// A 429 with server advice retries in the sampling loop and completes without a terminal error.
#[tokio::test(flavor = "current_thread")]
async fn responses_http_429_uses_retry_after() -> Result<()> {
skip_if_no_network!(Ok(()));
let mut telemetry = RetryTelemetryCapture::install();
let server = responses::start_mock_server().await;
let response_mock = responses::mount_response_sequence(
&server,
vec![
ResponseTemplate::new(429)
.insert_header("Retry-After", "1")
.set_body_json(json!({ "error": { "code": "rate_limit_exceeded" } })),
responses::sse_response(responses::sse(vec![
responses::ev_response_created("recovered"),
responses::ev_completed("recovered"),
])),
],
)
.await;
let test = test_codex()
.with_config(|config| {
config.model_provider.request_max_retries = Some(0);
config.model_provider.stream_max_retries = Some(1);
})
.build_with_auto_env(&server)
.await?;
submit_user_input(&test, "retry the rate-limited request").await?;
let retry = telemetry.next_retry().await;
assert!(retry.delay <= Duration::from_secs(1));
assert_eq!(
retry,
RetryTelemetryEvent {
attempt: 1,
delay: retry.delay,
layer: "stream".into(),
operation: "sampling".into(),
}
);
assert!(wait_for_retry(&mut telemetry, &retry).await >= Duration::from_secs(1));
loop {
match wait_for_event(&test.codex, |_| true).await {
EventMsg::Error(error) => panic!("unexpected terminal error: {error:?}"),
EventMsg::TurnComplete(completed) => {
assert_eq!(completed.error, None);
break;
}
_ => {}
}
}
assert_eq!(response_mock.requests().len(), 2);
assert_eq!(
telemetry.events.try_recv(),
@@ -425,31 +499,40 @@ async fn exhausted_http_retries_preserve_deadline_through_error_mapping() {
assert_eq!(outer_error.server_retry_delay(), Some(Duration::ZERO));
}
/// Headerless HTTP overloads currently exhaust request retries before emitting one terminal error.
/// Retry advice changes delays but never extends either retry budget or its displayed count.
#[test_case::test_case(None, 2, 3; "without_advice")]
#[test_case::test_case(Some("0"), 0, 3; "advice_without_stream_retries")]
#[test_case::test_case(Some("0"), 2, 9; "advice_with_stream_retries")]
#[tokio::test(flavor = "current_thread")]
async fn responses_http_overload_without_retry_after_exhausts_request_retries() -> Result<()> {
async fn responses_http_overload_respects_retry_limits(
retry_after: Option<&str>,
stream_max_retries: u64,
expected_requests: usize,
) -> Result<()> {
skip_if_no_network!(Ok(()));
let server = responses::start_mock_server().await;
let mut overload = ResponseTemplate::new(503)
.set_body_json(json!({ "error": { "code": "server_is_overloaded" } }));
if let Some(retry_after) = retry_after {
overload = overload.insert_header("Retry-After", retry_after);
}
Mock::given(method("POST"))
.and(path("/v1/responses"))
.respond_with(
ResponseTemplate::new(503)
.set_body_json(json!({ "error": { "code": "server_is_overloaded" } })),
)
.respond_with(overload)
.mount(&server)
.await;
let test = test_codex()
.with_config(|config| {
.with_config(move |config| {
config.model_provider.request_max_retries = Some(2);
config.model_provider.stream_max_retries = Some(2);
config.model_provider.stream_max_retries = Some(stream_max_retries);
})
.build_with_auto_env(&server)
.await?;
submit_user_input(&test, "reject the disabled model").await?;
let mut error_events = 0;
let mut stream_error_events = 0;
let mut reconnect_messages = Vec::new();
tokio::time::timeout(Duration::from_secs(10), async {
loop {
match wait_for_event(&test.codex, |_| true).await {
@@ -464,7 +547,7 @@ async fn responses_http_overload_without_retry_after_exhausts_request_retries()
"Selected model is at capacity. Please try a different model."
);
}
EventMsg::StreamError(_) => stream_error_events += 1,
EventMsg::StreamError(error) => reconnect_messages.push(error.message),
EventMsg::TurnComplete(event) => {
assert_eq!(
event.error.and_then(|error| error.codex_error_info),
@@ -480,7 +563,14 @@ async fn responses_http_overload_without_retry_after_exhausts_request_retries()
.expect("overload retries should finish the turn within 10 seconds");
assert_eq!(error_events, 1);
assert_eq!(stream_error_events, 0);
let expected_messages = if retry_after.is_some() {
(1..=stream_max_retries)
.map(|retry| format!("Reconnecting... {retry}/{stream_max_retries}"))
.collect::<Vec<_>>()
} else {
Vec::new()
};
assert_eq!(reconnect_messages, expected_messages);
let request_count = server
.received_requests()
.await
@@ -489,8 +579,8 @@ async fn responses_http_overload_without_retry_after_exhausts_request_retries()
.filter(|request| request.url.path() == "/v1/responses")
.count();
assert_eq!(
request_count, 3,
"headerless overload should exhaust the configured request retries"
request_count, expected_requests,
"overload should exhaust the configured retry budgets"
);
Ok(())
@@ -1709,6 +1799,79 @@ async fn websocket_connection_limit_retries_with_local_backoff() -> Result<()> {
Ok(())
}
/// Advised WebSocket failures reach HTTP after the existing retry budget. HTTP
/// recovery uses its own retry budget and honors the preceding advice deadlines.
#[tokio::test(flavor = "current_thread")]
async fn websocket_upgrade_rejection_uses_retry_after() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = responses::start_mock_server().await;
let prewarm_rejection = Mock::given(method("GET"))
.and(path("/v1/responses"))
.respond_with(ResponseTemplate::new(/*s*/ 503))
.up_to_n_times(/*n*/ 1)
.expect(/*r*/ 1)
.mount_as_scoped(&server)
.await;
Mock::given(method("GET"))
.and(path("/v1/responses"))
.respond_with(
ResponseTemplate::new(/*s*/ 503)
.insert_header("Retry-After", "1")
.set_body_json(json!({ "error": { "code": "server_is_overloaded" } })),
)
.mount(&server)
.await;
let http_overload = ResponseTemplate::new(/*s*/ 503)
.insert_header("Retry-After", "0")
.set_body_json(json!({ "error": { "code": "server_is_overloaded" } }));
let response_mock = responses::mount_response_sequence(
&server,
vec![
http_overload,
ResponseTemplate::new(/*s*/ 200)
.insert_header("content-type", "text/event-stream")
.set_body_string(responses::sse(vec![
responses::ev_response_created("recovered"),
responses::ev_completed("recovered"),
])),
],
)
.await;
let test = test_codex()
.with_config(|config| {
config.model_catalog = Some(bundled_models_response().expect("bundled models"));
config.model_provider.supports_websockets = true;
config.model_provider.request_max_retries = Some(0);
config.model_provider.stream_max_retries = Some(1);
})
.build_with_auto_env(&server)
.await?;
// Prewarm is best effort; start the advised failures in the sampling retry loop.
tokio::time::timeout(
Duration::from_secs(10),
prewarm_rejection.wait_until_satisfied(),
)
.await?;
drop(prewarm_rejection);
let start = Instant::now();
submit_user_input(&test, "retry the rejected websocket upgrade").await?;
wait_for_turn_completion(&test).await;
// Both WebSocket advice deadlines apply, including the one before transport fallback.
assert!(start.elapsed() >= Duration::from_secs(2));
let requests = server.received_requests().await.unwrap_or_default();
let methods = requests
.iter()
.filter(|request| request.url.path() == "/v1/responses")
.map(|request| request.method.as_str())
.collect::<Vec<_>>();
assert_eq!(methods, vec!["GET", "GET", "GET", "POST", "POST"]);
assert_eq!(response_mock.requests().len(), 2);
Ok(())
}
// TODO(anp) respect Retry-After
/// Nested websocket retry headers are currently ignored, leaving rate-limit errors terminal.
#[tokio::test(flavor = "current_thread")]
codex-rs/protocol/src/error.rs
+3 -2
View File
@@ -403,18 +403,19 @@ impl CodexErr {
| CodexErrorDetails::UnsupportedOperation(_)
| CodexErrorDetails::Sandbox(_)
| CodexErrorDetails::LandlockSandboxExecutableNotProvided
| CodexErrorDetails::RetryLimit(_)
| CodexErrorDetails::ContextWindowExceeded
| CodexErrorDetails::ThreadNotFound(_)
| CodexErrorDetails::AgentLimitReached { .. }
| CodexErrorDetails::Spawn
| CodexErrorDetails::SessionConfiguredNotFirstEvent
| CodexErrorDetails::UsageLimitReached(_)
| CodexErrorDetails::ServerOverloaded
| CodexErrorDetails::FlexUnavailable
| CodexErrorDetails::CyberPolicy { .. }
| CodexErrorDetails::BioPolicy { .. }
| CodexErrorDetails::MisalignmentPolicyViolation { .. } => None,
CodexErrorDetails::ServerOverloaded | CodexErrorDetails::RetryLimit(_) => {
self.server_retry_delay()
}
CodexErrorDetails::Stream(..)
| CodexErrorDetails::ContentFilter
| CodexErrorDetails::RateLimitExceeded(_)