From 8d7549e2f4ffbb5359b6ac34e39d4ee7658431db Mon Sep 17 00:00:00 2001 From: ashwinnathan-openai Date: Mon, 6 Jul 2026 19:26:34 +0000 Subject: [PATCH] [codex] Simplify concurrent reasoning summary rollout --- codex-rs/core/src/client.rs | 33 +++----- codex-rs/core/src/client_tests.rs | 3 + codex-rs/core/src/session/session.rs | 3 + codex-rs/core/src/session/tests.rs | 32 ++++++-- codex-rs/core/src/session/turn.rs | 82 +++++++++++++------ codex-rs/core/src/session/turn_tests.rs | 2 +- codex-rs/core/src/stream_events_utils.rs | 4 +- .../core/src/stream_events_utils_tests.rs | 7 +- codex-rs/core/src/turn_timing.rs | 2 +- codex-rs/core/tests/responses_headers.rs | 3 + codex-rs/core/tests/suite/client.rs | 4 + .../core/tests/suite/client_websockets.rs | 34 ++++++-- codex-rs/core/tests/suite/items.rs | 81 +++++++++++++++++- codex-rs/features/src/lib.rs | 8 ++ 14 files changed, 223 insertions(+), 75 deletions(-) diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index 63883c8d7c..0434414c04 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -206,6 +206,7 @@ struct ModelClientState { include_timing_metrics: bool, beta_features_header: Option, item_ids_enabled: bool, + parallel_reasoning_summaries_enabled: bool, include_attestation: bool, attestation_provider: Option>, 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, item_ids_enabled: bool, + parallel_reasoning_summaries_enabled: bool, attestation_provider: Option>, ) -> 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 diff --git a/codex-rs/core/src/client_tests.rs b/codex-rs/core/src/client_tests.rs index 6217f27400..2d1afac4a7 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, + /*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(), })), diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 48a9ee447e..0b4f3f95c9 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), + /*parallel_reasoning_summaries_enabled*/ config + .features + .enabled(Feature::ParallelReasoningSummaries), 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 9ae29a0917..3992feaafb 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, + /*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()); diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 5faf5b1b9c..6eba9015d8 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/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, ) -> 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 = None; let mut active_item: Option = None; let mut current_response_item_id: Option = 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::::new(); let mut active_tool_argument_diff_consumer: Option<( String, Box, @@ -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(_)) { diff --git a/codex-rs/core/src/session/turn_tests.rs b/codex-rs/core/src/session/turn_tests.rs index 051799cf85..7b45c32b5c 100644 --- a/codex-rs/core/src/session/turn_tests.rs +++ b/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; diff --git a/codex-rs/core/src/stream_events_utils.rs b/codex-rs/core/src/stream_events_utils.rs index b07c180b16..adc6357ed7 100644 --- a/codex-rs/core/src/stream_events_utils.rs +++ b/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, + item_was_streamed_to_client: bool, ) -> Result { 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(); diff --git a/codex-rs/core/src/stream_events_utils_tests.rs b/codex-rs/core/src/stream_events_utils_tests.rs index b3312ad49d..969bcd7699 100644 --- a/codex-rs/core/src/stream_events_utils_tests.rs +++ b/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(), diff --git a/codex-rs/core/src/turn_timing.rs b/codex-rs/core/src/turn_timing.rs index e291ce45cb..04b2f8eebc 100644 --- a/codex-rs/core/src/turn_timing.rs +++ b/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, diff --git a/codex-rs/core/tests/responses_headers.rs b/codex-rs/core/tests/responses_headers.rs index 5b7c2c65b8..165fa9034a 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, + /*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); diff --git a/codex-rs/core/tests/suite/client.rs b/codex-rs/core/tests/suite/client.rs index ef16a83cf2..f28e017b54 100644 --- a/codex-rs/core/tests/suite/client.rs +++ b/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); diff --git a/codex-rs/core/tests/suite/client_websockets.rs b/codex-rs/core/tests/suite/client_websockets.rs index 4c1313bb8f..702171423e 100755 --- a/codex-rs/core/tests/suite/client_websockets.rs +++ b/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, ); diff --git a/codex-rs/core/tests/suite/items.rs b/codex-rs/core/tests/suite/items.rs index dfd4337026..a9196e05f2 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::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"] diff --git a/codex-rs/features/src/lib.rs b/codex-rs/features/src/lib.rs index 7ab7df18d1..03ea6a62f8 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 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",