From fa8197b2ddd0a058245f8fa79febf05169371cf2 Mon Sep 17 00:00:00 2001 From: Hunter Moore Date: Mon, 20 Jul 2026 07:18:41 -0400 Subject: [PATCH 1/4] fix(codex): retry empty live completions --- src/providers/codex/mod.rs | 54 ++++++- src/providers/codex/translate/live_stream.rs | 85 ++++++++++- tests/smoke_cutover.rs | 151 +++++++++++++++++++ 3 files changed, 280 insertions(+), 10 deletions(-) diff --git a/src/providers/codex/mod.rs b/src/providers/codex/mod.rs index 13f7125b..7a2650c7 100644 --- a/src/providers/codex/mod.rs +++ b/src/providers/codex/mod.rs @@ -567,6 +567,8 @@ async fn live_stream_response_once( ) -> LiveStreamStart { let mut translator = LiveStreamTranslator::new(message_id, model.to_string()); let mut upstream_sse_body = Vec::new(); + // Keep protocol framing private until real output makes a transparent retry unsafe. + let mut pending_chunk = Vec::new(); let mut generation_started = false; while let Some(item) = upstream_events.recv().await { @@ -638,8 +640,17 @@ async fn live_stream_response_once( return LiveStreamStart::Response(map_codex_failure_to_response(&message)); } }; - if !chunk.is_empty() { - record_live_stream_progress(&ctx, &chunk); + pending_chunk.extend_from_slice(&chunk); + if terminal + && is_codex_success_terminal_event(&payload) + && !translator.has_semantic_output() + { + return LiveStreamStart::Retry { + error: empty_live_completion_error(), + }; + } + if translator.has_semantic_output() && !pending_chunk.is_empty() { + record_live_stream_progress(&ctx, &pending_chunk); if terminal { update_continuation_from_upstream( ctx.session_id.as_deref(), @@ -648,12 +659,12 @@ async fn live_stream_response_once( &upstream_sse_body, compact_boundary, ); - return LiveStreamStart::Response(single_live_stream_response(chunk)); + return LiveStreamStart::Response(single_live_stream_response(pending_chunk)); } return LiveStreamStart::Response(remaining_live_stream_response( upstream_events, translator, - chunk, + pending_chunk, ctx, turn_id, request_body, @@ -669,7 +680,11 @@ async fn live_stream_response_once( &upstream_sse_body, compact_boundary, ); - return LiveStreamStart::Response(empty_live_stream_response()); + if pending_chunk.is_empty() { + return LiveStreamStart::Response(empty_live_stream_response()); + } + record_live_stream_progress(&ctx, &pending_chunk); + return LiveStreamStart::Response(single_live_stream_response(pending_chunk)); } } @@ -684,6 +699,16 @@ async fn live_stream_response_once( } } +fn empty_live_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: client::CodexErrorOrigin::WebSocket, + } +} + fn codex_generation_event(payload: &serde_json::Value) -> bool { !matches!( payload.get("type").and_then(|value| value.as_str()), @@ -926,6 +951,13 @@ fn is_codex_terminal_event(payload: &serde_json::Value) -> bool { ) } +fn is_codex_success_terminal_event(payload: &serde_json::Value) -> bool { + matches!( + payload.get("type").and_then(|v| v.as_str()), + Some("response.completed") | Some("response.done") + ) +} + fn retryable_live_start_codex_error(err: &client::CodexError) -> bool { if err.origin == client::CodexErrorOrigin::WebSocketHandshake { return err.status == 0 || matches!(err.status, 429 | 500 | 502 | 503 | 504 | 529); @@ -1481,6 +1513,18 @@ mod tests { ); } + #[tokio::test] + async fn empty_live_completion_maps_to_explicit_service_unavailable() { + let err = empty_live_completion_error(); + + assert_eq!(err.status, 503); + assert_eq!(err.detail.as_deref(), Some(EMPTY_CODEX_COMPLETION_DETAIL)); + assert_eq!( + map_codex_error_to_response(&err).status(), + StatusCode::SERVICE_UNAVAILABLE + ); + } + #[test] fn live_start_statusless_websocket_handshake_error_is_retryable() { let err = client::CodexError { diff --git a/src/providers/codex/translate/live_stream.rs b/src/providers/codex/translate/live_stream.rs index 3ae5d341..25016c57 100644 --- a/src/providers/codex/translate/live_stream.rs +++ b/src/providers/codex/translate/live_stream.rs @@ -63,6 +63,7 @@ pub struct LiveStreamTranslator { web_searches: Vec, web_search_results: Vec, deferred_text: Vec<(usize, String)>, + semantic_output_started: bool, finished: bool, } @@ -82,6 +83,7 @@ impl LiveStreamTranslator { web_searches: Vec::new(), web_search_results: Vec::new(), deferred_text: Vec::new(), + semantic_output_started: false, finished: false, } } @@ -163,6 +165,10 @@ impl LiveStreamTranslator { self.finished } + pub fn has_semantic_output(&self) -> bool { + self.semantic_output_started + } + pub fn finish_after_closed_completed_tool_call( &mut self, traffic: Option<&TrafficCapture>, @@ -305,6 +311,7 @@ impl LiveStreamTranslator { "function_call" => { self.close_thinking(traffic, out); self.saw_tool_use = true; + self.semantic_output_started = true; let index = self.anthropic_index; self.anthropic_index += 1; let call_id = item @@ -364,6 +371,7 @@ impl LiveStreamTranslator { if delta.is_empty() { return; } + self.semantic_output_started = true; if self.thinking.map(|thinking| thinking.output_index) != Some(output_index) { self.close_thinking(traffic, out); let index = self.anthropic_index; @@ -411,6 +419,7 @@ impl LiveStreamTranslator { if delta.is_empty() { return; } + self.semantic_output_started = true; let output_index = payload .get("output_index") @@ -626,6 +635,7 @@ impl LiveStreamTranslator { == Some("web_search_call") { self.close_thinking(traffic, out); + self.semantic_output_started = true; let item = &payload["item"]; let index = self.anthropic_index; self.anthropic_index += 1; @@ -1188,6 +1198,7 @@ mod tests { assert!(out.contains("text_delta")); assert!(out.contains("hello")); assert!(!out.contains("message_stop")); + assert!(translator.has_semantic_output()); } #[test] @@ -1220,13 +1231,77 @@ mod tests { } #[test] - fn completed_response_with_null_incomplete_details_is_end_turn() { - let out = render(vec![json!({ - "type": "response.completed", - "response": {"id": "resp_1", "status": "completed", "incomplete_details": null, "usage": {}} - })]); + fn terminal_only_completion_remains_non_semantic() { + let mut translator = LiveStreamTranslator::new("msg_1", "gpt-5.5"); + let out = translator + .accept( + &json!({ + "type": "response.completed", + "response": {"id": "resp_1", "status": "completed", "incomplete_details": null, "usage": {}} + }), + None, + ) + .unwrap(); + let out = String::from_utf8(out).unwrap(); assert!(out.contains(r#""stop_reason":"end_turn""#)); assert!(!out.contains(r#""stop_reason":"max_tokens""#)); + assert!(!translator.has_semantic_output()); + } + + #[test] + fn tool_thinking_and_web_search_events_are_semantic() { + let mut tool = LiveStreamTranslator::new("msg_tool", "gpt-5.5"); + tool.accept( + &json!({ + "type": "response.output_item.added", + "output_index": 0, + "item": {"type": "function_call", "call_id": "call_1", "name": "Read"} + }), + None, + ) + .unwrap(); + assert!(tool.has_semantic_output()); + + let mut thinking = LiveStreamTranslator::new("msg_thinking", "gpt-5.5"); + thinking + .accept( + &json!({ + "type": "response.reasoning_summary_text.delta", + "output_index": 0, + "delta": "plan" + }), + None, + ) + .unwrap(); + assert!(thinking.has_semantic_output()); + + let mut web_search = LiveStreamTranslator::new("msg_search", "gpt-5.5"); + web_search + .accept( + &json!({ + "type": "response.output_item.added", + "output_index": 0, + "item": {"type": "web_search_call", "id": "ws_1"} + }), + None, + ) + .unwrap(); + assert!(!web_search.has_semantic_output()); + web_search + .accept( + &json!({ + "type": "response.output_item.done", + "output_index": 0, + "item": { + "type": "web_search_call", + "id": "ws_1", + "action": {"query": "claude-code-proxy"} + } + }), + None, + ) + .unwrap(); + assert!(web_search.has_semantic_output()); } #[test] diff --git a/tests/smoke_cutover.rs b/tests/smoke_cutover.rs index df700b2a..5a8735c0 100644 --- a/tests/smoke_cutover.rs +++ b/tests/smoke_cutover.rs @@ -538,6 +538,92 @@ async fn spawn_websocket_close_then_retry_upstream(captured: Arc>>, +) -> String { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let addr_str = format!("http://{addr}"); + + tokio::spawn(async move { + let mut handled = 0usize; + while handled < 3 { + let Ok((stream, _)) = listener.accept().await else { + return; + }; + let Ok(ws) = tokio_tungstenite::accept_async(stream).await else { + return; + }; + let (mut sender, mut receiver) = ws.split(); + + while handled < 3 { + let Some(text) = (loop { + match receiver.next().await { + Some(Ok(Message::Text(text))) => break Some(text), + Some(Ok(Message::Ping(data))) => { + let _ = sender.send(Message::Pong(data)).await; + } + Some(Ok(Message::Pong(_))) => {} + Some(Ok(_)) => {} + Some(Err(_)) | None => break None, + } + }) else { + break; + }; + if let Ok(json) = serde_json::from_str::(&text) { + let _ = captured.lock().map(|mut g| g.push(json)); + } + + if handled == 1 { + let event = json!({ + "type": "response.completed", + "response": { + "id": "resp_empty", + "status": "completed", + "incomplete_details": null, + "usage": {"input_tokens": 5, "output_tokens": 0} + } + }); + let _ = sender.send(Message::Text(event.to_string())).await; + handled += 1; + continue; + } + + let response_text = if handled == 0 { "first" } else { "retry" }; + let response_id = if handled == 0 { "resp_1" } else { "resp_retry" }; + let events = [ + json!({ + "type":"response.output_item.added", + "output_index":0, + "item":{"type":"message","id":format!("msg_empty_{handled}")} + }), + json!({ + "type":"response.output_text.delta", + "output_index":0, + "delta":response_text + }), + json!({ + "type":"response.output_item.done", + "output_index":0, + "item":{"type":"message"} + }), + json!({ + "type":"response.completed", + "response":{"id":response_id,"usage":{"input_tokens":5,"output_tokens":2}} + }), + ]; + + for event in &events { + let _ = sender.send(Message::Text(event.to_string())).await; + } + handled += 1; + } + } + }); + + addr_str +} + // --------------------------------------------------------------------------- // Health and routing smoke tests (no env var mutation needed) // --------------------------------------------------------------------------- @@ -1736,6 +1822,71 @@ async fn smoke_codex_websocket_stream_retries_empty_close_with_full_context() { clear_codex_websocket_pool_for_tests(); } +#[allow(clippy::await_holding_lock)] +#[tokio::test(flavor = "multi_thread")] +async fn smoke_codex_websocket_stream_retries_terminal_only_completion_with_full_context() { + let _guard = env_lock(); + let config = TempDir::new().unwrap(); + write_auth(config.path(), "codex"); + clear_codex_websocket_pool_for_tests(); + clear_all_continuations_for_tests(); + + let captured = Arc::new(Mutex::new(Vec::new())); + let upstream = spawn_websocket_empty_completion_then_retry_upstream(captured.clone()).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", "websocket"); + let _previous_response_env = EnvGuard::set("CCP_CODEX_PREVIOUS_RESPONSE_ID", "1"); + + let first = call_messages_body(json!({ + "model": "gpt-5.5", + "max_tokens": 64, + "stream": true, + "messages": [{"role":"user","content":"one"}] + })) + .await; + assert_eq!(first.status(), StatusCode::OK); + let _ = axum::body::to_bytes(first.into_body(), usize::MAX) + .await + .unwrap(); + + let second = call_messages_body(json!({ + "model": "gpt-5.5", + "max_tokens": 64, + "stream": true, + "messages": [ + {"role":"user","content":"one"}, + {"role":"assistant","content":"first"}, + {"role":"user","content":"two"} + ] + })) + .await; + assert_eq!(second.status(), StatusCode::OK); + let second_body = axum::body::to_bytes(second.into_body(), usize::MAX) + .await + .unwrap(); + assert!( + String::from_utf8_lossy(&second_body).contains("retry"), + "second response body: {}", + String::from_utf8_lossy(&second_body) + ); + + let guard = captured.lock().unwrap(); + assert_eq!(guard.len(), 3, "expected full-context retry request"); + assert!(guard[0].get("previous_response_id").is_none()); + assert_eq!(guard[1]["previous_response_id"], "resp_1"); + assert!(guard[2].get("previous_response_id").is_none()); + assert_eq!( + guard[2]["input"].as_array().map(Vec::len), + Some(3), + "retry request should send the full input" + ); + + clear_all_continuations_for_tests(); + clear_codex_websocket_pool_for_tests(); +} + #[allow(clippy::await_holding_lock)] #[tokio::test(flavor = "multi_thread")] async fn smoke_codex_websocket_previous_response_id_sends_delta_on_second_turn() { From 3f9bff911968d21bb0af5fe2f5e5a4269bde7397 Mon Sep 17 00:00:00 2001 From: Hunter Moore Date: Mon, 20 Jul 2026 07:23:15 -0400 Subject: [PATCH 2/4] test(codex): preserve signature-only live output --- src/providers/codex/translate/live_stream.rs | 25 ++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/src/providers/codex/translate/live_stream.rs b/src/providers/codex/translate/live_stream.rs index 25016c57..de79e163 100644 --- a/src/providers/codex/translate/live_stream.rs +++ b/src/providers/codex/translate/live_stream.rs @@ -956,6 +956,7 @@ impl LiveStreamTranslator { let Some(signature) = encode_reasoning_signature(&replay) else { return; }; + self.semantic_output_started = true; let index = self.anthropic_index; self.anthropic_index += 1; self.ensure_message_start(traffic, out); @@ -1512,4 +1513,28 @@ mod tests { assert!(signature_delta < thinking_stop); assert!(out.contains("ccp:codex:v1:")); } + + #[test] + fn signature_only_reasoning_is_semantic_output() { + let mut translator = LiveStreamTranslator::new("msg_1", "gpt-5.5"); + let mut out = Vec::new(); + for event in [ + json!({ + "type":"response.output_item.added", + "output_index":0, + "item":{"type":"reasoning","id":"rs_1","encrypted_content":"opaque"} + }), + json!({ + "type":"response.output_item.done", + "output_index":0, + "item":{"type":"reasoning","id":"rs_1"} + }), + ] { + out.extend(translator.accept(&event, None).unwrap()); + } + + let out = String::from_utf8(out).unwrap(); + assert!(out.contains(r#""type":"signature_delta""#)); + assert!(translator.has_semantic_output()); + } } From bb42ead8ba93ca0cbcd7d1822cb452b9224f8aab Mon Sep 17 00:00:00 2001 From: Hunter Moore Date: Mon, 20 Jul 2026 10:25:43 -0400 Subject: [PATCH 3/4] test(codex): prove bounded 503 on repeated empty completions Add a smoke test where the upstream answers every attempt with a terminal-only completion: the proxy must stop after the initial attempt plus MAX_RETRYABLE_LIVE_STREAM_RETRIES full-context retries and surface the explicit 503 instead of an empty end_turn. A zero-delay test hook in retry::sleep (same pattern as the existing *_for_tests helpers) keeps the exhaustion run instant; a Drop guard restores real backoff even on panic. Also document the pending_chunk single-flush invariant. --- src/providers/codex/mod.rs | 1 + tests/smoke_cutover.rs | 119 +++++++++++++++++++++++++++++++++++++ 2 files changed, 120 insertions(+) diff --git a/src/providers/codex/mod.rs b/src/providers/codex/mod.rs index 7a2650c7..a3627f25 100644 --- a/src/providers/codex/mod.rs +++ b/src/providers/codex/mod.rs @@ -568,6 +568,7 @@ async fn live_stream_response_once( let mut translator = LiveStreamTranslator::new(message_id, model.to_string()); let mut upstream_sse_body = Vec::new(); // Keep protocol framing private until real output makes a transparent retry unsafe. + // Every branch that consumes pending_chunk returns, so it is never flushed twice. let mut pending_chunk = Vec::new(); let mut generation_started = false; diff --git a/tests/smoke_cutover.rs b/tests/smoke_cutover.rs index 5a8735c0..5529796b 100644 --- a/tests/smoke_cutover.rs +++ b/tests/smoke_cutover.rs @@ -624,6 +624,58 @@ async fn spawn_websocket_empty_completion_then_retry_upstream( addr_str } +/// Upstream that answers every request with a terminal-only completion, +/// so the proxy's bounded retry loop always exhausts. +async fn spawn_websocket_always_empty_completion_upstream( + request_count: Arc, +) -> String { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let addr_str = format!("http://{addr}"); + + tokio::spawn(async move { + loop { + let Ok((stream, _)) = listener.accept().await else { + return; + }; + let Ok(ws) = tokio_tungstenite::accept_async(stream).await else { + return; + }; + let (mut sender, mut receiver) = ws.split(); + let request_count = request_count.clone(); + + tokio::spawn(async move { + while let Some(message) = receiver.next().await { + match message { + Ok(Message::Text(_)) => { + request_count.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + let event = json!({ + "type": "response.completed", + "response": { + "id": "resp_empty", + "status": "completed", + "incomplete_details": null, + "usage": {"input_tokens": 5, "output_tokens": 0} + } + }); + if sender.send(Message::Text(event.to_string())).await.is_err() { + return; + } + } + Ok(Message::Ping(data)) => { + let _ = sender.send(Message::Pong(data)).await; + } + Ok(_) => {} + Err(_) => return, + } + } + }); + } + }); + + addr_str +} + // --------------------------------------------------------------------------- // Health and routing smoke tests (no env var mutation needed) // --------------------------------------------------------------------------- @@ -1887,6 +1939,73 @@ async fn smoke_codex_websocket_stream_retries_terminal_only_completion_with_full clear_codex_websocket_pool_for_tests(); } +/// 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); + } +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test(flavor = "multi_thread")] +async fn smoke_codex_websocket_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"); + clear_codex_websocket_pool_for_tests(); + clear_all_continuations_for_tests(); + + let request_count = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let upstream = spawn_websocket_always_empty_completion_upstream(request_count.clone()).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", "websocket"); + + 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::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_RETRYABLE_LIVE_STREAM_RETRIES full-context retries. + assert_eq!( + request_count.load(std::sync::atomic::Ordering::SeqCst), + 11, + "retry loop must stay bounded" + ); + + clear_all_continuations_for_tests(); + clear_codex_websocket_pool_for_tests(); +} + #[allow(clippy::await_holding_lock)] #[tokio::test(flavor = "multi_thread")] async fn smoke_codex_websocket_previous_response_id_sends_delta_on_second_turn() { From 23d22eaa13cf3fb240c26aa6faee006c86e076ef Mon Sep 17 00:00:00 2001 From: Raine Virta Date: Tue, 28 Jul 2026 16:07:07 +0200 Subject: [PATCH 4/4] codex: capture only committed live events Live stream translation writes traffic artifacts while initial Anthropic framing is still buffered. Empty completion retries discard those bytes, but the capture previously presented them as events delivered to the client. Translate the buffered prefix without a live capture sink, then parse and record its Anthropic SSE events only when the prefix is committed downstream. Later stream events continue to use direct capture. Extend the retry smoke test to prove discarded end-turn framing stays out of downstream artifacts. Client-visible streaming and retry behavior is unchanged. --- src/providers/codex/mod.rs | 72 +++++++++++++++++++++++++------------- tests/smoke_cutover.rs | 36 ++++++++++--------- 2 files changed, 67 insertions(+), 41 deletions(-) diff --git a/src/providers/codex/mod.rs b/src/providers/codex/mod.rs index a3627f25..0717421e 100644 --- a/src/providers/codex/mod.rs +++ b/src/providers/codex/mod.rs @@ -20,6 +20,7 @@ use std::sync::Arc; use crate::anthropic::error::json_error; use crate::anthropic::schema::{CountTokensResponse, MessagesRequest}; +use crate::anthropic::sse::parse_sse_events; use crate::config; use crate::logging::create_logger; use crate::monitor::usage_from_anthropic_sse; @@ -595,7 +596,7 @@ async fn live_stream_response_once( generation_started = true; } append_upstream_sse_payload(&mut upstream_sse_body, &payload); - let (chunk, terminal) = match translate_live_stream_payload(&mut translator, &payload, &ctx) + let (chunk, terminal) = match translate_live_stream_payload(&mut translator, &payload, None) { Ok(result) => result, Err(message) => { @@ -651,6 +652,7 @@ async fn live_stream_response_once( }; } if translator.has_semantic_output() && !pending_chunk.is_empty() { + record_live_stream_downstream_capture(&ctx, &pending_chunk); record_live_stream_progress(&ctx, &pending_chunk); if terminal { update_continuation_from_upstream( @@ -684,6 +686,7 @@ async fn live_stream_response_once( if pending_chunk.is_empty() { return LiveStreamStart::Response(empty_live_stream_response()); } + record_live_stream_downstream_capture(&ctx, &pending_chunk); record_live_stream_progress(&ctx, &pending_chunk); return LiveStreamStart::Response(single_live_stream_response(pending_chunk)); } @@ -720,13 +723,31 @@ fn codex_generation_event(payload: &serde_json::Value) -> bool { fn translate_live_stream_payload( translator: &mut LiveStreamTranslator, payload: &serde_json::Value, - ctx: &RequestContext, + traffic: Option<&crate::traffic::TrafficCapture>, ) -> Result<(Vec, bool), String> { - let chunk = translator.accept(payload, ctx.traffic.as_deref())?; + let chunk = translator.accept(payload, traffic)?; let terminal = is_codex_terminal_event(payload) || translator.is_finished(); Ok((chunk, terminal)) } +fn record_live_stream_downstream_capture(ctx: &RequestContext, chunk: &[u8]) { + let Some(traffic) = ctx.traffic.as_ref() else { + return; + }; + for event in parse_sse_events(chunk) { + let Ok(data) = serde_json::from_str::(&event.data) else { + continue; + }; + traffic.write_json_event( + "050-downstream-event", + &serde_json::json!({ + "event": event.event.as_deref().unwrap_or("message"), + "data": data, + }), + ); + } +} + fn record_live_stream_progress(ctx: &RequestContext, chunk: &[u8]) { if let Some(monitor) = ctx.monitor.as_ref() { let (input_tokens, output_tokens) = usage_from_anthropic_sse(chunk); @@ -776,28 +797,31 @@ fn remaining_live_stream_response( match item { Ok(payload) => { append_upstream_sse_payload(&mut upstream_sse_body, &payload); - let (chunk, terminal) = - match translate_live_stream_payload(&mut translator, &payload, &ctx) { - Ok(result) => result, - Err(message) => { - abort_request_state( - ctx.session_id.as_deref(), - turn_id, - compact_boundary, - &request_body, - ); - let chunk = translator.error_chunk( - &message, - "api_error", - ctx.traffic.as_deref(), - ); - if !chunk.is_empty() { - record_live_stream_progress(&ctx, &chunk); - let _ = tx.send(Ok(Bytes::from(chunk))).await; - } - return; + let (chunk, terminal) = match translate_live_stream_payload( + &mut translator, + &payload, + ctx.traffic.as_deref(), + ) { + Ok(result) => result, + Err(message) => { + abort_request_state( + ctx.session_id.as_deref(), + turn_id, + compact_boundary, + &request_body, + ); + let chunk = translator.error_chunk( + &message, + "api_error", + ctx.traffic.as_deref(), + ); + if !chunk.is_empty() { + record_live_stream_progress(&ctx, &chunk); + let _ = tx.send(Ok(Bytes::from(chunk))).await; } - }; + return; + } + }; if !chunk.is_empty() { record_live_stream_progress(&ctx, &chunk); if tx.send(Ok(Bytes::from(chunk))).await.is_err() { diff --git a/tests/smoke_cutover.rs b/tests/smoke_cutover.rs index 5529796b..35b91da6 100644 --- a/tests/smoke_cutover.rs +++ b/tests/smoke_cutover.rs @@ -1879,6 +1879,7 @@ async fn smoke_codex_websocket_stream_retries_empty_close_with_full_context() { async fn smoke_codex_websocket_stream_retries_terminal_only_completion_with_full_context() { let _guard = env_lock(); let config = TempDir::new().unwrap(); + let state = TempDir::new().unwrap(); write_auth(config.path(), "codex"); clear_codex_websocket_pool_for_tests(); clear_all_continuations_for_tests(); @@ -1886,6 +1887,8 @@ async fn smoke_codex_websocket_stream_retries_terminal_only_completion_with_full let captured = Arc::new(Mutex::new(Vec::new())); let upstream = spawn_websocket_empty_completion_then_retry_upstream(captured.clone()).await; + let _traffic_env = EnvGuard::set("CCP_TRAFFIC_LOG", "1"); + let _state_env = EnvGuard::set("XDG_STATE_HOME", state.path()); 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", "websocket"); @@ -1924,6 +1927,22 @@ async fn smoke_codex_websocket_stream_retries_terminal_only_completion_with_full String::from_utf8_lossy(&second_body) ); + let downstream_end_turns = traffic_files(state.path()) + .into_iter() + .filter(|path| { + path.file_name() + .and_then(|name| name.to_str()) + .is_some_and(|name| name.ends_with("050-downstream-event.json")) + }) + .filter_map(|path| std::fs::read(path).ok()) + .filter_map(|bytes| serde_json::from_slice::(&bytes).ok()) + .filter(|event| event["data"]["delta"]["stop_reason"] == "end_turn") + .count(); + assert_eq!( + downstream_end_turns, 2, + "discarded empty attempts must not be captured as downstream events" + ); + let guard = captured.lock().unwrap(); assert_eq!(guard.len(), 3, "expected full-context retry request"); assert!(guard[0].get("previous_response_id").is_none()); @@ -1939,23 +1958,6 @@ async fn smoke_codex_websocket_stream_retries_terminal_only_completion_with_full clear_codex_websocket_pool_for_tests(); } -/// 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); - } -} - #[allow(clippy::await_holding_lock)] #[tokio::test(flavor = "multi_thread")] async fn smoke_codex_websocket_empty_completions_exhaust_to_service_unavailable() {