From 775ef7dcc7d50069e759f3e11454f412a3e39d95 Mon Sep 17 00:00:00 2001 From: ashwinnathan-openai Date: Mon, 6 Jul 2026 23:23:56 -0400 Subject: [PATCH] [codex] Support sequential cutoff reasoning summaries (#31306) ## Summary When enabled for the OpenAI provider, Codex sends `stream_options.reasoning_summary_delivery = "sequential_cutoff"` on HTTP and WebSocket requests, including prewarm, and renders completed summary sections from `reasoning_summary_text.done`. Flag-off and non-OpenAI behavior is unchanged. ## Expected rollout ```text reasoning 0 added summary 0 done summary 1 done summary 2 starts summary 2 cancelled / incomplete reasoning 0 done <-- cancel summary 2 work and mark it incomplete message 1 added message 1 text streams message 1 completed ``` Depends on [openai/openai#1096660](https://github.com/openai/openai/pull/1096660). --- codex-rs/codex-api/src/common.rs | 21 +++++ .../src/endpoint/responses_websocket.rs | 1 + codex-rs/codex-api/src/lib.rs | 2 + codex-rs/codex-api/src/sse/responses.rs | 38 ++++++++ codex-rs/codex-api/tests/clients.rs | 3 + codex-rs/core/config.schema.json | 6 ++ codex-rs/core/src/client.rs | 16 +++- codex-rs/core/src/client_common_tests.rs | 4 + codex-rs/core/src/client_tests.rs | 3 + codex-rs/core/src/session/session.rs | 3 + codex-rs/core/src/session/tests.rs | 9 ++ codex-rs/core/src/session/turn.rs | 45 +++++++++ codex-rs/core/src/turn_timing.rs | 1 + codex-rs/core/tests/responses_headers.rs | 3 + codex-rs/core/tests/suite/client.rs | 58 ++++++++++++ .../core/tests/suite/client_websockets.rs | 42 ++++++++- codex-rs/core/tests/suite/items.rs | 93 +++++++++++++++++++ codex-rs/features/src/lib.rs | 8 ++ codex-rs/memories/write/src/runtime.rs | 1 + codex-rs/otel/src/events/session_telemetry.rs | 1 + 20 files changed, 352 insertions(+), 6 deletions(-) diff --git a/codex-rs/codex-api/src/common.rs b/codex-rs/codex-api/src/common.rs index 254a5cb38f..e4600e26aa 100644 --- a/codex-rs/codex-api/src/common.rs +++ b/codex-rs/codex-api/src/common.rs @@ -104,6 +104,11 @@ pub enum ResponseEvent { delta: String, summary_index: i64, }, + ReasoningSummaryDone { + item_id: String, + text: String, + summary_index: i64, + }, ReasoningContentDelta { delta: String, content_index: i64, @@ -148,6 +153,17 @@ pub struct Reasoning { pub context: Option, } +#[derive(Debug, Serialize, Clone, PartialEq)] +#[serde(rename_all = "snake_case")] +pub enum ReasoningSummaryDelivery { + SequentialCutoff, +} + +#[derive(Debug, Serialize, Clone, PartialEq)] +pub struct StreamOptions { + pub reasoning_summary_delivery: ReasoningSummaryDelivery, +} + #[derive(Debug, Serialize, Default, Clone, PartialEq)] #[serde(rename_all = "snake_case")] pub enum TextFormatType { @@ -209,6 +225,8 @@ pub struct ResponsesApiRequest { pub reasoning: Option, pub store: bool, pub stream: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub stream_options: Option, pub include: Vec, #[serde(skip_serializing_if = "Option::is_none")] pub service_tier: Option, @@ -233,6 +251,7 @@ impl From<&ResponsesApiRequest> for ResponseCreateWsRequest { reasoning: request.reasoning.clone(), store: request.store, stream: request.stream, + stream_options: request.stream_options.clone(), include: request.include.clone(), service_tier: request.service_tier.clone(), prompt_cache_key: request.prompt_cache_key.clone(), @@ -258,6 +277,8 @@ pub struct ResponseCreateWsRequest { pub reasoning: Option, pub store: bool, pub stream: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub stream_options: Option, pub include: Vec, #[serde(skip_serializing_if = "Option::is_none")] pub service_tier: Option, diff --git a/codex-rs/codex-api/src/endpoint/responses_websocket.rs b/codex-rs/codex-api/src/endpoint/responses_websocket.rs index 447450fae1..cce056f638 100644 --- a/codex-rs/codex-api/src/endpoint/responses_websocket.rs +++ b/codex-rs/codex-api/src/endpoint/responses_websocket.rs @@ -851,6 +851,7 @@ mod tests { reasoning: None, store: false, stream: true, + stream_options: None, include: vec!["reasoning.encrypted_content".to_string()], service_tier: Some("priority".to_string()), prompt_cache_key: Some("cache-key".to_string()), diff --git a/codex-rs/codex-api/src/lib.rs b/codex-rs/codex-api/src/lib.rs index 72de4d0585..1d1a4c8c6e 100644 --- a/codex-rs/codex-api/src/lib.rs +++ b/codex-rs/codex-api/src/lib.rs @@ -34,11 +34,13 @@ pub use crate::common::RawMemory; pub use crate::common::RawMemoryMetadata; pub use crate::common::Reasoning; pub use crate::common::ReasoningContext; +pub use crate::common::ReasoningSummaryDelivery; pub use crate::common::ResponseCreateWsRequest; pub use crate::common::ResponseEvent; pub use crate::common::ResponseStream; pub use crate::common::ResponsesApiRequest; pub use crate::common::ResponsesWsRequest; +pub use crate::common::StreamOptions; pub use crate::common::TextControls; pub use crate::common::WS_REQUEST_HEADER_TRACEPARENT_CLIENT_METADATA_KEY; pub use crate::common::WS_REQUEST_HEADER_TRACESTATE_CLIENT_METADATA_KEY; diff --git a/codex-rs/codex-api/src/sse/responses.rs b/codex-rs/codex-api/src/sse/responses.rs index 17d31a02ee..265d3ced5c 100644 --- a/codex-rs/codex-api/src/sse/responses.rs +++ b/codex-rs/codex-api/src/sse/responses.rs @@ -167,6 +167,7 @@ pub struct ResponsesStreamEvent { item_id: Option, call_id: Option, delta: Option, + text: Option, summary_index: Option, content_index: Option, safety_buffering: Option, @@ -358,6 +359,17 @@ pub fn process_responses_event( })); } } + "response.reasoning_summary_text.done" => { + if let (Some(item_id), Some(text), Some(summary_index)) = + (event.item_id, event.text, event.summary_index) + { + return Ok(Some(ResponseEvent::ReasoningSummaryDone { + item_id, + text, + summary_index, + })); + } + } "response.reasoning_text.delta" => { if let (Some(delta), Some(content_index)) = (event.delta, event.content_index) { return Ok(Some(ResponseEvent::ReasoningContentDelta { @@ -796,6 +808,32 @@ mod tests { } } + #[tokio::test] + async fn parses_reasoning_summary_done() { + let events = run_sse(vec![ + json!({ + "type": "response.reasoning_summary_text.done", + "item_id": "reasoning-1", + "summary_index": 0, + "text": "Checking", + }), + json!({ + "type": "response.completed", + "response": { "id": "resp1" }, + }), + ]) + .await; + + assert_matches!( + &events[0], + ResponseEvent::ReasoningSummaryDone { + item_id, + text, + summary_index: 0, + } if item_id == "reasoning-1" && text == "Checking" + ); + } + #[tokio::test] async fn error_when_missing_completed() { let item1 = json!({ diff --git a/codex-rs/codex-api/tests/clients.rs b/codex-rs/codex-api/tests/clients.rs index 60a9852cfa..ab4ab1d08f 100644 --- a/codex-rs/codex-api/tests/clients.rs +++ b/codex-rs/codex-api/tests/clients.rs @@ -321,6 +321,7 @@ async fn responses_client_stream_request_preserves_item_ids() -> Result<()> { reasoning: None, store: false, stream: true, + stream_options: None, include: Vec::new(), service_tier: None, prompt_cache_key: None, @@ -407,6 +408,7 @@ async fn streaming_client_retries_on_transport_error() -> Result<()> { reasoning: None, store: false, stream: true, + stream_options: None, include: Vec::new(), service_tier: None, prompt_cache_key: None, @@ -526,6 +528,7 @@ async fn azure_store_sends_ids_and_headers() -> Result<()> { reasoning: None, store: true, stream: true, + stream_options: None, include: Vec::new(), service_tier: None, prompt_cache_key: None, diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index 876180dd7e..0a48882b09 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -469,6 +469,9 @@ "computer_use": { "type": "boolean" }, + "concurrent_reasoning_summaries": { + "type": "boolean" + }, "connectors": { "type": "boolean" }, @@ -4838,6 +4841,9 @@ "computer_use": { "type": "boolean" }, + "concurrent_reasoning_summaries": { + "type": "boolean" + }, "connectors": { "type": "boolean" }, diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index 43a758b0e8..716466b189 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -56,6 +56,7 @@ use codex_api::ResponsesWebsocketConnection as ApiWebSocketConnection; use codex_api::ResponsesWsRequest; use codex_api::SharedAuthProvider; use codex_api::SseTelemetry; +use codex_api::StreamOptions; use codex_api::TransportError; use codex_api::WebsocketTelemetry; use codex_api::auth_header_telemetry; @@ -203,6 +204,7 @@ struct ModelClientState { include_timing_metrics: bool, beta_features_header: Option, item_ids_enabled: bool, + concurrent_reasoning_summaries_enabled: bool, include_attestation: bool, attestation_provider: Option>, disable_websockets: AtomicBool, @@ -311,6 +313,7 @@ fn responses_request_properties_match( reasoning: previous_reasoning, store: previous_store, stream: previous_stream, + stream_options: _, include: previous_include, service_tier: previous_service_tier, prompt_cache_key: previous_prompt_cache_key, @@ -327,6 +330,7 @@ fn responses_request_properties_match( reasoning: current_reasoning, store: current_store, stream: current_stream, + stream_options: _, include: current_include, service_tier: current_service_tier, prompt_cache_key: current_prompt_cache_key, @@ -342,6 +346,8 @@ fn responses_request_properties_match( && previous_reasoning == current_reasoning && previous_store == current_store && previous_stream == current_stream + // Stream options control delivery for this response, not the context + // referenced by `previous_response_id`. && previous_include == current_include && previous_service_tier == current_service_tier && previous_prompt_cache_key == current_prompt_cache_key @@ -409,6 +415,7 @@ impl ModelClient { include_timing_metrics: bool, beta_features_header: Option, item_ids_enabled: bool, + concurrent_reasoning_summaries_enabled: bool, attestation_provider: Option>, ) -> Self { let model_provider = create_model_provider(provider_info, auth_manager); @@ -431,6 +438,7 @@ impl ModelClient { include_timing_metrics, beta_features_header, item_ids_enabled, + concurrent_reasoning_summaries_enabled, include_attestation, attestation_provider, disable_websockets: AtomicBool::new(false), @@ -820,7 +828,8 @@ impl ModelClient { responses_metadata: &CodexResponsesMetadata, ) -> Result { let mut input = prompt.get_formatted_input_for_request(model_info.use_responses_lite); - if !self.state.provider.info().is_openai() { + let is_openai = self.state.provider.info().is_openai(); + if !is_openai { input .iter_mut() .for_each(ResponseItem::clear_internal_chat_message_metadata_passthrough); @@ -848,6 +857,10 @@ impl ModelClient { } else { (prompt.base_instructions.text.clone(), Some(tools)) }; + let stream_options = (self.state.concurrent_reasoning_summaries_enabled && is_openai) + .then_some(StreamOptions { + reasoning_summary_delivery: codex_api::ReasoningSummaryDelivery::SequentialCutoff, + }); let reasoning = Self::build_reasoning(model_info, effort, summary); let include = if reasoning.is_some() { vec!["reasoning.encrypted_content".to_string()] @@ -882,6 +895,7 @@ impl ModelClient { reasoning, store: provider.is_azure_responses_endpoint(), stream: true, + stream_options, include, service_tier, prompt_cache_key, diff --git a/codex-rs/core/src/client_common_tests.rs b/codex-rs/core/src/client_common_tests.rs index 38c4aaecb5..b8d0a364a3 100644 --- a/codex-rs/core/src/client_common_tests.rs +++ b/codex-rs/core/src/client_common_tests.rs @@ -116,6 +116,7 @@ fn serializes_text_verbosity_when_set() { reasoning: None, store: false, stream: true, + stream_options: None, include: vec![], prompt_cache_key: None, service_tier: None, @@ -163,6 +164,7 @@ fn serializes_text_schema_with_strict_format() { reasoning: None, store: false, stream: true, + stream_options: None, include: vec![], prompt_cache_key: None, service_tier: None, @@ -224,6 +226,7 @@ fn omits_text_when_not_set() { reasoning: None, store: false, stream: true, + stream_options: None, include: vec![], prompt_cache_key: None, service_tier: None, @@ -247,6 +250,7 @@ fn serializes_flex_service_tier_when_set() { reasoning: None, store: false, stream: true, + stream_options: None, include: vec![], prompt_cache_key: None, service_tier: Some(ServiceTier::Flex.to_string()), diff --git a/codex-rs/core/src/client_tests.rs b/codex-rs/core/src/client_tests.rs index 82c464625d..c05f56b713 100644 --- a/codex-rs/core/src/client_tests.rs +++ b/codex-rs/core/src/client_tests.rs @@ -103,6 +103,7 @@ fn test_model_client_with_thread_id( /*include_timing_metrics*/ false, /*beta_features_header*/ None, /*item_ids_enabled*/ false, + /*concurrent_reasoning_summaries_enabled*/ false, /*attestation_provider*/ None, ) } @@ -147,6 +148,7 @@ async fn compact_uses_bearer_after_agent_identity_session_fallback() -> anyhow:: /*include_timing_metrics*/ false, /*beta_features_header*/ None, /*item_ids_enabled*/ false, + /*concurrent_reasoning_summaries_enabled*/ false, /*attestation_provider*/ None, ); let prompt = Prompt { @@ -821,6 +823,7 @@ fn model_client_with_counting_attestation( /*include_timing_metrics*/ false, /*beta_features_header*/ None, /*item_ids_enabled*/ false, + /*concurrent_reasoning_summaries_enabled*/ false, Some(Arc::new(CountingAttestationProvider { calls: attestation_calls.clone(), })), diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 2261eaa495..78a7f02588 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -1111,6 +1111,9 @@ impl Session { config.features.enabled(Feature::RuntimeMetrics), Self::build_model_client_beta_features_header(config.as_ref()), /*item_ids_enabled*/ config.features.enabled(Feature::ItemIds), + /*concurrent_reasoning_summaries_enabled*/ config + .features + .enabled(Feature::ConcurrentReasoningSummaries), attestation_provider, ) .with_prompt_cache_key_override( diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 6cc223cd7d..1ff469cf4c 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -490,6 +490,7 @@ fn test_model_client_session() -> crate::client::ModelClientSession { /*include_timing_metrics*/ false, /*beta_features_header*/ None, /*item_ids_enabled*/ false, + /*concurrent_reasoning_summaries_enabled*/ false, /*attestation_provider*/ None, ) .new_session() @@ -5460,6 +5461,10 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { config.features.enabled(Feature::RuntimeMetrics), Session::build_model_client_beta_features_header(config.as_ref()), /*item_ids_enabled*/ config.features.enabled(Feature::ItemIds), + /*concurrent_reasoning_summaries_enabled*/ + config + .features + .enabled(Feature::ConcurrentReasoningSummaries), /*attestation_provider*/ None, ), code_mode_service: crate::tools::code_mode::CodeModeService::new(Arc::new( @@ -7586,6 +7591,10 @@ where config.features.enabled(Feature::RuntimeMetrics), Session::build_model_client_beta_features_header(config.as_ref()), /*item_ids_enabled*/ config.features.enabled(Feature::ItemIds), + /*concurrent_reasoning_summaries_enabled*/ + config + .features + .enabled(Feature::ConcurrentReasoningSummaries), /*attestation_provider*/ None, ), code_mode_service: crate::tools::code_mode::CodeModeService::new(Arc::new( diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index f16b525002..708f42f0ff 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -1909,6 +1909,11 @@ async fn try_run_sampling_request( turn_context.provider.info().name.as_str(), ); let sampling_timing_guard = turn_context.turn_timing_state.begin_sampling(); + let uses_sequential_cutoff_reasoning_summaries = turn_context + .config + .features + .enabled(Feature::ConcurrentReasoningSummaries) + && turn_context.provider.info().is_openai(); let mut stream = client_session .stream( prompt, @@ -2291,6 +2296,9 @@ async fn try_run_sampling_request( delta, summary_index, } => { + if uses_sequential_cutoff_reasoning_summaries { + continue; + } if let Some(active) = active_item.as_ref() { if !active_item_is_streaming_to_client { continue; @@ -2309,6 +2317,9 @@ async fn try_run_sampling_request( } } ResponseEvent::ReasoningSummaryPartAdded { summary_index } => { + if uses_sequential_cutoff_reasoning_summaries { + continue; + } if let Some(active) = active_item.as_ref() { if !active_item_is_streaming_to_client { continue; @@ -2323,6 +2334,40 @@ async fn try_run_sampling_request( error_or_panic("ReasoningSummaryPartAdded without active item".to_string()); } } + ResponseEvent::ReasoningSummaryDone { + item_id, + text, + summary_index, + } => { + if !uses_sequential_cutoff_reasoning_summaries { + continue; + } + let Some(active) = active_item.as_ref() else { + continue; + }; + if !active_item_is_streaming_to_client || active.id() != item_id { + continue; + } + if summary_index > 0 { + sess.send_event( + &turn_context, + EventMsg::AgentReasoningSectionBreak(AgentReasoningSectionBreakEvent { + item_id: item_id.clone(), + summary_index, + }), + ) + .await; + } + let event = ReasoningContentDeltaEvent { + thread_id: sess.thread_id.to_string(), + turn_id: turn_context.sub_id.clone(), + item_id, + delta: text, + summary_index, + }; + sess.send_event(&turn_context, EventMsg::ReasoningContentDelta(event)) + .await; + } ResponseEvent::ReasoningContentDelta { delta, content_index, diff --git a/codex-rs/core/src/turn_timing.rs b/codex-rs/core/src/turn_timing.rs index 04f7aeb69b..d0f4f46cc3 100644 --- a/codex-rs/core/src/turn_timing.rs +++ b/codex-rs/core/src/turn_timing.rs @@ -334,6 +334,7 @@ fn response_event_records_turn_ttft(event: &ResponseEvent) -> bool { } ResponseEvent::OutputTextDelta(_) | ResponseEvent::ReasoningSummaryDelta { .. } + | ResponseEvent::ReasoningSummaryDone { .. } | ResponseEvent::ReasoningContentDelta { .. } => true, ResponseEvent::Created | ResponseEvent::ServerModel(_) diff --git a/codex-rs/core/tests/responses_headers.rs b/codex-rs/core/tests/responses_headers.rs index 5b7c2c65b8..4c5f906160 100644 --- a/codex-rs/core/tests/responses_headers.rs +++ b/codex-rs/core/tests/responses_headers.rs @@ -131,6 +131,7 @@ async fn responses_stream_includes_subagent_header_on_review() { /*include_timing_metrics*/ false, /*beta_features_header*/ None, /*item_ids_enabled*/ false, + /*concurrent_reasoning_summaries_enabled*/ false, /*attestation_provider*/ None, ); let responses_metadata = test_turn_responses_metadata(&client, thread_id, &session_source); @@ -265,6 +266,7 @@ async fn responses_stream_includes_subagent_header_on_other() { /*include_timing_metrics*/ false, /*beta_features_header*/ None, /*item_ids_enabled*/ false, + /*concurrent_reasoning_summaries_enabled*/ false, /*attestation_provider*/ None, ); let responses_metadata = test_turn_responses_metadata(&client, thread_id, &session_source); @@ -385,6 +387,7 @@ async fn responses_respects_model_info_overrides_from_config() { /*include_timing_metrics*/ false, /*beta_features_header*/ None, /*item_ids_enabled*/ false, + /*concurrent_reasoning_summaries_enabled*/ false, /*attestation_provider*/ None, ); let responses_metadata = test_turn_responses_metadata(&client, thread_id, &session_source); diff --git a/codex-rs/core/tests/suite/client.rs b/codex-rs/core/tests/suite/client.rs index c87539f940..92624bfcb3 100644 --- a/codex-rs/core/tests/suite/client.rs +++ b/codex-rs/core/tests/suite/client.rs @@ -1334,6 +1334,10 @@ async fn send_provider_auth_request(server: &MockServer, auth: ModelProviderAuth /*include_timing_metrics*/ false, /*beta_features_header*/ None, /*item_ids_enabled*/ config.features.enabled(Feature::ItemIds), + /*concurrent_reasoning_summaries_enabled*/ + config + .features + .enabled(Feature::ConcurrentReasoningSummaries), /*attestation_provider*/ None, ); let responses_metadata = test_turn_responses_metadata(&client, thread_id); @@ -2376,6 +2380,9 @@ async fn configured_reasoning_summary_is_sent() -> anyhow::Result<()> { let TestCodex { codex, .. } = test_codex() .with_config(|config| { config.model_reasoning_summary = Some(ReasoningSummary::Concise); + let _ = config + .features + .enable(Feature::ConcurrentReasoningSummaries); }) .build(&server) .await?; @@ -2412,6 +2419,55 @@ async fn configured_reasoning_summary_is_sent() -> anyhow::Result<()> { .and_then(|reasoning| reasoning.get("context")), None ); + pretty_assertions::assert_eq!( + request_body + .get("stream_options") + .and_then(|stream_options| stream_options.get("reasoning_summary_delivery")) + .and_then(|value| value.as_str()), + Some("sequential_cutoff") + ); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn sequential_cutoff_is_omitted_for_non_openai_provider() -> anyhow::Result<()> { + skip_if_no_network!(Ok(())); + let server = MockServer::start().await; + + let resp_mock = mount_sse_once( + &server, + sse(vec![ev_response_created("resp1"), ev_completed("resp1")]), + ) + .await; + let TestCodex { codex, .. } = test_codex() + .with_config(|config| { + config.model_provider.name = "mock".to_string(); + let _ = config + .features + .enable(Feature::ConcurrentReasoningSummaries); + }) + .build(&server) + .await?; + + codex + .submit(Op::UserInput { + items: vec![UserInput::Text { + text: "hello".into(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + responsesapi_client_metadata: None, + additional_context: Default::default(), + thread_settings: Default::default(), + }) + .await + .unwrap(); + + wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; + + let request_body = resp_mock.single_request().body_json(); + pretty_assertions::assert_eq!(request_body.get("stream_options"), None); Ok(()) } @@ -2583,6 +2639,7 @@ async fn reasoning_summary_is_omitted_when_disabled() -> anyhow::Result<()> { .and_then(|reasoning| reasoning.get("summary")), None ); + pretty_assertions::assert_eq!(request_body.get("stream_options"), None); Ok(()) } @@ -2948,6 +3005,7 @@ async fn azure_responses_request_includes_store_and_reasoning_ids() { /*include_timing_metrics*/ false, /*beta_features_header*/ None, /*item_ids_enabled*/ false, + /*concurrent_reasoning_summaries_enabled*/ false, /*attestation_provider*/ None, ); let responses_metadata = test_turn_responses_metadata(&client, thread_id); diff --git a/codex-rs/core/tests/suite/client_websockets.rs b/codex-rs/core/tests/suite/client_websockets.rs index d5218cdcab..c56a8680ab 100755 --- a/codex-rs/core/tests/suite/client_websockets.rs +++ b/codex-rs/core/tests/suite/client_websockets.rs @@ -386,7 +386,13 @@ async fn responses_websocket_request_prewarm_reuses_connection() { ]]) .await; - let harness = websocket_harness_with_options(&server, /*runtime_metrics_enabled*/ true).await; + let mut provider = websocket_provider(&server); + provider.name = ModelProviderInfo::create_openai_provider(/*base_url*/ None).name; + let harness = websocket_harness_with_provider_options( + provider, /*runtime_metrics_enabled*/ true, + /*concurrent_reasoning_summaries_enabled*/ true, + ) + .await; let mut client_session = harness.client.new_session(); let prompt = prompt_with_input(vec![message_item("hello")]); let responses_metadata = prewarm_metadata(&harness, /*turn_id*/ None); @@ -422,6 +428,10 @@ async fn responses_websocket_request_prewarm_reuses_connection() { assert_eq!(warmup["type"].as_str(), Some("response.create")); assert_eq!(warmup["generate"].as_bool(), Some(false)); + assert_eq!( + warmup["stream_options"]["reasoning_summary_delivery"].as_str(), + Some("sequential_cutoff") + ); assert_eq!(warmup["tools"], serde_json::json!([])); let warmup_turn_metadata: serde_json::Value = serde_json::from_str( warmup["client_metadata"]["x-codex-turn-metadata"] @@ -436,6 +446,10 @@ async fn responses_websocket_request_prewarm_reuses_connection() { assert_eq!(follow_up["type"].as_str(), Some("response.create")); assert_eq!(follow_up["previous_response_id"].as_str(), Some("warm-1")); assert_eq!(follow_up["input"], serde_json::json!([])); + assert_eq!( + follow_up["stream_options"]["reasoning_summary_delivery"].as_str(), + Some("sequential_cutoff") + ); server.shutdown().await; } @@ -990,8 +1004,11 @@ async fn responses_websocket_v2_incremental_requests_are_reused_across_turns() { // use OpenAI provider to check metadata logic let mut provider = websocket_provider(&server); provider.name = ModelProviderInfo::create_openai_provider(/*base_url*/ None).name; - let harness = - websocket_harness_with_provider_options(provider, /*runtime_metrics_enabled*/ false).await; + let harness = websocket_harness_with_provider_options( + provider, /*runtime_metrics_enabled*/ false, + /*concurrent_reasoning_summaries_enabled*/ false, + ) + .await; // Turn one: initiate let prompt_one = prompt_with_input(vec![message_item("hello")]); @@ -2195,13 +2212,18 @@ async fn websocket_harness_with_options( server: &WebSocketTestServer, runtime_metrics_enabled: bool, ) -> WebsocketTestHarness { - websocket_harness_with_provider_options(websocket_provider(server), runtime_metrics_enabled) - .await + websocket_harness_with_provider_options( + websocket_provider(server), + runtime_metrics_enabled, + /*concurrent_reasoning_summaries_enabled*/ false, + ) + .await } async fn websocket_harness_with_provider_options( provider: ModelProviderInfo, runtime_metrics_enabled: bool, + concurrent_reasoning_summaries_enabled: bool, ) -> WebsocketTestHarness { let codex_home = TempDir::new().unwrap(); let mut config = load_default_config_for_test(&codex_home).await; @@ -2212,6 +2234,12 @@ async fn websocket_harness_with_provider_options( .enable(Feature::RuntimeMetrics) .expect("test config should allow feature update"); } + if concurrent_reasoning_summaries_enabled { + config + .features + .enable(Feature::ConcurrentReasoningSummaries) + .expect("test config should allow feature update"); + } let config = Arc::new(config); let model_info = codex_core::test_support::construct_model_info_offline(MODEL, &config); let thread_id = ThreadId::new(); @@ -2251,6 +2279,10 @@ async fn websocket_harness_with_provider_options( runtime_metrics_enabled, /*beta_features_header*/ None, /*item_ids_enabled*/ config.features.enabled(Feature::ItemIds), + /*concurrent_reasoning_summaries_enabled*/ + config + .features + .enabled(Feature::ConcurrentReasoningSummaries), /*attestation_provider*/ None, ); diff --git a/codex-rs/core/tests/suite/items.rs b/codex-rs/core/tests/suite/items.rs index 515d496b8b..196277394a 100644 --- a/codex-rs/core/tests/suite/items.rs +++ b/codex-rs/core/tests/suite/items.rs @@ -1,6 +1,7 @@ #![cfg(not(target_os = "windows"))] use anyhow::Ok; +use codex_features::Feature; use codex_protocol::config_types::CollaborationMode; use codex_protocol::config_types::ModeKind; use codex_protocol::config_types::Settings; @@ -1134,6 +1135,98 @@ async fn reasoning_content_delta_has_item_metadata() -> anyhow::Result<()> { Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn sequential_cutoff_renders_done_summaries_for_active_reasoning_item() -> anyhow::Result<()> +{ + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + + let TestCodex { codex, .. } = test_codex() + .with_config(|config| { + config + .features + .enable(Feature::ConcurrentReasoningSummaries) + .expect("test config should allow feature update"); + }) + .build(&server) + .await?; + let summary_done = |item_id: &str, summary_index: i64, text: &str| { + serde_json::json!({ + "type": "response.reasoning_summary_text.done", + "item_id": item_id, + "summary_index": summary_index, + "text": text, + }) + }; + mount_sse_once( + &server, + sse(vec![ + ev_response_created("resp-1"), + ev_reasoning_item_added("reasoning-1", &[""]), + // Sequential cutoff renders atomic done events, not partial legacy events. + ev_reasoning_summary_text_delta("partial"), + summary_done("reasoning-1", 0, "step one"), + summary_done("reasoning-1", 1, "step two"), + ev_reasoning_item("reasoning-1", &["step one", "step two"], &[]), + ev_message_item_added("message-1", ""), + // Drop a stale completion rather than attaching it to the active message. + summary_done("reasoning-1", 2, "late step"), + ev_output_text_delta("Done"), + ev_assistant_message("message-1", "Done"), + ev_completed("resp-1"), + ]), + ) + .await; + + codex + .submit(Op::UserInput { + items: vec![UserInput::Text { + text: "reason through it".into(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + responsesapi_client_metadata: None, + additional_context: Default::default(), + thread_settings: Default::default(), + }) + .await?; + + let reasoning_item = wait_for_event_match(&codex, |ev| match ev { + EventMsg::ItemStarted(ItemStartedEvent { + item: TurnItem::Reasoning(item), + .. + }) => Some(item.clone()), + _ => None, + }) + .await; + + let mut summary_deltas = Vec::new(); + let mut summary_sections = Vec::new(); + loop { + match wait_for_event(&codex, |_| true).await { + EventMsg::ReasoningContentDelta(event) => { + summary_deltas.push((event.item_id, event.delta, event.summary_index)); + } + EventMsg::AgentReasoningSectionBreak(event) => { + summary_sections.push((event.item_id, event.summary_index)); + } + EventMsg::TurnComplete(_) => break, + _ => {} + } + } + + assert_eq!( + summary_deltas, + vec![ + (reasoning_item.id.clone(), "step one".to_string(), 0), + (reasoning_item.id, "step two".to_string(), 1), + ] + ); + assert_eq!(summary_sections, vec![("reasoning-1".to_string(), 1)]); + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn reasoning_raw_content_delta_respects_flag() -> anyhow::Result<()> { skip_if_no_network!(Ok(())); diff --git a/codex-rs/features/src/lib.rs b/codex-rs/features/src/lib.rs index 7ab7df18d1..ce5ca459b4 100644 --- a/codex-rs/features/src/lib.rs +++ b/codex-rs/features/src/lib.rs @@ -204,6 +204,8 @@ pub enum Feature { ResizeAllImages, /// Generate Responses API item IDs for client-created history items. ItemIds, + /// Request sequential cutoff reasoning summary delivery. + ConcurrentReasoningSummaries, /// Allow prompting and installing missing MCP dependencies. SkillMcpDependencyInstall, /// Removed compatibility flag for deleted skill env var dependency prompting. @@ -1184,6 +1186,12 @@ pub const FEATURES: &[FeatureSpec] = &[ stage: Stage::UnderDevelopment, default_enabled: false, }, + FeatureSpec { + id: Feature::ConcurrentReasoningSummaries, + key: "concurrent_reasoning_summaries", + stage: Stage::UnderDevelopment, + default_enabled: false, + }, FeatureSpec { id: Feature::SkillMcpDependencyInstall, key: "skill_mcp_dependency_install", diff --git a/codex-rs/memories/write/src/runtime.rs b/codex-rs/memories/write/src/runtime.rs index 6c7ad3978f..ae0903cfdc 100644 --- a/codex-rs/memories/write/src/runtime.rs +++ b/codex-rs/memories/write/src/runtime.rs @@ -261,6 +261,7 @@ impl MemoryStartupContext { config.features.enabled(Feature::RuntimeMetrics), /*beta_features_header*/ None, config.features.enabled(Feature::ItemIds), + /*concurrent_reasoning_summaries_enabled*/ false, /*attestation_provider*/ None, ); diff --git a/codex-rs/otel/src/events/session_telemetry.rs b/codex-rs/otel/src/events/session_telemetry.rs index 440f8f04a5..c470f8dc7c 100644 --- a/codex-rs/otel/src/events/session_telemetry.rs +++ b/codex-rs/otel/src/events/session_telemetry.rs @@ -1195,6 +1195,7 @@ impl SessionTelemetry { ResponseEvent::OutputTextDelta(_) => "text_delta".into(), ResponseEvent::ToolCallInputDelta { .. } => "tool_input_delta".into(), ResponseEvent::ReasoningSummaryDelta { .. } => "reasoning_summary_delta".into(), + ResponseEvent::ReasoningSummaryDone { .. } => "reasoning_summary_done".into(), ResponseEvent::ReasoningContentDelta { .. } => "reasoning_content_delta".into(), ResponseEvent::ReasoningSummaryPartAdded { .. } => { "reasoning_summary_part_added".into()