diff --git a/src/providers/codex/mod.rs b/src/providers/codex/mod.rs index 13f7125b..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; @@ -567,6 +568,9 @@ 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. + // Every branch that consumes pending_chunk returns, so it is never flushed twice. + let mut pending_chunk = Vec::new(); let mut generation_started = false; while let Some(item) = upstream_events.recv().await { @@ -592,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) => { @@ -638,8 +642,18 @@ 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_downstream_capture(&ctx, &pending_chunk); + record_live_stream_progress(&ctx, &pending_chunk); if terminal { update_continuation_from_upstream( ctx.session_id.as_deref(), @@ -648,12 +662,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 +683,12 @@ 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_downstream_capture(&ctx, &pending_chunk); + record_live_stream_progress(&ctx, &pending_chunk); + return LiveStreamStart::Response(single_live_stream_response(pending_chunk)); } } @@ -684,6 +703,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()), @@ -694,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); @@ -750,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() { @@ -926,6 +976,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 +1538,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..de79e163 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; @@ -946,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); @@ -1188,6 +1199,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 +1232,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] @@ -1437,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()); + } } diff --git a/tests/smoke_cutover.rs b/tests/smoke_cutover.rs index df700b2a..35b91da6 100644 --- a/tests/smoke_cutover.rs +++ b/tests/smoke_cutover.rs @@ -538,6 +538,144 @@ 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 +} + +/// 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) // --------------------------------------------------------------------------- @@ -1736,6 +1874,140 @@ 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(); + let state = 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 _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"); + 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 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()); + 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_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() {