From 7b779902b7336fcca95c10f67f0294f63dbfd501 Mon Sep 17 00:00:00 2001 From: mulfyx Date: Wed, 15 Jul 2026 02:31:21 +0500 Subject: [PATCH] stream Codex HTTP responses incrementally Decode fragmented Responses SSE frames as they arrive and keep Claude Code alive with Anthropic heartbeats during silent upstream gaps. Preserve the bounded HTTP retry window across control-only events, but close it after semantic output to avoid duplicate text or tool calls. Propagate downstream cancellation into active reads and retry backoff. Add integration coverage for incremental delivery, overload, rate limiting, transient failures, early EOF, retry exhaustion, and cancellation. --- src/providers/codex/client.rs | 827 ++++++++++++++++++- src/providers/codex/mod.rs | 239 +++++- src/providers/codex/translate/live_stream.rs | 35 +- tests/smoke_cutover.rs | 360 ++++++++ 4 files changed, 1446 insertions(+), 15 deletions(-) diff --git a/src/providers/codex/client.rs b/src/providers/codex/client.rs index 4995245..8ae1c21 100644 --- a/src/providers/codex/client.rs +++ b/src/providers/codex/client.rs @@ -209,6 +209,134 @@ pub struct CodexResponse { pub headers: Vec<(String, String)>, } +pub type CodexHttpEventReceiver = + tokio::sync::mpsc::Receiver>; + +const MAX_HTTP_SSE_FRAME_BYTES: usize = 8 * 1024 * 1024; + +#[derive(Default)] +struct HttpSseDecoder { + frame: Vec, + line_start: usize, + skip_lf: bool, +} + +struct DecodedHttpSseEvent { + event: Option, + payload: Option, +} + +struct HttpEventStreamState { + resp: reqwest::Response, + started_at: Instant, + body_json: String, + auth: StoredAuth, + auth_refresh_attempted: bool, + use_responses_lite: bool, +} + +impl HttpSseDecoder { + fn push(&mut self, input: &[u8]) -> Result, CodexError> { + let mut events = Vec::new(); + for &byte in input { + if self.skip_lf { + self.skip_lf = false; + if byte == b'\n' { + continue; + } + } + match byte { + b'\n' => self.end_line(&mut events)?, + b'\r' => { + self.end_line(&mut events)?; + self.skip_lf = true; + } + _ => self.push_byte(byte)?, + } + } + Ok(events) + } + + fn finish(&self) -> Result<(), CodexError> { + if self.frame.is_empty() { + Ok(()) + } else { + Err(http_sse_error( + "Codex SSE stream ended with an incomplete frame", + )) + } + } + + fn push_byte(&mut self, byte: u8) -> Result<(), CodexError> { + if self.frame.len() >= MAX_HTTP_SSE_FRAME_BYTES { + return Err(http_sse_error("Codex SSE frame exceeds the size limit")); + } + self.frame.push(byte); + Ok(()) + } + + fn end_line(&mut self, events: &mut Vec) -> Result<(), CodexError> { + if self.frame.len() == self.line_start { + if !self.frame.is_empty() + && let Some(event) = decode_http_sse_frame(&self.frame)? + { + events.push(event); + } + self.frame.clear(); + self.line_start = 0; + return Ok(()); + } + self.push_byte(b'\n')?; + self.line_start = self.frame.len(); + Ok(()) + } +} + +fn decode_http_sse_frame(frame: &[u8]) -> Result, CodexError> { + let frame = std::str::from_utf8(frame) + .map_err(|_| http_sse_error("Codex SSE frame contains invalid UTF-8"))?; + let mut event = None; + let mut data = Vec::new(); + for line in frame.lines() { + if line.starts_with(':') { + continue; + } + let (field, value) = line.split_once(':').unwrap_or((line, "")); + let value = value.strip_prefix(' ').unwrap_or(value); + match field { + "event" => event = Some(value.to_owned()), + "data" => data.push(value), + _ => {} + } + } + if data.is_empty() { + return Ok(None); + } + let data = data.join("\n"); + if data == "[DONE]" { + return Ok(Some(DecodedHttpSseEvent { + event, + payload: None, + })); + } + let payload = serde_json::from_str(&data) + .map_err(|_| http_sse_error("Codex SSE frame contains invalid JSON"))?; + Ok(Some(DecodedHttpSseEvent { + event, + payload: Some(payload), + })) +} + +fn http_sse_error(message: &str) -> CodexError { + CodexError { + status: 0, + message: message.to_string(), + detail: Some("http_response_sse".to_string()), + retry_after: None, + origin: CodexErrorOrigin::Http, + } +} + // --------------------------------------------------------------------------- // Client // --------------------------------------------------------------------------- @@ -296,6 +424,418 @@ impl CodexHttpClient { .await } + pub async fn stream_codex_http_events( + self: &Arc, + body: &ResponsesRequest, + ctx: &RequestContext, + ) -> Result { + let mut auth = self.auth_manager.get_auth().await.map_err(|e| CodexError { + status: 401, + message: "Auth error".to_string(), + detail: Some(e.to_string()), + retry_after: None, + origin: CodexErrorOrigin::Auth, + })?; + let body_json = serde_json::to_string(body).map_err(|e| CodexError { + status: 500, + message: "Failed to serialize request".to_string(), + detail: Some(e.to_string()), + retry_after: None, + origin: CodexErrorOrigin::Http, + })?; + let mut auth_refresh_attempted = false; + let use_responses_lite = body.client_metadata.is_some(); + let (resp, started_at) = self + .start_http_event_attempt( + &mut auth, + &body_json, + ctx, + use_responses_lite, + &mut auth_refresh_attempted, + ) + .await?; + + Ok(self.spawn_http_event_stream( + HttpEventStreamState { + resp, + started_at, + body_json, + auth, + auth_refresh_attempted, + use_responses_lite, + }, + ctx.clone(), + )) + } + + async fn start_http_event_attempt( + &self, + auth: &mut StoredAuth, + body_json: &str, + ctx: &RequestContext, + use_responses_lite: bool, + auth_refresh_attempted: &mut bool, + ) -> Result<(reqwest::Response, Instant), CodexError> { + loop { + let (resp, started_at) = self + .start_post_http(auth, body_json, ctx, use_responses_lite) + .await?; + + if resp.status().as_u16() == 401 && !*auth_refresh_attempted { + *auth_refresh_attempted = true; + *auth = self + .auth_manager + .force_refresh(&auth.access) + .await + .map_err(auth_refresh_error)?; + continue; + } + + if !resp.status().is_success() { + let response = self.collect_http_response(resp, started_at, ctx).await?; + return Err(codex_status_error( + response, + crate::config::CodexTransport::Http, + )); + } + + let status = resp.status().as_u16(); + let headers = response_headers(&resp); + if let Some(traffic) = ctx.traffic.as_deref() { + write_upstream_response_headers_capture( + traffic, + status, + started_at.elapsed(), + &headers, + ); + } + return Ok((resp, started_at)); + } + } + + pub async fn stream_codex_auto_events( + self: &Arc, + body: &ResponsesRequest, + ctx: &RequestContext, + continuation: Option<&super::continuation::ContinuationCandidate>, + ) -> Result { + let mut websocket = self + .stream_codex_websocket_events(body, ctx, continuation) + .await?; + let client = self.clone(); + let body = body.clone(); + let ctx = ctx.clone(); + let (tx, rx) = tokio::sync::mpsc::channel(64); + tokio::spawn(async move { + match websocket.recv().await { + Some(Err(err)) if should_fallback_to_http(&err) => { + match client.stream_codex_http_events(&body, &ctx).await { + Ok(http) => forward_codex_events(http, tx).await, + Err(err) => { + let _ = tx.send(Err(err)).await; + } + } + } + Some(item) => { + if tx.send(item).await.is_ok() { + forward_codex_events(websocket, tx).await; + } + } + None => { + let _ = tx + .send(Err(CodexError { + status: 0, + message: "WebSocket connection closed before the first Codex event" + .to_string(), + detail: Some( + super::websocket::WEBSOCKET_MISSING_TERMINAL_DETAIL.to_string(), + ), + retry_after: None, + origin: CodexErrorOrigin::WebSocket, + })) + .await; + } + } + }); + Ok(rx) + } + + fn spawn_http_event_stream( + self: &Arc, + state: HttpEventStreamState, + ctx: RequestContext, + ) -> CodexHttpEventReceiver { + let HttpEventStreamState { + mut resp, + mut started_at, + body_json, + mut auth, + mut auth_refresh_attempted, + use_responses_lite, + } = state; + let client = self.clone(); + let body_idle_timeout_ms = self.body_idle_timeout_ms; + let req_id = ctx.req_id.clone(); + let traffic = ctx.traffic.clone(); + let (tx, rx) = tokio::sync::mpsc::channel(64); + + tokio::spawn(async move { + let log = create_logger("codex"); + let mut retries = 0_u32; + let mut semantic_output_forwarded = false; + + if tx + .send(Ok(serde_json::json!({ + "type": "keepalive", + "_ccp_synthetic": true + }))) + .await + .is_err() + { + return; + } + + 'attempts: loop { + let mut decoder = HttpSseDecoder::default(); + let mut body_bytes = 0_u64; + let mut body_chunks = 0_u64; + let mut event_count = 0_u64; + let mut pending_events = Vec::new(); + + let mut retry_error = 'read_attempt: loop { + let chunk = tokio::select! { + _ = tx.closed() => { + log_http_stream_end( + &log, + "codex_http_stream_dropped", + &req_id, + started_at, + body_bytes, + body_chunks, + event_count, + None, + ); + return; + } + chunk = tokio::time::timeout( + Duration::from_millis(body_idle_timeout_ms), + resp.chunk(), + ) => chunk + }; + + let chunk = match chunk { + Ok(Ok(Some(chunk))) => chunk, + Ok(Ok(None)) => { + let error = match decoder.finish() { + Ok(()) => http_sse_error( + "Codex SSE stream ended before a terminal response event", + ), + Err(error) => error, + }; + log_http_stream_end( + &log, + "codex_http_stream_failed", + &req_id, + started_at, + body_bytes, + body_chunks, + event_count, + Some(&error.message), + ); + if !semantic_output_forwarded && retryable_http_stream_error(&error) { + break 'read_attempt error; + } + let _ = tx.send(Err(error)).await; + return; + } + Ok(Err(err)) => { + let error = CodexError { + status: 0, + message: format!( + "Transport error reading Codex response body: {err}" + ), + detail: Some("http_response_body".to_string()), + retry_after: None, + origin: CodexErrorOrigin::Http, + }; + log_http_stream_end( + &log, + "codex_http_stream_failed", + &req_id, + started_at, + body_bytes, + body_chunks, + event_count, + Some(&error.message), + ); + if !semantic_output_forwarded { + break 'read_attempt error; + } + let _ = tx.send(Err(error)).await; + return; + } + Err(_) => { + let error = CodexError { + status: 0, + message: format!( + "Timed out waiting {body_idle_timeout_ms}ms for the next Codex response body chunk" + ), + detail: Some("http_response_body".to_string()), + retry_after: None, + origin: CodexErrorOrigin::Http, + }; + log_http_stream_end( + &log, + "codex_http_stream_failed", + &req_id, + started_at, + body_bytes, + body_chunks, + event_count, + Some(&error.message), + ); + if !semantic_output_forwarded { + break 'read_attempt error; + } + let _ = tx.send(Err(error)).await; + return; + } + }; + + body_bytes = body_bytes.saturating_add(chunk.len() as u64); + body_chunks = body_chunks.saturating_add(1); + let events = match decoder.push(&chunk) { + Ok(events) => events, + Err(error) => { + log_http_stream_end( + &log, + "codex_http_stream_failed", + &req_id, + started_at, + body_bytes, + body_chunks, + event_count, + Some(&error.message), + ); + if !semantic_output_forwarded && retryable_http_stream_error(&error) { + break 'read_attempt error; + } + let _ = tx.send(Err(error)).await; + return; + } + }; + + for event in events { + let Some(payload) = event.payload else { + continue; + }; + event_count = event_count.saturating_add(1); + if let Some(traffic) = traffic.as_deref() { + write_codex_http_sse_event_capture( + traffic, + event.event.as_deref(), + &payload, + ); + } + + let failure = super::events::classify_event_failure(&payload); + if !semantic_output_forwarded + && let Some(failure) = failure.as_ref() + && failure.retryable() + { + pending_events.clear(); + break 'read_attempt codex_event_failure_error(failure.clone()); + } + + let terminal = event_closes_http_stream(&payload); + if !semantic_output_forwarded && failure.is_some() { + pending_events.clear(); + if tx.send(Ok(payload)).await.is_err() { + return; + } + } else if !semantic_output_forwarded + && http_event_starts_semantic_output(&payload) + { + semantic_output_forwarded = true; + for pending in pending_events.drain(..) { + if tx.send(Ok(pending)).await.is_err() { + return; + } + } + if tx.send(Ok(payload)).await.is_err() { + return; + } + } else if semantic_output_forwarded || http_event_is_control(&payload) { + if tx.send(Ok(payload)).await.is_err() { + return; + } + } else { + pending_events.push(payload); + } + + if terminal { + log_http_stream_end( + &log, + "codex_http_stream_completed", + &req_id, + started_at, + body_bytes, + body_chunks, + event_count, + None, + ); + return; + } + } + }; + + loop { + if retries >= MAX_BUFFERED_TRANSPORT_RETRIES { + let _ = tx.send(Err(retry_error)).await; + return; + } + let delay = compute_backoff_delay(retries, retry_error.retry_after.as_deref()); + if delay.exceeds_budget { + let _ = tx.send(Err(retry_error)).await; + return; + } + retries += 1; + tokio::select! { + _ = tx.closed() => return, + _ = sleep(delay.wait_ms) => {} + } + + let next_attempt = tokio::select! { + _ = tx.closed() => return, + result = client.start_http_event_attempt( + &mut auth, + &body_json, + &ctx, + use_responses_lite, + &mut auth_refresh_attempted, + ) => result + }; + match next_attempt { + Ok((next_resp, next_started_at)) => { + resp = next_resp; + started_at = next_started_at; + continue 'attempts; + } + Err(error) if retryable_http_stream_error(&error) => { + retry_error = error; + } + Err(error) => { + let _ = tx.send(Err(error)).await; + return; + } + } + } + } + }); + + rx + } + async fn post_codex_with_transport( &self, body: &ResponsesRequest, @@ -823,6 +1363,19 @@ impl CodexHttpClient { ctx: &RequestContext, use_responses_lite: bool, ) -> Result { + let (resp, started_at) = self + .start_post_http(auth, body_json, ctx, use_responses_lite) + .await?; + self.collect_http_response(resp, started_at, ctx).await + } + + async fn start_post_http( + &self, + auth: &StoredAuth, + body_json: &str, + ctx: &RequestContext, + use_responses_lite: bool, + ) -> Result<(reqwest::Response, Instant), CodexError> { let url = &self.base_url; let headers = build_codex_headers(auth, ctx, use_responses_lite)?; @@ -841,7 +1394,7 @@ impl CodexHttpClient { let send_fut = req_builder.body(body_json.to_string()).send(); let header_timeout_dur = Duration::from_millis(self.header_timeout_ms); - let mut resp = tokio::time::timeout(header_timeout_dur, send_fut) + let resp = tokio::time::timeout(header_timeout_dur, send_fut) .await .map_err(|_| CodexError { status: 0, @@ -873,6 +1426,15 @@ impl CodexHttpClient { } })?; + Ok((resp, started_at)) + } + + async fn collect_http_response( + &self, + mut resp: reqwest::Response, + started_at: Instant, + ctx: &RequestContext, + ) -> Result { let status = resp.status().as_u16(); let headers: Vec<(String, String)> = resp .headers() @@ -936,6 +1498,133 @@ impl CodexHttpClient { } } +async fn forward_codex_events( + mut source: tokio::sync::mpsc::Receiver>, + tx: tokio::sync::mpsc::Sender>, +) { + while let Some(item) = source.recv().await { + if tx.send(item).await.is_err() { + return; + } + } +} + +fn response_headers(resp: &reqwest::Response) -> Vec<(String, String)> { + resp.headers() + .iter() + .map(|(key, value)| (key.to_string(), value.to_str().unwrap_or("").to_string())) + .collect() +} + +fn event_closes_http_stream(payload: &serde_json::Value) -> bool { + matches!( + payload.get("type").and_then(|value| value.as_str()), + Some( + "response.completed" + | "response.incomplete" + | "response.done" + | "response.failed" + | "response.error" + | "error" + ) + ) +} + +fn http_event_is_control(payload: &serde_json::Value) -> bool { + matches!( + payload.get("type").and_then(|value| value.as_str()), + Some( + "keepalive" + | "response.created" + | "response.in_progress" + | "codex.rate_limits" + | "response.web_search_call.in_progress" + | "response.web_search_call.searching" + | "response.web_search_call.completed" + ) + ) +} + +fn http_event_starts_semantic_output(payload: &serde_json::Value) -> bool { + match payload.get("type").and_then(|value| value.as_str()) { + Some("response.output_item.added") => matches!( + payload + .pointer("/item/type") + .and_then(|value| value.as_str()), + Some("message" | "function_call") + ), + Some( + "response.reasoning_summary_text.delta" + | "response.output_text.delta" + | "response.function_call_arguments.delta", + ) => payload + .get("delta") + .and_then(|value| value.as_str()) + .is_some_and(|delta| !delta.is_empty()), + Some("response.output_item.done") => matches!( + payload + .pointer("/item/type") + .and_then(|value| value.as_str()), + Some("reasoning" | "message" | "function_call") + ), + Some("response.completed" | "response.incomplete" | "response.done") => true, + _ => false, + } +} + +fn codex_event_failure_error(failure: super::events::CodexEventFailure) -> CodexError { + CodexError { + status: failure.status, + message: failure.message.clone(), + detail: Some(failure.message), + retry_after: failure.retry_after, + origin: CodexErrorOrigin::Http, + } +} + +fn retryable_http_stream_error(error: &CodexError) -> bool { + if should_retry_codex_status(error.status) || is_retryable_transport_error(error) { + return true; + } + if error.status != 0 { + return false; + } + let message = error.message.to_ascii_lowercase(); + message.contains("ended before a terminal response event") + || message.contains("ended with an incomplete frame") +} + +#[allow(clippy::too_many_arguments)] +fn log_http_stream_end( + log: &crate::logging::Logger, + message: &str, + req_id: &str, + started_at: Instant, + body_bytes: u64, + body_chunks: u64, + event_count: u64, + error: Option<&str>, +) { + let mut fields = serde_json::Map::from_iter([ + ("reqId".to_string(), serde_json::json!(req_id)), + ("transport".to_string(), serde_json::json!("http")), + ("bodyBytes".to_string(), serde_json::json!(body_bytes)), + ("bodyChunks".to_string(), serde_json::json!(body_chunks)), + ("eventCount".to_string(), serde_json::json!(event_count)), + ( + "ms".to_string(), + serde_json::json!(started_at.elapsed().as_millis()), + ), + ]); + if let Some(error) = error { + fields.insert("error".to_string(), serde_json::json!(error)); + } + match message { + "codex_http_stream_completed" => log.info(message, Some(fields)), + _ => log.warn(message, Some(fields)), + } +} + fn write_codex_http_request_capture( traffic: &TrafficCapture, url: &str, @@ -968,6 +1657,21 @@ fn write_upstream_response_capture( elapsed: Duration, headers: &[(String, String)], body: &[u8], +) { + write_upstream_response_headers_capture(traffic, status, elapsed, headers); + if status >= 400 { + traffic.write_text("031-upstream-error-body", &String::from_utf8_lossy(body)); + } else { + traffic.write_bytes("032-upstream-response-body.sse", body); + write_codex_sse_event_capture(traffic, body); + } +} + +fn write_upstream_response_headers_capture( + traffic: &TrafficCapture, + status: u16, + elapsed: Duration, + headers: &[(String, String)], ) { traffic.write_json( "030-upstream-response-headers", @@ -977,12 +1681,22 @@ fn write_upstream_response_capture( "headers": headers_to_json_from_pairs(headers), }), ); - if status >= 400 { - traffic.write_text("031-upstream-error-body", &String::from_utf8_lossy(body)); - } else { - traffic.write_bytes("032-upstream-response-body.sse", body); - write_codex_sse_event_capture(traffic, body); +} + +fn write_codex_http_sse_event_capture( + traffic: &TrafficCapture, + event: Option<&str>, + payload: &serde_json::Value, +) { + let mut payload = payload.clone(); + if let Some(event) = event + && let Some(object) = payload.as_object_mut() + { + object + .entry("_sse_event") + .or_insert_with(|| serde_json::json!(event)); } + traffic.write_json_event("040-upstream-event", &payload); } fn write_codex_sse_event_capture(traffic: &TrafficCapture, body: &[u8]) { @@ -1358,6 +2072,107 @@ mod tests { client } + async fn write_http_chunk(stream: &mut tokio::net::TcpStream, body: &[u8]) { + stream + .write_all(format!("{:x}\r\n", body.len()).as_bytes()) + .await + .unwrap(); + stream.write_all(body).await.unwrap(); + stream.write_all(b"\r\n").await.unwrap(); + stream.flush().await.unwrap(); + } + + #[test] + fn http_sse_decoder_handles_fragmented_crlf_and_done_marker() { + let mut decoder = HttpSseDecoder::default(); + assert!( + decoder + .push(b"event: response.output_text.delta\r") + .unwrap() + .is_empty() + ); + let events = decoder + .push(b"\ndata: {\"type\":\"response.output_text.delta\",\"delta\":\"ok\"}\r\n\r\n") + .unwrap(); + assert_eq!(events.len(), 1); + assert_eq!( + events[0].event.as_deref(), + Some("response.output_text.delta") + ); + assert_eq!( + events[0] + .payload + .as_ref() + .and_then(|payload| payload.get("delta")) + .and_then(|value| value.as_str()), + Some("ok") + ); + + let done = decoder.push(b"data: [DONE]\n\n").unwrap(); + assert_eq!(done.len(), 1); + assert!(done[0].payload.is_none()); + decoder.finish().unwrap(); + } + + #[tokio::test] + async fn http_stream_forwards_event_before_terminal_body() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let (release_tx, release_rx) = tokio::sync::oneshot::channel(); + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut request = [0_u8; 16 * 1024]; + assert!(stream.read(&mut request).await.unwrap() > 0); + stream + .write_all( + b"HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\ntransfer-encoding: chunked\r\nconnection: close\r\n\r\n", + ) + .await + .unwrap(); + write_http_chunk( + &mut stream, + b"data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"message\",\"id\":\"msg_up\"}}\n\n", + ) + .await; + release_rx.await.unwrap(); + write_http_chunk( + &mut stream, + b"data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_1\",\"status\":\"completed\",\"usage\":{}}}\n\n", + ) + .await; + }); + + let client = Arc::new(http_test_client(format!("http://{addr}/responses"), 1_000)); + client.auth_manager().set_test_auth(http_test_auth()); + let mut events = client + .stream_codex_http_events(&buffered_test_request(), &http_test_context()) + .await + .unwrap(); + + let synthetic = events.recv().await.unwrap().unwrap(); + assert_eq!( + synthetic.get("type").and_then(|value| value.as_str()), + Some("keepalive") + ); + let first_upstream = tokio::time::timeout(Duration::from_millis(200), events.recv()) + .await + .expect("first upstream event must arrive before the response completes") + .unwrap() + .unwrap(); + assert_eq!( + first_upstream.get("type").and_then(|value| value.as_str()), + Some("response.output_item.added") + ); + + release_tx.send(()).unwrap(); + let terminal = events.recv().await.unwrap().unwrap(); + assert_eq!( + terminal.get("type").and_then(|value| value.as_str()), + Some("response.completed") + ); + server.await.unwrap(); + } + #[tokio::test] async fn buffered_http_retries_retryable_status() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); diff --git a/src/providers/codex/mod.rs b/src/providers/codex/mod.rs index ead8d11..fb99346 100644 --- a/src/providers/codex/mod.rs +++ b/src/providers/codex/mod.rs @@ -14,10 +14,12 @@ use axum::response::{IntoResponse, Response}; use bytes::Bytes; use http::StatusCode; use std::sync::Arc; +use std::time::{Duration, Instant}; use crate::anthropic::error::json_error; use crate::anthropic::schema::{CountTokensResponse, MessagesRequest}; use crate::config; +use crate::logging::create_logger; use crate::monitor::usage_from_anthropic_sse; use crate::provider::{CliHandlers, Provider, RequestContext}; use crate::registry; @@ -41,6 +43,7 @@ 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 LIVE_STREAM_HEARTBEAT_INTERVAL: Duration = Duration::from_secs(15); use self::translate::stream::translate_stream_bytes_with_traffic; // --------------------------------------------------------------------------- @@ -136,23 +139,71 @@ impl Provider for CodexProvider { previous_response_id_enabled, ); let turn_id = continuation.turn_id; + let configured_transport = config::codex_transport(); + let transport = configured_transport.as_str(); + let upstream_started_at = Instant::now(); + let log = create_logger("codex"); + let req_id = ctx.req_id.clone(); + log.info( + "codex_upstream_request_started", + Some(serde_json::Map::from_iter([ + ("reqId".to_string(), serde_json::json!(&req_id)), + ("transport".to_string(), serde_json::json!(transport)), + ("model".to_string(), serde_json::json!(&resolved.model)), + ("stream".to_string(), serde_json::json!(want_stream)), + ( + "responsesLite".to_string(), + serde_json::json!(use_responses_lite), + ), + ( + "previousResponseIdEnabled".to_string(), + serde_json::json!(previous_response_id_enabled), + ), + ( + "hasPreviousResponseId".to_string(), + serde_json::json!(continuation.previous_response_id.is_some()), + ), + ( + "inputDeltaCount".to_string(), + serde_json::json!(continuation.input_delta.as_ref().map(Vec::len)), + ), + ("turnId".to_string(), serde_json::json!(turn_id)), + ])), + ); // Post to upstream with continuation let client = self.client.clone(); if let Some(monitor) = ctx.monitor.as_ref() { monitor.upstream_started(&ctx.req_id); } - if want_stream && matches!(config::codex_transport(), config::CodexTransport::WebSocket) { + if want_stream { let stream_request = translated.clone(); - return live_stream_response( + let response = live_stream_response( client, message_id, model, ctx, stream_request, continuation, + configured_transport, ) .await; + log.info( + "codex_upstream_response_ready", + Some(serde_json::Map::from_iter([ + ("reqId".to_string(), serde_json::json!(&req_id)), + ("transport".to_string(), serde_json::json!(transport)), + ( + "status".to_string(), + serde_json::json!(response.status().as_u16()), + ), + ( + "ms".to_string(), + serde_json::json!(upstream_started_at.elapsed().as_millis()), + ), + ])), + ); + return response; } let upstream = match client @@ -161,10 +212,43 @@ impl Provider for CodexProvider { { Ok(r) => r, Err(e) => { + log.warn( + "codex_upstream_request_failed", + Some(serde_json::Map::from_iter([ + ("reqId".to_string(), serde_json::json!(&req_id)), + ("transport".to_string(), serde_json::json!(transport)), + ("status".to_string(), serde_json::json!(e.status)), + ( + "origin".to_string(), + serde_json::json!(format!("{:?}", e.origin)), + ), + ("error".to_string(), serde_json::json!(&e.message)), + ( + "ms".to_string(), + serde_json::json!(upstream_started_at.elapsed().as_millis()), + ), + ])), + ); abort_continuation(ctx.session_id.as_deref(), turn_id); return map_codex_error_to_response(&e); } }; + log.info( + "codex_upstream_response_received", + Some(serde_json::Map::from_iter([ + ("reqId".to_string(), serde_json::json!(&req_id)), + ("transport".to_string(), serde_json::json!(transport)), + ("status".to_string(), serde_json::json!(upstream.status)), + ( + "bodyBytes".to_string(), + serde_json::json!(upstream.body.len()), + ), + ( + "ms".to_string(), + serde_json::json!(upstream_started_at.elapsed().as_millis()), + ), + ])), + ); if want_stream { let sse_bytes = match translate_stream_bytes_with_traffic( @@ -315,6 +399,7 @@ async fn live_stream_response( ctx: RequestContext, request_body: translate::request::ResponsesRequest, continuation: ContinuationCandidate, + transport: config::CodexTransport, ) -> Response { let model = model.to_string(); let turn_id = continuation.turn_id; @@ -322,10 +407,22 @@ async fn live_stream_response( let mut continuation = Some(continuation); loop { - let upstream_events = match client - .stream_codex_websocket_events(&request_body, &ctx, continuation.as_ref()) - .await - { + let upstream_events = match transport { + config::CodexTransport::Http => { + client.stream_codex_http_events(&request_body, &ctx).await + } + config::CodexTransport::WebSocket => { + client + .stream_codex_websocket_events(&request_body, &ctx, continuation.as_ref()) + .await + } + config::CodexTransport::Auto => { + client + .stream_codex_auto_events(&request_body, &ctx, continuation.as_ref()) + .await + } + }; + let upstream_events = match upstream_events { Ok(events) => events, Err(err) if retryable_live_start_codex_error(&err) => { let dropped = drop_live_continuation_for_retry(&mut continuation); @@ -554,7 +651,31 @@ fn remaining_live_stream_response( abort_continuation(ctx.session_id.as_deref(), turn_id); return; } - while let Some(item) = upstream_events.recv().await { + let mut heartbeat = tokio::time::interval(LIVE_STREAM_HEARTBEAT_INTERVAL); + heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + heartbeat.tick().await; + loop { + let item = tokio::select! { + _ = tx.closed() => { + abort_continuation(ctx.session_id.as_deref(), turn_id); + return; + } + item = upstream_events.recv() => item, + _ = heartbeat.tick() => { + let chunk = translator.ping_chunk(ctx.traffic.as_deref()); + if !chunk.is_empty() { + record_live_stream_progress(&ctx, &chunk); + if tx.send(Ok(Bytes::from(chunk))).await.is_err() { + abort_continuation(ctx.session_id.as_deref(), turn_id); + return; + } + } + continue; + } + }; + let Some(item) = item else { + break; + }; match item { Ok(payload) => { append_upstream_sse_payload(&mut upstream_sse_body, &payload); @@ -624,7 +745,7 @@ fn remaining_live_stream_response( return; } let chunk = translator.error_chunk( - "WebSocket connection closed before terminal Codex response event", + "Upstream event stream closed before terminal Codex response event", "api_error", ctx.traffic.as_deref(), ); @@ -1023,6 +1144,108 @@ mod tests { assert_eq!(state.active[0].output_tokens, Some(48)); } + #[tokio::test] + async fn live_stream_response_emits_downstream_frames_before_terminal_event() { + use http_body_util::BodyExt as _; + + let body = request_with_tools(serde_json::json!([])); + let request_body = translate_request( + &body, + TranslateOptions { + session_id: None, + service_tier: None, + model: "gpt-5.6-sol".to_string(), + use_responses_lite: true, + }, + ) + .unwrap(); + let ctx = RequestContext { + req_id: "incremental-http".to_string(), + session_id: None, + session_seq: None, + provider: "codex".to_string(), + traffic: None, + monitor: None, + }; + let (tx, rx) = tokio::sync::mpsc::channel(8); + tx.send(Ok(serde_json::json!({"type": "keepalive"}))) + .await + .unwrap(); + + let response = match live_stream_response_once( + rx, + "msg_test".to_string(), + "claude-opus-4-8", + ctx, + None, + request_body, + ) + .await + { + LiveStreamStart::Response(response) => response, + LiveStreamStart::Retry { error } => panic!("unexpected retry: {error}"), + }; + let mut body = response.into_body(); + let first = tokio::time::timeout(Duration::from_millis(200), body.frame()) + .await + .expect("initial downstream frame must be available immediately") + .unwrap() + .unwrap() + .into_data() + .unwrap(); + let first = String::from_utf8(first.to_vec()).unwrap(); + assert!(first.contains("event: message_start")); + assert!(first.contains("event: ping")); + + tx.send(Ok(serde_json::json!({ + "type": "response.output_item.added", + "output_index": 0, + "item": {"type": "message", "id": "msg_up"} + }))) + .await + .unwrap(); + let second = tokio::time::timeout(Duration::from_millis(200), body.frame()) + .await + .expect("content block must arrive before the terminal event") + .unwrap() + .unwrap() + .into_data() + .unwrap(); + assert!( + String::from_utf8(second.to_vec()) + .unwrap() + .contains("event: content_block_start") + ); + + for payload in [ + serde_json::json!({ + "type": "response.output_text.delta", + "output_index": 0, + "delta": "done" + }), + serde_json::json!({ + "type": "response.output_item.done", + "output_index": 0, + "item": {"type": "message"} + }), + serde_json::json!({ + "type": "response.completed", + "response": { + "id": "resp_1", + "status": "completed", + "incomplete_details": null, + "usage": {"input_tokens": 1, "output_tokens": 1} + } + }), + ] { + tx.send(Ok(payload)).await.unwrap(); + } + drop(tx); + while let Some(frame) = body.frame().await { + frame.unwrap(); + } + } + #[test] fn supported_models_includes_fast_variants() { let provider = CodexProvider::new(); diff --git a/src/providers/codex/translate/live_stream.rs b/src/providers/codex/translate/live_stream.rs index cfc4844..0e3f898 100644 --- a/src/providers/codex/translate/live_stream.rs +++ b/src/providers/codex/translate/live_stream.rs @@ -107,8 +107,11 @@ impl LiveStreamTranslator { { return Err("rate limit reached".to_string()); } + self.emit_ping(traffic, &mut out); + } + "keepalive" | "response.created" | "response.in_progress" => { + self.emit_ping(traffic, &mut out); } - "keepalive" => {} "response.failed" | "response.error" | "error" => { return Err(error_message(payload)); } @@ -167,6 +170,14 @@ impl LiveStreamTranslator { self.finished } + pub fn ping_chunk(&mut self, traffic: Option<&TrafficCapture>) -> Vec { + let mut out = Vec::new(); + if !self.finished { + self.emit_ping(traffic, &mut out); + } + out + } + pub fn finish_after_closed_completed_tool_call( &mut self, traffic: Option<&TrafficCapture>, @@ -237,6 +248,11 @@ impl LiveStreamTranslator { ); } + fn emit_ping(&mut self, traffic: Option<&TrafficCapture>, out: &mut Vec) { + self.ensure_message_start(traffic, out); + self.emit(traffic, out, "ping", &serde_json::json!({"type": "ping"})); + } + fn emit( &self, traffic: Option<&TrafficCapture>, @@ -1408,6 +1424,23 @@ mod tests { assert_eq!(err, "rate limit reached"); } + #[test] + fn progress_events_start_message_and_emit_pings() { + let mut translator = LiveStreamTranslator::new("msg_1", "gpt-5.5"); + let first = String::from_utf8( + translator + .accept(&json!({"type": "response.created"}), None) + .unwrap(), + ) + .unwrap(); + assert_eq!(first.matches("event: message_start").count(), 1); + assert_eq!(first.matches("event: ping").count(), 1); + + let second = String::from_utf8(translator.ping_chunk(None)).unwrap(); + assert!(!second.contains("event: message_start")); + assert_eq!(second.matches("event: ping").count(), 1); + } + #[test] fn live_stream_emits_signature_delta_before_thinking_stop() { let out = render(vec![ diff --git a/tests/smoke_cutover.rs b/tests/smoke_cutover.rs index afb5927..0b7991b 100644 --- a/tests/smoke_cutover.rs +++ b/tests/smoke_cutover.rs @@ -11,6 +11,7 @@ use futures_util::{SinkExt, StreamExt}; use http_body_util::BodyExt; use serde_json::{Value, json}; use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex, OnceLock}; use std::time::Duration; use tempfile::TempDir; @@ -174,6 +175,63 @@ where addr_str } +#[allow(clippy::await_holding_lock)] +async fn assert_codex_http_presemantic_retry(first_response: Vec) { + let _guard = env_lock(); + clear_all_continuations_for_tests(); + let config = TempDir::new().unwrap(); + write_auth(config.path(), "codex"); + + let attempts = Arc::new(AtomicUsize::new(0)); + let first_response = Arc::new(first_response); + let upstream = spawn_http_upstream({ + let attempts = attempts.clone(); + let first_response = first_response.clone(); + move |_body: Value| { + if attempts.fetch_add(1, Ordering::SeqCst) == 0 { + first_response.as_ref().clone() + } else { + concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_retry\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"message\",\"id\":\"msg_retry\"}}\n\n", + "data: {\"type\":\"response.output_text.delta\",\"output_index\":0,\"delta\":\"retry succeeded\"}\n\n", + "data: {\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{\"type\":\"message\"}}\n\n", + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_retry\",\"status\":\"completed\",\"usage\":{\"input_tokens\":5,\"output_tokens\":2}}}\n\n" + ) + .as_bytes() + .to_vec() + } + } + }) + .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":"hello"}] + })) + .await; + + assert_eq!(response.status(), StatusCode::OK); + let body = tokio::time::timeout( + Duration::from_secs(2), + axum::body::to_bytes(response.into_body(), usize::MAX), + ) + .await + .expect("retried stream must finish") + .unwrap(); + let text = String::from_utf8_lossy(&body); + assert_eq!(attempts.load(Ordering::SeqCst), 2, "stream body: {text}"); + assert!(text.contains("retry succeeded"), "stream body: {text}"); + assert!(!text.contains("event: error"), "stream body: {text}"); + assert_eq!(text.matches("event: message_start").count(), 1); + assert_eq!(text.matches("event: message_stop").count(), 1); +} + /// Spawn a mock WebSocket server that accepts one connection, captures the /// first text message, and responds with Codex WebSocket events that /// accumulate to `"codex websocket ok"`. @@ -788,6 +846,308 @@ async fn smoke_codex_http_stream_traffic_captures_downstream_events() { assert!(downstream.get("data").is_some()); } +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_stream_returns_before_upstream_completion() { + let _guard = env_lock(); + let config = TempDir::new().unwrap(); + write_auth(config.path(), "codex"); + + let release = Arc::new(tokio::sync::Notify::new()); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let upstream = format!("http://{addr}"); + let mock = axum::Router::new().fallback({ + let release = release.clone(); + move |_body: String| { + let release = release.clone(); + async move { + let stream = futures_util::stream::unfold(0_u8, move |state| { + let release = release.clone(); + async move { + match state { + 0 => Some(( + Ok::<_, std::convert::Infallible>(bytes::Bytes::from_static( + b"data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"message\",\"id\":\"msg_up\"}}\n\n", + )), + 1, + )), + 1 => { + release.notified().await; + Some(( + Ok(bytes::Bytes::from_static(concat!( + "data: {\"type\":\"response.output_text.delta\",\"output_index\":0,\"delta\":\"incremental ok\"}\n\n", + "data: {\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{\"type\":\"message\"}}\n\n", + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_1\",\"status\":\"completed\",\"usage\":{\"input_tokens\":5,\"output_tokens\":2}}}\n\n" + ).as_bytes())), + 2, + )) + } + _ => None, + } + } + }); + http::Response::builder() + .status(StatusCode::OK) + .header("content-type", "text/event-stream") + .body(Body::from_stream(stream)) + .unwrap() + } + } + }); + tokio::spawn(async move { + axum::serve(listener, mock).await.ok(); + }); + + 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 = tokio::time::timeout( + Duration::from_millis(500), + call_messages_body(json!({ + "model": "gpt-5.5", + "max_tokens": 64, + "stream": true, + "messages": [{"role":"user","content":"hello"}] + })), + ) + .await + .expect("CCP must return streaming headers before upstream completion"); + assert_eq!(response.status(), StatusCode::OK); + + let mut body = response.into_body(); + let first = tokio::time::timeout(Duration::from_millis(200), body.frame()) + .await + .expect("initial Anthropic heartbeat must arrive before upstream completion") + .unwrap() + .unwrap() + .into_data() + .unwrap(); + let first = String::from_utf8(first.to_vec()).unwrap(); + assert!(first.contains("event: message_start")); + assert!(first.contains("event: ping")); + + release.notify_one(); + let rest = axum::body::to_bytes(body, usize::MAX).await.unwrap(); + let rest = String::from_utf8(rest.to_vec()).unwrap(); + assert!(rest.contains("incremental ok"), "stream body: {rest}"); + assert!(rest.contains("event: message_stop"), "stream body: {rest}"); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_retries_overload_after_control_events() { + assert_codex_http_presemantic_retry( + concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_failed\"}}\n\n", + "data: {\"type\":\"response.in_progress\",\"response\":{\"id\":\"resp_failed\"}}\n\n", + "data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"type\":\"overloaded_error\",\"message\":\"Our servers are currently overloaded. Please try again later.\",\"retry_after\":0}}}\n\n" + ) + .as_bytes() + .to_vec(), + ) + .await; +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_retries_rate_limit_after_control_events() { + assert_codex_http_presemantic_retry( + concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_limited\"}}\n\n", + "data: {\"type\":\"codex.rate_limits\",\"rate_limits\":{\"limit_reached\":true,\"primary\":{\"reset_after_seconds\":0}}}\n\n" + ) + .as_bytes() + .to_vec(), + ) + .await; +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_retries_transient_failure_after_control_events() { + assert_codex_http_presemantic_retry( + concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_transient\"}}\n\n", + "data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"type\":\"server_error\",\"status\":503,\"message\":\"temporarily unavailable\",\"retry_after\":0}}}\n\n" + ) + .as_bytes() + .to_vec(), + ) + .await; +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_retries_presemantic_eof() { + assert_codex_http_presemantic_retry( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_truncated\"}}\n\n" + .as_bytes() + .to_vec(), + ) + .await; +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_does_not_retry_overload_after_semantic_output() { + let _guard = env_lock(); + let config = TempDir::new().unwrap(); + write_auth(config.path(), "codex"); + + let attempts = Arc::new(AtomicUsize::new(0)); + let upstream = spawn_http_upstream({ + let attempts = attempts.clone(); + move |_body: Value| { + let attempt = attempts.fetch_add(1, Ordering::SeqCst); + if attempt == 0 { + concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_partial\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"message\",\"id\":\"msg_partial\"}}\n\n", + "data: {\"type\":\"response.output_text.delta\",\"output_index\":0,\"delta\":\"partial output\"}\n\n", + "data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"type\":\"overloaded_error\",\"message\":\"overloaded after output\",\"retry_after\":0}}}\n\n" + ) + .as_bytes() + .to_vec() + } else { + panic!("semantic output must close the full-request retry window"); + } + } + }) + .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":"hello"}] + })) + .await; + + assert_eq!(response.status(), StatusCode::OK); + let body = tokio::time::timeout( + Duration::from_secs(2), + axum::body::to_bytes(response.into_body(), usize::MAX), + ) + .await + .expect("failed semantic stream must terminate") + .unwrap(); + let text = String::from_utf8_lossy(&body); + assert_eq!(attempts.load(Ordering::SeqCst), 1, "stream body: {text}"); + assert!(text.contains("partial output"), "stream body: {text}"); + assert!( + text.contains("overloaded after output"), + "stream body: {text}" + ); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_stops_after_retry_limit() { + let _guard = env_lock(); + clear_all_continuations_for_tests(); + let config = TempDir::new().unwrap(); + write_auth(config.path(), "codex"); + + let attempts = Arc::new(AtomicUsize::new(0)); + let upstream = spawn_http_upstream({ + let attempts = attempts.clone(); + move |_body: Value| { + attempts.fetch_add(1, Ordering::SeqCst); + concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_exhausted\"}}\n\n", + "data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"type\":\"overloaded_error\",\"message\":\"overloaded until retry limit\",\"retry_after\":0}}}\n\n" + ) + .as_bytes() + .to_vec() + } + }) + .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":"hello"}] + })) + .await; + + assert_eq!(response.status(), StatusCode::OK); + let body = tokio::time::timeout( + Duration::from_secs(2), + axum::body::to_bytes(response.into_body(), usize::MAX), + ) + .await + .expect("exhausted stream must terminate") + .unwrap(); + let text = String::from_utf8_lossy(&body); + assert_eq!(attempts.load(Ordering::SeqCst), 4, "stream body: {text}"); + assert!(text.contains("event: error"), "stream body: {text}"); + assert!( + text.contains("overloaded until retry limit"), + "stream body: {text}" + ); + assert_eq!(text.matches("event: message_start").count(), 1); + assert_eq!(text.matches("event: message_stop").count(), 0); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_cancels_retry_backoff_when_downstream_drops() { + let _guard = env_lock(); + clear_all_continuations_for_tests(); + let config = TempDir::new().unwrap(); + write_auth(config.path(), "codex"); + + let attempts = Arc::new(AtomicUsize::new(0)); + let upstream = spawn_http_upstream({ + let attempts = attempts.clone(); + move |_body: Value| { + let attempt = attempts.fetch_add(1, Ordering::SeqCst); + if attempt == 0 { + concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_cancel\"}}\n\n", + "data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"type\":\"overloaded_error\",\"message\":\"cancel during retry backoff\",\"retry_after\":0.1}}}\n\n" + ) + .as_bytes() + .to_vec() + } else { + panic!("downstream cancellation must prevent another upstream attempt"); + } + } + }) + .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":"hello"}] + })) + .await; + + assert_eq!(response.status(), StatusCode::OK); + let mut body = response.into_body(); + tokio::time::timeout(Duration::from_millis(200), body.frame()) + .await + .expect("initial heartbeat must arrive") + .expect("stream must contain a frame") + .expect("initial frame must be valid"); + drop(body); + + tokio::time::sleep(Duration::from_millis(300)).await; + assert_eq!(attempts.load(Ordering::SeqCst), 1); +} + #[allow(clippy::await_holding_lock)] #[tokio::test] async fn smoke_codex_http_truncated_upstream_writes_reducer_diagnostic() {