diff --git a/backend/src/proxy/dispatch/responses/service.rs b/backend/src/proxy/dispatch/responses/service.rs index 97f2ffa7e..23d5a2d99 100644 --- a/backend/src/proxy/dispatch/responses/service.rs +++ b/backend/src/proxy/dispatch/responses/service.rs @@ -158,7 +158,8 @@ impl ResponseDispatchService { request: &mut CodexResponsesRequest, ) -> Option { prepare_variant_identity(request); - if let Some(previous_response_id) = request.previous_response_id().map(ToString::to_string) { + if let Some(previous_response_id) = request.previous_response_id().map(ToString::to_string) + { if request.prompt_cache_key().is_none() { let conversation_id = self .session_affinity @@ -273,9 +274,7 @@ impl ResponseDispatchService { implicit_resume: &mut Option, ) { if let Some(previous_response_id) = request.previous_response_id() { - self.session_affinity - .forget(previous_response_id) - .await; + self.session_affinity.forget(previous_response_id).await; } if let Some(snapshot) = implicit_resume.take() { snapshot.restore(request); @@ -376,8 +375,7 @@ impl ResponseDispatchService { let tuple_schema = request.tuple_schema.clone(); let image_generation_requested = request.expects_image_generation(); let now = Utc::now(); - let explicit_previous_response_id = - request.previous_response_id().map(ToString::to_string); + let explicit_previous_response_id = request.previous_response_id().map(ToString::to_string); let mut implicit_resume = self.prepare_response_session(&mut request).await; let preferred_account_id = self.preferred_account_id_for_request(&request, now).await; let mut acquire_request = AccountAcquireRequest::new(request.model(), now); diff --git a/backend/src/proxy/dispatch/session_affinity.rs b/backend/src/proxy/dispatch/session_affinity.rs index 70f3a4f06..f0238eec1 100644 --- a/backend/src/proxy/dispatch/session_affinity.rs +++ b/backend/src/proxy/dispatch/session_affinity.rs @@ -161,9 +161,7 @@ fn compute_variant_hash_with_identity( request: &CodexResponsesRequest, identity: Option<&str>, ) -> String { - let tools_json = request - .tools() - .map_or_else(|| "[]".to_string(), tools_json); + let tools_json = request.tools().map_or_else(|| "[]".to_string(), tools_json); let mut hasher = Sha256::new(); hasher.update(request.instructions().as_bytes()); hasher.update(b"\0"); @@ -270,7 +268,11 @@ pub fn ensure_prompt_cache_key(request: &mut CodexResponsesRequest) { /// 按原版 `stable-conversation-key.ts` 的规则派生稳定 conversation key。 pub fn derive_stable_conversation_key(request: &CodexResponsesRequest) -> Option { - let instructions = request.instructions().chars().take(2000).collect::(); + let instructions = request + .instructions() + .chars() + .take(2000) + .collect::(); let first_user_text = first_user_text(request.input()); let normalized_first_user_text = normalize_conversation_anchor_text(&first_user_text); let first_user_text = if normalized_first_user_text.is_empty() { diff --git a/backend/src/proxy/openai/responses.rs b/backend/src/proxy/openai/responses.rs index 691637531..3b002348b 100644 --- a/backend/src/proxy/openai/responses.rs +++ b/backend/src/proxy/openai/responses.rs @@ -114,7 +114,10 @@ pub fn build_codex_request( /// transport 控制字段,避免上游返回代理无法解析的 SSE 形态。业务语义字段 /// (`reasoning`/`text`/`store`/`prompt_cache_key`/`previous_response_id`/ /// `include`/`client_metadata`/未知字段)一律原样透传。 -pub fn build_compact_request(mut body: Map, headers: &HeaderMap) -> CodexCompactRequest { +pub fn build_compact_request( + mut body: Map, + headers: &HeaderMap, +) -> CodexCompactRequest { body.remove(COMPACT_STREAM_KEY); for key in TRANSPORT_ONLY_KEYS { body.remove(key); @@ -277,7 +280,14 @@ pub async fn compact_responses( headers: HeaderMap, body: Bytes, ) -> Response { - handle_compact_responses(state, request_id, client_ip_string(client_ip), headers, body).await + handle_compact_responses( + state, + request_id, + client_ip_string(client_ip), + headers, + body, + ) + .await } /// 从可选的 `ClientIp` extension 提取 IP 字符串。 @@ -387,7 +397,9 @@ fn parse_responses_body(body: &Bytes) -> Option> { } fn request_model(body: &Map) -> &str { - body.get("model").and_then(Value::as_str).unwrap_or_default() + body.get("model") + .and_then(Value::as_str) + .unwrap_or_default() } async fn validated_responses_model( diff --git a/backend/src/upstream/protocol/responses.rs b/backend/src/upstream/protocol/responses.rs index 6dc99d042..b862a4d2b 100644 --- a/backend/src/upstream/protocol/responses.rs +++ b/backend/src/upstream/protocol/responses.rs @@ -5,11 +5,9 @@ use serde_json::{json, Map, Value}; use crate::{ infra::time::elapsed_millis_i64, - upstream::{ - protocol::{ - schema::reconvert_tuple_values, - sse::{parse_sse_events, SseError, SseEvent}, - }, + upstream::protocol::{ + schema::reconvert_tuple_values, + sse::{parse_sse_events, SseError, SseEvent}, }, }; @@ -534,7 +532,8 @@ impl CodexResponsesRequest { /// 只做最小规范化:`input` 缺省为 `[]`、`stream` 缺省 `true`、`store` 缺省 `false`。 /// 其余字段(含未知字段)原样保留在 `body` 中透传上游。 pub fn from_body(mut body: Map) -> Self { - body.entry("input").or_insert_with(|| Value::Array(Vec::new())); + body.entry("input") + .or_insert_with(|| Value::Array(Vec::new())); body.entry("stream").or_insert(Value::Bool(true)); body.entry("store").or_insert(Value::Bool(false)); Self { @@ -565,7 +564,10 @@ impl CodexResponsesRequest { ) -> Self { let mut body = Map::new(); body.insert("model".to_string(), Value::String(model.into())); - body.insert("instructions".to_string(), Value::String(instructions.into())); + body.insert( + "instructions".to_string(), + Value::String(instructions.into()), + ); body.insert("input".to_string(), Value::Array(input)); Self::from_body(body) } @@ -579,22 +581,32 @@ impl CodexResponsesRequest { /// 模型名。 pub fn model(&self) -> &str { - self.body.get("model").and_then(Value::as_str).unwrap_or_default() + self.body + .get("model") + .and_then(Value::as_str) + .unwrap_or_default() } /// 设置模型名(模型后缀路由归一)。 pub fn set_model(&mut self, model: impl Into) { - self.body.insert("model".to_string(), Value::String(model.into())); + self.body + .insert("model".to_string(), Value::String(model.into())); } /// 指令文本(缺省空串)。 pub fn instructions(&self) -> &str { - self.body.get("instructions").and_then(Value::as_str).unwrap_or_default() + self.body + .get("instructions") + .and_then(Value::as_str) + .unwrap_or_default() } /// 输入条目切片(非数组时为空)。 pub fn input(&self) -> &[Value] { - self.body.get("input").and_then(Value::as_array).map_or(&[], Vec::as_slice) + self.body + .get("input") + .and_then(Value::as_array) + .map_or(&[], Vec::as_slice) } /// 替换输入条目(implicit resume / 字符串输入兼容)。 @@ -604,7 +616,10 @@ impl CodexResponsesRequest { /// 是否流式返回。 pub fn stream(&self) -> bool { - self.body.get("stream").and_then(Value::as_bool).unwrap_or(true) + self.body + .get("stream") + .and_then(Value::as_bool) + .unwrap_or(true) } /// 设置流式标志。 @@ -614,7 +629,10 @@ impl CodexResponsesRequest { /// 是否要求上游存储响应。 pub fn store(&self) -> bool { - self.body.get("store").and_then(Value::as_bool).unwrap_or(false) + self.body + .get("store") + .and_then(Value::as_bool) + .unwrap_or(false) } /// reasoning 配置(透传,不规整)。 @@ -657,7 +675,8 @@ impl CodexResponsesRequest { pub fn set_service_tier(&mut self, service_tier: Option) { match service_tier { Some(value) => { - self.body.insert("service_tier".to_string(), Value::String(value)); + self.body + .insert("service_tier".to_string(), Value::String(value)); } None => { self.body.remove("service_tier"); @@ -667,7 +686,9 @@ impl CodexResponsesRequest { /// 前一个 response ID。 pub fn previous_response_id(&self) -> Option<&str> { - self.body.get("previous_response_id").and_then(Value::as_str) + self.body + .get("previous_response_id") + .and_then(Value::as_str) } /// 设置 / 清除前一个 response ID。 @@ -755,7 +776,10 @@ impl Serialize for CodexCompactRequest { impl CodexCompactRequest { /// 模型名。 pub fn model(&self) -> &str { - self.body.get("model").and_then(Value::as_str).unwrap_or_default() + self.body + .get("model") + .and_then(Value::as_str) + .unwrap_or_default() } /// 设置模型名(模型后缀路由归一)。 diff --git a/backend/tests/upstream/protocol/mod.rs b/backend/tests/upstream/protocol/mod.rs index aaa10eaa0..5614329a8 100644 --- a/backend/tests/upstream/protocol/mod.rs +++ b/backend/tests/upstream/protocol/mod.rs @@ -7,9 +7,9 @@ use codex_proxy_rs::upstream::protocol::events::{ rate_limit_quota, retry_after_seconds_from_body, RateLimitWindow, TokenUsage, }; use codex_proxy_rs::upstream::protocol::responses::{ - http_sse_fallback_allowed, response_body_has_first_event, - response_from_codex_sse, response_sse_event_is_terminal, transport_for_request, - CodexResponsesRequest, CodexTransport, CollectedResponse, + http_sse_fallback_allowed, response_body_has_first_event, response_from_codex_sse, + response_sse_event_is_terminal, transport_for_request, CodexResponsesRequest, CodexTransport, + CollectedResponse, }; use codex_proxy_rs::upstream::protocol::sse::parse_sse_events; use codex_proxy_rs::upstream::protocol::websocket::{ diff --git a/backend/tests/upstream/protocol/openai_responses.rs b/backend/tests/upstream/protocol/openai_responses.rs index e1896789a..d983e5aa6 100644 --- a/backend/tests/upstream/protocol/openai_responses.rs +++ b/backend/tests/upstream/protocol/openai_responses.rs @@ -351,10 +351,16 @@ fn build_codex_request_should_pass_through_body_fields_verbatim() { json!({"type": "function", "function": {"name": "lookup"}}) ); assert_eq!(upstream["parallel_tool_calls"], true); - assert_eq!(upstream["tools"], json!([{"type": "function", "name": "lookup"}])); + assert_eq!( + upstream["tools"], + json!([{"type": "function", "name": "lookup"}]) + ); assert_eq!(upstream["prompt_cache_key"], "pcache"); assert_eq!(upstream["include"], json!(["reasoning.encrypted_content"])); - assert_eq!(upstream["client_metadata"], json!({"safe": "yes", "drop": 42})); + assert_eq!( + upstream["client_metadata"], + json!({"safe": "yes", "drop": 42}) + ); // transport-only 字段不进上游 body。 assert!(upstream.get("use_websocket").is_none()); @@ -383,8 +389,14 @@ fn build_codex_request_should_prefer_body_context_fields_then_fall_back_to_heade .cloned() .unwrap(); let mut headers = HeaderMap::new(); - headers.insert("x-codex-turn-metadata", " meta-from-header ".parse().unwrap()); - headers.insert("x-codex-beta-features", " beta-from-header ".parse().unwrap()); + headers.insert( + "x-codex-turn-metadata", + " meta-from-header ".parse().unwrap(), + ); + headers.insert( + "x-codex-beta-features", + " beta-from-header ".parse().unwrap(), + ); headers.insert( "x-responsesapi-include-timing-metrics", " true ".parse().unwrap(), @@ -402,7 +414,10 @@ fn build_codex_request_should_prefer_body_context_fields_then_fall_back_to_heade assert_eq!(codex.turn_metadata.as_deref(), Some("meta-from-header")); assert_eq!(codex.include_timing_metrics.as_deref(), Some("true")); assert_eq!(codex.codex_window_id.as_deref(), Some("window-from-header")); - assert_eq!(codex.parent_thread_id.as_deref(), Some("parent-from-header")); + assert_eq!( + codex.parent_thread_id.as_deref(), + Some("parent-from-header") + ); } #[test] @@ -485,14 +500,13 @@ fn openai_streaming_response_with_previous_response_should_require_websocket() { "stream": true }); - let codex = build_codex_request( - body.as_object().unwrap().clone(), - &HeaderMap::new(), - None, - ); + let codex = build_codex_request(body.as_object().unwrap().clone(), &HeaderMap::new(), None); assert_eq!(codex.previous_response_id(), Some("resp_previous")); assert!(codex.stream()); assert!(!codex.force_http_sse); - assert_eq!(transport_for_request(&codex), CodexTransport::WebSocketRequired); + assert_eq!( + transport_for_request(&codex), + CodexTransport::WebSocketRequired + ); }