fix(codex): retry empty buffered completions

The buffered transport path (CCP_CODEX_TRANSPORT=http, and non-stream
requests generally) translated an upstream body that ended in a
successful terminal event with no semantic output into an empty 200
end_turn. Detect that shape via the reducer (text, thinking, tool, and
web-search events all count as semantic), retry with previous_response_id
dropped so the resend carries full context, and surface an explicit 503
after the bounded retries exhaust — mirroring the live-stream semantics
of #70. response.incomplete stays non-retryable.

Includes a zero-delay retry hook (same *_for_tests pattern as the
websocket pool/continuation helpers) so the exhaustion smoke test runs
instantly.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: HM-Silvey

src/providers/codex/mod.rs

@@ -41,6 +41,8 @@ use self::translate::reducer::finish_metadata_from_upstream;
 use self::translate::request::{TranslateOptions, has_hosted_web_search, 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;
 
 // ---------------------------------------------------------------------------
@@ -155,15 +157,37 @@ 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_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_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_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 {
@@ -662,6 +686,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::TextStart { .. }
+            | 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()),
@@ -769,6 +832,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 | 403 => json_error(
@@ -940,6 +1006,93 @@ 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<u8> {
+        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 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",

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;
 }