From 1fa37ac92f293ba014eeb36832314030648acf8f Mon Sep 17 00:00:00 2001 From: Zyy Date: Wed, 28 Oct 2026 11:04:29 +0800 Subject: [PATCH] fix(openai): observe reasoning effort updates across continuations Resolve reasoning effort from configuration updates and preserve overrides in response-scoped session metadata without rewriting upstream requests. Cover HTTP and WebSocket continuations, new-chain resets, and baseline changes. Co-authored-by: Codex --- .../gateway-core/src/engine/provider.rs | 2 +- .../gateway-protocol/src/openai/codex.rs | 35 +++- .../gateway-protocol/tests/openai/codex.rs | 65 ++++++- .../openai/src/provider/execution.rs | 4 + .../providers/openai/src/provider/mod.rs | 14 +- .../openai/src/provider/observation.rs | 2 +- .../providers/openai/src/request_identity.rs | 1 + .../src/transport/protocol/responses.rs | 8 +- .../openai/tests/provider/contract/mod.rs | 177 ++++++++++++++++++ .../openai/tests/transport/request.rs | 4 +- docs/api.md | 4 + 11 files changed, 298 insertions(+), 18 deletions(-) diff --git a/backend/crates/gateway-core/src/engine/provider.rs b/backend/crates/gateway-core/src/engine/provider.rs index 90a8388a0..3296ecd98 100644 --- a/backend/crates/gateway-core/src/engine/provider.rs +++ b/backend/crates/gateway-core/src/engine/provider.rs @@ -1015,7 +1015,7 @@ pub struct ProviderRequest { pub struct ProviderRequestObservation { /// 独立 Provider 端点声明的请求模型，仅用于观测，不参与文本模型目录路由 pub requested_model: Option, - /// 客户端原始请求中的推理强度 + /// Provider 从客户端请求与续接配置中解析的推理强度 pub reasoning_effort: Option, pub reasoning_preset: Option, pub request_kind: Option, diff --git a/backend/crates/gateway-protocol/src/openai/codex.rs b/backend/crates/gateway-protocol/src/openai/codex.rs index c983cdf43..9dd915ffb 100644 --- a/backend/crates/gateway-protocol/src/openai/codex.rs +++ b/backend/crates/gateway-protocol/src/openai/codex.rs @@ -105,8 +105,10 @@ fn codex_identity_field( /// 从 OpenAI Responses 请求中提取的稳定 Codex 请求语义 #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct CodexResponsesRequestSemantics { - /// 客户端原始选择的推理强度 + /// 按最新配置更新与请求基线解析的推理强度 pub reasoning_effort: Option, + /// 对话内持续生效的推理强度更新，供原生续接继承 + pub reasoning_effort_override: Option, /// Codex turn metadata 中的请求类型 pub request_kind: Option, /// Codex turn metadata 中的子代理类型 @@ -126,24 +128,31 @@ pub fn codex_responses_request_semantics( codex_responses_request_semantics_with_turn_metadata( body, non_empty_string(context.get("turn_metadata")), + None, ) } -/// 调用方已解析权威 turn metadata 时，从 Responses body 中提取 Codex 语义 +/// 结合权威 turn metadata 与原生续接配置，从 Responses body 中提取 Codex 语义 #[must_use] pub fn codex_responses_request_semantics_with_turn_metadata( body: &Map, turn_metadata: Option<&str>, + previous_reasoning_effort_override: Option<&str>, ) -> CodexResponsesRequestSemantics { - let reasoning_effort = body - .get("reasoning") - .and_then(Value::as_object) - .and_then(|reasoning| reasoning.get("effort")) - .and_then(|effort| { - non_empty_string(Some(effort)) + // 配置更新不改变顶层基线，只有 previous_response_id 续接才继承上一响应的更新 + let reasoning_effort_override = input_items(body) + .iter() + .rev() + .find(|item| item.get("type").and_then(Value::as_str) == Some("configuration_update")) + .map(|item| reasoning_effort(item.get("reasoning"))) + .unwrap_or_else(|| { + previous_reasoning_effort_override + .filter(|_| non_empty_string(body.get("previous_response_id")).is_some()) .map(ToOwned::to_owned) - .or_else(|| effort.as_u64().map(|value| value.to_string())) }); + let reasoning_effort = reasoning_effort_override + .clone() + .or_else(|| reasoning_effort(body.get("reasoning"))); let parsed_turn_metadata = turn_metadata .or_else(|| request_turn_metadata(body)) .and_then(|value| serde_json::from_str::(value).ok()); @@ -168,6 +177,7 @@ pub fn codex_responses_request_semantics_with_turn_metadata( CodexResponsesRequestSemantics { reasoning_effort, + reasoning_effort_override, request_kind, subagent_kind, reasoning_preset, @@ -175,6 +185,13 @@ pub fn codex_responses_request_semantics_with_turn_metadata( } } +fn reasoning_effort(reasoning: Option<&Value>) -> Option { + let effort = reasoning?.get("effort")?; + non_empty_string(Some(effort)) + .map(ToOwned::to_owned) + .or_else(|| effort.as_u64().map(|value| value.to_string())) +} + fn request_turn_metadata(body: &Map) -> Option<&str> { non_empty_string(body.get("turnMetadata")) .or_else(|| non_empty_string(body.get("turn_metadata"))) diff --git a/backend/crates/gateway-protocol/tests/openai/codex.rs b/backend/crates/gateway-protocol/tests/openai/codex.rs index 15d417a92..eeb7e3f48 100644 --- a/backend/crates/gateway-protocol/tests/openai/codex.rs +++ b/backend/crates/gateway-protocol/tests/openai/codex.rs @@ -1,7 +1,8 @@ //! 验证 Codex 会话身份优先级、轮次元数据与多代理模式解析 use gateway_protocol::openai::{ - codex_responses_request_semantics, codex_session_id, codex_thread_id, + codex_responses_request_semantics, codex_responses_request_semantics_with_turn_metadata, + codex_session_id, codex_thread_id, }; use serde_json::{Map, Value, json}; @@ -24,6 +25,56 @@ fn reasoning_effort_should_observe_unsigned_numbers_without_changing_the_body() } } +#[test] +fn reasoning_effort_should_follow_the_latest_configuration_update() { + for (input, previous_id, inherited, expected) in [ + (json!([]), None, None, "medium"), + (json!([]), Some("resp_previous"), Some("high"), "high"), + (json!([]), None, Some("high"), "medium"), + ( + json!([ + {"type":"configuration_update","reasoning":{"effort":"high"}}, + {"type":"message","role":"user","content":"first turn"}, + {"type":"configuration_update","reasoning":{"effort":"low"}}, + {"type":"message","role":"user","content":"next turn"} + ]), + Some("resp_previous"), + Some("max"), + "low", + ), + ( + json!([{"type":"configuration_update","reasoning":{"effort":64}}]), + None, + None, + "64", + ), + ( + json!([{"type":"configuration_update","reasoning":{"effort":" future-value "}}]), + None, + None, + "future-value", + ), + ( + json!([{"type":"message","role":"user","reasoning":{"effort":"high"}, + "content":"configuration_update: high"}]), + None, + None, + "medium", + ), + ] { + let body = object(json!({ + "reasoning":{"effort":"medium"}, + "input":input, + "previous_response_id":previous_id, + })); + let original = body.clone(); + let semantics = + codex_responses_request_semantics_with_turn_metadata(&body, None, inherited); + assert_eq!(semantics.reasoning_effort.as_deref(), Some(expected)); + assert_eq!(body, original); + } +} + #[test] fn codex_session_should_prefer_the_connection_identity_over_body_metadata() { let body = object( @@ -147,7 +198,7 @@ fn codex_semantics_should_use_protocol_context_before_body_metadata() { #[test] fn codex_semantics_should_detect_ultra_preset_from_proactive_multi_agent() { - let body = object(json!({ + let mut body = object(json!({ "model": "gpt-test", "reasoning": {"effort": "max"}, "input": [ @@ -167,6 +218,16 @@ fn codex_semantics_should_detect_ultra_preset_from_proactive_multi_agent() { assert_eq!(semantics.reasoning_effort.as_deref(), Some("max")); assert_eq!(semantics.reasoning_preset, Some("ultra")); assert!(!semantics.compact); + + for (effort, expected_preset) in [("high", None), ("max", Some("ultra"))] { + body["input"] + .as_array_mut() + .unwrap() + .push(json!({"type":"configuration_update","reasoning":{"effort":effort}})); + let semantics = codex_responses_request_semantics(&body, &Map::new()); + assert_eq!(semantics.reasoning_effort.as_deref(), Some(effort)); + assert_eq!(semantics.reasoning_preset, expected_preset); + } } fn object(value: Value) -> Map { diff --git a/backend/crates/providers/openai/src/provider/execution.rs b/backend/crates/providers/openai/src/provider/execution.rs index 055374672..3bf80093d 100644 --- a/backend/crates/providers/openai/src/provider/execution.rs +++ b/backend/crates/providers/openai/src/provider/execution.rs @@ -425,6 +425,8 @@ pub(super) struct OpenAiSessionState { pub(super) turn_state: Option, #[serde(default)] pub(super) client_turn_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub(super) reasoning_effort_override: Option, pub(super) continuation_scope: OpenAiContinuationScope, } @@ -442,6 +444,7 @@ pub(super) struct OpenAiSessionCapture { pub(super) conversation_id: Option, pub(super) turn_state: Option, pub(super) client_turn_id: Option, + pub(super) reasoning_effort_override: Option, pub(super) response_store: bool, pub(super) continuation_scope: Option, } @@ -488,6 +491,7 @@ fn encode_openai_session_capture( conversation_id: capture.conversation_id.clone(), turn_state: capture.turn_state.clone(), client_turn_id: capture.client_turn_id.clone(), + reasoning_effort_override: capture.reasoning_effort_override.clone(), continuation_scope, }) } diff --git a/backend/crates/providers/openai/src/provider/mod.rs b/backend/crates/providers/openai/src/provider/mod.rs index 5ca2730a0..24cb1cca8 100644 --- a/backend/crates/providers/openai/src/provider/mod.rs +++ b/backend/crates/providers/openai/src/provider/mod.rs @@ -494,7 +494,12 @@ impl Provider for CodexProvider { let Ok(encoded) = encode_generate_request(request, "observability", None) else { return ProviderRequestObservation::default(); }; - let semantics = encoded.semantics(); + let previous_session = decode_openai_session_state(request); + let semantics = encoded.semantics( + previous_session + .as_ref() + .and_then(|state| state.reasoning_effort_override.as_deref()), + ); let reasoning_effort = semantics.reasoning_effort.clone(); let previous_response_id = encoded.previous_response_id(); let continuation = ContinuationRequestObservation { @@ -1077,6 +1082,13 @@ impl CodexProvider { conversation_id: upstream_request.local_conversation_id.clone(), turn_state: upstream_request.turn_state.clone(), client_turn_id: upstream_request.client_turn_id.clone(), + reasoning_effort_override: upstream_request + .semantics( + previous_session + .as_ref() + .and_then(|state| state.reasoning_effort_override.as_deref()), + ) + .reasoning_effort_override, response_store, continuation_scope: None, }); diff --git a/backend/crates/providers/openai/src/provider/observation.rs b/backend/crates/providers/openai/src/provider/observation.rs index 8b2aeee79..38865c5ac 100644 --- a/backend/crates/providers/openai/src/provider/observation.rs +++ b/backend/crates/providers/openai/src/provider/observation.rs @@ -432,7 +432,7 @@ pub(super) fn openai_response_request_summary( let body = request.body(); let input = body.get("input"); let tools = body.get("tools"); - let semantics = request.semantics(); + let semantics = request.semantics(None); json!({ "model": request.model(), "stream": request.stream(), diff --git a/backend/crates/providers/openai/src/request_identity.rs b/backend/crates/providers/openai/src/request_identity.rs index bc322c8a1..ca5c6279b 100644 --- a/backend/crates/providers/openai/src/request_identity.rs +++ b/backend/crates/providers/openai/src/request_identity.rs @@ -23,6 +23,7 @@ pub(crate) fn follows_session_with_headers( gateway_protocol::openai::codex_responses_request_semantics_with_turn_metadata( body, metadata.or_else(|| context.get("turn_metadata").and_then(Value::as_str)), + None, ) .subagent_kind .is_some() diff --git a/backend/crates/providers/openai/src/transport/protocol/responses.rs b/backend/crates/providers/openai/src/transport/protocol/responses.rs index 8841170b8..628dcc2da 100644 --- a/backend/crates/providers/openai/src/transport/protocol/responses.rs +++ b/backend/crates/providers/openai/src/transport/protocol/responses.rs @@ -760,10 +760,14 @@ impl CodexResponsesRequest { } /// 提取 Codex 请求类型、子代理类型、推理预设与压缩语义 - pub fn semantics(&self) -> CodexRequestSemantics { + pub fn semantics( + &self, + previous_reasoning_effort_override: Option<&str>, + ) -> CodexRequestSemantics { let mut semantics = codex_responses_request_semantics_with_turn_metadata( self.body(), self.turn_metadata.as_deref(), + previous_reasoning_effort_override, ); // 官方 generate=false 只准备连接与上下文，不是模型推理 // 预热分类参与用量筛选，不能仅凭客户端的 request_kind 提示排除真实推理 @@ -782,7 +786,7 @@ impl CodexResponsesRequest { /// 它不拆分根会话的账号首选项； /// `thread_spawn` 仍用它派生独立 WebSocket/continuation transport identity pub fn subagent_kind(&self) -> Option { - self.semantics() + self.semantics(None) .subagent_kind .filter(|value| !value.trim().is_empty()) .or_else(|| { diff --git a/backend/crates/providers/openai/tests/provider/contract/mod.rs b/backend/crates/providers/openai/tests/provider/contract/mod.rs index e82f0ee9b..78993268c 100644 --- a/backend/crates/providers/openai/tests/provider/contract/mod.rs +++ b/backend/crates/providers/openai/tests/provider/contract/mod.rs @@ -10256,6 +10256,183 @@ fn request_observation_preserves_the_raw_reasoning_effort() { } } +#[tokio::test] +async fn reasoning_effort_observation_follows_configuration_updates_across_continuations() { + fn completed(index: usize) -> Value { + json!({"type":"response.completed","response":{ + "id":format!("resp_effort_{index}"),"model":"gpt-5.4","status":"completed", + "output":[],"reasoning":{"effort":if index == 4 { "high" } else { "medium" }}, + "usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2} + }}) + } + + for use_websocket in [false, true] { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let http = MockServer::start().await; + let count = std::sync::atomic::AtomicUsize::new(0); + Mock::given(method("POST")) + .and(path("/codex/responses")) + .respond_with(move |_: &wiremock::Request| { + let index = count.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(format!("data: {}\n\n", completed(index))) + }) + .mount(&http) + .await; + let (base_url, websocket_server) = if use_websocket { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base_url = format!("http://{}", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let mut websocket = accept_codex_test_websocket(socket).await; + let mut frames = Vec::new(); + for index in 0..5 { + let message = websocket.next().await.unwrap().unwrap(); + let frame: Value = serde_json::from_str(message.to_text().unwrap()).unwrap(); + assert_eq!(frame["type"], "response.create"); + frames.push(frame); + websocket + .send(Message::Text(completed(index).to_string().into())) + .await + .unwrap(); + } + frames + }); + (base_url, Some(server)) + } else { + (http.uri(), None) + }; + let provider = provider_with_base_url(&store, base_url); + let client_key = ClientApiKeyId::new("key_openai_contract").unwrap(); + let mut session_state = Some( + ProviderSessionState::new( + "openai", + json!({ + "account_id":"acct_provider_contract", + "conversation_id":"effort-observation", + "continuation_scope":if use_websocket { "connection_local" } else { "persisted" }, + }) + .as_object() + .unwrap() + .clone(), + ) + .unwrap(), + ); + let mut originals = Vec::new(); + for (index, (baseline, update, expected)) in [ + ("medium", Some("high"), "high"), + ("medium", None, "high"), + ("medium", Some("low"), "low"), + ("medium", None, "medium"), + ("high", None, "high"), + ] + .into_iter() + .enumerate() + { + let mut input = Vec::new(); + if let Some(effort) = update { + input.push(json!({"type":"configuration_update","reasoning":{"effort":effort}})); + } + input.push(json!({"type":"message","role":"user","content":"next turn"})); + let mut body = json!({ + "model":"gpt-5.4","input":input,"reasoning":{"effort":baseline}, + "session_id":"effort-observation","store":!use_websocket, + }); + let previous = (index > 0 && index != 3).then(|| format!("resp_effort_{}", index - 1)); + if let Some(previous) = &previous { + body["previous_response_id"] = json!(previous); + } + let mut protocol_context = + Map::from_iter([("use_websocket".to_owned(), json!(use_websocket))]); + if use_websocket { + protocol_context.insert( + "downstream_websocket_connection_id".to_owned(), + json!("ws_effort"), + ); + } + let mut generate = GenerateRequest::from_protocol_payload( + ProtocolPayload::json_object("openai", body.as_object().unwrap().clone()) + .unwrap() + .with_context(protocol_context), + ); + if let Some(state) = session_state.take() { + generate = generate.with_provider_session_state(state); + } + let operation = Operation::Generate(generate); + let observation = provider.request_observation(&operation, &client_key); + assert_eq!(observation.reasoning_effort.as_deref(), Some(expected)); + let binding = previous.as_ref().map(|previous| { + ContinuationBinding::Pinned(NativeContinuationPin::new( + PreviousResponseId::new(previous), + PreviousResponseId::new(previous), + client_key.clone(), + ProviderKind::new("openai").unwrap(), + ProviderAccountId::new("acct_provider_contract").unwrap(), + )) + }); + let context = AttemptContext::new( + RequestAttemptContext::new( + ModelRequestId::new(format!("req_effort_{index}")).unwrap(), + client_key.clone(), + ), + NonZeroU32::new(1).unwrap(), + SystemTime::now() + Duration::from_secs(30), + account_policy(), + AccountAttemptContext::new(BTreeSet::new(), None, None) + .with_account_scope(contract_account_scope()), + binding, + CancellationToken::new(), + ) + .with_continuation_attempt(if previous.is_some() { + ContinuationAttempt::Native + } else { + ContinuationAttempt::None + }); + let mut stream = Arc::clone(&provider) + .execute(planned_request("openai", operation), context) + .await + .unwrap(); + while let Some(event) = timeout(Duration::from_secs(5), stream.next()) + .await + .unwrap() + { + let event = event.unwrap_or_else(|error| { + panic!("use_websocket={use_websocket}, turn={index}: {error:?}") + }); + if let Some(update) = event.session_update() { + session_state = Some(update.clone()); + } + } + assert!(session_state.is_some()); + originals.push(body); + } + let forwarded = if let Some(server) = websocket_server { + timeout(Duration::from_secs(5), server) + .await + .unwrap() + .unwrap() + } else { + http.received_requests() + .await + .unwrap() + .iter() + .map(captured_request_body) + .collect() + }; + assert_eq!(forwarded.len(), originals.len()); + for (forwarded, original) in forwarded.iter().zip(&originals) { + assert_eq!(forwarded["input"], original["input"]); + assert_eq!(forwarded["reasoning"], original["reasoning"]); + assert_eq!( + forwarded.get("previous_response_id"), + original.get("previous_response_id") + ); + } + } +} + #[tokio::test] async fn provider_compiles_catalog_presentation_for_codex_models() { let store = Arc::new(MemoryAccountStore::default()); diff --git a/backend/crates/providers/openai/tests/transport/request.rs b/backend/crates/providers/openai/tests/transport/request.rs index edd0cc252..9e43753a0 100644 --- a/backend/crates/providers/openai/tests/transport/request.rs +++ b/backend/crates/providers/openai/tests/transport/request.rs @@ -775,7 +775,7 @@ fn observability_semantics_should_reuse_codex_turn_metadata() { let semantics = encode_generate_request(&request, "observability", None) .expect("observability request should encode") - .semantics(); + .semantics(None); assert_eq!(semantics.request_kind.as_deref(), Some("compaction")); assert_eq!(semantics.subagent_kind.as_deref(), Some("review")); @@ -845,7 +845,7 @@ fn observability_semantics_should_use_the_transparent_openai_payload() { let semantics = encode_generate_request(&request, "observability", None) .expect("transparent OpenAI request should encode") - .semantics(); + .semantics(None); assert_eq!(semantics.reasoning_preset, Some("ultra")); assert!(semantics.compact); diff --git a/docs/api.md b/docs/api.md index 44970b8d3..570daf96e 100644 --- a/docs/api.md +++ b/docs/api.md @@ -2118,6 +2118,10 @@ Dashboard 返回原始 `asOf` 及 `asOfDisplay`，作为当前数据的查询锚 ### 记录范围与统计口径 +OpenAI 推理强度优先取 `input` 中最后一条 `type: "configuration_update"` 项的 `reasoning.effort`， +原生续接未再次更新时沿用已关联响应的配置，否则取请求顶层 `reasoning.effort`。 +该字段是请求观测，不改写上游请求；网关未持有的续接历史无法用于推断档位 + 请求健康可用性与服务成功率按 `成功 /（成功 + 服务失败 + 未完成）` 计算 未完成保留独立计数，客户端取消与调用方错误不计入该分母，中间尝试失败不重复计为最终失败请求 健康时间线以 15 分钟分桶，可用性不低于 95% 且有效样本不少于 10 个时显示稳定