[codex] Simplify concurrent reasoning summary rollout

codex-rs/core/src/client.rs

@@ -206,6 +206,7 @@ struct ModelClientState {
     include_timing_metrics: bool,
     beta_features_header: Option<String>,
     item_ids_enabled: bool,
+    parallel_reasoning_summaries_enabled: bool,
     include_attestation: bool,
     attestation_provider: Option<Arc<dyn AttestationProvider>>,
     disable_websockets: AtomicBool,
@@ -399,16 +400,6 @@ fn sideband_websocket_auth_headers(api_auth: &dyn AuthProvider) -> ApiHeaderMap
 }
 
 impl ModelClient {
-    pub(crate) fn should_request_concurrent_reasoning_summaries(
-        provider_info: &ModelProviderInfo,
-        model_info: &ModelInfo,
-        summary: ReasoningSummaryConfig,
-    ) -> bool {
-        provider_info.is_openai()
-            && model_info.supports_reasoning_summaries
-            && summary != ReasoningSummaryConfig::None
-    }
-
     #[allow(clippy::too_many_arguments)]
     /// Creates a new session-scoped `ModelClient`.
     ///
@@ -426,6 +417,7 @@ impl ModelClient {
         include_timing_metrics: bool,
         beta_features_header: Option<String>,
         item_ids_enabled: bool,
+        parallel_reasoning_summaries_enabled: bool,
         attestation_provider: Option<Arc<dyn AttestationProvider>>,
     ) -> Self {
         let model_provider = create_model_provider(provider_info, auth_manager);
@@ -448,6 +440,7 @@ impl ModelClient {
                 include_timing_metrics,
                 beta_features_header,
                 item_ids_enabled,
+                parallel_reasoning_summaries_enabled,
                 include_attestation,
                 attestation_provider,
                 disable_websockets: AtomicBool::new(false),
@@ -865,16 +858,11 @@ impl ModelClient {
         } else {
             (prompt.base_instructions.text.clone(), Some(tools))
         };
-        // Keep setup and background requests byte-for-byte compatible; only normal turns
-        // consume concurrent summary delivery in the turn event loop.
-        let stream_options = (matches!(
-            responses_metadata.request_kind,
-            Some(CodexResponsesRequestKind::Turn)
-        ) && Self::should_request_concurrent_reasoning_summaries(
-            self.state.provider.info(),
-            model_info,
-            summary,
-        ))
+        let stream_options = (self.state.parallel_reasoning_summaries_enabled
+            && matches!(
+                responses_metadata.request_kind,
+                Some(CodexResponsesRequestKind::Turn)
+            ))
         .then_some(StreamOptions {
             reasoning_summary_delivery: codex_api::ReasoningSummaryDelivery::ConcurrentCutoff,
         });
@@ -1109,6 +1097,10 @@ impl ModelClientSession {
         Arc::clone(&self.turn_state)
     }
 
+    pub(crate) fn uses_concurrent_reasoning_summaries(&self) -> bool {
+        self.client.state.parallel_reasoning_summaries_enabled
+    }
+
     fn reset_websocket_session(&mut self) {
         self.websocket_session.connection = None;
         self.websocket_session.last_request = None;
@@ -1571,7 +1563,6 @@ impl ModelClientSession {
             };
             if warmup {
                 ws_payload.generate = Some(false);
-                ws_payload.stream_options = None;
             }
 
             match self

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,
+        /*parallel_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,
+        /*parallel_reasoning_summaries_enabled*/ false,
         /*attestation_provider*/ None,
     );
     let prompt = Prompt {
@@ -943,6 +945,7 @@ fn model_client_with_counting_attestation(
         /*include_timing_metrics*/ false,
         /*beta_features_header*/ None,
         /*item_ids_enabled*/ false,
+        /*parallel_reasoning_summaries_enabled*/ false,
         Some(Arc::new(CountingAttestationProvider {
             calls: attestation_calls.clone(),
         })),

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),
+                    /*parallel_reasoning_summaries_enabled*/ config
+                        .features
+                        .enabled(Feature::ParallelReasoningSummaries),
                     attestation_provider,
                 )
                 .with_prompt_cache_key_override(

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,
+        /*parallel_reasoning_summaries_enabled*/ false,
         /*attestation_provider*/ None,
     )
     .new_session()
@@ -5443,6 +5444,8 @@ 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),
+            /*parallel_reasoning_summaries_enabled*/
+            config.features.enabled(Feature::ParallelReasoningSummaries),
             /*attestation_provider*/ None,
         ),
         code_mode_service: crate::tools::code_mode::CodeModeService::new(Arc::new(
@@ -7569,6 +7572,8 @@ where
             config.features.enabled(Feature::RuntimeMetrics),
             Session::build_model_client_beta_features_header(config.as_ref()),
             /*item_ids_enabled*/ config.features.enabled(Feature::ItemIds),
+            /*parallel_reasoning_summaries_enabled*/
+            config.features.enabled(Feature::ParallelReasoningSummaries),
             /*attestation_provider*/ None,
         ),
         code_mode_service: crate::tools::code_mode::CodeModeService::new(Arc::new(
@@ -8761,9 +8766,13 @@ async fn handle_output_item_done_records_image_save_history_message() {
         tool_runtime: test_tool_runtime(Arc::clone(&session), Arc::clone(&turn_context)),
         cancellation_token: CancellationToken::new(),
     };
-    handle_output_item_done(&mut ctx, item.clone(), /*previously_active_item*/ None)
-        .await
-        .expect("image generation item should succeed");
+    handle_output_item_done(
+        &mut ctx,
+        item.clone(),
+        /*item_was_streamed_to_client*/ false,
+    )
+    .await
+    .expect("image generation item should succeed");
 
     let history = session.clone_history().await;
     let image_output_path = crate::stream_events_utils::image_generation_artifact_path(
@@ -8818,9 +8827,13 @@ async fn handle_output_item_done_skips_image_save_message_when_save_fails() {
         tool_runtime: test_tool_runtime(Arc::clone(&session), Arc::clone(&turn_context)),
         cancellation_token: CancellationToken::new(),
     };
-    handle_output_item_done(&mut ctx, item.clone(), /*previously_active_item*/ None)
-        .await
-        .expect("image generation item should still complete");
+    handle_output_item_done(
+        &mut ctx,
+        item.clone(),
+        /*item_was_streamed_to_client*/ false,
+    )
+    .await
+    .expect("image generation item should still complete");
 
     let history = session.clone_history().await;
     let expected = vec![item];
@@ -10449,9 +10462,10 @@ async fn tool_calls_reopen_mailbox_delivery_for_current_turn() {
         cancellation_token: CancellationToken::new(),
     };
 
-    let output = handle_output_item_done(&mut ctx, item, /*previously_active_item*/ None)
-        .await
-        .expect("tool call should be handled");
+    let output =
+        handle_output_item_done(&mut ctx, item, /*item_was_streamed_to_client*/ false)
+            .await
+            .expect("tool call should be handled");
 
     assert!(output.needs_follow_up);
     assert!(output.tool_future.is_some());

codex-rs/core/src/session/turn.rs

@@ -6,7 +6,6 @@ use std::sync::atomic::Ordering;
 
 use crate::SkillInjections;
 use crate::build_skill_injections;
-use crate::client::ModelClient;
 use crate::client::ModelClientSession;
 use crate::client_common::Prompt;
 use crate::client_common::ResponseEvent;
@@ -1780,7 +1779,7 @@ async fn emit_turn_item_in_plan_mode(
     sess: &Session,
     turn_context: &TurnContext,
     turn_item: TurnItem,
-    previously_active_item: Option<&TurnItem>,
+    item_was_streamed_to_client: bool,
     state: &mut PlanModeStreamState,
 ) {
     match turn_item {
@@ -1788,7 +1787,7 @@ async fn emit_turn_item_in_plan_mode(
             emit_agent_message_in_plan_mode(sess, turn_context, agent_message, state).await;
         }
         _ => {
-            if previously_active_item.is_none() {
+            if !item_was_streamed_to_client {
                 sess.emit_turn_item_started(turn_context, &turn_item).await;
             }
             sess.emit_turn_item_completed(turn_context, turn_item).await;
@@ -1803,7 +1802,7 @@ async fn handle_assistant_item_done_in_plan_mode(
     turn_store: &codex_extension_api::ExtensionData,
     item: &ResponseItem,
     state: &mut PlanModeStreamState,
-    previously_active_item: Option<&TurnItem>,
+    item_was_streamed_to_client: bool,
     last_agent_message: &mut Option<String>,
 ) -> bool {
     if let ResponseItem::Message { role, .. } = item
@@ -1826,7 +1825,7 @@ async fn handle_assistant_item_done_in_plan_mode(
                 sess,
                 turn_context,
                 finalized_turn_item.turn_item,
-                previously_active_item,
+                item_was_streamed_to_client,
                 state,
             )
             .await;
@@ -1910,15 +1909,7 @@ 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_concurrent_reasoning_summaries = matches!(
-        responses_metadata.request_kind,
-        Some(CodexResponsesRequestKind::Turn)
-    )
-        && ModelClient::should_request_concurrent_reasoning_summaries(
-            turn_context.provider.info(),
-            &turn_context.model_info,
-            turn_context.reasoning_summary,
-        );
+    let uses_concurrent_reasoning_summaries = client_session.uses_concurrent_reasoning_summaries();
     let mut stream = client_session
         .stream(
             prompt,
@@ -1939,6 +1930,10 @@ async fn try_run_sampling_request(
     let mut last_agent_message: Option<String> = None;
     let mut active_item: Option<TurnItem> = None;
     let mut current_response_item_id: Option<String> = None;
+    // Concurrent cutoff can complete an older reasoning item after a newer item starts.
+    // Keep only the IDs needed to avoid emitting a second start when those done events
+    // are reordered back into response order.
+    let mut streamed_response_item_ids = HashSet::<String>::new();
     let mut active_tool_argument_diff_consumer: Option<(
         String,
         Box<dyn ToolArgumentDiffConsumer>,
@@ -1997,28 +1992,60 @@ async fn try_run_sampling_request(
         match event {
             ResponseEvent::Created => {}
             ResponseEvent::OutputItemDone { item, .. } => {
-                if let Some((_, mut consumer)) = active_tool_argument_diff_consumer.take()
+                let completed_item_id = item.id().map(str::to_owned);
+                let completes_current_response_item = match (
+                    completed_item_id.as_deref(),
+                    current_response_item_id.as_deref(),
+                ) {
+                    (Some(completed_item_id), Some(current_response_item_id)) => {
+                        completed_item_id == current_response_item_id
+                    }
+                    // Preserve the old fallback for providers that omit item IDs.
+                    (None, _) | (_, None) => true,
+                };
+                if completes_current_response_item
+                    && let Some((_, mut consumer)) = active_tool_argument_diff_consumer.take()
                     && let Ok(Some(event)) = consumer.finish()
                 {
                     sess.send_event(&turn_context, event).await;
                 }
-                let previously_active_item = active_item.take();
-                let previously_streamed_item = if active_item_is_streaming_to_client {
-                    previously_active_item
+
+                let completes_active_item =
+                    match (completed_item_id.as_deref(), active_item.as_ref()) {
+                        (Some(completed_item_id), Some(active_item)) => {
+                            completed_item_id == active_item.id()
+                        }
+                        // Preserve the old fallback for providers that omit item IDs.
+                        (None, Some(_)) => true,
+                        _ => false,
+                    };
+                let item_was_streamed_to_client = completed_item_id
+                    .as_deref()
+                    .is_some_and(|item_id| streamed_response_item_ids.remove(item_id))
+                    || (completed_item_id.is_none()
+                        && completes_active_item
+                        && active_item_is_streaming_to_client);
+                let previously_active_item = if completes_active_item {
+                    active_item.take()
                 } else {
                     None
                 };
-                active_item_is_streaming_to_client = false;
-                if let Some(previous) = previously_streamed_item.as_ref()
-                    && matches!(previous, TurnItem::AgentMessage(_))
+                if completes_active_item {
+                    active_item_is_streaming_to_client = false;
+                }
+                let completed_or_active_item_id = completed_item_id
+                    .clone()
+                    .or_else(|| previously_active_item.as_ref().map(TurnItem::id));
+                if item_was_streamed_to_client
+                    && matches!(&item, ResponseItem::Message { .. })
+                    && let Some(item_id) = completed_or_active_item_id.as_deref()
                 {
-                    let item_id = previous.id();
                     flush_assistant_text_segments_for_item(
                         &sess,
                         &turn_context,
                         plan_mode_state.as_mut(),
                         &mut assistant_message_stream_parsers,
-                        &item_id,
+                        item_id,
                     )
                     .await;
                 }
@@ -2029,7 +2056,7 @@ async fn try_run_sampling_request(
                         turn_store.as_ref(),
                         &item,
                         state,
-                        previously_streamed_item.as_ref(),
+                        item_was_streamed_to_client,
                         &mut last_agent_message,
                     )
                     .await
@@ -2068,7 +2095,7 @@ async fn try_run_sampling_request(
                 };
 
                 let output_result =
-                    match handle_output_item_done(&mut ctx, item, previously_streamed_item)
+                    match handle_output_item_done(&mut ctx, item, item_was_streamed_to_client)
                         .instrument(handle_responses)
                         .await
                     {
@@ -2140,6 +2167,11 @@ async fn try_run_sampling_request(
                         seeded_item_id = Some(item_id);
                     }
                     if stream_item_to_client {
+                        streamed_response_item_ids.insert(
+                            item.id()
+                                .map(str::to_owned)
+                                .unwrap_or_else(|| turn_item.id()),
+                        );
                         if let Some(state) = plan_mode_state.as_mut()
                             && matches!(turn_item, TurnItem::AgentMessage(_))
                         {

codex-rs/core/src/session/turn_tests.rs

@@ -54,7 +54,7 @@ async fn plan_mode_uses_contributed_turn_item_for_last_agent_message() {
         &turn_store,
         &item,
         &mut state,
-        /*previously_active_item*/ None,
+        /*item_was_streamed_to_client*/ false,
         &mut last_agent_message,
     )
     .await;

codex-rs/core/src/stream_events_utils.rs

@@ -405,7 +405,7 @@ pub(crate) async fn finalize_non_tool_response_item(
 pub(crate) async fn handle_output_item_done(
     ctx: &mut HandleOutputCtx,
     item: ResponseItem,
-    previously_active_item: Option<TurnItem>,
+    item_was_streamed_to_client: bool,
 ) -> Result<OutputItemResult> {
     let mut output = OutputItemResult::default();
     let plan_mode = ctx.turn_context.collaboration_mode.mode == ModeKind::Plan;
@@ -456,7 +456,7 @@ pub(crate) async fn handle_output_item_done(
                 .as_ref()
                 .map(|finalized| finalized.facts.clone());
             if let Some(finalized_turn_item) = finalized_turn_item {
-                if previously_active_item.is_none() {
+                if !item_was_streamed_to_client {
                     let mut started_item = finalized_turn_item.turn_item.clone();
                     if let TurnItem::ImageGeneration(item) = &mut started_item {
                         item.status = "in_progress".to_string();

codex-rs/core/src/stream_events_utils_tests.rs

@@ -304,9 +304,10 @@ async fn handle_output_item_done_returns_contributed_last_agent_message() {
         cancellation_token: CancellationToken::new(),
     };
 
-    let output = handle_output_item_done(&mut ctx, item, /*previously_active_item*/ None)
-        .await
-        .expect("assistant message should complete");
+    let output =
+        handle_output_item_done(&mut ctx, item, /*item_was_streamed_to_client*/ false)
+            .await
+            .expect("assistant message should complete");
 
     assert_eq!(
         output.last_agent_message.as_deref(),

codex-rs/core/src/turn_timing.rs

@@ -334,7 +334,6 @@ fn response_event_records_turn_ttft(event: &ResponseEvent) -> bool {
         }
         ResponseEvent::OutputTextDelta(_)
         | ResponseEvent::ReasoningSummaryDelta { .. }
-        | ResponseEvent::ReasoningSummaryDone { .. }
         | ResponseEvent::ReasoningContentDelta { .. } => true,
         ResponseEvent::Created
         | ResponseEvent::ServerModel(_)
@@ -344,6 +343,7 @@ fn response_event_records_turn_ttft(event: &ResponseEvent) -> bool {
         | ResponseEvent::ServerReasoningIncluded(_)
         | ResponseEvent::ToolCallInputDelta { .. }
         | ResponseEvent::Completed { .. }
+        | ResponseEvent::ReasoningSummaryDone { .. }
         | ResponseEvent::ReasoningSummaryPartAdded { .. }
         | ResponseEvent::RateLimits(_)
         | ResponseEvent::ModelsEtag(_) => false,

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,
+        /*parallel_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,
+        /*parallel_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,
+        /*parallel_reasoning_summaries_enabled*/ false,
         /*attestation_provider*/ None,
     );
     let responses_metadata = test_turn_responses_metadata(&client, thread_id, &session_source);

codex-rs/core/tests/suite/client.rs

@@ -1334,6 +1334,8 @@ 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),
+        /*parallel_reasoning_summaries_enabled*/
+        config.features.enabled(Feature::ParallelReasoningSummaries),
         /*attestation_provider*/ None,
     );
     let responses_metadata = test_turn_responses_metadata(&client, thread_id);
@@ -2376,6 +2378,7 @@ 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::ParallelReasoningSummaries);
         })
         .build(&server)
         .await?;
@@ -2956,6 +2959,7 @@ async fn azure_responses_request_includes_store_and_reasoning_ids() {
         /*include_timing_metrics*/ false,
         /*beta_features_header*/ None,
         /*item_ids_enabled*/ false,
+        /*parallel_reasoning_summaries_enabled*/ false,
         /*attestation_provider*/ None,
     );
     let responses_metadata = test_turn_responses_metadata(&client, thread_id);

codex-rs/core/tests/suite/client_websockets.rs

@@ -388,10 +388,11 @@ async fn responses_websocket_request_prewarm_reuses_connection() {
 
     let mut provider = websocket_provider(&server);
     provider.name = "OpenAI".to_string();
-    let mut harness =
-        websocket_harness_with_provider_options(provider, /*runtime_metrics_enabled*/ true).await;
-    harness.model_info.supports_reasoning_summaries = true;
-    harness.summary = ReasoningSummary::Concise;
+    let harness = websocket_harness_with_provider_options(
+        provider, /*runtime_metrics_enabled*/ true,
+        /*parallel_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);
@@ -446,7 +447,6 @@ async fn responses_websocket_request_prewarm_reuses_connection() {
         follow_up["stream_options"]["reasoning_summary_delivery"].as_str(),
         Some("concurrent_cutoff")
     );
-
     server.shutdown().await;
 }
 
@@ -1000,8 +1000,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,
+        /*parallel_reasoning_summaries_enabled*/ false,
+    )
+    .await;
 
     // Turn one: initiate
     let prompt_one = prompt_with_input(vec![message_item("hello")]);
@@ -2205,13 +2208,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,
+        /*parallel_reasoning_summaries_enabled*/ false,
+    )
+    .await
 }
 
 async fn websocket_harness_with_provider_options(
     provider: ModelProviderInfo,
     runtime_metrics_enabled: bool,
+    parallel_reasoning_summaries_enabled: bool,
 ) -> WebsocketTestHarness {
     let codex_home = TempDir::new().unwrap();
     let mut config = load_default_config_for_test(&codex_home).await;
@@ -2222,6 +2230,12 @@ async fn websocket_harness_with_provider_options(
             .enable(Feature::RuntimeMetrics)
             .expect("test config should allow feature update");
     }
+    if parallel_reasoning_summaries_enabled {
+        config
+            .features
+            .enable(Feature::ParallelReasoningSummaries)
+            .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();
@@ -2261,6 +2275,8 @@ async fn websocket_harness_with_provider_options(
         runtime_metrics_enabled,
         /*beta_features_header*/ None,
         /*item_ids_enabled*/ config.features.enabled(Feature::ItemIds),
+        /*parallel_reasoning_summaries_enabled*/
+        config.features.enabled(Feature::ParallelReasoningSummaries),
         /*attestation_provider*/ None,
     );
 

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::ReasoningSummary;
@@ -1085,6 +1086,63 @@ async fn plan_mode_handles_missing_plan_close_tag() -> anyhow::Result<()> {
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn reasoning_content_delta_has_item_metadata() -> anyhow::Result<()> {
+    skip_if_no_network!(Ok(()));
+
+    let server = start_mock_server().await;
+
+    let TestCodex { codex, .. } = test_codex().build(&server).await?;
+
+    let stream = sse(vec![
+        ev_response_created("resp-1"),
+        ev_reasoning_item_added("reasoning-1", &[""]),
+        ev_reasoning_summary_text_delta("step one"),
+        ev_reasoning_item("reasoning-1", &["step one"], &[]),
+        ev_completed("resp-1"),
+    ]);
+    let response_mock = mount_sse_once(&server, stream).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 delta_event = wait_for_event_match(&codex, |ev| match ev {
+        EventMsg::ReasoningContentDelta(event) => Some(event.clone()),
+        _ => None,
+    })
+    .await;
+    assert_eq!(delta_event.item_id, reasoning_item.id);
+    assert_eq!(delta_event.delta, "step one");
+    assert_eq!(
+        response_mock
+            .single_request()
+            .body_json()
+            .get("stream_options"),
+        None
+    );
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn concurrent_cutoff_emits_only_current_reasoning_summary_done() -> anyhow::Result<()> {
     skip_if_no_network!(Ok(()));
@@ -1097,6 +1155,7 @@ async fn concurrent_cutoff_emits_only_current_reasoning_summary_done() -> anyhow
         })
         .with_config(|config| {
             config.model_reasoning_summary = Some(ReasoningSummary::Auto);
+            let _ = config.features.enable(Feature::ParallelReasoningSummaries);
         })
         .build(&server)
         .await?;
@@ -1124,14 +1183,14 @@ async fn concurrent_cutoff_emits_only_current_reasoning_summary_done() -> anyhow
             // Once a later output item starts, summaries for the earlier item are stale.
             summary_done("reasoning-1", 1, "late step"),
             ev_output_text_delta("Done"),
-            with_output_index(
-                ev_reasoning_item("reasoning-1", &["step one", "late step"], &[]),
-                /*output_index*/ 0,
-            ),
             with_output_index(
                 ev_assistant_message("message-1", "Done"),
                 /*output_index*/ 1,
             ),
+            with_output_index(
+                ev_reasoning_item("reasoning-1", &["step one", "late step"], &[]),
+                /*output_index*/ 0,
+            ),
             ev_completed("resp-1"),
         ]),
     )
@@ -1159,10 +1218,22 @@ async fn concurrent_cutoff_emits_only_current_reasoning_summary_done() -> anyhow
     })
     .await;
 
+    let mut started_item_ids = vec![reasoning_item.id.clone()];
+    let mut completed_item_ids = Vec::new();
     let mut summary_deltas = Vec::new();
     let mut summary_sections = Vec::new();
     loop {
         match wait_for_event(&codex, |_| true).await {
+            EventMsg::ItemStarted(ItemStartedEvent { item, .. })
+                if matches!(&item, TurnItem::Reasoning(_) | TurnItem::AgentMessage(_)) =>
+            {
+                started_item_ids.push(item.id());
+            }
+            EventMsg::ItemCompleted(ItemCompletedEvent { item, .. })
+                if matches!(&item, TurnItem::Reasoning(_) | TurnItem::AgentMessage(_)) =>
+            {
+                completed_item_ids.push(item.id());
+            }
             EventMsg::ReasoningContentDelta(event) => {
                 summary_deltas.push((event.item_id, event.delta, event.summary_index));
             }
@@ -1178,6 +1249,8 @@ async fn concurrent_cutoff_emits_only_current_reasoning_summary_done() -> anyhow
         summary_deltas,
         vec![(reasoning_item.id, "step one".to_string(), 0)]
     );
+    assert_eq!(started_item_ids, vec!["reasoning-1", "message-1"]);
+    assert_eq!(completed_item_ids, vec!["reasoning-1", "message-1"]);
     assert_eq!(summary_sections, Vec::new());
     assert_eq!(
         response_mock.single_request().body_json()["stream_options"]["reasoning_summary_delivery"]

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 concurrent cutoff reasoning summary delivery.
+    ParallelReasoningSummaries,
     /// 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::ParallelReasoningSummaries,
+        key: "parallel_reasoning_summaries",
+        stage: Stage::UnderDevelopment,
+        default_enabled: false,
+    },
     FeatureSpec {
         id: Feature::SkillMcpDependencyInstall,
         key: "skill_mcp_dependency_install",