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() {