starred/codex-proxy-rs
1
0
Fork 0
mirror of https://github.com/zyycn/codex-proxy-rs.git synced 2026-11-06 08:06:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(openai): observe reasoning effort updates across continuations

Browse Source 
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 <noreply@openai.com>
This commit is contained in:
Zyy
2026-10-28 11:04:29 +08:00
co-authored by Codex
parent 0ceac908f4
commit 1fa37ac92f
11 changed files with 298 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/crates/gateway-core/src/engine/provider.rs
+1 -1
View File
@@ -1015,7 +1015,7 @@ pub struct ProviderRequest {
pub struct ProviderRequestObservation {
/// 独立 Provider 端点声明的请求模型，仅用于观测，不参与文本模型目录路由
pub requested_model: Option<PublicModelId>,
/// 客户端原始请求中的推理强度
/// Provider 从客户端请求与续接配置中解析的推理强度
pub reasoning_effort: Option<String>,
pub reasoning_preset: Option<String>,
pub request_kind: Option<String>,
backend/crates/gateway-protocol/src/openai/codex.rs
+26 -9
View File
@@ -105,8 +105,10 @@ fn codex_identity_field(
/// 从 OpenAI Responses 请求中提取的稳定 Codex 请求语义
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct CodexResponsesRequestSemantics {
/// 客户端原始选择的推理强度
/// 按最新配置更新与请求基线解析的推理强度
pub reasoning_effort: Option<String>,
/// 对话内持续生效的推理强度更新，供原生续接继承
pub reasoning_effort_override: Option<String>,
/// Codex turn metadata 中的请求类型
pub request_kind: Option<String>,
/// 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<String, Value>,
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>(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<String> {
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<String, Value>) -> Option<&str> {
non_empty_string(body.get("turnMetadata"))
.or_else(|| non_empty_string(body.get("turn_metadata")))
backend/crates/gateway-protocol/tests/openai/codex.rs
+63 -2
View File
@@ -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<String, Value> {
backend/crates/providers/openai/src/provider/execution.rs
+4
View File
@@ -425,6 +425,8 @@ pub(super) struct OpenAiSessionState {
pub(super) turn_state: Option<String>,
#[serde(default)]
pub(super) client_turn_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub(super) reasoning_effort_override: Option<String>,
pub(super) continuation_scope: OpenAiContinuationScope,
}
@@ -442,6 +444,7 @@ pub(super) struct OpenAiSessionCapture {
pub(super) conversation_id: Option<String>,
pub(super) turn_state: Option<String>,
pub(super) client_turn_id: Option<String>,
pub(super) reasoning_effort_override: Option<String>,
pub(super) response_store: bool,
pub(super) continuation_scope: Option<OpenAiContinuationScope>,
}
@@ -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,
})
}
backend/crates/providers/openai/src/provider/mod.rs
+13 -1
View File
@@ -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,
});
backend/crates/providers/openai/src/provider/observation.rs
+1 -1
View File
@@ -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(),
backend/crates/providers/openai/src/request_identity.rs
+1
View File
@@ -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()
backend/crates/providers/openai/src/transport/protocol/responses.rs
+6 -2
View File
@@ -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<String> {
self.semantics()
self.semantics(None)
.subagent_kind
.filter(|value| !value.trim().is_empty())
.or_else(|| {
backend/crates/providers/openai/tests/provider/contract/mod.rs
+177
View File
@@ -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());
backend/crates/providers/openai/tests/transport/request.rs
+2 -2
View File
@@ -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);
docs/api.md
+4
View File
@@ -2118,6 +2118,10 @@ Dashboard 返回原始 `asOf` 及 `asOfDisplay`，作为当前数据的查询锚
### 记录范围与统计口径
OpenAI 推理强度优先取 `input` 中最后一条 `type: "configuration_update"` 项的 `reasoning.effort`，
原生续接未再次更新时沿用已关联响应的配置，否则取请求顶层 `reasoning.effort`。
该字段是请求观测，不改写上游请求；网关未持有的续接历史无法用于推断档位
请求健康可用性与服务成功率按 `成功 /（成功 + 服务失败 + 未完成）` 计算
未完成保留独立计数，客户端取消与调用方错误不计入该分母，中间尝试失败不重复计为最终失败请求
健康时间线以 15 分钟分桶，可用性不低于 95% 且有效样本不少于 10 个时显示稳定