mirror of
https://github.com/openai/codex.git
synced 2026-11-12 21:20:34 +00:00
[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).
This commit is contained in:
@@ -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<ReasoningContext>,
|
||||
}
|
||||
|
||||
#[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<Reasoning>,
|
||||
pub store: bool,
|
||||
pub stream: bool,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub stream_options: Option<StreamOptions>,
|
||||
pub include: Vec<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub service_tier: Option<String>,
|
||||
@@ -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<Reasoning>,
|
||||
pub store: bool,
|
||||
pub stream: bool,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub stream_options: Option<StreamOptions>,
|
||||
pub include: Vec<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub service_tier: Option<String>,
|
||||
|
||||
@@ -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()),
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -167,6 +167,7 @@ pub struct ResponsesStreamEvent {
|
||||
item_id: Option<String>,
|
||||
call_id: Option<String>,
|
||||
delta: Option<String>,
|
||||
text: Option<String>,
|
||||
summary_index: Option<i64>,
|
||||
content_index: Option<i64>,
|
||||
safety_buffering: Option<Value>,
|
||||
@@ -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!({
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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"
|
||||
},
|
||||
|
||||
@@ -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<String>,
|
||||
item_ids_enabled: bool,
|
||||
concurrent_reasoning_summaries_enabled: bool,
|
||||
include_attestation: bool,
|
||||
attestation_provider: Option<Arc<dyn AttestationProvider>>,
|
||||
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<String>,
|
||||
item_ids_enabled: bool,
|
||||
concurrent_reasoning_summaries_enabled: bool,
|
||||
attestation_provider: Option<Arc<dyn AttestationProvider>>,
|
||||
) -> 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<ResponsesApiRequest> {
|
||||
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,
|
||||
|
||||
@@ -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()),
|
||||
|
||||
@@ -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(),
|
||||
})),
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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(_)
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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,
|
||||
);
|
||||
|
||||
|
||||
@@ -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(()));
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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,
|
||||
);
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user