diff --git a/src/providers/codex/client.rs b/src/providers/codex/client.rs index 8f37ab81..83bbc013 100644 --- a/src/providers/codex/client.rs +++ b/src/providers/codex/client.rs @@ -1,3 +1,4 @@ +use std::collections::HashSet; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -355,6 +356,8 @@ pub type CodexHttpEventReceiver = tokio::sync::mpsc::Receiver>; const MAX_HTTP_SSE_FRAME_BYTES: usize = 8 * 1024 * 1024; +const MAX_PENDING_HTTP_EVENT_BYTES: usize = 2 * MAX_HTTP_SSE_FRAME_BYTES; +const MIN_PENDING_HTTP_EVENT_BYTES: usize = 256; #[derive(Default)] struct HttpSseDecoder { @@ -480,6 +483,16 @@ fn http_sse_error(message: &str) -> CodexError { } } +fn http_sse_protocol_error(message: &str) -> CodexError { + CodexError { + status: 0, + message: format!("Codex HTTP tool event protocol error: {message}"), + detail: Some("http_response_sse_protocol".to_string()), + retry_after: None, + origin: CodexErrorOrigin::Http, + } +} + pub(crate) struct OwnerAwareCodexResponse { response: CodexResponse, pub(crate) socket_id: Option, @@ -1307,6 +1320,7 @@ impl CodexHttpClient { tokio::spawn(async move { let log = create_logger("codex"); let mut semantic_output_forwarded = false; + let mut replay_committed = false; if tx .send(Ok(serde_json::json!({ @@ -1324,7 +1338,8 @@ impl CodexHttpClient { 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 pending_events = PendingHttpEvents::default(); + let mut provisional_tools = PendingHttpProvisionalTools::default(); let mut retry_error = 'read_attempt: loop { let chunk = tokio::select! { @@ -1366,7 +1381,7 @@ impl CodexHttpClient { event_count, Some(&error.message), ); - if !semantic_output_forwarded && retryable_http_stream_error(&error) { + if !replay_committed && retryable_http_stream_error(&error) { break 'read_attempt error; } let _ = tx.send(Err(error)).await; @@ -1392,7 +1407,7 @@ impl CodexHttpClient { event_count, Some(&error.message), ); - if !semantic_output_forwarded { + if !replay_committed { break 'read_attempt error; } let _ = tx.send(Err(error)).await; @@ -1418,7 +1433,7 @@ impl CodexHttpClient { event_count, Some(&error.message), ); - if !semantic_output_forwarded { + if !replay_committed { break 'read_attempt error; } let _ = tx.send(Err(error)).await; @@ -1441,7 +1456,7 @@ impl CodexHttpClient { event_count, Some(&error.message), ); - if !semantic_output_forwarded && retryable_http_stream_error(&error) { + if !replay_committed && retryable_http_stream_error(&error) { break 'read_attempt error; } let _ = tx.send(Err(error)).await; @@ -1450,7 +1465,7 @@ impl CodexHttpClient { }; for event in events { - let Some(payload) = event.payload else { + let Some(mut payload) = event.payload else { continue; }; event_count = event_count.saturating_add(1); @@ -1464,17 +1479,91 @@ impl CodexHttpClient { let event_kind = super::events::classify_stream_event(&payload); let failure = super::events::classify_event_failure(&payload); + let provisional_event = match provisional_tools.observe(&mut payload) { + Ok(event) => event, + Err(error) => { + let _ = tx.send(Err(error)).await; + return; + } + }; + if let Err(error) = ensure_pending_http_buffer_budget( + pending_events.event_bytes, + provisional_tools.buffered_bytes(), + ) { + let _ = tx.send(Err(error)).await; + return; + } + + if let PendingHttpProvisionalToolEvent::Provisional { stalled_read_call } = + provisional_event + { + if !semantic_output_forwarded + && let Some(call) = stalled_read_call.and_then(|call_index| { + provisional_tools.take_stalled_read(call_index, &pending_events) + }) + { + for pending in pending_events.take() { + if tx.send(Ok(pending)).await.is_err() { + return; + } + } + let output_index = call.output_index.unwrap_or(usize::MAX); + let call_id = call.call_id.unwrap_or_default(); + let added = serde_json::json!({ + "type":"response.output_item.added", + "output_index":output_index, + "item":{"type":"function_call","call_id":call_id,"name":"Read"} + }); + let delta = serde_json::json!({ + "type":"response.function_call_arguments.delta", + "output_index":output_index, + "delta":call.read_args + }); + if tx.send(Ok(added)).await.is_err() + || tx.send(Ok(delta)).await.is_err() + { + return; + } + log_http_stream_end( + &log, + "codex_http_stream_local_finish", + &req_id, + started_at, + body_bytes, + body_chunks, + event_count, + None, + ); + return; + } + continue; + } + + let commits_replay = http_event_commits_replay(&payload, event_kind); + replay_committed |= commits_replay; + if !semantic_output_forwarded + && !replay_committed && let Some(failure) = failure.as_ref() && failure.retryable() { pending_events.clear(); break 'read_attempt codex_event_failure_error(failure.clone()); } + if !semantic_output_forwarded + && replay_committed + && let Some(failure) = failure.as_ref() + { + let _ = tx + .send(Err(codex_event_failure_error(failure.clone()))) + .await; + return; + } let terminal = super::events::event_is_terminal(&payload); match event_kind { super::events::CodexStreamEventKind::TerminalFailure => { + provisional_tools.clear(); if !semantic_output_forwarded { pending_events.clear(); } @@ -1483,7 +1572,8 @@ impl CodexHttpClient { } } super::events::CodexStreamEventKind::TerminalSuccess => { - for pending in pending_events.drain(..) { + provisional_tools.clear(); + for pending in pending_events.take() { if tx.send(Ok(pending)).await.is_err() { return; } @@ -1492,31 +1582,32 @@ impl CodexHttpClient { return; } } - super::events::CodexStreamEventKind::Semantic => { - if !semantic_output_forwarded { - 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; - } - } super::events::CodexStreamEventKind::Control => { if tx.send(Ok(payload)).await.is_err() { return; } } - super::events::CodexStreamEventKind::Structural => { + super::events::CodexStreamEventKind::Semantic + | super::events::CodexStreamEventKind::Structural => { if semantic_output_forwarded { if tx.send(Ok(payload)).await.is_err() { return; } } else { - pending_events.push(payload); + if let Err(error) = pending_events + .push(payload, provisional_tools.buffered_bytes()) + { + let _ = tx.send(Err(error)).await; + return; + } + if commits_replay { + semantic_output_forwarded = true; + for pending in pending_events.take() { + if tx.send(Ok(pending)).await.is_err() { + return; + } + } + } } } } @@ -2534,6 +2625,335 @@ fn response_headers(resp: &reqwest::Response) -> Vec<(String, String)> { .collect() } +#[derive(Default)] +struct PendingHttpEvents { + events: Vec, + event_bytes: usize, +} + +impl PendingHttpEvents { + fn push( + &mut self, + payload: serde_json::Value, + pending_tool_bytes: usize, + ) -> Result<(), CodexError> { + let payload_bytes = serde_json::to_vec(&payload) + .map_err(|_| pending_http_event_buffer_error("event encoding"))? + .len() + .max(MIN_PENDING_HTTP_EVENT_BYTES); + let next_event_bytes = self.event_bytes.saturating_add(payload_bytes); + ensure_pending_http_buffer_budget(next_event_bytes, pending_tool_bytes)?; + self.event_bytes = next_event_bytes; + self.events.push(payload); + Ok(()) + } + + fn clear(&mut self) { + self.events.clear(); + self.event_bytes = 0; + } + + fn take(&mut self) -> Vec { + self.event_bytes = 0; + std::mem::take(&mut self.events) + } + + fn has_no_open_output_items(&self) -> bool { + let mut open_output_items = HashSet::new(); + for payload in &self.events { + match payload.get("type").and_then(serde_json::Value::as_str) { + Some("response.output_item.added") => { + let Some(output_index) = http_event_output_index(payload) else { + return false; + }; + if !open_output_items.insert(output_index) { + return false; + } + } + Some("response.output_item.done") => { + let Some(output_index) = http_event_output_index(payload) else { + return false; + }; + if !open_output_items.remove(&output_index) { + return false; + } + } + _ => {} + } + } + open_output_items.is_empty() + } +} + +#[derive(Default)] +struct PendingHttpProvisionalFunctionCall { + output_index: Option, + item_id: Option, + call_id: Option, + name: Option, + read_args: String, +} + +#[derive(Debug, Default)] +enum PendingHttpProvisionalToolEvent { + #[default] + Other, + Provisional { + stalled_read_call: Option, + }, + AuthoritativeDone, +} + +#[derive(Default)] +struct PendingHttpProvisionalTools { + calls: Vec, +} + +impl PendingHttpProvisionalTools { + fn observe( + &mut self, + payload: &mut serde_json::Value, + ) -> Result { + let event_type = payload.get("type").and_then(serde_json::Value::as_str); + match event_type { + Some("response.output_item.added") + if payload + .pointer("/item/type") + .and_then(serde_json::Value::as_str) + == Some("function_call") => + { + let item = &payload["item"]; + self.calls.push(PendingHttpProvisionalFunctionCall { + output_index: http_event_output_index(payload), + item_id: item + .get("id") + .and_then(serde_json::Value::as_str) + .map(str::to_owned), + call_id: item + .get("call_id") + .and_then(serde_json::Value::as_str) + .map(str::to_owned), + name: item + .get("name") + .and_then(serde_json::Value::as_str) + .map(str::to_owned), + read_args: String::new(), + }); + return Ok(PendingHttpProvisionalToolEvent::Provisional { + stalled_read_call: None, + }); + } + Some("response.function_call_arguments.delta") => { + let Some(call_index) = self.resolve(payload, None) else { + return Ok(PendingHttpProvisionalToolEvent::Provisional { + stalled_read_call: None, + }); + }; + let delta = payload + .get("delta") + .and_then(serde_json::Value::as_str) + .unwrap_or(""); + let call = &mut self.calls[call_index]; + let stalled_read_call = if call.name.as_deref() == Some("Read") && !delta.is_empty() + { + let next_len = call.read_args.len().saturating_add(delta.len()); + ensure_buffered_tool_args_limit("Read", next_len)?; + call.read_args.push_str(delta); + super::translate::read_rewrite::repair_whitespace_stalled_read_args( + "Read", + &call.read_args, + None, + ) + .is_some() + .then_some(call_index) + } else { + None + }; + return Ok(PendingHttpProvisionalToolEvent::Provisional { stalled_read_call }); + } + Some("response.function_call_arguments.done") => { + return Ok(PendingHttpProvisionalToolEvent::Provisional { + stalled_read_call: None, + }); + } + Some("response.output_item.done") + if payload + .pointer("/item/type") + .and_then(serde_json::Value::as_str) + == Some("function_call") => + { + let completed = super::translate::live_stream::validate_completed_function_call( + &payload["item"], + ) + .map_err(|error| { + match error { + super::translate::live_stream::CompletedFunctionCallValidationError::TooLarge { + .. + } => { + let message = error.to_string(); + CodexError { + status: 413, + message: message.clone(), + detail: Some(message), + retry_after: None, + origin: CodexErrorOrigin::Http, + } + } + super::translate::live_stream::CompletedFunctionCallValidationError::Protocol( + message, + ) => http_sse_protocol_error(&message), + } + })?; + payload["item"]["arguments"] = serde_json::Value::String(completed.arguments); + if let Some(call_index) = self.resolve(payload, payload.get("item")) { + self.calls.remove(call_index); + } + return Ok(PendingHttpProvisionalToolEvent::AuthoritativeDone); + } + _ => {} + } + Ok(PendingHttpProvisionalToolEvent::Other) + } + + fn take_stalled_read( + &mut self, + call_index: usize, + pending_events: &PendingHttpEvents, + ) -> Option { + if self.calls.len() != 1 || !pending_events.has_no_open_output_items() { + return None; + } + let call = self.calls.get(call_index)?; + if call.name.as_deref() != Some("Read") + || call + .call_id + .as_deref() + .is_none_or(|call_id| call_id.trim().is_empty()) + || call.read_args.is_empty() + { + return None; + } + Some(self.calls.remove(call_index)) + } + + fn buffered_bytes(&self) -> usize { + self.calls.iter().fold(0, |total, call| { + total + .saturating_add(std::mem::size_of::()) + .saturating_add(call.item_id.as_ref().map_or(0, String::capacity)) + .saturating_add(call.call_id.as_ref().map_or(0, String::capacity)) + .saturating_add(call.name.as_ref().map_or(0, String::capacity)) + .saturating_add(call.read_args.capacity()) + }) + } + + fn clear(&mut self) { + self.calls.clear(); + } + + fn resolve( + &self, + payload: &serde_json::Value, + item: Option<&serde_json::Value>, + ) -> Option { + let output_index = match payload.get("output_index") { + Some(_) => Some(http_event_output_index(payload)?), + None => None, + }; + let item_id = match item + .and_then(|item| item.get("id")) + .or_else(|| payload.get("item_id")) + { + Some(value) => Some(value.as_str()?), + None => None, + }; + let call_id = match item + .and_then(|item| item.get("call_id")) + .or_else(|| payload.get("call_id")) + { + Some(value) => Some(value.as_str()?), + None => None, + }; + let has_explicit_alias = output_index.is_some() || item_id.is_some() || call_id.is_some(); + let mut matches = self.calls.iter().enumerate().filter_map(|(index, call)| { + (output_index.is_none_or(|value| call.output_index == Some(value)) + && item_id.is_none_or(|value| call.item_id.as_deref() == Some(value)) + && call_id.is_none_or(|value| call.call_id.as_deref() == Some(value))) + .then_some(index) + }); + let first = matches.next(); + if matches.next().is_some() { + None + } else if has_explicit_alias { + first + } else { + first.or_else(|| (self.calls.len() == 1).then_some(0)) + } + } +} + +fn http_event_output_index(payload: &serde_json::Value) -> Option { + payload + .get("output_index") + .and_then(serde_json::Value::as_u64) + .and_then(|index| usize::try_from(index).ok()) +} + +fn http_event_commits_replay( + payload: &serde_json::Value, + event_kind: super::events::CodexStreamEventKind, +) -> bool { + if event_kind != super::events::CodexStreamEventKind::Semantic { + return false; + } + matches!( + payload.get("type").and_then(serde_json::Value::as_str), + Some( + "response.reasoning_summary_text.delta" + | "response.output_text.delta" + | "response.output_item.done" + ) + ) +} + +fn ensure_buffered_tool_args_limit(name: &str, len: usize) -> Result<(), CodexError> { + if len <= super::translate::live_stream::BUFFERED_TOOL_MAX_ARGS_BYTES { + return Ok(()); + } + let tool = if name.is_empty() { "tool" } else { name }; + Err(CodexError { + status: 413, + message: format!("Buffered {tool} tool arguments exceeded safe limits"), + detail: Some(format!( + "Buffered {tool} tool arguments exceeded safe limits" + )), + retry_after: None, + origin: CodexErrorOrigin::Http, + }) +} + +fn pending_http_event_buffer_error(limit: &str) -> CodexError { + CodexError { + status: 413, + message: format!("Codex pending HTTP event buffer exceeded its {limit} limit"), + detail: Some(format!( + "Codex pending HTTP event buffer exceeded its {limit} limit" + )), + retry_after: None, + origin: CodexErrorOrigin::Http, + } +} + +fn ensure_pending_http_buffer_budget( + pending_event_bytes: usize, + provisional_tool_bytes: usize, +) -> Result<(), CodexError> { + if pending_event_bytes.saturating_add(provisional_tool_bytes) > MAX_PENDING_HTTP_EVENT_BYTES { + return Err(pending_http_event_buffer_error("aggregate bytes")); + } + Ok(()) +} + fn codex_event_failure_error(failure: super::events::CodexEventFailure) -> CodexError { CodexError { status: failure.status, @@ -2585,7 +3005,9 @@ fn log_http_stream_end( fields.insert("error".to_string(), serde_json::json!(error)); } match message { - "codex_http_stream_completed" => log.info(message, Some(fields)), + "codex_http_stream_completed" | "codex_http_stream_local_finish" => { + log.info(message, Some(fields)) + } _ => log.warn(message, Some(fields)), } } @@ -3433,6 +3855,259 @@ mod tests { ))); } + #[test] + fn pending_http_event_buffer_caps_retained_read_metadata() { + let mut tools = PendingHttpProvisionalTools::default(); + for index in 0..4 { + tools.calls.push(PendingHttpProvisionalFunctionCall { + output_index: Some(index), + item_id: None, + call_id: Some(format!("call_{index}")), + name: Some("Read".to_string()), + read_args: "x".repeat(MAX_PENDING_HTTP_EVENT_BYTES / 4), + }); + } + let error = ensure_pending_http_buffer_budget(0, tools.buffered_bytes()).unwrap_err(); + + assert_eq!( + error.message, + "Codex pending HTTP event buffer exceeded its aggregate bytes limit" + ); + assert_eq!(error.detail.as_deref(), Some(error.message.as_str())); + assert_eq!(error.status, 413); + assert!(!retryable_http_stream_error(&error)); + } + + #[test] + fn pending_http_event_buffer_accepts_more_than_legacy_event_count() { + let mut pending = PendingHttpEvents::default(); + for index in 0..2_048 { + pending + .push(serde_json::json!({"type":"x","index":index}), 0) + .unwrap(); + } + assert_eq!(pending.event_bytes, 2_048 * MIN_PENDING_HTTP_EVENT_BYTES); + assert_eq!(pending.events.len(), 2_048); + } + + #[test] + fn provisional_tool_calls_tolerate_missing_and_ambiguous_aliases() { + let mut tools = PendingHttpProvisionalTools::default(); + let mut first_added = serde_json::json!({ + "type":"response.output_item.added", + "item":{"type":"function_call","call_id":"call_1","name":"Bash"} + }); + tools.observe(&mut first_added).unwrap(); + let mut second_added = serde_json::json!({ + "type":"response.output_item.added", + "item":{"type":"function_call","call_id":"call_2","name":"Bash"} + }); + tools.observe(&mut second_added).unwrap(); + let mut ambiguous_delta = serde_json::json!({ + "type":"response.function_call_arguments.delta", + "delta":"{\"command\":" + }); + let event = tools.observe(&mut ambiguous_delta).unwrap(); + assert!(matches!( + event, + PendingHttpProvisionalToolEvent::Provisional { + stalled_read_call: None + } + )); + assert_eq!(tools.calls.len(), 2); + } + + #[test] + fn provisional_tool_conflicting_alias_does_not_fall_back_to_sole_call() { + let mut tools = PendingHttpProvisionalTools::default(); + let mut added = serde_json::json!({ + "type":"response.output_item.added", + "output_index":0, + "item":{"type":"function_call","call_id":"call_read_0","name":"Read"} + }); + tools.observe(&mut added).unwrap(); + + let mut conflicting_delta = serde_json::json!({ + "type":"response.function_call_arguments.delta", + "output_index":1, + "delta":format!("{{\"file_path\":\"/tmp/conflicting\"}}{}", " ".repeat(1_024)) + }); + let event = tools.observe(&mut conflicting_delta).unwrap(); + + assert!(matches!( + event, + PendingHttpProvisionalToolEvent::Provisional { + stalled_read_call: None + } + )); + assert!(tools.calls[0].read_args.is_empty()); + } + + #[test] + fn provisional_read_with_whitespace_call_id_cannot_locally_finish() { + let mut tools = PendingHttpProvisionalTools::default(); + let mut added = serde_json::json!({ + "type":"response.output_item.added", + "output_index":0, + "item":{"type":"function_call","call_id":" \t ","name":"Read"} + }); + tools.observe(&mut added).unwrap(); + let mut repair_delta = serde_json::json!({ + "type":"response.function_call_arguments.delta", + "output_index":0, + "delta":format!("{{\"file_path\":\"/tmp/blank-id\"}}{}", " ".repeat(1_024)) + }); + let event = tools.observe(&mut repair_delta).unwrap(); + let PendingHttpProvisionalToolEvent::Provisional { + stalled_read_call: Some(call_index), + } = event + else { + panic!("stalled Read delta was not observed"); + }; + + assert!( + tools + .take_stalled_read(call_index, &PendingHttpEvents::default()) + .is_none() + ); + assert_eq!(tools.calls.len(), 1); + } + + #[test] + fn completed_function_call_normalizes_empty_args_and_preserves_valid_objects() { + let mut tools = PendingHttpProvisionalTools::default(); + for arguments in ["", " \t\n "] { + let mut done = serde_json::json!({ + "type":"response.output_item.done", + "item":{ + "type":"function_call", + "call_id":"call_empty", + "name":"Bash", + "arguments":arguments + } + }); + assert!(matches!( + tools.observe(&mut done).unwrap(), + PendingHttpProvisionalToolEvent::AuthoritativeDone + )); + assert_eq!( + done.pointer("/item/arguments") + .and_then(serde_json::Value::as_str), + Some("{}") + ); + } + + let arguments = r#" { "nested": {"items": [1, true, null]} } "#; + let mut done = serde_json::json!({ + "type":"response.output_item.done", + "item":{ + "type":"function_call", + "call_id":"call_nested", + "name":"Bash", + "arguments":arguments + } + }); + assert!(matches!( + tools.observe(&mut done).unwrap(), + PendingHttpProvisionalToolEvent::AuthoritativeDone + )); + assert_eq!( + done.pointer("/item/arguments") + .and_then(serde_json::Value::as_str), + Some(arguments) + ); + } + + #[test] + fn completed_function_call_rejects_invalid_argument_shapes_before_commit() { + for arguments in ["{", "null", "[]", r#""string""#, "1", "false"] { + let mut tools = PendingHttpProvisionalTools::default(); + let mut done = serde_json::json!({ + "type":"response.output_item.done", + "item":{ + "type":"function_call", + "call_id":"call_invalid", + "name":"Bash", + "arguments":arguments + } + }); + + let error = tools.observe(&mut done).unwrap_err(); + assert_eq!(error.status, 0, "arguments={arguments:?}"); + assert_eq!( + error.detail.as_deref(), + Some("http_response_sse_protocol"), + "arguments={arguments:?}" + ); + let expected_message = if arguments == "{" { + "arguments are not valid JSON" + } else { + "arguments must be a JSON object" + }; + assert!( + error.message.contains(expected_message), + "arguments={arguments:?}: {error:?}" + ); + assert!( + !retryable_http_stream_error(&error), + "arguments={arguments:?}: {error:?}" + ); + } + } + + #[test] + fn completed_function_call_requires_nonblank_identifiers_and_string_arguments() { + for (field, value) in [ + ("call_id", serde_json::json!("")), + ("call_id", serde_json::json!(" \t")), + ("name", serde_json::json!("")), + ("name", serde_json::json!(" \t")), + ("arguments", serde_json::json!(null)), + ("arguments", serde_json::json!({})), + ] { + let mut item = serde_json::json!({ + "type":"function_call", + "call_id":"call_required", + "name":"Bash", + "arguments":"{}" + }); + item[field] = value; + let mut done = serde_json::json!({ + "type":"response.output_item.done", + "item":item + }); + + let error = PendingHttpProvisionalTools::default() + .observe(&mut done) + .unwrap_err(); + assert_eq!(error.detail.as_deref(), Some("http_response_sse_protocol")); + assert!(!retryable_http_stream_error(&error)); + } + } + + #[test] + fn completed_function_call_argument_limit_is_non_retryable() { + let mut done = serde_json::json!({ + "type":"response.output_item.done", + "item":{ + "type":"function_call", + "call_id":"call_too_large", + "name":"Bash", + "arguments":"x".repeat(super::super::translate::live_stream::BUFFERED_TOOL_MAX_ARGS_BYTES + 1) + } + }); + let error = PendingHttpProvisionalTools::default() + .observe(&mut done) + .unwrap_err(); + + assert_eq!(error.status, 413); + assert_eq!( + error.message, + "Buffered Bash tool arguments exceeded safe limits" + ); + assert!(!retryable_http_stream_error(&error)); + } + #[tokio::test] async fn http_stream_forwards_event_before_terminal_body() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); @@ -5002,6 +5677,35 @@ mod tests { ); } + #[test] + fn stalled_read_probe_does_not_record_offset_rewrite() { + let call_id = "call_pending_probe_offset"; + let mut tools = PendingHttpProvisionalTools::default(); + let mut added = serde_json::json!({ + "type":"response.output_item.added", + "output_index":0, + "item":{"type":"function_call","call_id":call_id,"name":"Read"} + }); + tools.observe(&mut added).unwrap(); + let mut delta = serde_json::json!({ + "type":"response.function_call_arguments.delta", + "output_index":0, + "delta":format!( + "{{\"file_path\":\"/tmp/probe\",\"offset\":1200000{}", + " ".repeat(1_024) + ) + }); + let observation = tools.observe(&mut delta).unwrap(); + + assert!(matches!( + observation, + PendingHttpProvisionalToolEvent::Provisional { + stalled_read_call: Some(0) + } + )); + assert!(super::super::translate::read_rewrite::read_offset_rewrite(call_id).is_none()); + } + #[test] fn codex_headers_include_responses_lite_when_requested() { let auth = StoredAuth { diff --git a/src/providers/codex/mod.rs b/src/providers/codex/mod.rs index c09f9302..af5ad325 100644 --- a/src/providers/codex/mod.rs +++ b/src/providers/codex/mod.rs @@ -868,6 +868,14 @@ async fn live_stream_response_once( let payload = match item { Ok(payload) => payload, Err(err) => { + if err.origin == client::CodexErrorOrigin::Http { + abort_request_state( + ctx.session_id.as_deref(), + &request_continuation, + compaction.attempt, + ); + return LiveStreamStart::Response(map_codex_error_to_response(&err)); + } if retryable_live_start_codex_error(&err) { return provider_retry(&upstream_events, err); } @@ -1354,6 +1362,10 @@ fn codex_event_failure_error( fn codex_stream_error_type(err: &client::CodexError) -> &'static str { match err.status { + 400 | 422 => "invalid_request_error", + 401 => "authentication_error", + 403 => "permission_error", + 413 => "request_too_large", 429 => "rate_limit_error", 529 => "overloaded_error", _ if codex_error_message(err) @@ -1425,6 +1437,30 @@ fn map_codex_error_to_response(err: &client::CodexError) -> Response { "permission_error", err.detail.as_deref().unwrap_or("Permission denied"), ), + 400 | 422 => { + let response = json_error( + StatusCode::from_u16(err.status).unwrap_or(StatusCode::BAD_REQUEST), + "invalid_request_error", + codex_error_message(err), + ); + if let Some(retry_after) = err.retry_after.as_deref() { + ([(http::header::RETRY_AFTER, retry_after)], response).into_response() + } else { + response + } + } + 413 => { + let response = json_error( + StatusCode::PAYLOAD_TOO_LARGE, + "request_too_large", + codex_error_message(err), + ); + if let Some(retry_after) = err.retry_after.as_deref() { + ([(http::header::RETRY_AFTER, retry_after)], response).into_response() + } else { + response + } + } 429 => { let response = json_error( StatusCode::TOO_MANY_REQUESTS, @@ -2053,6 +2089,57 @@ mod tests { ); } + #[tokio::test] + async fn codex_http_error_mapping_preserves_typed_client_contract() { + for (status, expected_status, expected_type) in [ + (400, StatusCode::BAD_REQUEST, "invalid_request_error"), + ( + 422, + StatusCode::UNPROCESSABLE_ENTITY, + "invalid_request_error", + ), + (413, StatusCode::PAYLOAD_TOO_LARGE, "request_too_large"), + ] { + let err = client::CodexError { + status, + message: format!("status {status}"), + detail: Some(format!("detail {status}")), + retry_after: None, + origin: client::CodexErrorOrigin::Http, + }; + let response = map_codex_error_to_response(&err); + assert_eq!(response.status(), expected_status); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let body: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(body["error"]["type"], expected_type); + assert_eq!(body["error"]["message"], format!("detail {status}")); + } + } + + #[test] + fn codex_stream_error_type_preserves_typed_client_contract() { + for (status, expected_type) in [ + (400, "invalid_request_error"), + (422, "invalid_request_error"), + (401, "authentication_error"), + (403, "permission_error"), + (413, "request_too_large"), + (429, "rate_limit_error"), + (529, "overloaded_error"), + ] { + let err = client::CodexError { + status, + message: format!("status {status}"), + detail: Some(format!("detail {status}")), + retry_after: None, + origin: client::CodexErrorOrigin::Http, + }; + assert_eq!(codex_stream_error_type(&err), expected_type); + } + } + #[tokio::test] async fn statusless_codex_error_returns_source_message() { let err = client::CodexError { diff --git a/src/providers/codex/translate/live_stream.rs b/src/providers/codex/translate/live_stream.rs index d8e868f9..659cf74e 100644 --- a/src/providers/codex/translate/live_stream.rs +++ b/src/providers/codex/translate/live_stream.rs @@ -7,14 +7,101 @@ use crate::providers::codex::events::{ use crate::traffic::TrafficCapture; use super::IncompleteResponsePolicy; -use super::read_rewrite::sanitize_read_args; +use super::read_rewrite::{repair_whitespace_stalled_read_args, sanitize_read_args}; use super::reasoning_signature::{PendingReasoning, encode_reasoning_signature}; use super::reducer::{ CodexUsage, STOP_END_TURN, STOP_MAX_TOKENS, STOP_TOOL_USE, map_codex_usage_to_anthropic, }; -const BUFFERED_READ_REPAIR_TRAILING_WHITESPACE_BYTES: usize = 1_024; -const BUFFERED_TOOL_MAX_ARGS_BYTES: usize = 5_000_000; +pub(crate) const BUFFERED_TOOL_MAX_ARGS_BYTES: usize = 5_000_000; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct CompletedFunctionCall { + pub call_id: String, + pub name: String, + pub arguments: String, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum CompletedFunctionCallValidationError { + Protocol(String), + TooLarge { tool: String }, +} + +impl std::fmt::Display for CompletedFunctionCallValidationError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Protocol(message) => formatter.write_str(message), + Self::TooLarge { tool } => { + write!( + formatter, + "Buffered {tool} tool arguments exceeded safe limits" + ) + } + } + } +} + +pub(crate) fn validate_completed_function_call( + item: &serde_json::Value, +) -> Result { + let call_id = item + .get("call_id") + .and_then(serde_json::Value::as_str) + .filter(|call_id| !call_id.trim().is_empty()) + .ok_or_else(|| { + CompletedFunctionCallValidationError::Protocol( + "completed function call is missing call_id".to_string(), + ) + })?; + let name = item + .get("name") + .and_then(serde_json::Value::as_str) + .filter(|name| !name.trim().is_empty()) + .ok_or_else(|| { + CompletedFunctionCallValidationError::Protocol( + "completed function call is missing name".to_string(), + ) + })?; + let arguments = item + .get("arguments") + .and_then(serde_json::Value::as_str) + .ok_or_else(|| { + CompletedFunctionCallValidationError::Protocol( + "completed function call is missing string arguments".to_string(), + ) + })?; + + if arguments.len() > BUFFERED_TOOL_MAX_ARGS_BYTES { + return Err(CompletedFunctionCallValidationError::TooLarge { + tool: name.to_string(), + }); + } + if arguments.trim().is_empty() { + return Ok(CompletedFunctionCall { + call_id: call_id.to_string(), + name: name.to_string(), + arguments: "{}".to_string(), + }); + } + + let value = serde_json::from_str::(arguments).map_err(|_| { + CompletedFunctionCallValidationError::Protocol( + "completed function call arguments are not valid JSON".to_string(), + ) + })?; + if !value.is_object() { + return Err(CompletedFunctionCallValidationError::Protocol( + "completed function call arguments must be a JSON object".to_string(), + )); + } + + Ok(CompletedFunctionCall { + call_id: call_id.to_string(), + name: name.to_string(), + arguments: arguments.to_string(), + }) +} enum LiveBlock { Text { @@ -179,7 +266,7 @@ impl LiveStreamTranslator { self.tool_arguments_done(payload); } "response.output_item.done" => { - self.output_item_done(payload, traffic, &mut out); + self.output_item_done(payload, traffic, &mut out)?; } "response.completed" | "response.incomplete" | "response.done" => { self.finish(payload, traffic, &mut out); @@ -649,8 +736,8 @@ impl LiveStreamTranslator { payload: &serde_json::Value, traffic: Option<&TrafficCapture>, out: &mut Vec, - ) { - let output_index = output_index(payload); + ) -> Result<(), String> { + let mut output_index = output_index(payload); if let Some(item) = payload .get("item") .and_then(|item| item.get("type")) @@ -669,7 +756,7 @@ impl LiveStreamTranslator { if !had_active_summary { self.emit_signature_only_reasoning(output_index, traffic, out); } - return; + return Ok(()); } if payload @@ -692,11 +779,28 @@ impl LiveStreamTranslator { id: super::web_search_compat::server_tool_use_id_from_codex_web_search_id(raw_id), query: web_search_query(item), }); - return; + return Ok(()); } + let completed_function_call = if payload + .pointer("/item/type") + .and_then(|value| value.as_str()) + == Some("function_call") + { + let completed = validate_completed_function_call(&payload["item"]) + .map_err(|error| error.to_string())?; + let Some(resolved_output_index) = self.open_tool_output_index_for_done(payload) else { + self.emit_completed_function_call(&completed, traffic, out); + return Ok(()); + }; + output_index = resolved_output_index; + Some(completed) + } else { + None + }; + let Some(mut state) = self.blocks_by_output_index.remove(&output_index) else { - return; + return Ok(()); }; match &mut state { @@ -730,11 +834,9 @@ impl LiveStreamTranslator { .. } => { self.semantic_output_started = true; - if let Some(final_args) = payload - .get("item") - .and_then(|item| item.get("arguments")) - .and_then(|v| v.as_str()) - .filter(|s| !s.is_empty()) + if let Some(final_args) = completed_function_call + .as_ref() + .map(|completed| completed.arguments.as_str()) && (args_accum.is_empty() || (!*had_delta && !*emitted_args)) { *args_accum = final_args.to_string(); @@ -769,6 +871,94 @@ impl LiveStreamTranslator { ); } } + Ok(()) + } + + fn emit_completed_function_call( + &mut self, + completed: &CompletedFunctionCall, + traffic: Option<&TrafficCapture>, + out: &mut Vec, + ) { + let arguments = sanitize_read_args( + &completed.name, + &completed.arguments, + Some(&completed.call_id), + ); + + self.close_thinking(traffic, out); + self.saw_tool_use = true; + self.semantic_output_started = true; + let index = self.anthropic_index; + self.anthropic_index += 1; + self.ensure_message_start(traffic, out); + self.emit( + traffic, + out, + "content_block_start", + &serde_json::json!({ + "type": "content_block_start", + "index": index, + "content_block": { + "type": "tool_use", + "id": completed.call_id, + "name": completed.name, + "input": {} + } + }), + ); + self.emit( + traffic, + out, + "content_block_delta", + &serde_json::json!({ + "type": "content_block_delta", + "index": index, + "delta": { + "type": "input_json_delta", + "partial_json": arguments + } + }), + ); + self.emit( + traffic, + out, + "content_block_stop", + &serde_json::json!({ + "type": "content_block_stop", + "index": index, + }), + ); + } + + fn open_tool_output_index_for_done(&self, payload: &serde_json::Value) -> Option { + let call_id = payload + .pointer("/item/call_id") + .and_then(serde_json::Value::as_str)?; + if let Some(output_index) = payload + .get("output_index") + .and_then(serde_json::Value::as_u64) + .and_then(|value| usize::try_from(value).ok()) + && matches!( + self.blocks_by_output_index.get(&output_index), + Some(LiveBlock::Tool { call_id: open_call_id, .. }) if open_call_id == call_id + ) + { + return Some(output_index); + } + + let mut matching = + self.blocks_by_output_index + .iter() + .filter_map(|(output_index, block)| { + matches!( + block, + LiveBlock::Tool { call_id: open_call_id, .. } if open_call_id == call_id + ) + .then_some(*output_index) + }); + let output_index = matching.next()?; + matching.next().is_none().then_some(output_index) } fn web_search_annotation(&mut self, payload: &serde_json::Value) { @@ -1114,74 +1304,6 @@ fn parse_codex_usage(response: &serde_json::Value) -> CodexUsage { } } -fn repair_whitespace_stalled_read_args( - name: &str, - args: &str, - call_id: Option<&str>, -) -> Option { - if name != "Read" { - return None; - } - let trimmed = args.trim_end(); - let trailing_whitespace = args.len().saturating_sub(trimmed.len()); - if trailing_whitespace < BUFFERED_READ_REPAIR_TRAILING_WHITESPACE_BYTES { - return None; - } - parse_read_args_candidate(trimmed, call_id).or_else(|| { - let with_brace = format!("{trimmed}}}"); - parse_read_args_candidate(&with_brace, call_id) - }) -} - -fn parse_read_args_candidate(args: &str, call_id: Option<&str>) -> Option { - let parsed: serde_json::Value = serde_json::from_str(args).ok()?; - if !is_valid_read_args(&parsed) { - return None; - } - Some(sanitize_read_args( - "Read", - &serde_json::to_string(&parsed).ok()?, - call_id, - )) -} - -fn is_valid_read_args(value: &serde_json::Value) -> bool { - let Some(obj) = value.as_object() else { - return false; - }; - for key in obj.keys() { - if !matches!(key.as_str(), "file_path" | "offset" | "limit" | "pages") { - return false; - } - } - let Some(file_path) = obj.get("file_path").and_then(|v| v.as_str()) else { - return false; - }; - if file_path.is_empty() { - return false; - } - if let Some(offset) = obj.get("offset").and_then(|v| v.as_i64()) - && offset < 0 - { - return false; - } - if let Some(limit) = obj.get("limit").and_then(|v| v.as_i64()) - && limit <= 0 - { - return false; - } - if obj.get("offset").is_some_and(|v| !v.is_i64()) { - return false; - } - if obj.get("limit").is_some_and(|v| !v.is_i64()) { - return false; - } - if obj.get("pages").is_some_and(|v| !v.is_string()) { - return false; - } - true -} - #[cfg(test)] mod tests { use super::*; @@ -1449,6 +1571,233 @@ mod tests { assert!(web_search.has_semantic_output()); } + #[test] + fn standalone_function_done_does_not_close_a_different_open_tool() { + let mut translator = LiveStreamTranslator::new("msg_1", "gpt-5.5"); + let mut rendered = Vec::new(); + for event in [ + json!({ + "type":"response.output_item.added", + "output_index":0, + "item":{"type":"function_call","call_id":"call_a","name":"Bash"} + }), + json!({ + "type":"response.function_call_arguments.delta", + "output_index":0, + "delta":"{\"command\":\"echo a\"}" + }), + ] { + rendered.extend(translator.accept(&event, None).unwrap()); + } + + let standalone_b = translator + .accept( + &json!({ + "type":"response.output_item.done", + "item":{ + "type":"function_call", + "call_id":"call_b", + "name":"Bash", + "arguments":"{\"command\":\"echo b\"}" + } + }), + None, + ) + .unwrap(); + assert!(String::from_utf8_lossy(&standalone_b).contains("call_b")); + assert!(translator.blocks_by_output_index.contains_key(&0)); + rendered.extend(standalone_b); + + rendered.extend( + translator + .accept( + &json!({ + "type":"response.output_item.done", + "output_index":0, + "item":{ + "type":"function_call", + "call_id":"call_a", + "name":"Bash", + "arguments":"{\"command\":\"echo a\"}" + } + }), + None, + ) + .unwrap(), + ); + assert!(!translator.blocks_by_output_index.contains_key(&0)); + + let rendered = String::from_utf8(rendered).unwrap(); + assert_eq!(rendered.matches("call_a").count(), 1); + assert_eq!(rendered.matches("call_b").count(), 1); + assert_eq!(rendered.matches("event: content_block_stop").count(), 2); + } + + #[test] + fn completed_function_call_rejects_invalid_authoritative_fields_before_emission() { + for (item, expected_error) in [ + ( + json!({"type":"function_call","name":"Bash","arguments":"{}"}), + "missing call_id", + ), + ( + json!({"type":"function_call","call_id":"call_missing_name","arguments":"{}"}), + "missing name", + ), + ( + json!({"type":"function_call","call_id":"call_missing_arguments","name":"Bash"}), + "missing string arguments", + ), + ( + json!({"type":"function_call","call_id":"call_malformed","name":"Bash","arguments":"{"}), + "arguments are not valid JSON", + ), + ( + json!({"type":"function_call","call_id":"call_nonobject","name":"Bash","arguments":"[]"}), + "arguments must be a JSON object", + ), + ] { + let mut translator = LiveStreamTranslator::new("msg_1", "gpt-5.5"); + let error = translator + .accept( + &json!({"type":"response.output_item.done","item":item}), + None, + ) + .unwrap_err(); + + assert!(error.contains(expected_error), "error: {error}"); + assert!(!translator.saw_tool_use); + assert!(!translator.has_semantic_output()); + assert!(!translator.message_started); + assert!(translator.blocks_by_output_index.is_empty()); + } + } + + #[test] + fn completed_function_call_rejects_oversized_arguments_before_emission() { + let mut translator = LiveStreamTranslator::new("msg_1", "gpt-5.5"); + let error = translator + .accept( + &json!({ + "type":"response.output_item.done", + "item":{ + "type":"function_call", + "call_id":"call_oversized", + "name":"Bash", + "arguments":"x".repeat(BUFFERED_TOOL_MAX_ARGS_BYTES + 1) + } + }), + None, + ) + .unwrap_err(); + + assert_eq!(error, "Buffered Bash tool arguments exceeded safe limits"); + assert!(!translator.saw_tool_use); + assert!(!translator.has_semantic_output()); + assert!(!translator.message_started); + assert!(translator.blocks_by_output_index.is_empty()); + } + + #[test] + fn completed_function_call_normalizes_blank_arguments_before_emission() { + let mut translator = LiveStreamTranslator::new("msg_1", "gpt-5.5"); + let out = translator + .accept( + &json!({ + "type":"response.output_item.done", + "item":{ + "type":"function_call", + "call_id":"call_blank", + "name":"Bash", + "arguments":" \t\n " + } + }), + None, + ) + .unwrap(); + let out = String::from_utf8(out).unwrap(); + + assert!(out.contains("call_blank")); + assert!(out.contains(r#""partial_json":"{}""#)); + assert!(out.contains("event: content_block_stop")); + } + + #[test] + fn matched_completed_function_call_propagates_invalid_authoritative_arguments() { + let mut translator = LiveStreamTranslator::new("msg_1", "gpt-5.5"); + translator + .accept( + &json!({ + "type":"response.output_item.added", + "output_index":0, + "item":{"type":"function_call","call_id":"call_matched","name":"Bash"} + }), + None, + ) + .unwrap(); + + let error = translator + .accept( + &json!({ + "type":"response.output_item.done", + "output_index":0, + "item":{ + "type":"function_call", + "call_id":"call_matched", + "name":"Bash", + "arguments":"{" + } + }), + None, + ) + .unwrap_err(); + + assert!(error.contains("arguments are not valid JSON")); + assert!(translator.blocks_by_output_index.contains_key(&0)); + } + + #[test] + fn opencode_style_added_then_done_function_call_remains_valid() { + let mut translator = LiveStreamTranslator::new("msg_1", "gpt-5.6-luna"); + let added = translator + .accept( + &json!({ + "type":"response.output_item.added", + "output_index":0, + "item":{"type":"function_call","call_id":"call_opencode","name":"Bash"} + }), + None, + ) + .unwrap(); + let done = translator + .accept( + &json!({ + "type":"response.output_item.done", + "output_index":0, + "item":{ + "type":"function_call", + "call_id":"call_opencode", + "name":"Bash", + "arguments":"{\"command\":\"echo opencode\"}" + } + }), + None, + ) + .unwrap(); + let completed = translator + .accept( + &json!({"type":"response.completed","response":{"status":"completed","usage":{}}}), + None, + ) + .unwrap(); + let out = String::from_utf8([added, done, completed].concat()).unwrap(); + + assert!(out.contains("call_opencode")); + assert!(out.contains("echo opencode")); + assert_eq!(out.matches("event: content_block_stop").count(), 1); + assert!(out.contains("event: message_stop")); + } + #[test] fn buffers_read_tool_args_until_done() { let out = render(vec![ @@ -1493,7 +1842,7 @@ mod tests { &json!({ "type": "response.output_item.added", "output_index": 0, - "item": {"type":"function_call","call_id":"call_1","name":"Read"} + "item": {"type":"function_call","call_id":"call_stalled_offset_commit","name":"Read"} }), None, ) @@ -1505,7 +1854,7 @@ mod tests { &json!({ "type": "response.function_call_arguments.delta", "output_index": 0, - "delta": format!("{{\"file_path\":\"/tmp/a\",\"pages\":\"\"{}", " ".repeat(1024)) + "delta": format!("{{\"file_path\":\"/tmp/a\",\"offset\":1200000{}", " ".repeat(1024)) }), None, ) @@ -1516,6 +1865,12 @@ mod tests { assert!(rendered.contains(r#""stop_reason":"tool_use""#)); assert!(rendered.contains("message_stop")); assert!(translator.is_finished()); + assert_eq!( + super::super::read_rewrite::read_offset_rewrite("call_stalled_offset_commit") + .as_ref() + .map(|rewrite| rewrite.offset), + Some(1_200_000) + ); } #[test] diff --git a/src/providers/codex/translate/read_rewrite.rs b/src/providers/codex/translate/read_rewrite.rs index dd5e397d..7bb0fd97 100644 --- a/src/providers/codex/translate/read_rewrite.rs +++ b/src/providers/codex/translate/read_rewrite.rs @@ -6,6 +6,7 @@ use serde_json::Value; const MAX_REWRITE_NOTES: usize = 4_096; const READ_OFFSET_REWRITE_THRESHOLD: i64 = 1_000_000; +const BUFFERED_READ_REPAIR_TRAILING_WHITESPACE_BYTES: usize = 1_024; #[derive(Debug, Clone, PartialEq, Eq)] pub struct ReadOffsetRewrite { @@ -74,6 +75,74 @@ pub fn sanitize_read_args(name: &str, args: &str, call_id: Option<&str>) -> Stri } } +pub(crate) fn repair_whitespace_stalled_read_args( + name: &str, + args: &str, + call_id: Option<&str>, +) -> Option { + if name != "Read" { + return None; + } + let trimmed = args.trim_end(); + let trailing_whitespace = args.len().saturating_sub(trimmed.len()); + if trailing_whitespace < BUFFERED_READ_REPAIR_TRAILING_WHITESPACE_BYTES { + return None; + } + parse_read_args_candidate(trimmed, call_id).or_else(|| { + let with_brace = format!("{trimmed}}}"); + parse_read_args_candidate(&with_brace, call_id) + }) +} + +fn parse_read_args_candidate(args: &str, call_id: Option<&str>) -> Option { + let parsed: Value = serde_json::from_str(args).ok()?; + if !is_valid_read_args(&parsed) { + return None; + } + Some(sanitize_read_args( + "Read", + &serde_json::to_string(&parsed).ok()?, + call_id, + )) +} + +fn is_valid_read_args(value: &Value) -> bool { + let Some(obj) = value.as_object() else { + return false; + }; + for key in obj.keys() { + if !matches!(key.as_str(), "file_path" | "offset" | "limit" | "pages") { + return false; + } + } + let Some(file_path) = obj.get("file_path").and_then(Value::as_str) else { + return false; + }; + if file_path.is_empty() { + return false; + } + if let Some(offset) = obj.get("offset").and_then(Value::as_i64) + && offset < 0 + { + return false; + } + if let Some(limit) = obj.get("limit").and_then(Value::as_i64) + && limit <= 0 + { + return false; + } + if obj.get("offset").is_some_and(|value| !value.is_i64()) { + return false; + } + if obj.get("limit").is_some_and(|value| !value.is_i64()) { + return false; + } + if obj.get("pages").is_some_and(|value| !value.is_string()) { + return false; + } + true +} + pub fn read_offset_rewrite(call_id: &str) -> Option { READ_OFFSET_REWRITES .lock() @@ -136,4 +205,18 @@ mod tests { assert_eq!(parsed.get("offset").and_then(|v| v.as_i64()), Some(1_300)); assert!(read_offset_rewrite("call_keep_test").is_none()); } + + #[test] + fn repairs_whitespace_stalled_read_args() { + let repaired = repair_whitespace_stalled_read_args( + "Read", + &format!( + "{{\"file_path\":\"/tmp/a\",\"pages\":\"\"{}", + " ".repeat(1_024) + ), + Some("call_repair_test"), + ); + + assert_eq!(repaired.as_deref(), Some(r#"{"file_path":"/tmp/a"}"#)); + } } diff --git a/tests/smoke_cutover.rs b/tests/smoke_cutover.rs index 12f1622f..05dc0cd4 100644 --- a/tests/smoke_cutover.rs +++ b/tests/smoke_cutover.rs @@ -257,6 +257,45 @@ async fn spawn_retrying_truncated_http_upstream( format!("http://{addr}") } +#[allow(clippy::await_holding_lock)] +async fn collect_http_stream_after_truncated_attempt( + first_body: Vec, + success_body: Vec, +) -> (usize, StatusCode, String) { + 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_retrying_truncated_http_upstream(first_body, success_body, attempts.clone()).await; + let _config_env = EnvGuard::set("CCP_CONFIG_DIR", config.path()); + let _base_url_env = EnvGuard::set("CCP_CODEX_BASE_URL", &upstream); + let _transport_env = EnvGuard::set("CCP_CODEX_TRANSPORT", "http"); + let response = call_messages_body(json!({ + "model": "gpt-5.5", + "max_tokens": 64, + "stream": true, + "messages": [{"role":"user","content":"hello"}] + })) + .await; + + let status = response.status(); + let body = tokio::time::timeout( + Duration::from_secs(4), + axum::body::to_bytes(response.into_body(), usize::MAX), + ) + .await + .expect("retried stream must finish") + .unwrap(); + ( + attempts.load(Ordering::SeqCst), + status, + String::from_utf8(body.to_vec()).unwrap(), + ) +} + #[allow(clippy::await_holding_lock)] async fn assert_codex_http_retries_structural_body_error(first_body: Vec) { let _guard = env_lock(); @@ -1847,6 +1886,1088 @@ async fn smoke_codex_http_retries_body_error_after_structural_tool() { .await; } +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_does_not_retry_after_reasoning_output() { + let first_body = concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_reasoning_failed\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"reasoning\",\"id\":\"reasoning_failed\"}}\n\n", + "data: {\"type\":\"response.reasoning_summary_text.delta\",\"output_index\":0,\"delta\":\"reasoning before reset\"}\n\n", + "data: {\"type\":\"response.reasoning_summary_part.added\",\"output_index\":0}\n\n" + ) + .as_bytes() + .to_vec(); + let (attempts, status, text) = + collect_http_stream_after_truncated_attempt(first_body, Vec::new()).await; + assert_eq!(status, StatusCode::OK, "stream body: {text}"); + assert_eq!(attempts, 1, "stream body: {text}"); + assert!( + text.contains("reasoning before reset"), + "stream body: {text}" + ); + assert!( + !text.contains("reasoning after retry"), + "stream body: {text}" + ); + assert!(!text.contains("answer after retry"), "stream body: {text}"); + assert!(text.contains("event: error"), "stream body: {text}"); + assert_eq!(text.matches("event: message_start").count(), 1); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_does_not_retry_reasoning_held_behind_open_tool() { + let first_body = concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_reasoning_held\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_open\",\"name\":\"Bash\"}}\n\n", + "data: {\"type\":\"response.function_call_arguments.delta\",\"output_index\":0,\"delta\":\"{\\\"command\\\":\\\"echo held\"}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":1,\"item\":{\"type\":\"reasoning\",\"id\":\"reasoning_held\"}}\n\n", + "data: {\"type\":\"response.reasoning_summary_text.delta\",\"output_index\":1,\"delta\":\"reasoning held behind tool\"}\n\n", + "data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"type\":\"overloaded_error\",\"message\":\"overloaded while tool remained open\",\"retry_after\":0}}}\n\n" + ) + .as_bytes() + .to_vec(); + + let (attempts, status, text) = + collect_http_stream_after_truncated_attempt(first_body, Vec::new()).await; + assert_eq!(attempts, 1, "response body: {text}"); + assert_eq!(status, StatusCode::OK, "response body: {text}"); + assert!( + text.contains("reasoning held behind tool"), + "response body: {text}" + ); + assert!(text.contains("overloaded while tool remained open")); + assert!(text.contains("event: error"), "response body: {text}"); + assert!(!text.contains("call_open"), "response body: {text}"); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_preserves_cyber_policy_behind_open_tool_barrier() { + let first_body = concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_policy_held\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_policy_open\",\"name\":\"Bash\"}}\n\n", + "data: {\"type\":\"response.function_call_arguments.delta\",\"output_index\":0,\"delta\":\"{\\\"command\\\":\\\"echo held\"}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":1,\"item\":{\"type\":\"reasoning\",\"id\":\"reasoning_policy_held\"}}\n\n", + "data: {\"type\":\"response.reasoning_summary_text.delta\",\"output_index\":1,\"delta\":\"reasoning before policy result\"}\n\n", + "data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"code\":\"cyber_policy\",\"message\":\"request rejected by policy\"}}}\n\n" + ) + .as_bytes() + .to_vec(); + + let (attempts, status, text) = + collect_http_stream_after_truncated_attempt(first_body, Vec::new()).await; + assert_eq!(attempts, 1, "response body: {text}"); + assert_eq!(status, StatusCode::OK, "response body: {text}"); + assert!(text.contains("reasoning before policy result")); + assert!(text.contains("request rejected by policy")); + assert!(text.contains("event: error"), "response body: {text}"); + assert!(!text.contains("call_policy_open"), "response body: {text}"); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_drops_provisional_tool_after_completed_tool_on_terminal() { + let _guard = env_lock(); + clear_all_continuations_for_tests(); + let config = TempDir::new().unwrap(); + write_auth(config.path(), "codex"); + + let upstream = spawn_http_upstream(|_body: Value| { + concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_open_terminal\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_open_terminal\",\"name\":\"Bash\"}}\n\n", + "data: {\"type\":\"response.function_call_arguments.delta\",\"output_index\":0,\"delta\":\"{\\\"command\\\":\\\"echo incomplete\"}\n\n", + "data: {\"type\":\"response.output_item.done\",\"item\":{\"type\":\"function_call\",\"call_id\":\"call_complete_terminal\",\"name\":\"Bash\",\"arguments\":\"{\\\"command\\\":\\\"echo complete\\\"}\"}}\n\n", + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_open_terminal\",\"status\":\"completed\",\"usage\":{}}}\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 = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let text = String::from_utf8(body.to_vec()).unwrap(); + assert!( + text.contains("call_complete_terminal"), + "response body: {text}" + ); + assert!( + text.contains("event: message_stop"), + "response body: {text}" + ); + assert!( + !text.contains("call_open_terminal"), + "response body: {text}" + ); + assert!(!text.contains("echo incomplete"), "response body: {text}"); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_finishes_reasoning_when_terminal_follows_partial_tool() { + let _guard = env_lock(); + clear_all_continuations_for_tests(); + let config = TempDir::new().unwrap(); + write_auth(config.path(), "codex"); + + let upstream = spawn_http_upstream(|_body: Value| { + concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_reasoning_then_tool\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"reasoning\",\"id\":\"reasoning_then_tool\"}}\n\n", + "data: {\"type\":\"response.reasoning_summary_text.delta\",\"output_index\":0,\"delta\":\"visible reasoning\"}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":1,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_after_reasoning\",\"name\":\"Bash\"}}\n\n", + "data: {\"type\":\"response.function_call_arguments.delta\",\"output_index\":1,\"delta\":\"{\\\"command\\\":\\\"echo partial\"}\n\n", + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_reasoning_then_tool\",\"status\":\"completed\",\"usage\":{}}}\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 = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let text = String::from_utf8(body.to_vec()).unwrap(); + assert!(text.contains("visible reasoning"), "stream body: {text}"); + assert!(!text.contains("echo partial"), "stream body: {text}"); + assert!(!text.contains("event: error"), "stream body: {text}"); + assert_eq!(text.matches("event: message_stop").count(), 1); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_does_not_retry_after_stalled_read_repair() { + let repair_event = serde_json::json!({ + "type": "response.function_call_arguments.delta", + "output_index": 0, + "delta": format!("{{\"file_path\":\"/tmp/stalled\"}}{}", " ".repeat(1_024)) + }); + let mut first_body = concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_stalled_read\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_stalled_read\",\"name\":\"Read\"}}\n\n" + ) + .as_bytes() + .to_vec(); + first_body.extend_from_slice(format!("data: {repair_event}\n\n").as_bytes()); + + let (attempts, status, text) = + collect_http_stream_after_truncated_attempt(first_body, Vec::new()).await; + assert_eq!(status, StatusCode::OK, "stream body: {text}"); + assert_eq!(attempts, 1, "stream body: {text}"); + assert!(text.contains("call_stalled_read"), "stream body: {text}"); + assert!(text.contains("/tmp/stalled"), "stream body: {text}"); + assert!(!text.contains("event: error"), "stream body: {text}"); + assert_eq!(text.matches("event: message_stop").count(), 1); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_conflicting_read_delta_retries_without_local_finish_or_leakage() { + let conflicting_delta = serde_json::json!({ + "type": "response.function_call_arguments.delta", + "output_index": 1, + "delta": format!("{{\"file_path\":\"/tmp/conflicting\"}}{}", " ".repeat(1_024)) + }); + let mut first_body = concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_conflicting_delta\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_read_0\",\"name\":\"Read\"}}\n\n" + ) + .as_bytes() + .to_vec(); + first_body.extend_from_slice(format!("data: {conflicting_delta}\n\n").as_bytes()); + + let success_body = concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_conflicting_retry\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"message\",\"id\":\"msg_conflicting_retry\"}}\n\n", + "data: {\"type\":\"response.output_text.delta\",\"output_index\":0,\"delta\":\"retry after conflicting read delta\"}\n\n", + "data: {\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{\"type\":\"message\"}}\n\n", + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_conflicting_retry\",\"status\":\"completed\",\"usage\":{\"input_tokens\":5,\"output_tokens\":2}}}\n\n" + ) + .as_bytes() + .to_vec(); + + let (attempts, status, text) = + collect_http_stream_after_truncated_attempt(first_body, success_body).await; + assert_eq!(attempts, 2, "stream body: {text}"); + assert_eq!(status, StatusCode::OK, "stream body: {text}"); + assert!( + text.contains("retry after conflicting read delta"), + "stream body: {text}" + ); + assert!(!text.contains("call_read_0"), "stream body: {text}"); + assert!(!text.contains("/tmp/conflicting"), "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); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_whitespace_read_call_id_retries_without_local_finish_or_leakage() { + let repair_delta = serde_json::json!({ + "type": "response.function_call_arguments.delta", + "output_index": 0, + "delta": format!("{{\"file_path\":\"/tmp/blank-id\"}}{}", " ".repeat(1_024)) + }); + let added = serde_json::json!({ + "type":"response.output_item.added", + "output_index":0, + "item":{"type":"function_call","call_id":" \t ","name":"Read"} + }); + let mut first_body = + format!("data: {{\"type\":\"response.created\",\"response\":{{\"id\":\"resp_blank_id\"}}}}\n\ndata: {added}\n\n") + .into_bytes(); + first_body.extend_from_slice(format!("data: {repair_delta}\n\n").as_bytes()); + + let success_body = concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_blank_id_retry\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"message\",\"id\":\"msg_blank_id_retry\"}}\n\n", + "data: {\"type\":\"response.output_text.delta\",\"output_index\":0,\"delta\":\"retry after blank read id\"}\n\n", + "data: {\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{\"type\":\"message\"}}\n\n", + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_blank_id_retry\",\"status\":\"completed\",\"usage\":{\"input_tokens\":5,\"output_tokens\":2}}}\n\n" + ) + .as_bytes() + .to_vec(); + + let (attempts, status, text) = + collect_http_stream_after_truncated_attempt(first_body, success_body).await; + assert_eq!(attempts, 2, "stream body: {text}"); + assert_eq!(status, StatusCode::OK, "stream body: {text}"); + assert!( + text.contains("retry after blank read id"), + "stream body: {text}" + ); + assert!(!text.contains("/tmp/blank-id"), "stream body: {text}"); + assert!( + !text.contains(r#""type":"tool_use""#), + "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); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_retries_stalled_read_behind_another_open_tool() { + let repair_event = serde_json::json!({ + "type": "response.function_call_arguments.delta", + "output_index": 1, + "delta": format!("{{\"file_path\":\"/tmp/held\"}}{}", " ".repeat(1_024)) + }); + let mut first_body = concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_repair_held\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_open\",\"name\":\"Bash\"}}\n\n", + "data: {\"type\":\"response.function_call_arguments.delta\",\"output_index\":0,\"delta\":\"{\\\"command\\\":\\\"echo held\"}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":1,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_stalled_held\",\"name\":\"Read\"}}\n\n" + ) + .as_bytes() + .to_vec(); + first_body.extend_from_slice(format!("data: {repair_event}\n\n").as_bytes()); + let success_body = concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_repair_retry\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"message\",\"id\":\"msg_repair_retry\"}}\n\n", + "data: {\"type\":\"response.output_text.delta\",\"output_index\":0,\"delta\":\"retry after held read\"}\n\n", + "data: {\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{\"type\":\"message\"}}\n\n", + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_repair_retry\",\"status\":\"completed\",\"usage\":{\"input_tokens\":5,\"output_tokens\":2}}}\n\n" + ) + .as_bytes() + .to_vec(); + + let (attempts, status, text) = + collect_http_stream_after_truncated_attempt(first_body, success_body).await; + assert_eq!(attempts, 2, "stream body: {text}"); + assert_eq!(status, StatusCode::OK, "stream body: {text}"); + assert!( + text.contains("retry after held read"), + "stream body: {text}" + ); + assert!(!text.contains("call_open"), "stream body: {text}"); + assert!(!text.contains("call_stalled_held"), "stream body: {text}"); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_does_not_locally_finish_stalled_read_with_open_shells() { + let repair_event = serde_json::json!({ + "type": "response.function_call_arguments.delta", + "output_index": 2, + "delta": format!("{{\"file_path\":\"/tmp/shells\"}}{}", " ".repeat(1_024)) + }); + let mut first_body = concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_repair_shells\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"reasoning\",\"id\":\"reasoning_shell\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":1,\"item\":{\"type\":\"message\",\"id\":\"message_shell\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":2,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_stalled_shells\",\"name\":\"Read\"}}\n\n" + ) + .as_bytes() + .to_vec(); + first_body.extend_from_slice(format!("data: {repair_event}\n\n").as_bytes()); + let success_body = concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_shells_retry\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"message\",\"id\":\"msg_shells_retry\"}}\n\n", + "data: {\"type\":\"response.output_text.delta\",\"output_index\":0,\"delta\":\"retry after open shells\"}\n\n", + "data: {\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{\"type\":\"message\"}}\n\n", + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_shells_retry\",\"status\":\"completed\",\"usage\":{\"input_tokens\":5,\"output_tokens\":2}}}\n\n" + ) + .as_bytes() + .to_vec(); + + let (attempts, status, text) = + collect_http_stream_after_truncated_attempt(first_body, success_body).await; + assert_eq!(attempts, 2, "stream body: {text}"); + assert_eq!(status, StatusCode::OK, "stream body: {text}"); + assert!( + text.contains("retry after open shells"), + "stream body: {text}" + ); + assert!(!text.contains("call_stalled_shells"), "stream body: {text}"); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_retries_body_error_mid_tool_arguments_without_leakage() { + let first_body = concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_tool_failed\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_failed\",\"name\":\"Bash\"}}\n\n", + "data: {\"type\":\"response.function_call_arguments.delta\",\"output_index\":0,\"delta\":\"{\\\"command\\\":\\\"echo failed attempt\"}\n\n" + ) + .as_bytes() + .to_vec(); + let success_body = concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_tool_success\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_success\",\"name\":\"Bash\"}}\n\n", + "data: {\"type\":\"response.function_call_arguments.delta\",\"output_index\":0,\"delta\":\"{\\\"command\\\":\\\"echo successful attempt\\\"}\"}\n\n", + "data: {\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_success\",\"name\":\"Bash\",\"arguments\":\"{\\\"command\\\":\\\"echo successful attempt\\\"}\"}}\n\n", + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_tool_success\",\"status\":\"completed\",\"usage\":{\"input_tokens\":5,\"output_tokens\":4}}}\n\n" + ) + .as_bytes() + .to_vec(); + + let (attempts, status, text) = + collect_http_stream_after_truncated_attempt(first_body, success_body).await; + assert_eq!(status, StatusCode::OK, "stream body: {text}"); + assert_eq!(attempts, 2, "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); + assert_eq!(text.matches(r#""type":"tool_use""#).count(), 1); + assert!(text.contains("call_success"), "stream body: {text}"); + assert!( + text.contains("echo successful attempt"), + "stream body: {text}" + ); + assert!(!text.contains("call_failed"), "stream body: {text}"); + assert!(!text.contains("echo failed attempt"), "stream body: {text}"); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_emits_standalone_function_done_without_output_index() { + let _guard = env_lock(); + clear_all_continuations_for_tests(); + let config = TempDir::new().unwrap(); + write_auth(config.path(), "codex"); + + let upstream = spawn_http_upstream(|_body: Value| { + concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_standalone\"}}\n\n", + "data: {\"type\":\"response.output_item.done\",\"item\":{\"type\":\"function_call\",\"call_id\":\"call_standalone\",\"name\":\"Bash\",\"arguments\":\"{\\\"command\\\":\\\"echo standalone\\\"}\"}}\n\n", + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_standalone\",\"status\":\"completed\",\"usage\":{}}}\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 = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let text = String::from_utf8(body.to_vec()).unwrap(); + assert!(text.contains("call_standalone"), "stream body: {text}"); + assert!(text.contains("echo standalone"), "stream body: {text}"); + assert_eq!(text.matches(r#""type":"tool_use""#).count(), 1); + assert_eq!(text.matches("event: message_stop").count(), 1); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_normalizes_empty_authoritative_tool_args() { + let _guard = env_lock(); + clear_all_continuations_for_tests(); + let config = TempDir::new().unwrap(); + let state = 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); + let events = [ + json!({ + "type":"response.output_item.added", + "output_index":0, + "item":{"type":"function_call","call_id":"call_empty","name":"Bash"} + }), + json!({ + "type":"response.output_item.done", + "output_index":0, + "item":{"type":"function_call","call_id":"call_empty","name":"Bash","arguments":""} + }), + json!({ + "type":"response.output_item.done", + "item":{"type":"function_call","call_id":"call_whitespace","name":"Bash","arguments":" \t\n "} + }), + json!({ + "type":"response.completed", + "response":{"id":"resp_empty_args","status":"completed","usage":{}} + }), + ]; + let mut body = Vec::new(); + for event in events { + body.extend_from_slice(format!("data: {event}\n\n").as_bytes()); + } + body + } + }) + .await; + + let _traffic_env = EnvGuard::set("CCP_TRAFFIC_LOG", "1"); + let _state_env = EnvGuard::set("XDG_STATE_HOME", state.path()); + let _config_env = EnvGuard::set("CCP_CONFIG_DIR", config.path()); + let _base_url_env = EnvGuard::set("CCP_CODEX_BASE_URL", &upstream); + let _transport_env = EnvGuard::set("CCP_CODEX_TRANSPORT", "http"); + let response = call_messages_body(json!({ + "model": "gpt-5.5", + "max_tokens": 64, + "stream": true, + "messages": [{"role":"user","content":"hello"}] + })) + .await; + + assert_eq!(attempts.load(Ordering::SeqCst), 1); + assert_eq!(response.status(), StatusCode::OK); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let text = String::from_utf8(body.to_vec()).unwrap(); + assert_eq!( + text.matches(r#""type":"tool_use""#).count(), + 2, + "stream body: {text}" + ); + assert_eq!( + text.matches(r#""partial_json":"{}""#).count(), + 2, + "stream body: {text}" + ); + assert_eq!(text.matches("event: content_block_stop").count(), 2); + assert_eq!(text.matches("event: message_stop").count(), 1); + + let files = traffic_files(state.path()); + let upstream_done_arguments = files + .iter() + .filter(|path| { + path.file_name() + .and_then(|name| name.to_str()) + .is_some_and(|name| name.ends_with("040-upstream-event.json")) + }) + .filter_map(|path| serde_json::from_slice::(&std::fs::read(path).ok()?).ok()) + .filter(|event| event["type"] == "response.output_item.done") + .filter_map(|event| { + event + .pointer("/item/arguments") + .and_then(Value::as_str) + .map(str::to_owned) + }) + .collect::>(); + assert!( + upstream_done_arguments + .iter() + .any(|arguments| arguments.is_empty()) + ); + assert!( + upstream_done_arguments + .iter() + .any(|arguments| arguments == " \t\n ") + ); + + let downstream_normalized = files + .iter() + .filter(|path| { + path.file_name() + .and_then(|name| name.to_str()) + .is_some_and(|name| name.ends_with("050-downstream-event.json")) + }) + .filter_map(|path| serde_json::from_slice::(&std::fs::read(path).ok()?).ok()) + .filter(|event| { + event.pointer("/data/delta/partial_json") == Some(&Value::String("{}".to_string())) + }) + .count(); + assert_eq!(downstream_normalized, 2); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_rejects_invalid_authoritative_tool_before_emission_or_replay() { + 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); + let added = json!({ + "type":"response.output_item.added", + "output_index":0, + "item":{"type":"function_call","call_id":"call_invalid","name":"Bash"} + }); + let done = json!({ + "type":"response.output_item.done", + "output_index":0, + "item":{"type":"function_call","call_id":"call_invalid","name":"Bash","arguments":"{"} + }); + format!("data: {added}\n\ndata: {done}\n\n").into_bytes() + } + }) + .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!(attempts.load(Ordering::SeqCst), 1); + assert_eq!(response.status(), StatusCode::BAD_GATEWAY); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let value: Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(value["error"]["type"], "api_error"); + let text = String::from_utf8(body.to_vec()).unwrap(); + assert!( + text.contains("completed function call arguments are not valid JSON"), + "stream body: {text}" + ); + assert!(!text.contains("call_invalid"), "stream body: {text}"); + assert!( + !text.contains(r#""type":"tool_use""#), + "stream body: {text}" + ); + assert!( + !text.contains("event: content_block_stop"), + "stream body: {text}" + ); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_rejects_oversized_authoritative_tool_args() { + 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); + let arguments = "x".repeat(5_000_001); + let added = json!({ + "type":"response.output_item.added", + "output_index":0, + "item":{"type":"function_call","call_id":"call_too_large","name":"Bash"} + }); + let done = json!({ + "type":"response.output_item.done", + "output_index":0, + "item":{"type":"function_call","call_id":"call_too_large","name":"Bash","arguments":arguments} + }); + format!("data: {added}\n\ndata: {done}\n\n").into_bytes() + } + }) + .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!(attempts.load(Ordering::SeqCst), 1); + assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let value: Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(value["error"]["type"], "request_too_large"); + assert_eq!( + value["error"]["message"], + "Buffered Bash tool arguments exceeded safe limits" + ); + assert!(!String::from_utf8_lossy(&body).contains("call_too_large")); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_emits_multiple_standalone_function_done_in_arrival_order() { + let _guard = env_lock(); + clear_all_continuations_for_tests(); + let config = TempDir::new().unwrap(); + write_auth(config.path(), "codex"); + + let upstream = spawn_http_upstream(|_body: Value| { + concat!( + "data: {\"type\":\"response.output_item.done\",\"item\":{\"type\":\"function_call\",\"call_id\":\"call_first\",\"name\":\"Bash\",\"arguments\":\"{\\\"command\\\":\\\"echo first\\\"}\"}}\n\n", + "data: {\"type\":\"response.output_item.done\",\"item\":{\"type\":\"function_call\",\"call_id\":\"call_second\",\"name\":\"Bash\",\"arguments\":\"{\\\"command\\\":\\\"echo second\\\"}\"}}\n\n", + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_multiple_standalone\",\"status\":\"completed\",\"usage\":{}}}\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 = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let text = String::from_utf8(body.to_vec()).unwrap(); + let first = text.find("call_first").unwrap(); + let second = text.find("call_second").unwrap(); + assert!(first < second, "stream body: {text}"); + assert_eq!(text.matches(r#""type":"tool_use""#).count(), 2); + assert_eq!(text.matches("event: message_stop").count(), 1); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_keeps_completed_tool_when_another_provisional_tool_truncates() { + let first_body = concat!( + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_partial_a\",\"name\":\"Bash\"}}\n\n", + "data: {\"type\":\"response.function_call_arguments.delta\",\"output_index\":0,\"delta\":\"{\\\"command\\\":\\\"echo partial\"}\n\n", + "data: {\"type\":\"response.output_item.done\",\"item\":{\"type\":\"function_call\",\"call_id\":\"call_done_b\",\"name\":\"Bash\",\"arguments\":\"{\\\"command\\\":\\\"echo done\\\"}\"}}\n\n" + ) + .as_bytes() + .to_vec(); + + let (attempts, status, text) = + collect_http_stream_after_truncated_attempt(first_body, Vec::new()).await; + assert_eq!(attempts, 1, "stream body: {text}"); + assert_eq!(status, StatusCode::OK, "stream body: {text}"); + assert!(text.contains("call_done_b"), "stream body: {text}"); + assert!(!text.contains("call_partial_a"), "stream body: {text}"); + assert!(text.contains("event: error"), "stream body: {text}"); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_retries_parallel_partial_tools_in_output_order() { + let first_body = concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_parallel_failed\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_failed_left\",\"name\":\"Bash\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":1,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_failed_right\",\"name\":\"Bash\"}}\n\n", + "data: {\"type\":\"response.function_call_arguments.delta\",\"output_index\":1,\"delta\":\"{\\\"command\\\":\\\"echo failed right\"}\n\n", + "data: {\"type\":\"response.function_call_arguments.delta\",\"output_index\":0,\"delta\":\"{\\\"command\\\":\\\"echo failed left\"}\n\n" + ) + .as_bytes() + .to_vec(); + let success_body = concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_parallel_success\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_success_left\",\"name\":\"Bash\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":1,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_success_right\",\"name\":\"Bash\"}}\n\n", + "data: {\"type\":\"response.function_call_arguments.delta\",\"output_index\":1,\"delta\":\"{\\\"command\\\":\\\"echo right\\\"}\"}\n\n", + "data: {\"type\":\"response.function_call_arguments.delta\",\"output_index\":0,\"delta\":\"{\\\"command\\\":\\\"echo left\\\"}\"}\n\n", + "data: {\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_success_left\",\"name\":\"Bash\",\"arguments\":\"{\\\"command\\\":\\\"echo left\\\"}\"}}\n\n", + "data: {\"type\":\"response.output_item.done\",\"output_index\":1,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_success_right\",\"name\":\"Bash\",\"arguments\":\"{\\\"command\\\":\\\"echo right\\\"}\"}}\n\n", + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_parallel_success\",\"status\":\"completed\",\"usage\":{\"input_tokens\":5,\"output_tokens\":4}}}\n\n" + ) + .as_bytes() + .to_vec(); + + let (attempts, status, text) = + collect_http_stream_after_truncated_attempt(first_body, success_body).await; + assert_eq!(status, StatusCode::OK, "stream body: {text}"); + assert_eq!(attempts, 2, "stream body: {text}"); + assert!(!text.contains("event: error"), "stream body: {text}"); + assert!(!text.contains("call_failed_left"), "stream body: {text}"); + assert!(!text.contains("call_failed_right"), "stream body: {text}"); + let left = text + .find("call_success_left") + .unwrap_or_else(|| panic!("left tool: {text}")); + let right = text + .find("call_success_right") + .unwrap_or_else(|| panic!("right tool: {text}")); + assert!(left < right, "stream body: {text}"); + assert_eq!(text.matches(r#""type":"tool_use""#).count(), 2); + assert_eq!(text.matches("event: message_stop").count(), 1); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_emits_completed_tool_before_another_provisional_tool_closes() { + let _guard = env_lock(); + clear_all_continuations_for_tests(); + let config = TempDir::new().unwrap(); + write_auth(config.path(), "codex"); + + let release = Arc::new(tokio::sync::Notify::new()); + let first_batch_sent = 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(); + let first_batch_sent = first_batch_sent.clone(); + move |_body: String| { + let release = release.clone(); + let first_batch_sent = first_batch_sent.clone(); + async move { + let stream = futures_util::stream::unfold(0_u8, move |state| { + let release = release.clone(); + let first_batch_sent = first_batch_sent.clone(); + async move { + match state { + 0 => { + first_batch_sent.notify_one(); + Some(( + Ok::<_, std::convert::Infallible>(bytes::Bytes::from_static( + concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_interleaved\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_open_a\",\"name\":\"Bash\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":1,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_done_b\",\"name\":\"Bash\"}}\n\n", + "data: {\"type\":\"response.function_call_arguments.delta\",\"output_index\":0,\"delta\":\"{\\\"command\\\":\\\"echo a\"}\n\n", + "data: {\"type\":\"response.function_call_arguments.delta\",\"output_index\":1,\"delta\":\"{\\\"command\\\":\\\"echo b\\\"}\"}\n\n", + "data: {\"type\":\"response.output_item.done\",\"output_index\":1,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_done_b\",\"name\":\"Bash\",\"arguments\":\"{\\\"command\\\":\\\"echo b\\\"}\"}}\n\n" + ) + .as_bytes(), + )), + 1, + )) + } + 1 => { + release.notified().await; + Some(( + Ok(bytes::Bytes::from_static( + concat!( + "data: {\"type\":\"response.function_call_arguments.delta\",\"output_index\":0,\"delta\":\"\\\"}\"}\n\n", + "data: {\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_open_a\",\"name\":\"Bash\",\"arguments\":\"{\\\"command\\\":\\\"echo a\\\"}\"}}\n\n", + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_interleaved\",\"status\":\"completed\",\"usage\":{\"input_tokens\":5,\"output_tokens\":4}}}\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 mut response = Box::pin(call_messages_body(json!({ + "model": "gpt-5.5", + "max_tokens": 64, + "stream": true, + "messages": [{"role":"user","content":"hello"}] + }))); + + let response = tokio::select! { + _ = first_batch_sent.notified() => tokio::time::timeout(Duration::from_millis(500), &mut response) + .await + .expect("completed tool must be emitted before another provisional tool closes"), + response = &mut response => response, + }; + assert_eq!(response.status(), StatusCode::OK); + release.notify_one(); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let text = String::from_utf8(body.to_vec()).unwrap(); + assert!(text.contains("call_done_b"), "stream body: {text}"); + assert!(text.contains("call_open_a"), "stream body: {text}"); + assert_eq!(text.matches(r#""type":"tool_use""#).count(), 2); + assert_eq!(text.matches("event: message_stop").count(), 1); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_discards_large_provisional_bash_deltas() { + 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); + let delta = "x".repeat(4_200_000); + let mut body = concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_buffer_limit\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_buffer_limit_0\",\"name\":\"Bash\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":1,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_buffer_limit_1\",\"name\":\"Bash\"}}\n\n" + ) + .as_bytes() + .to_vec(); + for output_index in 0..2 { + let event = serde_json::json!({ + "type":"response.function_call_arguments.delta", + "output_index":output_index, + "delta":delta + }); + body.extend_from_slice(format!("data: {event}\n\n").as_bytes()); + } + body.extend_from_slice( + b"data: {\"type\":\"response.output_item.done\",\"item\":{\"type\":\"function_call\",\"call_id\":\"call_buffer_limit_done\",\"name\":\"Bash\",\"arguments\":\"{\\\"command\\\":\\\"echo completed\\\"}\"}}\n\n", + ); + body.extend_from_slice( + b"data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_buffer_limit\",\"status\":\"completed\",\"usage\":{}}}\n\n", + ); + body + } + }) + .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!(attempts.load(Ordering::SeqCst), 1); + assert_eq!(response.status(), StatusCode::OK); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let text = String::from_utf8(body.to_vec()).unwrap(); + assert!( + text.contains("call_buffer_limit_done"), + "stream body: {text}" + ); + assert!(!text.contains("http_pending_event_buffer_limit")); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_caps_pending_events_plus_retained_read_args() { + 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); + let padding = "x".repeat(7_000_000); + let mut body = b"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_combined_budget\"}}\n\n".to_vec(); + for output_index in 0..2 { + let event = serde_json::json!({ + "type":"response.output_item.added", + "output_index":output_index, + "item":{"type":"message","id":format!("msg_budget_{output_index}"),"padding":padding.clone()} + }); + body.extend_from_slice(format!("data: {event}\n\n").as_bytes()); + } + body.extend_from_slice( + b"data: {\"type\":\"response.output_item.added\",\"output_index\":2,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_read_budget\",\"name\":\"Read\"}}\n\n", + ); + let delta = serde_json::json!({ + "type":"response.function_call_arguments.delta", + "output_index":2, + "delta":"y".repeat(3_100_000) + }); + body.extend_from_slice(format!("data: {delta}\n\n").as_bytes()); + body + } + }) + .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!(attempts.load(Ordering::SeqCst), 1); + assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let value: Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(value["error"]["type"], "request_too_large"); + assert_eq!( + value["error"]["message"], + "Codex pending HTTP event buffer exceeded its aggregate bytes limit" + ); + assert!(!String::from_utf8_lossy(&body).contains("call_read_budget")); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_accepts_more_than_legacy_pending_event_count() { + 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); + let arguments = serde_json::json!({ + "command": format!("printf %s {}", "x".repeat(1_500)) + }) + .to_string(); + assert!(arguments.len() > 1_024); + + let mut body = concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_many_tool_deltas\"}}\n\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_many_tool_deltas\",\"name\":\"Bash\"}}\n\n" + ) + .as_bytes() + .to_vec(); + for delta in arguments.chars() { + let event = serde_json::json!({ + "type":"response.function_call_arguments.delta", + "output_index":0, + "delta":delta.to_string() + }); + body.extend_from_slice(format!("data: {event}\n\n").as_bytes()); + } + let done = serde_json::json!({ + "type":"response.output_item.done", + "output_index":0, + "item":{ + "type":"function_call", + "call_id":"call_many_tool_deltas", + "name":"Bash", + "arguments":arguments + } + }); + body.extend_from_slice(format!("data: {done}\n\n").as_bytes()); + body.extend_from_slice( + b"data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_many_tool_deltas\",\"status\":\"completed\",\"usage\":{\"input_tokens\":5,\"output_tokens\":4}}}\n\n", + ); + body + } + }) + .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!(attempts.load(Ordering::SeqCst), 1); + assert_eq!(response.status(), StatusCode::OK); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let text = String::from_utf8(body.to_vec()).unwrap(); + assert!(!text.contains("Codex pending HTTP event buffer exceeded")); + assert_eq!(text.matches(r#""type":"tool_use""#).count(), 1); + assert_eq!(text.matches("event: message_stop").count(), 1); +} + #[allow(clippy::await_holding_lock)] #[tokio::test] async fn smoke_codex_http_incomplete_after_text_is_an_error() {