From 1d9d216c28c2108e01c62e3087872c50fcac2736 Mon Sep 17 00:00:00 2001 From: Hunter Moore Date: Mon, 20 Jul 2026 10:43:46 -0400 Subject: [PATCH] retry empty buffered codex completions Buffered Codex transports translated successful terminal events with no semantic output into empty HTTP 200 end_turn responses. Retry these responses instead so callers receive a usable completion or an explicit failure. Use reducer events to distinguish output text, thinking, tool calls, and web searches from terminal-only bodies. Message item framing without a text delta remains empty because the accumulator omits such blocks. Drop previous_response_id before each retry so WebSocket requests resend full context. Bound retries to ten after the initial attempt and return 503 after exhaustion. Abort continuation and pending compaction state when processing fails. Tests remove retry delays and cover buffered streaming, non-streaming, exhaustion, and empty message scaffolding. This changes empty successful completions from HTTP 200 end_turn responses to bounded retries and a 503. --- src/providers/codex/mod.rs | 194 +++++++++++++++++++++++++++++-- src/retry.rs | 12 ++ tests/smoke_cutover.rs | 229 +++++++++++++++++++++++++++++++++++++ 3 files changed, 428 insertions(+), 7 deletions(-) diff --git a/src/providers/codex/mod.rs b/src/providers/codex/mod.rs index 107628b1..1b5425a9 100644 --- a/src/providers/codex/mod.rs +++ b/src/providers/codex/mod.rs @@ -50,6 +50,8 @@ use self::translate::request::{ }; const MAX_RETRYABLE_LIVE_STREAM_RETRIES: u32 = 10; +const MAX_EMPTY_COMPLETION_RETRIES: u32 = 10; +const EMPTY_CODEX_COMPLETION_DETAIL: &str = "empty_codex_completion"; use self::translate::stream::translate_stream_bytes_with_traffic; // --------------------------------------------------------------------------- @@ -225,16 +227,44 @@ impl Provider for CodexProvider { .await; } - let upstream = match client - .post_codex(&translated, &ctx, Some(&continuation)) - .await - { - Ok(r) => r, - Err(e) => { + let mut continuation = Some(continuation); + let mut attempt = 0_u32; + let upstream = loop { + let response = match client + .post_codex(&translated, &ctx, continuation.as_ref()) + .await + { + Ok(r) => r, + Err(e) => { + abort_compaction_attempt( + ctx.session_id.as_deref(), + compact_boundary, + &translated, + ); + abort_continuation(ctx.session_id.as_deref(), turn_id); + return map_codex_error_to_response(&e); + } + }; + if !is_empty_codex_success_completion(&response.body) { + break response; + } + // A successful terminal event with no output would translate into + // an empty end_turn; retry with full context instead. + let error = empty_buffered_completion_error(); + drop_live_continuation_for_retry(&mut continuation); + if attempt >= MAX_EMPTY_COMPLETION_RETRIES { + abort_compaction_attempt(ctx.session_id.as_deref(), compact_boundary, &translated); + abort_continuation(ctx.session_id.as_deref(), turn_id); + return map_codex_error_to_response(&error); + } + let delay = compute_backoff_delay(attempt, None); + if delay.exceeds_budget { abort_compaction_attempt(ctx.session_id.as_deref(), compact_boundary, &translated); abort_continuation(ctx.session_id.as_deref(), turn_id); - return map_codex_error_to_response(&e); + return map_codex_error_to_response(&error); } + attempt += 1; + sleep(delay.wait_ms).await; }; if want_stream { @@ -841,6 +871,45 @@ where (headers, Body::from_stream(stream)).into_response() } +fn empty_buffered_completion_error() -> client::CodexError { + client::CodexError { + status: 503, + message: "Codex completed without producing output".to_string(), + detail: Some(EMPTY_CODEX_COMPLETION_DETAIL.to_string()), + retry_after: None, + origin: match config::codex_transport() { + config::CodexTransport::Http => client::CodexErrorOrigin::BufferedHttp, + _ => client::CodexErrorOrigin::BufferedWebSocket, + }, + } +} + +/// True when the buffered upstream body ended in a successful terminal event +/// without ever producing semantic output (text, thinking, tool, web search). +fn is_empty_codex_success_completion(upstream_sse: &[u8]) -> bool { + use self::translate::reducer::{ReducerEvent, TERM_COMPLETED, TERM_DONE}; + + let Ok(events) = self::translate::reducer::reduce_upstream_bytes(upstream_sse) else { + return false; + }; + let mut saw_success_terminal = false; + for event in &events { + match event { + ReducerEvent::TextDelta { text, .. } if !text.is_empty() => return false, + ReducerEvent::ThinkingStart { .. } + | ReducerEvent::ToolStart { .. } + | ReducerEvent::WebSearch { .. } => return false, + ReducerEvent::Finish { terminal_type, .. } + if terminal_type == TERM_COMPLETED || terminal_type == TERM_DONE => + { + saw_success_terminal = true; + } + _ => {} + } + } + saw_success_terminal +} + fn is_codex_terminal_event(payload: &serde_json::Value) -> bool { matches!( payload.get("type").and_then(|v| v.as_str()), @@ -955,6 +1024,9 @@ fn map_codex_error_to_response(err: &client::CodexError) -> Response { if is_context_window_overflow(message) { return map_codex_failure_to_response(message); } + if err.detail.as_deref() == Some(EMPTY_CODEX_COMPLETION_DETAIL) { + return json_error(StatusCode::SERVICE_UNAVAILABLE, "api_error", &err.message); + } match err.status { 401 => json_error( @@ -1127,6 +1199,114 @@ fn format_auth_saved_output(auth_path: &str, account_id: Option<&str>) -> String mod tests { use super::*; + fn upstream_sse(events: &[serde_json::Value]) -> Vec { + let mut bytes = Vec::new(); + for event in events { + bytes.extend_from_slice(format!("data: {event}\n\n").as_bytes()); + } + bytes + } + + #[test] + fn terminal_only_completed_upstream_is_empty_completion() { + let body = upstream_sse(&[serde_json::json!({ + "type": "response.completed", + "response": {"id": "resp_1", "status": "completed", "incomplete_details": null, "usage": {"input_tokens": 5, "output_tokens": 0}} + })]); + assert!(is_empty_codex_success_completion(&body)); + } + + #[test] + fn terminal_only_done_upstream_is_empty_completion() { + let body = upstream_sse(&[serde_json::json!({ + "type": "response.done", + "response": {"id": "resp_1", "usage": {}} + })]); + assert!(is_empty_codex_success_completion(&body)); + } + + #[test] + fn empty_message_item_is_empty_completion() { + let body = upstream_sse(&[ + serde_json::json!({ + "type": "response.output_item.added", + "output_index": 0, + "item": {"type": "message", "id": "msg_1"} + }), + serde_json::json!({ + "type": "response.output_item.done", + "output_index": 0, + "item": {"type": "message"} + }), + serde_json::json!({ + "type": "response.completed", + "response": {"id": "resp_1", "usage": {}} + }), + ]); + assert!(is_empty_codex_success_completion(&body)); + } + + #[test] + fn upstream_with_text_is_not_empty_completion() { + let body = upstream_sse(&[ + serde_json::json!({ + "type": "response.output_item.added", + "output_index": 0, + "item": {"type": "message", "id": "msg_1"} + }), + serde_json::json!({ + "type": "response.output_text.delta", + "output_index": 0, + "delta": "hello" + }), + serde_json::json!({ + "type": "response.output_item.done", + "output_index": 0, + "item": {"type": "message"} + }), + serde_json::json!({ + "type": "response.completed", + "response": {"id": "resp_1", "usage": {}} + }), + ]); + assert!(!is_empty_codex_success_completion(&body)); + } + + #[test] + fn upstream_with_tool_call_is_not_empty_completion() { + let body = upstream_sse(&[ + serde_json::json!({ + "type": "response.output_item.added", + "output_index": 0, + "item": {"type": "function_call", "call_id": "call_1", "name": "Read", "arguments": ""} + }), + serde_json::json!({ + "type": "response.output_item.done", + "output_index": 0, + "item": {"type": "function_call", "call_id": "call_1", "name": "Read", "arguments": "{}"} + }), + serde_json::json!({ + "type": "response.completed", + "response": {"id": "resp_1", "usage": {}} + }), + ]); + assert!(!is_empty_codex_success_completion(&body)); + } + + #[test] + fn terminal_only_incomplete_upstream_is_not_empty_completion() { + let body = upstream_sse(&[serde_json::json!({ + "type": "response.incomplete", + "response": {"id": "resp_1", "incomplete_details": {"reason": "max_output_tokens"}, "usage": {}} + })]); + assert!(!is_empty_codex_success_completion(&body)); + } + + #[test] + fn upstream_without_terminal_event_is_not_empty_completion() { + assert!(!is_empty_codex_success_completion(&upstream_sse(&[]))); + } + fn request_with_tools(tools: serde_json::Value) -> MessagesRequest { serde_json::from_value(serde_json::json!({ "model": "gpt-5.6-luna", diff --git a/src/retry.rs b/src/retry.rs index 48b61964..70603683 100644 --- a/src/retry.rs +++ b/src/retry.rs @@ -1,3 +1,4 @@ +use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; pub const RETRY_INITIAL_DELAY_MS: u64 = 2000; @@ -39,7 +40,18 @@ pub fn compute_backoff_delay(attempt: u32, retry_after: Option<&str>) -> Backoff } } +static ZERO_RETRY_DELAY_FOR_TESTS: AtomicBool = AtomicBool::new(false); + +/// Make retry sleeps return immediately so exhaustion paths can be exercised +/// in tests without waiting out the real backoff schedule. +pub fn set_zero_retry_delay_for_tests(enabled: bool) { + ZERO_RETRY_DELAY_FOR_TESTS.store(enabled, Ordering::SeqCst); +} + pub async fn sleep(ms: u64) { + if ZERO_RETRY_DELAY_FOR_TESTS.load(Ordering::SeqCst) { + return; + } tokio::time::sleep(Duration::from_millis(ms)).await; } diff --git a/tests/smoke_cutover.rs b/tests/smoke_cutover.rs index d0a6aeb9..2055b751 100644 --- a/tests/smoke_cutover.rs +++ b/tests/smoke_cutover.rs @@ -693,6 +693,235 @@ async fn smoke_codex_http_messages_uses_mock_upstream() { assert_eq!(sent["stream"], true); } +/// Resets the retry-delay override even when the test panics, so later tests +/// in this process keep real backoff behavior. +struct ZeroRetryDelayGuard; + +impl ZeroRetryDelayGuard { + fn enable() -> Self { + claude_code_proxy::retry::set_zero_retry_delay_for_tests(true); + ZeroRetryDelayGuard + } +} + +impl Drop for ZeroRetryDelayGuard { + fn drop(&mut self) { + claude_code_proxy::retry::set_zero_retry_delay_for_tests(false); + } +} + +fn empty_completion_sse() -> Vec { + concat!( + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_empty\",", + "\"status\":\"completed\",\"incomplete_details\":null,", + "\"usage\":{\"input_tokens\":5,\"output_tokens\":0}}}\n\n" + ) + .as_bytes() + .to_vec() +} + +fn empty_message_completion_sse() -> Vec { + concat!( + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,", + "\"item\":{\"type\":\"message\",\"id\":\"msg_empty\"}}\n\n", + "data: {\"type\":\"response.output_item.done\",\"output_index\":0,", + "\"item\":{\"type\":\"message\"}}\n\n", + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_empty\",", + "\"status\":\"completed\",\"incomplete_details\":null,", + "\"usage\":{\"input_tokens\":5,\"output_tokens\":0}}}\n\n" + ) + .as_bytes() + .to_vec() +} + +fn buffered_success_sse(text: &str) -> Vec { + format!( + concat!( + "data: {{\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{{\"type\":\"message\",\"id\":\"msg_up\"}}}}\n\n", + "data: {{\"type\":\"response.output_text.delta\",\"output_index\":0,\"delta\":\"{text}\"}}\n\n", + "data: {{\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{{\"type\":\"message\"}}}}\n\n", + "data: {{\"type\":\"response.completed\",\"response\":{{\"id\":\"resp_1\",\"usage\":{{\"input_tokens\":5,\"output_tokens\":2}}}}}}\n\n" + ), + text = text + ) + .into_bytes() +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_retries_empty_completion() { + let _guard = env_lock(); + let _delay_guard = ZeroRetryDelayGuard::enable(); + let config = TempDir::new().unwrap(); + write_auth(config.path(), "codex"); + + let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let upstream = spawn_http_upstream({ + let attempts = attempts.clone(); + move |_body: Value| { + let attempt = attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + if attempt == 0 { + empty_completion_sse() + } else { + buffered_success_sse("buffered retry ok") + } + } + }) + .await; + + let _config_env = EnvGuard::set("CCP_CONFIG_DIR", config.path()); + let _base_url_env = EnvGuard::set("CCP_CODEX_BASE_URL", &upstream); + let _transport_env = EnvGuard::set("CCP_CODEX_TRANSPORT", "http"); + + let response = call_messages("gpt-5.5").await; + let status = response.status(); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let body_text = String::from_utf8_lossy(&body); + + assert_eq!(status, StatusCode::OK, "body: {body_text}"); + let value: Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(value["content"][0]["text"], "buffered retry ok"); + assert_eq!( + attempts.load(std::sync::atomic::Ordering::SeqCst), + 2, + "empty completion must trigger one retry" + ); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_retries_empty_message_completion() { + let _guard = env_lock(); + let _delay_guard = ZeroRetryDelayGuard::enable(); + let config = TempDir::new().unwrap(); + write_auth(config.path(), "codex"); + + let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let upstream = spawn_http_upstream({ + let attempts = attempts.clone(); + move |_body: Value| { + let attempt = attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + if attempt == 0 { + empty_message_completion_sse() + } else { + buffered_success_sse("empty message retry ok") + } + } + }) + .await; + + let _config_env = EnvGuard::set("CCP_CONFIG_DIR", config.path()); + let _base_url_env = EnvGuard::set("CCP_CODEX_BASE_URL", &upstream); + let _transport_env = EnvGuard::set("CCP_CODEX_TRANSPORT", "http"); + + let response = call_messages("gpt-5.5").await; + let status = response.status(); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let body_text = String::from_utf8_lossy(&body); + + assert_eq!(status, StatusCode::OK, "body: {body_text}"); + let value: Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(value["content"][0]["text"], "empty message retry ok"); + assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 2); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_stream_retries_empty_completion() { + let _guard = env_lock(); + let _delay_guard = ZeroRetryDelayGuard::enable(); + let config = TempDir::new().unwrap(); + write_auth(config.path(), "codex"); + + let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let upstream = spawn_http_upstream({ + let attempts = attempts.clone(); + move |_body: Value| { + let attempt = attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + if attempt == 0 { + empty_completion_sse() + } else { + buffered_success_sse("buffered stream retry ok") + } + } + }) + .await; + + let _config_env = EnvGuard::set("CCP_CONFIG_DIR", config.path()); + let _base_url_env = EnvGuard::set("CCP_CODEX_BASE_URL", &upstream); + let _transport_env = EnvGuard::set("CCP_CODEX_TRANSPORT", "http"); + + let response = call_messages_body(json!({ + "model": "gpt-5.5", + "max_tokens": 64, + "stream": true, + "messages": [{"role":"user","content":"one"}] + })) + .await; + let status = response.status(); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let body_text = String::from_utf8_lossy(&body); + + assert_eq!(status, StatusCode::OK, "body: {body_text}"); + assert!( + body_text.contains("buffered stream retry ok"), + "expected retried text in SSE body: {body_text}" + ); + assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 2); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_empty_completions_exhaust_to_service_unavailable() { + let _guard = env_lock(); + let _delay_guard = ZeroRetryDelayGuard::enable(); + let config = TempDir::new().unwrap(); + write_auth(config.path(), "codex"); + + let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let upstream = spawn_http_upstream({ + let attempts = attempts.clone(); + move |_body: Value| { + attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + empty_completion_sse() + } + }) + .await; + + let _config_env = EnvGuard::set("CCP_CONFIG_DIR", config.path()); + let _base_url_env = EnvGuard::set("CCP_CODEX_BASE_URL", &upstream); + let _transport_env = EnvGuard::set("CCP_CODEX_TRANSPORT", "http"); + + let response = call_messages("gpt-5.5").await; + let status = response.status(); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let body_text = String::from_utf8_lossy(&body); + + assert_eq!( + status, + StatusCode::SERVICE_UNAVAILABLE, + "exhausted empty completions must surface an explicit error: {body_text}" + ); + assert!( + body_text.contains("Codex completed without producing output"), + "unexpected exhaustion body: {body_text}" + ); + // Initial attempt plus MAX_EMPTY_COMPLETION_RETRIES retries. + assert_eq!( + attempts.load(std::sync::atomic::Ordering::SeqCst), + 11, + "retry loop must stay bounded" + ); +} + #[allow(clippy::await_holding_lock)] #[tokio::test] async fn smoke_codex_http_server_compaction_replays_native_history() {