From 65742c56eefe4e6a3fac2f0da134d64dc8850dbf Mon Sep 17 00:00:00 2001 From: Felipe Coury Date: Tue, 19 May 2026 13:02:54 -0300 Subject: [PATCH] fix(core): retry unsupported reasoning summaries --- codex-rs/core/src/client.rs | 210 +++++++++++++++++- codex-rs/core/tests/suite/client.rs | 80 +++++++ .../core/tests/suite/client_websockets.rs | 77 +++++++ 3 files changed, 359 insertions(+), 8 deletions(-) diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index f604a63458..959c25303d 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -24,6 +24,7 @@ //! fails, normal stream retry/fallback logic handles recovery on the same turn. use std::collections::HashMap; +use std::collections::HashSet; use std::sync::Arc; use std::sync::Mutex as StdMutex; use std::sync::OnceLock; @@ -177,6 +178,7 @@ struct ModelClientState { include_attestation: bool, attestation_provider: Option>, disable_websockets: AtomicBool, + reasoning_summary_unsupported_models: StdMutex>, cached_websocket_session: StdMutex, } @@ -281,6 +283,14 @@ enum WebsocketStreamOutcome { FallbackToHttp, } +enum PreparedResponseStream { + Ready { + stream: codex_api::ResponseStream, + buffered_events: Vec>, + }, + RetryWithoutReasoningSummary, +} + /// Result of opening a WebRTC Realtime call. /// /// The SDP answer goes back to the client. The call id and auth headers stay on the server so the @@ -346,6 +356,7 @@ impl ModelClient { include_attestation, attestation_provider, disable_websockets: AtomicBool::new(false), + reasoning_summary_unsupported_models: StdMutex::new(HashSet::new()), cached_websocket_session: StdMutex::new(WebsocketSession::default()), }), } @@ -688,11 +699,13 @@ impl ModelClient { } fn build_reasoning( + &self, model_info: &ModelInfo, effort: Option, summary: ReasoningSummaryConfig, ) -> Option { if model_info.supports_reasoning_summaries { + let summary = self.reasoning_summary_for_model(model_info, summary); Some(Reasoning { effort: effort.or(model_info.default_reasoning_level), summary: if summary == ReasoningSummaryConfig::None { @@ -706,6 +719,99 @@ impl ModelClient { } } + fn reasoning_summary_for_model( + &self, + model_info: &ModelInfo, + summary: ReasoningSummaryConfig, + ) -> ReasoningSummaryConfig { + if summary == ReasoningSummaryConfig::None + || !self + .state + .reasoning_summary_unsupported_models + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .contains(&model_info.slug) + { + summary + } else { + ReasoningSummaryConfig::None + } + } + + fn should_retry_without_reasoning_summary( + &self, + model_info: &ModelInfo, + summary: ReasoningSummaryConfig, + err: &ApiError, + ) -> bool { + if summary == ReasoningSummaryConfig::None || !reasoning_summary_is_unsupported(err) { + return false; + } + + let inserted = self + .state + .reasoning_summary_unsupported_models + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert(model_info.slug.clone()); + if inserted { + warn!( + model = %model_info.slug, + "retrying request without reasoning summary after model rejected reasoning.summary" + ); + } + true + } + + async fn prepare_response_stream( + &self, + mut stream: codex_api::ResponseStream, + model_info: &ModelInfo, + summary: ReasoningSummaryConfig, + inference_trace_attempt: &InferenceTraceAttempt, + ) -> PreparedResponseStream { + let mut buffered_events = Vec::new(); + loop { + let Some(event) = stream.next().await else { + return PreparedResponseStream::Ready { + stream, + buffered_events, + }; + }; + match event { + Ok(event) => { + let can_retry_after = + response_event_can_precede_reasoning_summary_rejection(&event); + buffered_events.push(Ok(event)); + if !can_retry_after { + return PreparedResponseStream::Ready { + stream, + buffered_events, + }; + } + } + Err(err) => { + if self.should_retry_without_reasoning_summary(model_info, summary, &err) { + let response_debug_context = + extract_response_debug_context_from_api_error(&err); + let err = map_api_error(err); + inference_trace_attempt.record_failed( + &err, + response_debug_context.request_id.as_deref(), + /*output_items*/ &[], + ); + return PreparedResponseStream::RetryWithoutReasoningSummary; + } + buffered_events.push(Err(err)); + return PreparedResponseStream::Ready { + stream, + buffered_events, + }; + } + } + } + } + fn build_responses_request( &self, provider: &codex_api::Provider, @@ -718,7 +824,7 @@ impl ModelClient { let instructions = &prompt.base_instructions.text; let input = prompt.get_formatted_input(); let tools = create_tools_json_for_responses_api(&prompt.tools)?; - let reasoning = Self::build_reasoning(model_info, effort, summary); + let reasoning = self.build_reasoning(model_info, effort, summary); let include = if reasoning.is_some() { vec!["reasoning.encrypted_content".to_string()] } else { @@ -1263,8 +1369,25 @@ impl ModelClientSession { match stream_result { Ok(stream) => { - let (stream, _) = map_response_stream( + let (stream, buffered_events) = match self + .client + .prepare_response_stream( + stream, + model_info, + summary, + &inference_trace_attempt, + ) + .await + { + PreparedResponseStream::Ready { + stream, + buffered_events, + } => (stream, buffered_events), + PreparedResponseStream::RetryWithoutReasoningSummary => continue, + }; + let (stream, _) = map_response_stream_with_buffered_events( stream, + buffered_events, session_telemetry.clone(), inference_trace_attempt, ); @@ -1293,12 +1416,18 @@ impl ModelClientSession { Err(err) => { let response_debug_context = extract_response_debug_context_from_api_error(&err); + let should_retry_without_reasoning_summary = self + .client + .should_retry_without_reasoning_summary(model_info, summary, &err); let err = map_api_error(err); inference_trace_attempt.record_failed( &err, response_debug_context.request_id.as_deref(), /*output_items*/ &[], ); + if should_retry_without_reasoning_summary { + continue; + } return Err(err); } } @@ -1423,22 +1552,52 @@ impl ModelClientSession { "websocket connection is unavailable".to_string(), )) })?; - let stream_result = websocket_connection + let stream_result = match websocket_connection .stream_request(ws_request, self.websocket_session.connection_reused()) .await - .map_err(|err| { + { + Ok(stream_result) => stream_result, + Err(err) => { let response_debug_context = extract_response_debug_context_from_api_error(&err); + let should_retry_without_reasoning_summary = self + .client + .should_retry_without_reasoning_summary(model_info, summary, &err); let err = map_api_error(err); inference_trace_attempt.record_failed( &err, response_debug_context.request_id.as_deref(), /*output_items*/ &[], ); - err - })?; - let (stream, last_request_rx) = map_response_stream( + if should_retry_without_reasoning_summary { + self.reset_websocket_session(); + continue; + } + return Err(err); + } + }; + let (stream_result, buffered_events) = match self + .client + .prepare_response_stream( + stream_result, + model_info, + summary, + &inference_trace_attempt, + ) + .await + { + PreparedResponseStream::Ready { + stream, + buffered_events, + } => (stream, buffered_events), + PreparedResponseStream::RetryWithoutReasoningSummary => { + self.reset_websocket_session(); + continue; + } + }; + let (stream, last_request_rx) = map_response_stream_with_buffered_events( stream_result, + buffered_events, session_telemetry.clone(), inference_trace_attempt, ); @@ -1616,6 +1775,39 @@ impl ModelClientSession { } } +fn reasoning_summary_is_unsupported(err: &ApiError) -> bool { + let ApiError::Transport(TransportError::Http { status, body, .. }) = err else { + return false; + }; + if *status != StatusCode::BAD_REQUEST { + return false; + } + + let Some(body) = body.as_deref() else { + return false; + }; + let Ok(payload) = serde_json::from_str::(body) else { + return false; + }; + let Some(error) = payload.get("error") else { + return false; + }; + error.get("code").and_then(serde_json::Value::as_str) == Some("unsupported_parameter") + && error.get("param").and_then(serde_json::Value::as_str) == Some("reasoning.summary") +} + +fn response_event_can_precede_reasoning_summary_rejection(event: &ResponseEvent) -> bool { + matches!( + event, + ResponseEvent::Created + | ResponseEvent::ServerModel(_) + | ResponseEvent::ModelVerifications(_) + | ResponseEvent::ServerReasoningIncluded(_) + | ResponseEvent::RateLimits(_) + | ResponseEvent::ModelsEtag(_) + ) +} + /// Parses per-turn metadata into an HTTP header value. /// /// Invalid values are treated as absent so callers can compare and propagate @@ -1712,8 +1904,9 @@ fn parent_thread_id_header_value(session_source: &SessionSource) -> Option>, session_telemetry: SessionTelemetry, inference_trace_attempt: InferenceTraceAttempt, ) -> (ResponseStream, oneshot::Receiver) { @@ -1725,6 +1918,7 @@ fn map_response_stream( rx_event, upstream_request_id: None, }; + let api_stream = futures::stream::iter(buffered_events).chain(api_stream); map_response_events( upstream_request_id, api_stream, diff --git a/codex-rs/core/tests/suite/client.rs b/codex-rs/core/tests/suite/client.rs index f6731a0ea4..f877765a16 100644 --- a/codex-rs/core/tests/suite/client.rs +++ b/codex-rs/core/tests/suite/client.rs @@ -58,6 +58,7 @@ use core_test_support::responses::ev_completed_with_tokens; use core_test_support::responses::ev_message_item_added; use core_test_support::responses::ev_output_text_delta; use core_test_support::responses::ev_response_created; +use core_test_support::responses::mount_response_sequence; use core_test_support::responses::mount_sse_once; use core_test_support::responses::mount_sse_once_match; use core_test_support::responses::mount_sse_sequence; @@ -1855,6 +1856,85 @@ async fn configured_reasoning_summary_is_sent() -> anyhow::Result<()> { Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn unsupported_reasoning_summary_is_retried_without_summary() -> anyhow::Result<()> { + skip_if_no_network!(Ok(())); + let server = MockServer::start().await; + + let responses = mount_response_sequence( + &server, + vec![ + ResponseTemplate::new(400).set_body_json(json!({ + "error": { + "type": "invalid_request_error", + "code": "unsupported_parameter", + "message": "Unsupported parameter: 'reasoning.summary' is not supported with the 'gpt-5.4' model.", + "param": "reasoning.summary" + } + })), + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_raw( + sse(vec![ev_response_created("resp1"), ev_completed("resp1")]), + "text/event-stream", + ), + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_raw( + sse(vec![ev_response_created("resp2"), ev_completed("resp2")]), + "text/event-stream", + ), + ], + ) + .await; + let TestCodex { codex, .. } = test_codex() + .with_config(|config| { + config.model_reasoning_summary = Some(ReasoningSummary::Concise); + }) + .build(&server) + .await?; + + for text in ["hello", "again"] { + codex + .submit(Op::UserInput { + environments: None, + items: vec![UserInput::Text { + text: text.into(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + responsesapi_client_metadata: None, + thread_settings: Default::default(), + }) + .await + .unwrap(); + + wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; + } + + let requests = responses.requests(); + assert_eq!(requests.len(), 3); + pretty_assertions::assert_eq!( + requests[0] + .body_json() + .get("reasoning") + .and_then(|reasoning| reasoning.get("summary")) + .and_then(|value| value.as_str()), + Some("concise") + ); + for request in &requests[1..] { + pretty_assertions::assert_eq!( + request + .body_json() + .get("reasoning") + .and_then(|reasoning| reasoning.get("summary")), + None + ); + } + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn user_turn_explicit_reasoning_summary_overrides_model_catalog_default() -> anyhow::Result<()> { diff --git a/codex-rs/core/tests/suite/client_websockets.rs b/codex-rs/core/tests/suite/client_websockets.rs index feee9d4227..5bd71c4701 100755 --- a/codex-rs/core/tests/suite/client_websockets.rs +++ b/codex-rs/core/tests/suite/client_websockets.rs @@ -1318,6 +1318,83 @@ async fn responses_websocket_invalid_request_error_with_status_is_forwarded() { server.shutdown().await; } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn responses_websocket_retries_without_unsupported_reasoning_summary() { + skip_if_no_network!(); + + let unsupported_reasoning_summary_error = json!({ + "type": "error", + "status": 400, + "error": { + "type": "invalid_request_error", + "code": "unsupported_parameter", + "message": "Unsupported parameter: 'reasoning.summary' is not supported with the 'gpt-5.3-codex-spark' model.", + "param": "reasoning.summary" + } + }); + + let server = start_websocket_server(vec![ + vec![ + vec![ + ev_response_created("resp-prewarm"), + ev_completed("resp-prewarm"), + ], + vec![unsupported_reasoning_summary_error], + ], + vec![vec![ev_response_created("resp-1"), ev_completed("resp-1")]], + ]) + .await; + let mut builder = test_codex().with_config(|config| { + config.model_provider.request_max_retries = Some(0); + config.model_provider.stream_max_retries = Some(0); + config.model_reasoning_summary = Some(ReasoningSummary::Concise); + }); + let test = builder + .build_with_websocket_server(&server) + .await + .expect("build websocket codex"); + + test.codex + .submit(Op::UserInput { + environments: None, + items: vec![UserInput::Text { + text: "hello".into(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + responsesapi_client_metadata: None, + thread_settings: Default::default(), + }) + .await + .expect("submission should succeed"); + wait_for_event(&test.codex, |msg| matches!(msg, EventMsg::TurnComplete(_))).await; + + let connections = server.connections(); + assert_eq!(connections.len(), 2); + let first_connection = connections.first().expect("missing first connection"); + let second_connection = connections.get(1).expect("missing second connection"); + assert_eq!(first_connection.len(), 2); + assert_eq!(second_connection.len(), 1); + assert_eq!( + first_connection + .get(1) + .expect("missing rejected request") + .body_json()["reasoning"]["summary"] + .as_str(), + Some("concise") + ); + assert_eq!( + second_connection + .first() + .expect("missing retried request") + .body_json()["reasoning"] + .get("summary"), + None + ); + + server.shutdown().await; +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn responses_websocket_connection_limit_error_reconnects_and_completes() { skip_if_no_network!();