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

fix(core): retry model capacity errors

Browse Source
This commit is contained in:
Peter Steinberger
2026-07-09 20:28:57 -07:00
parent 1f0566d3f5
commit 3d1218206d
11 changed files with 400 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/compact_remote_v2.rs
+44 -18
View File
@@ -26,6 +26,7 @@ use codex_analytics::CompactionImplementation;
use codex_analytics::CompactionPhase;
use codex_analytics::CompactionReason;
use codex_analytics::CompactionTrigger;
use codex_async_utils::OrCancelExt;
use codex_protocol::error::CodexErr;
use codex_protocol::error::Result as CodexResult;
use codex_protocol::items::ContextCompactionItem;
@@ -42,6 +43,7 @@ use codex_rollout_trace::InferenceTraceContext;
use codex_utils_output_truncation::approx_token_count;
use codex_utils_output_truncation::truncate_text;
use futures::StreamExt;
use tokio_util::sync::CancellationToken;
#[path = "compact_remote_v2_attempt.rs"]
mod attempt;
@@ -63,6 +65,7 @@ pub(crate) async fn run_inline_remote_auto_compact_task(
initial_context_injection: InitialContextInjection,
reason: CompactionReason,
phase: CompactionPhase,
cancellation_token: CancellationToken,
) -> CodexResult<()> {
let compaction_metadata = CompactionTurnMetadata::new(
CompactionTrigger::Auto,
@@ -77,6 +80,7 @@ pub(crate) async fn run_inline_remote_auto_compact_task(
Some(client_session),
initial_context_injection,
compaction_metadata,
cancellation_token,
)
.await
}
@@ -84,6 +88,7 @@ pub(crate) async fn run_inline_remote_auto_compact_task(
pub(crate) async fn run_remote_compact_task(
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
cancellation_token: CancellationToken,
) -> CodexResult<()> {
// Standalone compaction is its own request boundary, so it captures a fresh step.
let step_context = sess.capture_step_context(Arc::clone(&turn_context)).await;
@@ -109,10 +114,12 @@ pub(crate) async fn run_remote_compact_task(
/*client_session*/ None,
InitialContextInjection::DoNotInject,
compaction_metadata,
cancellation_token,
)
.await
}
#[allow(clippy::too_many_arguments)]
async fn run_remote_compact_task_inner(
sess: &Arc<Session>,
step_context: &Arc<StepContext>,
@@ -120,6 +127,7 @@ async fn run_remote_compact_task_inner(
client_session: Option<&mut ModelClientSession>,
initial_context_injection: InitialContextInjection,
compaction_metadata: CompactionTurnMetadata,
cancellation_token: CancellationToken,
) -> CodexResult<()> {
let turn_context = &step_context.turn;
let trigger = compaction_metadata.trigger();
@@ -163,6 +171,7 @@ async fn run_remote_compact_task_inner(
initial_context_injection,
compaction_metadata,
&mut analytics_details,
&cancellation_token,
)
.await;
let status = compaction_status_from_result(&result);
@@ -201,6 +210,7 @@ async fn run_remote_compact_task_inner_impl(
initial_context_injection: InitialContextInjection,
compaction_metadata: CompactionTurnMetadata,
analytics_details: &mut CompactionAnalyticsDetails,
cancellation_token: &CancellationToken,
) -> CodexResult<()> {
let turn_context = &step_context.turn;
let context_compaction_item = ContextCompactionItem::new();
@@ -222,6 +232,7 @@ async fn run_remote_compact_task_inner_impl(
&compaction_trace,
compaction_metadata,
analytics_details,
cancellation_token,
)
.await;
let (attempt, compaction_turn_context) = match attempt {
@@ -248,6 +259,7 @@ async fn run_remote_compact_task_inner_impl(
&fallback_compaction_trace,
compaction_metadata,
analytics_details,
cancellation_token,
)
.await;
record_model_fallback(
@@ -333,6 +345,7 @@ async fn run_remote_compaction_request_v2(
client_session: &mut ModelClientSession,
prompt: &Prompt,
responses_metadata: &CodexResponsesMetadata,
cancellation_token: &CancellationToken,
) -> CodexResult<RemoteCompactionV2Output> {
let max_retries = turn_context
.provider
@@ -340,36 +353,49 @@ async fn run_remote_compaction_request_v2(
.stream_max_retries()
.min(MAX_REMOTE_COMPACTION_V2_STREAM_RETRIES);
let mut retries = 0;
let mut server_overloaded_retries = 0;
loop {
let result = match client_session
.stream(
prompt,
&turn_context.model_info,
&turn_context.session_telemetry,
turn_context.reasoning_effort.clone(),
turn_context.reasoning_summary,
turn_context.config.service_tier.clone(),
responses_metadata,
&InferenceTraceContext::disabled(),
)
.await
{
Ok(stream) => collect_compaction_output(stream).await,
Err(err) => Err(err),
};
let result = async {
let stream = client_session
.stream(
prompt,
&turn_context.model_info,
&turn_context.session_telemetry,
turn_context.reasoning_effort.clone(),
turn_context.reasoning_summary,
turn_context.config.service_tier.clone(),
responses_metadata,
&InferenceTraceContext::disabled(),
)
.await?;
collect_compaction_output(stream).await
}
.or_cancel(cancellation_token)
.await
.map_err(|_| CodexErr::TurnAborted)?;
match result {
Ok(compaction_output) => return Ok(compaction_output),
Err(err) if !err.is_retryable() => return Err(err),
Err(err) => {
let should_retry_server_overload = matches!(&err, CodexErr::ServerOverloaded)
&& !crate::guardian::is_guardian_reviewer_source(&turn_context.session_source);
if !err.is_retryable() && !should_retry_server_overload {
return Err(err);
}
let retry_counter = if should_retry_server_overload {
&mut server_overloaded_retries
} else {
&mut retries
};
handle_retryable_response_stream_error(
&mut retries,
retry_counter,
max_retries,
err,
client_session,
sess,
turn_context,
ResponsesStreamRequest::RemoteCompactionV2,
cancellation_token.child_token(),
)
.await?;
}
codex-rs/core/src/compact_remote_v2_attempt.rs
+2
View File
@@ -34,6 +34,7 @@ pub(super) async fn run_remote_compact_v2_attempt(
compaction_trace: &CompactionTraceContext,
compaction_metadata: CompactionTurnMetadata,
analytics_details: &mut CompactionAnalyticsDetails,
cancellation_token: &CancellationToken,
) -> CodexResult<RemoteCompactV2Attempt> {
let turn_context = &step_context.turn;
let mut history = sess.clone_history().await;
@@ -105,6 +106,7 @@ pub(super) async fn run_remote_compact_v2_attempt(
client_session,
&prompt,
&responses_metadata,
cancellation_token,
)
.await;
trace_attempt.record_result(
codex-rs/core/src/responses_retry.rs
+28 -4
View File
@@ -6,11 +6,20 @@ use crate::client::ModelClientSession;
use crate::session::session::Session;
use crate::session::turn_context::TurnContext;
use crate::util::backoff;
use crate::util::jitter;
use codex_protocol::error::CodexErr;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::WarningEvent;
use tokio_util::sync::CancellationToken;
use tracing::warn;
const SERVER_OVERLOADED_MAX_RETRIES: u64 = 3;
const SERVER_OVERLOADED_RETRY_DELAYS: [Duration; 3] = [
Duration::from_secs(30),
Duration::from_secs(120),
Duration::from_secs(300),
];
#[derive(Debug, Clone, Copy)]
pub(crate) enum ResponsesStreamRequest {
Sampling,
@@ -19,6 +28,7 @@ pub(crate) enum ResponsesStreamRequest {
/// Handles a retryable stream error and returns `Ok(())` when the caller should
/// retry the request loop.
#[allow(clippy::too_many_arguments)]
pub(crate) async fn handle_retryable_response_stream_error(
retries: &mut u64,
max_retries: u64,
@@ -27,8 +37,16 @@ pub(crate) async fn handle_retryable_response_stream_error(
sess: &Session,
turn_context: &TurnContext,
request: ResponsesStreamRequest,
cancellation_token: CancellationToken,
) -> Result<(), CodexErr> {
if *retries >= max_retries
let is_server_overloaded = matches!(&err, CodexErr::ServerOverloaded);
let max_retries = if is_server_overloaded {
SERVER_OVERLOADED_MAX_RETRIES
} else {
max_retries
};
if !is_server_overloaded
&& *retries >= max_retries
&& client_session.try_switch_fallback_transport(
&turn_context.session_telemetry,
&turn_context.model_info,
@@ -49,6 +67,9 @@ pub(crate) async fn handle_retryable_response_stream_error(
*retries += 1;
let retry_count = *retries;
let delay = match &err {
CodexErr::ServerOverloaded => {
jitter(SERVER_OVERLOADED_RETRY_DELAYS[retry_count.saturating_sub(1) as usize])
}
CodexErr::Stream(_, requested_delay) => {
requested_delay.unwrap_or_else(|| backoff(retry_count))
}
@@ -58,7 +79,8 @@ pub(crate) async fn handle_retryable_response_stream_error(
// In release builds, hide the first websocket retry notification to reduce noisy
// transient reconnect messages. In debug builds, keep full visibility for diagnosis.
let report_error = retry_count > 1
let report_error = is_server_overloaded
|| retry_count > 1
|| cfg!(debug_assertions)
|| !sess.services.model_client.responses_websocket_enabled();
if report_error {
@@ -71,8 +93,10 @@ pub(crate) async fn handle_retryable_response_stream_error(
)
.await;
}
tokio::time::sleep(delay).await;
return Ok(());
return tokio::select! {
() = tokio::time::sleep(delay) => Ok(()),
() = cancellation_token.cancelled() => Err(CodexErr::TurnAborted),
};
}
Err(err)
codex-rs/core/src/session/mod.rs
+5 -2
View File
@@ -3853,8 +3853,11 @@ impl Session {
codex_error: CodexErr,
) {
let additional_details = codex_error.to_string();
let codex_error_info = CodexErrorInfo::ResponseStreamDisconnected {
http_status_code: codex_error.http_status_code_value(),
let codex_error_info = match &codex_error {
CodexErr::ServerOverloaded => CodexErrorInfo::ServerOverloaded,
_ => CodexErrorInfo::ResponseStreamDisconnected {
http_status_code: codex_error.http_status_code_value(),
},
};
let event = EventMsg::StreamError(StreamErrorEvent {
message: message.into(),
codex-rs/core/src/session/turn.rs
+29 -4
View File
@@ -153,7 +153,14 @@ pub(crate) async fn run_turn(
// new user message are recorded. Estimate pending incoming items (context
// diffs/full reinjection + user input) and trigger compaction preemptively
// when they would push the thread over the compaction threshold.
if let Err(err) = run_pre_sampling_compact(&sess, &turn_context, &mut client_session).await {
if let Err(err) = run_pre_sampling_compact(
&sess,
&turn_context,
&mut client_session,
&cancellation_token,
)
.await
{
if matches!(err, CodexErr::TurnAborted) {
return Err(err);
}
@@ -354,6 +361,7 @@ pub(crate) async fn run_turn(
InitialContextInjection::BeforeLastUserMessage(Arc::clone(&world_state)),
CompactionReason::ContextLimit,
CompactionPhase::MidTurn,
&cancellation_token,
)
.await
{
@@ -799,8 +807,10 @@ async fn run_pre_sampling_compact(
sess: &Arc<Session>,
turn_context: &Arc<TurnContext>,
client_session: &mut ModelClientSession,
cancellation_token: &CancellationToken,
) -> CodexResult<()> {
maybe_run_previous_model_inline_compact(sess, turn_context, client_session).await?;
maybe_run_previous_model_inline_compact(sess, turn_context, client_session, cancellation_token)
.await?;
let token_status =
super::context_window::context_window_token_status(sess.as_ref(), turn_context.as_ref())
.await;
@@ -816,6 +826,7 @@ async fn run_pre_sampling_compact(
InitialContextInjection::DoNotInject,
CompactionReason::ContextLimit,
CompactionPhase::PreTurn,
cancellation_token,
)
.await?;
}
@@ -860,6 +871,7 @@ async fn maybe_run_previous_model_inline_compact(
sess: &Arc<Session>,
turn_context: &Arc<TurnContext>,
client_session: &mut ModelClientSession,
cancellation_token: &CancellationToken,
) -> CodexResult<()> {
let Some(previous_turn_settings) = sess.previous_turn_settings().await else {
return Ok(());
@@ -893,6 +905,7 @@ async fn maybe_run_previous_model_inline_compact(
InitialContextInjection::DoNotInject,
CompactionReason::CompHashChanged,
CompactionPhase::PreTurn,
cancellation_token,
)
.await?;
return Ok(());
@@ -940,6 +953,7 @@ async fn maybe_run_previous_model_inline_compact(
InitialContextInjection::DoNotInject,
CompactionReason::ModelDownshift,
CompactionPhase::PreTurn,
cancellation_token,
)
.await?;
}
@@ -959,6 +973,7 @@ async fn run_auto_compact(
initial_context_injection: InitialContextInjection,
reason: CompactionReason,
phase: CompactionPhase,
cancellation_token: &CancellationToken,
) -> CodexResult<()> {
let turn_context = &step_context.turn;
if turn_context.config.features.enabled(Feature::TokenBudget) {
@@ -992,6 +1007,7 @@ async fn run_auto_compact(
initial_context_injection,
reason,
phase,
cancellation_token.child_token(),
)
.await?;
return Ok(());
@@ -1138,6 +1154,7 @@ async fn run_sampling_request(
);
let max_retries = turn_context.provider.info().stream_max_retries();
let mut retries = 0;
let mut server_overloaded_retries = 0;
let mut initial_input = Some(input);
let mut original_input = None;
loop {
@@ -1188,18 +1205,26 @@ async fn run_sampling_request(
original_input = Some(prompt.input);
}
if !err.is_retryable() {
let should_retry_server_overload = matches!(&err, CodexErr::ServerOverloaded)
&& !crate::guardian::is_guardian_reviewer_source(&turn_context.session_source);
if !err.is_retryable() && !should_retry_server_overload {
return Err(err);
}
let retry_counter = if should_retry_server_overload {
&mut server_overloaded_retries
} else {
&mut retries
};
handle_retryable_response_stream_error(
&mut retries,
retry_counter,
max_retries,
err,
client_session,
&sess,
&turn_context,
ResponsesStreamRequest::Sampling,
cancellation_token.child_token(),
)
.await?;
turn_context.turn_timing_state.record_sampling_retry();
codex-rs/core/src/tasks/compact.rs
+7 -2
View File
@@ -29,7 +29,7 @@ impl SessionTask for CompactTask {
session: Arc<SessionTaskContext>,
ctx: Arc<TurnContext>,
_input: Vec<TurnInput>,
_cancellation_token: CancellationToken,
cancellation_token: CancellationToken,
) -> SessionTaskResult {
let session = session.clone_session();
if ctx.config.features.enabled(Feature::TokenBudget) {
@@ -48,7 +48,12 @@ impl SessionTask for CompactTask {
"remote_v2",
/*manual*/ true,
);
crate::compact_remote_v2::run_remote_compact_task(session.clone(), ctx).await
crate::compact_remote_v2::run_remote_compact_task(
session.clone(),
ctx,
cancellation_token,
)
.await
} else {
emit_compact_metric(
&session.services.session_telemetry,
codex-rs/core/src/util.rs
+5 -1
View File
@@ -85,8 +85,12 @@ pub(crate) fn emit_feedback_auth_recovery_tags(
pub fn backoff(attempt: u64) -> Duration {
let exp = BACKOFF_FACTOR.powi(attempt.saturating_sub(1) as i32);
let base = (INITIAL_DELAY_MS as f64 * exp) as u64;
jitter(Duration::from_millis(base))
}
pub(crate) fn jitter(delay: Duration) -> Duration {
let jitter = rand::rng().random_range(0.9..1.1);
Duration::from_millis((base as f64 * jitter) as u64)
delay.mul_f64(jitter)
}
pub(crate) fn error_or_panic(message: impl std::string::ToString) {
codex-rs/core/tests/common/test_codex.rs
+8 -1
View File
@@ -303,6 +303,7 @@ pub struct TestCodexBuilder {
supports_openai_form_elicitation: bool,
external_time_provider: Option<Arc<dyn TimeProvider>>,
code_mode_host_program: Option<PathBuf>,
session_source: Option<SessionSource>,
}
impl TestCodexBuilder {
@@ -326,6 +327,11 @@ impl TestCodexBuilder {
})
}
pub fn with_session_source(mut self, session_source: SessionSource) -> Self {
self.session_source = Some(session_source);
self
}
pub fn with_model_info_override<T>(self, model: &str, override_model_info: T) -> Self
where
T: FnOnce(&mut ModelInfo) + Send + 'static,
@@ -693,7 +699,7 @@ impl TestCodexBuilder {
allow_provider_model_fallback: false,
initial_history: InitialHistory::New,
history_mode: None,
session_source: None,
session_source: self.session_source.clone(),
thread_source: None,
dynamic_tools: Vec::new(),
metrics_service_name: None,
@@ -1255,6 +1261,7 @@ pub fn test_codex() -> TestCodexBuilder {
supports_openai_form_elicitation: false,
external_time_provider: None,
code_mode_host_program: None,
session_source: None,
}
}
codex-rs/core/tests/suite/client.rs
+94
View File
@@ -42,6 +42,7 @@ use codex_protocol::models::ReasoningItemReasoningSummary;
use codex_protocol::models::ResponseItem;
use codex_protocol::models::WebSearchAction;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::protocol::CodexErrorInfo;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::RolloutItem;
@@ -81,6 +82,7 @@ use serde_json::json;
use std::io::Write;
use std::num::NonZeroU64;
use std::sync::Arc;
use std::time::Duration;
use tempfile::TempDir;
use toml::toml;
use uuid::Uuid;
@@ -3521,6 +3523,98 @@ async fn incomplete_response_emits_content_filter_error_message() -> anyhow::Res
Ok(())
}
#[tokio::test]
async fn server_overload_retries_use_a_separate_budget_on_the_same_turn() -> anyhow::Result<()> {
skip_if_no_network!(Ok(()));
let server = MockServer::start().await;
let responses = mount_sse_sequence(
&server,
vec![
sse_failed(
"resp_stream_error",
"server_error",
"temporary stream failure",
),
sse_failed(
"resp_overloaded_1",
"server_is_overloaded",
"selected model is at capacity",
),
sse_failed(
"resp_overloaded_2",
"server_is_overloaded",
"selected model is at capacity",
),
sse_failed(
"resp_overloaded_3",
"server_is_overloaded",
"selected model is at capacity",
),
sse(vec![
ev_response_created("resp_retried"),
ev_completed("resp_retried"),
]),
],
)
.await;
let test = test_codex()
.with_config(|config| {
config.model_provider.request_max_retries = Some(0);
config.model_provider.stream_max_retries = Some(5);
})
.build(&server)
.await?;
tokio::time::pause();
test.codex
.submit(Op::UserInput {
items: vec![UserInput::Text {
text: "keep working".into(),
text_elements: Vec::new(),
}],
final_output_json_schema: None,
responsesapi_client_metadata: None,
additional_context: Default::default(),
thread_settings: Default::default(),
})
.await?;
let mut server_overload_retry_messages = Vec::new();
loop {
let event = test.codex.next_event().await?.msg;
match event {
EventMsg::StreamError(event) => {
if event.codex_error_info == Some(CodexErrorInfo::ServerOverloaded) {
server_overload_retry_messages.push(event.message);
}
tokio::time::advance(Duration::from_secs(360)).await;
}
EventMsg::TurnComplete(_) => break,
_ => {}
}
}
let requests = responses.requests();
assert_eq!(
server_overload_retry_messages,
vec![
"Reconnecting... 1/3",
"Reconnecting... 2/3",
"Reconnecting... 3/3",
]
);
assert_eq!(requests.len(), 5);
let turn_id = requests[0].body_json()["client_metadata"]["turn_id"].clone();
assert!(
requests
.iter()
.all(|request| request.body_json()["client_metadata"]["turn_id"] == turn_id)
);
Ok(())
}
/// We try to avoid setting env vars in tests because std::env::set_var() is
/// process-wide and unsafe. Though for this test, we want to simulate the
/// presence of an environment variable that the provider will read for auth, so
codex-rs/core/tests/suite/compact_remote.rs
+177 -6
View File
@@ -16,6 +16,7 @@ use codex_protocol::dynamic_tools::DynamicToolSpec;
use codex_protocol::items::TurnItem;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::CodexErrorInfo;
use codex_protocol::protocol::ConversationStartParams;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::ItemCompletedEvent;
@@ -26,6 +27,8 @@ use codex_protocol::protocol::RealtimeEvent;
use codex_protocol::protocol::RealtimeOutputModality;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::RolloutLine;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::user_input::UserInput;
use core_test_support::PathBufExt;
use core_test_support::apps_test_server::configure_search_capable_model;
@@ -1018,8 +1021,8 @@ async fn remote_compact_v2_reuses_compaction_trigger_for_followups() -> Result<(
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn remote_compact_v2_retries_failures_with_stream_retry_budget() -> Result<()> {
#[tokio::test]
async fn remote_compact_v2_retries_capacity_after_stream_retry_budget() -> Result<()> {
skip_if_no_network!(Ok(()));
let harness = TestCodexHarness::with_builder(
@@ -1049,6 +1052,21 @@ async fn remote_compact_v2_retries_failures_with_stream_retry_budget() -> Result
"encrypted_content": "FAILED_COMPACT_SUMMARY",
}
})])),
responses::sse_response(responses::sse_failed(
"resp-overloaded-1",
"server_is_overloaded",
"selected model is at capacity",
)),
responses::sse_response(responses::sse_failed(
"resp-overloaded-2",
"server_is_overloaded",
"selected model is at capacity",
)),
responses::sse_response(responses::sse_failed(
"resp-overloaded-3",
"server_is_overloaded",
"selected model is at capacity",
)),
responses::sse_response(responses::sse(vec![
serde_json::json!({
"type": "response.output_item.done",
@@ -1080,9 +1098,32 @@ async fn remote_compact_v2_retries_failures_with_stream_retry_budget() -> Result
})
.await?;
wait_for_turn_complete(&codex).await;
tokio::time::pause();
codex.submit(Op::Compact).await?;
wait_for_turn_complete(&codex).await;
let mut server_overload_retry_messages = Vec::new();
loop {
match codex.next_event().await?.msg {
EventMsg::StreamError(event) => {
if event.codex_error_info == Some(CodexErrorInfo::ServerOverloaded) {
server_overload_retry_messages.push(event.message);
}
tokio::time::advance(Duration::from_secs(360)).await;
}
EventMsg::TurnComplete(_) => break,
_ => {}
}
}
assert_eq!(
server_overload_retry_messages,
vec![
"Reconnecting... 1/3",
"Reconnecting... 2/3",
"Reconnecting... 3/3",
]
);
tokio::time::resume();
codex
.submit(Op::UserInput {
@@ -1100,12 +1141,12 @@ async fn remote_compact_v2_retries_failures_with_stream_retry_budget() -> Result
let response_requests = responses_mock.requests();
assert_eq!(
5,
8,
response_requests.len(),
"expected initial turn, failed open, failed stream, compact retry, and follow-up turn"
"expected initial turn, two stream failures, three capacity failures, compact retry, and follow-up turn"
);
for compact_request in &response_requests[1..=3] {
for compact_request in &response_requests[1..=6] {
assert_eq!("/v1/responses", compact_request.path());
assert!(
compact_request
@@ -1130,6 +1171,136 @@ async fn remote_compact_v2_retries_failures_with_stream_retry_budget() -> Result
Ok(())
}
#[tokio::test]
async fn remote_compact_v2_leaves_guardian_capacity_retries_to_guardian() -> Result<()> {
skip_if_no_network!(Ok(()));
let harness = TestCodexHarness::with_builder(
test_codex()
.with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
.with_session_source(SessionSource::SubAgent(SubAgentSource::Other(
"guardian".to_string(),
)))
.with_config(|config| {
let _ = config.features.enable(Feature::RemoteCompactionV2);
config.model_provider.request_max_retries = Some(0);
config.model_provider.stream_max_retries = Some(5);
}),
)
.await?;
let codex = harness.test().codex.clone();
let responses_mock = responses::mount_sse_sequence(
harness.server(),
vec![
responses::sse(vec![
responses::ev_assistant_message("m1", "FIRST_REMOTE_REPLY"),
responses::ev_completed("resp-1"),
]),
responses::sse_failed(
"resp-overloaded",
"server_is_overloaded",
"temporary reviewer overload",
),
],
)
.await;
codex
.submit(Op::UserInput {
items: vec![UserInput::Text {
text: "hello guardian compact".into(),
text_elements: Vec::new(),
}],
final_output_json_schema: None,
responsesapi_client_metadata: None,
additional_context: Default::default(),
thread_settings: Default::default(),
})
.await?;
wait_for_turn_complete(&codex).await;
tokio::time::pause();
codex.submit(Op::Compact).await?;
let mut terminal_error_info = None;
loop {
match codex.next_event().await?.msg {
EventMsg::Error(event) => terminal_error_info = event.codex_error_info,
EventMsg::StreamError(_) => tokio::time::advance(Duration::from_secs(360)).await,
EventMsg::TurnComplete(_) => break,
_ => {}
}
}
tokio::time::resume();
assert_eq!(terminal_error_info, Some(CodexErrorInfo::ServerOverloaded));
assert_eq!(responses_mock.requests().len(), 2);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn remote_compact_v2_interrupts_pending_request() -> Result<()> {
skip_if_no_network!(Ok(()));
let harness = TestCodexHarness::with_builder(
test_codex()
.with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
.with_config(|config| {
let _ = config.features.enable(Feature::RemoteCompactionV2);
}),
)
.await?;
let codex = harness.test().codex.clone();
let responses_mock = responses::mount_response_sequence(
harness.server(),
vec![
responses::sse_response(responses::sse(vec![
responses::ev_assistant_message("m1", "FIRST_REMOTE_REPLY"),
responses::ev_completed("resp-1"),
])),
responses::sse_response(responses::sse(vec![
serde_json::json!({
"type": "response.output_item.done",
"item": {
"type": "compaction",
"encrypted_content": "DELAYED_COMPACT_SUMMARY",
}
}),
responses::ev_completed("resp-compact"),
]))
.set_delay(Duration::from_secs(30)),
],
)
.await;
codex
.submit(Op::UserInput {
items: vec![UserInput::Text {
text: "hello interruptible compact".into(),
text_elements: Vec::new(),
}],
final_output_json_schema: None,
responsesapi_client_metadata: None,
additional_context: Default::default(),
thread_settings: Default::default(),
})
.await?;
wait_for_turn_complete(&codex).await;
codex.submit(Op::Compact).await?;
tokio::time::timeout(Duration::from_secs(2), async {
while responses_mock.requests().len() < 2 {
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await?;
codex.submit(Op::Interrupt).await?;
wait_for_event(&codex, |event| matches!(event, EventMsg::TurnAborted(_))).await;
assert_eq!(responses_mock.requests().len(), 2);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn remote_compact_v2_accepts_additional_output_items_before_compaction() -> Result<()> {
skip_if_no_network!(Ok(()));
codex-rs/protocol/src/error.rs
+1 -1
View File
@@ -114,7 +114,7 @@ pub enum CodexErr {
InvalidImageRequest(),
#[error("{0}")]
UsageLimitReached(UsageLimitReachedError),
#[error("Selected model is at capacity. Please try a different model.")]
#[error("Selected model is at capacity. Please try again later.")]
ServerOverloaded,
#[error("{message}")]
CyberPolicy { message: String },