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

style: format responses proxy code

Browse Source 
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
zyycn
2026-07-05 05:11:27 +08:00
co-authored by Codex
parent d123704172
commit be3d648fee
6 changed files with 93 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/proxy/dispatch/responses/service.rs
+4 -6
View File
@@ -158,7 +158,8 @@ impl ResponseDispatchService {
request: &mut CodexResponsesRequest,
) -> Option<ImplicitResumeSnapshot> {
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<ImplicitResumeSnapshot>,
) {
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);
backend/src/proxy/dispatch/session_affinity.rs
+6 -4
View File
@@ -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<String> {
let instructions = request.instructions().chars().take(2000).collect::<String>();
let instructions = request
.instructions()
.chars()
.take(2000)
.collect::<String>();
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() {
backend/src/proxy/openai/responses.rs
+15 -3
View File
@@ -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<String, Value>, headers: &HeaderMap) -> CodexCompactRequest {
pub fn build_compact_request(
mut body: Map<String, Value>,
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<Map<String, Value>> {
}
fn request_model(body: &Map<String, Value>) -> &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(
backend/src/upstream/protocol/responses.rs
+40 -16
View File
@@ -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<String, Value>) -> 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<String>) {
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<String>) {
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()
}
/// 设置模型名（模型后缀路由归一）。
backend/tests/upstream/protocol/mod.rs
+3 -3
View File
@@ -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::{
backend/tests/upstream/protocol/openai_responses.rs
+25 -11
View File
@@ -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
);
}