mirror of
https://github.com/openai/codex.git
synced 2026-11-06 20:20:34 +00:00
[codex] Tighten concurrent reasoning summary handling
This commit is contained in:
+57
-21
@@ -23,6 +23,7 @@
|
||||
//! WebSocket prewarm is treated as the first websocket connection attempt for a turn. If it
|
||||
//! fails, normal stream retry/fallback logic handles recovery on the same turn.
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
use std::sync::Mutex as StdMutex;
|
||||
@@ -112,7 +113,6 @@ use crate::client_common::Prompt;
|
||||
use crate::client_common::ResponseEvent;
|
||||
use crate::client_common::ResponseStream;
|
||||
use crate::feedback_tags;
|
||||
use crate::response_event_buffer::OutputItemDoneBuffer;
|
||||
use crate::responses_metadata::CodexResponsesMetadata;
|
||||
use crate::responses_metadata::CodexResponsesRequestKind;
|
||||
use crate::responses_metadata::subagent_header_value;
|
||||
@@ -206,7 +206,7 @@ struct ModelClientState {
|
||||
include_timing_metrics: bool,
|
||||
beta_features_header: Option<String>,
|
||||
item_ids_enabled: bool,
|
||||
parallel_reasoning_summaries_enabled: bool,
|
||||
concurrent_reasoning_summaries_enabled: bool,
|
||||
include_attestation: bool,
|
||||
attestation_provider: Option<Arc<dyn AttestationProvider>>,
|
||||
disable_websockets: AtomicBool,
|
||||
@@ -417,7 +417,7 @@ impl ModelClient {
|
||||
include_timing_metrics: bool,
|
||||
beta_features_header: Option<String>,
|
||||
item_ids_enabled: bool,
|
||||
parallel_reasoning_summaries_enabled: bool,
|
||||
concurrent_reasoning_summaries_enabled: bool,
|
||||
attestation_provider: Option<Arc<dyn AttestationProvider>>,
|
||||
) -> Self {
|
||||
let model_provider = create_model_provider(provider_info, auth_manager);
|
||||
@@ -440,7 +440,7 @@ impl ModelClient {
|
||||
include_timing_metrics,
|
||||
beta_features_header,
|
||||
item_ids_enabled,
|
||||
parallel_reasoning_summaries_enabled,
|
||||
concurrent_reasoning_summaries_enabled,
|
||||
include_attestation,
|
||||
attestation_provider,
|
||||
disable_websockets: AtomicBool::new(false),
|
||||
@@ -858,7 +858,7 @@ impl ModelClient {
|
||||
} else {
|
||||
(prompt.base_instructions.text.clone(), Some(tools))
|
||||
};
|
||||
let stream_options = (self.state.parallel_reasoning_summaries_enabled
|
||||
let stream_options = (self.state.concurrent_reasoning_summaries_enabled
|
||||
&& matches!(
|
||||
responses_metadata.request_kind,
|
||||
Some(CodexResponsesRequestKind::Turn)
|
||||
@@ -1098,7 +1098,7 @@ impl ModelClientSession {
|
||||
}
|
||||
|
||||
pub(crate) fn uses_concurrent_reasoning_summaries(&self) -> bool {
|
||||
self.client.state.parallel_reasoning_summaries_enabled
|
||||
self.client.state.concurrent_reasoning_summaries_enabled
|
||||
}
|
||||
|
||||
fn reset_websocket_session(&mut self) {
|
||||
@@ -1899,7 +1899,51 @@ fn add_responses_lite_header(headers: &mut ApiHeaderMap, use_responses_lite: boo
|
||||
const RESPONSE_STREAM_CHANNEL_CAPACITY: usize = 1600;
|
||||
const STREAM_DROPPED_REASON: &str = "response stream dropped before provider terminal event";
|
||||
|
||||
async fn send_output_item_done_events(
|
||||
/// Restores response order for completed items without delaying incremental events.
|
||||
#[derive(Default)]
|
||||
struct OutputItemDoneBuffer {
|
||||
next_output_index: usize,
|
||||
pending: BTreeMap<usize, ResponseItem>,
|
||||
}
|
||||
|
||||
impl OutputItemDoneBuffer {
|
||||
fn push(
|
||||
&mut self,
|
||||
item: ResponseItem,
|
||||
output_index: Option<usize>,
|
||||
) -> Vec<(ResponseItem, Option<usize>)> {
|
||||
let Some(output_index) = output_index else {
|
||||
return vec![(item, None)];
|
||||
};
|
||||
|
||||
if output_index < self.next_output_index {
|
||||
return Vec::new();
|
||||
}
|
||||
|
||||
self.pending.insert(output_index, item);
|
||||
self.take_ready()
|
||||
}
|
||||
|
||||
fn finish(&mut self) -> Vec<(ResponseItem, Option<usize>)> {
|
||||
let pending = std::mem::take(&mut self.pending);
|
||||
pending
|
||||
.into_iter()
|
||||
.map(|(output_index, item)| (item, Some(output_index)))
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn take_ready(&mut self) -> Vec<(ResponseItem, Option<usize>)> {
|
||||
let mut ready = Vec::new();
|
||||
while let Some(item) = self.pending.remove(&self.next_output_index) {
|
||||
ready.push((item, Some(self.next_output_index)));
|
||||
self.next_output_index += 1;
|
||||
}
|
||||
ready
|
||||
}
|
||||
}
|
||||
|
||||
/// Keep completed-item history and downstream events in the same order.
|
||||
async fn forward_completed_items(
|
||||
tx_event: &mpsc::Sender<Result<ResponseEvent>>,
|
||||
items_added: &mut Vec<ResponseItem>,
|
||||
ready: Vec<(ResponseItem, Option<usize>)>,
|
||||
@@ -1994,7 +2038,7 @@ where
|
||||
Some(buffer) => buffer.push(item, output_index),
|
||||
None => vec![(item, output_index)],
|
||||
};
|
||||
if !send_output_item_done_events(&tx_event, &mut items_added, ready).await {
|
||||
if !forward_completed_items(&tx_event, &mut items_added, ready).await {
|
||||
inference_trace_attempt.record_cancelled(
|
||||
STREAM_DROPPED_REASON,
|
||||
upstream_request_id,
|
||||
@@ -2009,12 +2053,8 @@ where
|
||||
end_turn,
|
||||
}) => {
|
||||
if let Some(buffer) = output_item_done_buffer.as_mut() {
|
||||
if !send_output_item_done_events(
|
||||
&tx_event,
|
||||
&mut items_added,
|
||||
buffer.finish(),
|
||||
)
|
||||
.await
|
||||
if !forward_completed_items(&tx_event, &mut items_added, buffer.finish())
|
||||
.await
|
||||
{
|
||||
inference_trace_attempt.record_cancelled(
|
||||
STREAM_DROPPED_REASON,
|
||||
@@ -2076,12 +2116,8 @@ where
|
||||
}
|
||||
Err(err) => {
|
||||
if let Some(buffer) = output_item_done_buffer.as_mut()
|
||||
&& !send_output_item_done_events(
|
||||
&tx_event,
|
||||
&mut items_added,
|
||||
buffer.finish(),
|
||||
)
|
||||
.await
|
||||
&& !forward_completed_items(&tx_event, &mut items_added, buffer.finish())
|
||||
.await
|
||||
{
|
||||
inference_trace_attempt.record_cancelled(
|
||||
STREAM_DROPPED_REASON,
|
||||
@@ -2114,7 +2150,7 @@ where
|
||||
}
|
||||
}
|
||||
if let Some(buffer) = output_item_done_buffer.as_mut()
|
||||
&& !send_output_item_done_events(&tx_event, &mut items_added, buffer.finish()).await
|
||||
&& !forward_completed_items(&tx_event, &mut items_added, buffer.finish()).await
|
||||
{
|
||||
inference_trace_attempt.record_cancelled(
|
||||
STREAM_DROPPED_REASON,
|
||||
|
||||
@@ -103,7 +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,
|
||||
/*concurrent_reasoning_summaries_enabled*/ false,
|
||||
/*attestation_provider*/ None,
|
||||
)
|
||||
}
|
||||
@@ -148,7 +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,
|
||||
/*concurrent_reasoning_summaries_enabled*/ false,
|
||||
/*attestation_provider*/ None,
|
||||
);
|
||||
let prompt = Prompt {
|
||||
@@ -945,7 +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,
|
||||
/*concurrent_reasoning_summaries_enabled*/ false,
|
||||
Some(Arc::new(CountingAttestationProvider {
|
||||
calls: attestation_calls.clone(),
|
||||
})),
|
||||
|
||||
@@ -12,7 +12,6 @@ mod client_common;
|
||||
mod realtime_context;
|
||||
mod realtime_conversation;
|
||||
mod realtime_prompt;
|
||||
mod response_event_buffer;
|
||||
mod responses_metadata;
|
||||
mod responses_retry;
|
||||
pub(crate) mod session;
|
||||
|
||||
@@ -1,46 +0,0 @@
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
use codex_protocol::models::ResponseItem;
|
||||
|
||||
/// Restores response order for completed items without delaying incremental events.
|
||||
#[derive(Default)]
|
||||
pub(crate) struct OutputItemDoneBuffer {
|
||||
next_output_index: usize,
|
||||
pending: BTreeMap<usize, ResponseItem>,
|
||||
}
|
||||
|
||||
impl OutputItemDoneBuffer {
|
||||
pub(crate) fn push(
|
||||
&mut self,
|
||||
item: ResponseItem,
|
||||
output_index: Option<usize>,
|
||||
) -> Vec<(ResponseItem, Option<usize>)> {
|
||||
let Some(output_index) = output_index else {
|
||||
return vec![(item, None)];
|
||||
};
|
||||
|
||||
if output_index < self.next_output_index {
|
||||
return vec![(item, Some(output_index))];
|
||||
}
|
||||
|
||||
self.pending.insert(output_index, item);
|
||||
self.take_ready()
|
||||
}
|
||||
|
||||
pub(crate) fn finish(&mut self) -> Vec<(ResponseItem, Option<usize>)> {
|
||||
let pending = std::mem::take(&mut self.pending);
|
||||
pending
|
||||
.into_iter()
|
||||
.map(|(output_index, item)| (item, Some(output_index)))
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn take_ready(&mut self) -> Vec<(ResponseItem, Option<usize>)> {
|
||||
let mut ready = Vec::new();
|
||||
while let Some(item) = self.pending.remove(&self.next_output_index) {
|
||||
ready.push((item, Some(self.next_output_index)));
|
||||
self.next_output_index += 1;
|
||||
}
|
||||
ready
|
||||
}
|
||||
}
|
||||
@@ -1111,9 +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
|
||||
/*concurrent_reasoning_summaries_enabled*/ config
|
||||
.features
|
||||
.enabled(Feature::ParallelReasoningSummaries),
|
||||
.enabled(Feature::ConcurrentReasoningSummaries),
|
||||
attestation_provider,
|
||||
)
|
||||
.with_prompt_cache_key_override(
|
||||
|
||||
@@ -490,7 +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,
|
||||
/*concurrent_reasoning_summaries_enabled*/ false,
|
||||
/*attestation_provider*/ None,
|
||||
)
|
||||
.new_session()
|
||||
@@ -5444,8 +5444,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),
|
||||
/*parallel_reasoning_summaries_enabled*/
|
||||
config.features.enabled(Feature::ParallelReasoningSummaries),
|
||||
/*concurrent_reasoning_summaries_enabled*/
|
||||
config
|
||||
.features
|
||||
.enabled(Feature::ConcurrentReasoningSummaries),
|
||||
/*attestation_provider*/ None,
|
||||
),
|
||||
code_mode_service: crate::tools::code_mode::CodeModeService::new(Arc::new(
|
||||
@@ -7572,8 +7574,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),
|
||||
/*parallel_reasoning_summaries_enabled*/
|
||||
config.features.enabled(Feature::ParallelReasoningSummaries),
|
||||
/*concurrent_reasoning_summaries_enabled*/
|
||||
config
|
||||
.features
|
||||
.enabled(Feature::ConcurrentReasoningSummaries),
|
||||
/*attestation_provider*/ None,
|
||||
),
|
||||
code_mode_service: crate::tools::code_mode::CodeModeService::new(Arc::new(
|
||||
|
||||
@@ -1938,7 +1938,7 @@ async fn try_run_sampling_request(
|
||||
// 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 streamed_response_item_ids = HashSet::new();
|
||||
let mut active_tool_argument_diff_consumer: Option<(
|
||||
String,
|
||||
Box<dyn ToolArgumentDiffConsumer>,
|
||||
@@ -1998,11 +1998,16 @@ async fn try_run_sampling_request(
|
||||
ResponseEvent::Created => {}
|
||||
ResponseEvent::OutputItemDone { item, .. } => {
|
||||
let completed_item_id = item.id().map(str::to_owned);
|
||||
let is_current_response_item = response_item_ids_match_when_present(
|
||||
completed_item_id.as_deref(),
|
||||
current_response_item_id.as_deref(),
|
||||
);
|
||||
if is_current_response_item
|
||||
let completes_active_tool_call = active_tool_argument_diff_consumer
|
||||
.as_ref()
|
||||
.is_some_and(|(active_call_id, _)| {
|
||||
matches!(
|
||||
&item,
|
||||
ResponseItem::CustomToolCall { call_id, .. }
|
||||
if call_id == active_call_id
|
||||
)
|
||||
});
|
||||
if completes_active_tool_call
|
||||
&& let Some((_, mut consumer)) = active_tool_argument_diff_consumer.take()
|
||||
&& let Ok(Some(event)) = consumer.finish()
|
||||
{
|
||||
@@ -2011,10 +2016,18 @@ async fn try_run_sampling_request(
|
||||
|
||||
let active_item_id = active_item.as_ref().map(TurnItem::id);
|
||||
let matches_active_item = active_item.is_some()
|
||||
&& response_item_ids_match_when_present(
|
||||
completed_item_id.as_deref(),
|
||||
active_item_id.as_deref(),
|
||||
);
|
||||
&& if uses_concurrent_reasoning_summaries {
|
||||
completed_item_id
|
||||
.as_deref()
|
||||
.is_some_and(|completed_item_id| {
|
||||
active_item_id.as_deref() == Some(completed_item_id)
|
||||
})
|
||||
} else {
|
||||
response_item_ids_match_when_present(
|
||||
completed_item_id.as_deref(),
|
||||
active_item_id.as_deref(),
|
||||
)
|
||||
};
|
||||
let item_was_streamed_to_client = match completed_item_id.as_deref() {
|
||||
Some(item_id) => streamed_response_item_ids.remove(item_id),
|
||||
None => matches_active_item && active_item_is_streaming_to_client,
|
||||
@@ -2155,11 +2168,9 @@ 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(item_id) = item.id() {
|
||||
streamed_response_item_ids.insert(item_id.to_owned());
|
||||
}
|
||||
if let Some(state) = plan_mode_state.as_mut()
|
||||
&& matches!(turn_item, TurnItem::AgentMessage(_))
|
||||
{
|
||||
@@ -2369,7 +2380,6 @@ async fn try_run_sampling_request(
|
||||
// Concurrent cutoff can cancel a partial summary at the next output-item
|
||||
// boundary. Present only atomic parts that finish while their item is current.
|
||||
if !uses_concurrent_reasoning_summaries
|
||||
|| defer_streamed_turn_items_for_contributors
|
||||
|| current_response_item_id.as_deref() != Some(item_id.as_str())
|
||||
{
|
||||
continue;
|
||||
|
||||
@@ -131,7 +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,
|
||||
/*concurrent_reasoning_summaries_enabled*/ false,
|
||||
/*attestation_provider*/ None,
|
||||
);
|
||||
let responses_metadata = test_turn_responses_metadata(&client, thread_id, &session_source);
|
||||
@@ -266,7 +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,
|
||||
/*concurrent_reasoning_summaries_enabled*/ false,
|
||||
/*attestation_provider*/ None,
|
||||
);
|
||||
let responses_metadata = test_turn_responses_metadata(&client, thread_id, &session_source);
|
||||
@@ -387,7 +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,
|
||||
/*concurrent_reasoning_summaries_enabled*/ false,
|
||||
/*attestation_provider*/ None,
|
||||
);
|
||||
let responses_metadata = test_turn_responses_metadata(&client, thread_id, &session_source);
|
||||
|
||||
@@ -41,6 +41,8 @@ use core_test_support::assert_regex_match;
|
||||
use core_test_support::responses::ev_assistant_message;
|
||||
use core_test_support::responses::ev_completed;
|
||||
use core_test_support::responses::ev_function_call;
|
||||
use core_test_support::responses::ev_reasoning_item;
|
||||
use core_test_support::responses::ev_reasoning_item_added;
|
||||
use core_test_support::responses::ev_response_created;
|
||||
use core_test_support::responses::ev_shell_command_call_with_args;
|
||||
use core_test_support::responses::mount_sse_sequence;
|
||||
@@ -1167,18 +1169,27 @@ async fn apply_patch_custom_tool_streaming_emits_updated_changes() -> Result<()>
|
||||
.features
|
||||
.enable(Feature::ApplyPatchStreamingEvents)
|
||||
.expect("enable apply_patch streaming events");
|
||||
config
|
||||
.features
|
||||
.enable(Feature::ConcurrentReasoningSummaries)
|
||||
.expect("enable concurrent reasoning summaries");
|
||||
})
|
||||
})
|
||||
.await?;
|
||||
let test = harness.test();
|
||||
let codex = test.codex.clone();
|
||||
let call_id = "apply-patch-streaming";
|
||||
let with_output_index = |mut event: serde_json::Value, output_index: usize| {
|
||||
event["output_index"] = serde_json::json!(output_index);
|
||||
event
|
||||
};
|
||||
let patch = "*** Begin Patch\n*** Add File: streamed.txt\n+hello\n+world\n*** End Patch";
|
||||
mount_sse_sequence(
|
||||
harness.server(),
|
||||
vec![
|
||||
sse(vec![
|
||||
ev_response_created("resp-1"),
|
||||
ev_reasoning_item_added("reasoning-1", &[""]),
|
||||
json!({
|
||||
"type": "response.output_item.added",
|
||||
"item": {
|
||||
@@ -1198,12 +1209,20 @@ async fn apply_patch_custom_tool_streaming_emits_updated_changes() -> Result<()>
|
||||
"call_id": call_id,
|
||||
"delta": "*** Add File: streamed.txt\n+hello",
|
||||
}),
|
||||
// The earlier reasoning item can complete while tool input keeps streaming.
|
||||
with_output_index(
|
||||
ev_reasoning_item("reasoning-1", &[], &[]),
|
||||
/*output_index*/ 0,
|
||||
),
|
||||
json!({
|
||||
"type": "response.custom_tool_call_input.delta",
|
||||
"call_id": call_id,
|
||||
"delta": "\n+world\n*** End Patch",
|
||||
}),
|
||||
ev_apply_patch_custom_tool_call(call_id, patch),
|
||||
with_output_index(
|
||||
ev_apply_patch_custom_tool_call(call_id, patch),
|
||||
/*output_index*/ 1,
|
||||
),
|
||||
ev_completed("resp-1"),
|
||||
]),
|
||||
sse(vec![
|
||||
|
||||
@@ -1334,8 +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),
|
||||
/*parallel_reasoning_summaries_enabled*/
|
||||
config.features.enabled(Feature::ParallelReasoningSummaries),
|
||||
/*concurrent_reasoning_summaries_enabled*/
|
||||
config
|
||||
.features
|
||||
.enabled(Feature::ConcurrentReasoningSummaries),
|
||||
/*attestation_provider*/ None,
|
||||
);
|
||||
let responses_metadata = test_turn_responses_metadata(&client, thread_id);
|
||||
@@ -2378,7 +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::ParallelReasoningSummaries);
|
||||
let _ = config
|
||||
.features
|
||||
.enable(Feature::ConcurrentReasoningSummaries);
|
||||
})
|
||||
.build(&server)
|
||||
.await?;
|
||||
@@ -2959,7 +2963,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,
|
||||
/*concurrent_reasoning_summaries_enabled*/ false,
|
||||
/*attestation_provider*/ None,
|
||||
);
|
||||
let responses_metadata = test_turn_responses_metadata(&client, thread_id);
|
||||
|
||||
@@ -386,11 +386,10 @@ async fn responses_websocket_request_prewarm_reuses_connection() {
|
||||
]])
|
||||
.await;
|
||||
|
||||
let mut provider = websocket_provider(&server);
|
||||
provider.name = "OpenAI".to_string();
|
||||
let harness = websocket_harness_with_provider_options(
|
||||
provider, /*runtime_metrics_enabled*/ true,
|
||||
/*parallel_reasoning_summaries_enabled*/ true,
|
||||
websocket_provider(&server),
|
||||
/*runtime_metrics_enabled*/ true,
|
||||
/*concurrent_reasoning_summaries_enabled*/ true,
|
||||
)
|
||||
.await;
|
||||
let mut client_session = harness.client.new_session();
|
||||
@@ -1002,7 +1001,7 @@ async fn responses_websocket_v2_incremental_requests_are_reused_across_turns() {
|
||||
provider.name = ModelProviderInfo::create_openai_provider(/*base_url*/ None).name;
|
||||
let harness = websocket_harness_with_provider_options(
|
||||
provider, /*runtime_metrics_enabled*/ false,
|
||||
/*parallel_reasoning_summaries_enabled*/ false,
|
||||
/*concurrent_reasoning_summaries_enabled*/ false,
|
||||
)
|
||||
.await;
|
||||
|
||||
@@ -2211,7 +2210,7 @@ async fn websocket_harness_with_options(
|
||||
websocket_harness_with_provider_options(
|
||||
websocket_provider(server),
|
||||
runtime_metrics_enabled,
|
||||
/*parallel_reasoning_summaries_enabled*/ false,
|
||||
/*concurrent_reasoning_summaries_enabled*/ false,
|
||||
)
|
||||
.await
|
||||
}
|
||||
@@ -2219,7 +2218,7 @@ async fn websocket_harness_with_options(
|
||||
async fn websocket_harness_with_provider_options(
|
||||
provider: ModelProviderInfo,
|
||||
runtime_metrics_enabled: bool,
|
||||
parallel_reasoning_summaries_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;
|
||||
@@ -2230,10 +2229,10 @@ async fn websocket_harness_with_provider_options(
|
||||
.enable(Feature::RuntimeMetrics)
|
||||
.expect("test config should allow feature update");
|
||||
}
|
||||
if parallel_reasoning_summaries_enabled {
|
||||
if concurrent_reasoning_summaries_enabled {
|
||||
config
|
||||
.features
|
||||
.enable(Feature::ParallelReasoningSummaries)
|
||||
.enable(Feature::ConcurrentReasoningSummaries)
|
||||
.expect("test config should allow feature update");
|
||||
}
|
||||
let config = Arc::new(config);
|
||||
@@ -2275,8 +2274,10 @@ 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),
|
||||
/*concurrent_reasoning_summaries_enabled*/
|
||||
config
|
||||
.features
|
||||
.enabled(Feature::ConcurrentReasoningSummaries),
|
||||
/*attestation_provider*/ None,
|
||||
);
|
||||
|
||||
|
||||
@@ -1101,7 +1101,7 @@ async fn reasoning_content_delta_has_item_metadata() -> anyhow::Result<()> {
|
||||
ev_reasoning_item("reasoning-1", &["step one"], &[]),
|
||||
ev_completed("resp-1"),
|
||||
]);
|
||||
let response_mock = mount_sse_once(&server, stream).await;
|
||||
mount_sse_once(&server, stream).await;
|
||||
|
||||
codex
|
||||
.submit(Op::UserInput {
|
||||
@@ -1132,14 +1132,6 @@ async fn reasoning_content_delta_has_item_metadata() -> anyhow::Result<()> {
|
||||
.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(())
|
||||
}
|
||||
|
||||
@@ -1155,7 +1147,9 @@ 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);
|
||||
let _ = config
|
||||
.features
|
||||
.enable(Feature::ConcurrentReasoningSummaries);
|
||||
})
|
||||
.build(&server)
|
||||
.await?;
|
||||
@@ -1171,7 +1165,7 @@ async fn concurrent_cutoff_emits_only_current_reasoning_summary_done() -> anyhow
|
||||
event["output_index"] = serde_json::json!(output_index);
|
||||
event
|
||||
};
|
||||
let response_mock = mount_sse_once(
|
||||
mount_sse_once(
|
||||
&server,
|
||||
sse(vec![
|
||||
ev_response_created("resp-1"),
|
||||
@@ -1252,12 +1246,6 @@ async fn concurrent_cutoff_emits_only_current_reasoning_summary_done() -> anyhow
|
||||
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"]
|
||||
.as_str(),
|
||||
Some("concurrent_cutoff")
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -205,7 +205,7 @@ pub enum Feature {
|
||||
/// Generate Responses API item IDs for client-created history items.
|
||||
ItemIds,
|
||||
/// Request concurrent cutoff reasoning summary delivery.
|
||||
ParallelReasoningSummaries,
|
||||
ConcurrentReasoningSummaries,
|
||||
/// Allow prompting and installing missing MCP dependencies.
|
||||
SkillMcpDependencyInstall,
|
||||
/// Removed compatibility flag for deleted skill env var dependency prompting.
|
||||
@@ -1187,8 +1187,8 @@ pub const FEATURES: &[FeatureSpec] = &[
|
||||
default_enabled: false,
|
||||
},
|
||||
FeatureSpec {
|
||||
id: Feature::ParallelReasoningSummaries,
|
||||
key: "parallel_reasoning_summaries",
|
||||
id: Feature::ConcurrentReasoningSummaries,
|
||||
key: "concurrent_reasoning_summaries",
|
||||
stage: Stage::UnderDevelopment,
|
||||
default_enabled: false,
|
||||
},
|
||||
|
||||
@@ -261,7 +261,7 @@ impl MemoryStartupContext {
|
||||
config.features.enabled(Feature::RuntimeMetrics),
|
||||
/*beta_features_header*/ None,
|
||||
config.features.enabled(Feature::ItemIds),
|
||||
config.features.enabled(Feature::ParallelReasoningSummaries),
|
||||
/*concurrent_reasoning_summaries_enabled*/ false,
|
||||
/*attestation_provider*/ None,
|
||||
);
|
||||
|
||||
|
||||
Reference in New Issue
Block a user