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: improve OpenAI protocol fidelity

Browse Source 
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
zyycn
2026-07-03 21:57:05 +08:00
co-authored by Codex
parent c03baed4f7
commit 85dc14db3c
10 changed files with 618 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/proxy/dispatch/responses/live_stream.rs
+3 -9
View File
@@ -16,7 +16,8 @@ use crate::{
upstream::{
protocol::{
responses::{
reconvert_responses_sse_event_tuple_values, update_first_response_event_ms,
reconvert_responses_sse_event_tuple_values, response_sse_event_is_terminal,
update_first_response_event_ms,
},
sse::{
encode_sse_event, parse_sse_events, sse_body_has_done, sse_frame_end,
@@ -292,14 +293,7 @@ async fn send_live_response_stream_tail(
}
fn sse_body_has_terminal_event(body: &str) -> bool {
parse_sse_events(body).is_ok_and(|events| {
events.iter().any(|event| {
matches!(
event.event.as_deref(),
Some("response.completed" | "response.failed" | "error")
)
})
})
parse_sse_events(body).is_ok_and(|events| events.iter().any(response_sse_event_is_terminal))
}
fn missing_sse_event_separator(body: &str) -> Option<&'static str> {
backend/src/proxy/openai/chat.rs
+217 -72
View File
@@ -12,7 +12,7 @@ use axum::{
use chrono::Utc;
use futures::{stream as futures_stream, StreamExt};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use serde_json::{json, Map, Value};
use std::convert::Infallible;
use thiserror::Error;
use uuid::Uuid;
@@ -211,32 +211,27 @@ impl ChatCompletionStreamTranslator {
Some("response.output_text.delta") => {
self.push_text_delta(&value, output);
}
Some("response.reasoning_summary_text.delta") if self.include_reasoning => {
if let Some(delta) = value.get("delta").and_then(Value::as_str) {
self.has_content = true;
output.push_str(&self.frame(json!({
"choices": [{
"delta": {"reasoning_content": delta},
"finish_reason": Value::Null,
"index": 0,
}],
})));
}
Some("response.reasoning_summary_text.delta" | "response.reasoning_text.delta")
if self.include_reasoning =>
{
self.push_reasoning_delta(&value, output);
}
Some("response.output_item.added") => {
self.push_function_call_start(&value, output);
}
Some("response.function_call_arguments.delta") => {
Some(
"response.function_call_arguments.delta" | "response.custom_tool_call_input.delta",
) => {
self.push_function_call_delta(&value, output);
}
Some("response.function_call_arguments.done") => {
self.push_function_call_done(&value, output);
}
Some("response.output_item.done") => {
self.push_image_generation_done(&value, output);
self.push_output_item_done(&value, output);
}
Some("response.completed") => {
self.push_completed(&value, output);
Some("response.completed" | "response.incomplete") => {
self.push_terminal_response(&value, output);
}
_ => {}
}
@@ -262,21 +257,47 @@ impl ChatCompletionStreamTranslator {
})));
}
fn push_reasoning_delta(&mut self, value: &Value, output: &mut String) {
let Some(delta) = value.get("delta").and_then(Value::as_str) else {
return;
};
self.has_content = true;
output.push_str(&self.frame(json!({
"choices": [{
"delta": {"reasoning_content": delta},
"finish_reason": Value::Null,
"index": 0,
}],
})));
}
fn push_function_call_start(&mut self, value: &Value, output: &mut String) {
let Some((item_id, call_id, name)) = function_call_item(value.get("item")) else {
return;
};
self.ensure_function_call_started(item_id, call_id, name, output);
}
fn ensure_function_call_started(
&mut self,
item_id: String,
call_id: String,
name: String,
output: &mut String,
) {
self.function_call_items
.entry(item_id)
.or_insert_with(|| FunctionCallInfo {
call_id: call_id.clone(),
name: name.clone(),
});
if self.tool_call_indices.contains_key(&call_id) {
return;
}
self.has_tool_calls = true;
self.has_content = true;
let index = self.next_tool_call_index;
self.next_tool_call_index += 1;
self.function_call_items.insert(
item_id,
FunctionCallInfo {
call_id: call_id.clone(),
name: name.clone(),
},
);
self.tool_call_indices.insert(call_id.clone(), index);
output.push_str(&self.frame(json!({
"choices": [{
@@ -336,12 +357,19 @@ impl ChatCompletionStreamTranslator {
})));
}
fn push_image_generation_done(&mut self, value: &Value, output: &mut String) {
let Some(item) = value.get("item") else {
fn push_output_item_done(&mut self, value: &Value, output: &mut String) {
if self.push_image_generation_done(value, output) {
return;
}
self.push_function_call_output_item_done(value, output);
}
fn push_image_generation_done(&mut self, value: &Value, output: &mut String) -> bool {
let Some(item) = value.get("item") else {
return false;
};
if item.get("type").and_then(Value::as_str) != Some("image_generation_call") {
return;
return false;
}
self.has_tool_calls = true;
self.has_content = true;
@@ -380,9 +408,35 @@ impl ChatCompletionStreamTranslator {
"index": 0,
}],
})));
true
}
fn push_completed(&mut self, value: &Value, output: &mut String) {
fn push_function_call_output_item_done(&mut self, value: &Value, output: &mut String) {
let Some((item_id, call_id, name, arguments)) =
tool_call_output_item(value.get("item"), &self.function_call_items)
else {
return;
};
self.ensure_function_call_started(item_id, call_id.clone(), name, output);
if self.call_ids_with_deltas.contains(&call_id) {
return;
}
let index = self.tool_call_indices.get(&call_id).copied().unwrap_or(0);
output.push_str(&self.frame(json!({
"choices": [{
"delta": {
"tool_calls": [{
"function": {"arguments": arguments},
"index": index,
}],
},
"finish_reason": Value::Null,
"index": 0,
}],
})));
}
fn push_terminal_response(&mut self, value: &Value, output: &mut String) {
self.flush_tuple_text(output);
if !self.has_content {
output.push_str(&self.frame(json!({
@@ -394,10 +448,11 @@ impl ChatCompletionStreamTranslator {
})));
}
let response = value.get("response");
let finish_reason = openai_finish_reason_for_response(response, self.has_tool_calls);
output.push_str(&self.frame(json!({
"choices": [{
"delta": {},
"finish_reason": if self.has_tool_calls { "tool_calls" } else { "stop" },
"finish_reason": finish_reason,
"index": 0,
}],
"usage": openai_stream_usage(response.and_then(|response| response.get("usage"))),
@@ -498,7 +553,7 @@ pub fn translate_chat_to_codex(
text: prepared.text,
tuple_schema: prepared.tuple_schema,
tools: codex_tools(request.tools, request.functions),
tool_choice: request.tool_choice,
tool_choice: codex_tool_choice(request.tool_choice),
parallel_tool_calls: request.parallel_tool_calls,
reasoning: request.reasoning_effort.map(|effort| {
json!({
@@ -676,18 +731,57 @@ fn reconvert_tuple_text(text: &str, tuple_schema: &Value) -> Option<String> {
}
fn codex_tools(tools: Option<Vec<Value>>, functions: Option<Vec<Value>>) -> Option<Vec<Value>> {
if let Some(tools) = tools.filter(|tools| !tools.is_empty()) {
return Some(tools);
let mut codex_tools = Vec::new();
if let Some(tools) = tools {
codex_tools.extend(tools.into_iter().map(codex_tool));
}
if let Some(functions) = functions {
codex_tools.extend(functions.into_iter().map(codex_function_tool));
}
(!codex_tools.is_empty()).then_some(codex_tools)
}
functions
.filter(|functions| !functions.is_empty())
.map(|functions| {
functions
.into_iter()
.map(|function| json!({"type": "function", "function": function}))
.collect()
})
fn codex_tool(tool: Value) -> Value {
if tool.get("type").and_then(Value::as_str) != Some("function") {
return tool;
}
if let Some(function) = tool.get("function").filter(|value| value.is_object()) {
return codex_function_tool(function.clone());
}
codex_function_tool(tool)
}
fn codex_function_tool(function: Value) -> Value {
let mut object = Map::new();
object.insert("type".to_string(), Value::String("function".to_string()));
for field in ["name", "description", "parameters", "strict"] {
if let Some(value) = function.get(field) {
object.insert(field.to_string(), value.clone());
}
}
Value::Object(object)
}
fn codex_tool_choice(tool_choice: Option<Value>) -> Option<Value> {
tool_choice.map(codex_tool_choice_value)
}
fn codex_tool_choice_value(choice: Value) -> Value {
if choice.get("type").and_then(Value::as_str) != Some("function") {
return choice;
}
let name = choice
.get("name")
.or_else(|| choice.pointer("/function/name"))
.and_then(Value::as_str)
.filter(|name| !name.trim().is_empty());
let Some(name) = name else {
return choice;
};
json!({
"type": "function",
"name": name,
})
}
struct PreparedResponseFormat {
@@ -830,7 +924,7 @@ pub fn chat_completion_from_codex_sse(
tool_calls.push(tool_call);
}
}
Some("response.completed") => {
Some("response.completed" | "response.incomplete") => {
completed_response = value.get("response").cloned();
}
_ => {}
@@ -851,6 +945,7 @@ pub fn chat_completion_from_codex_sse(
};
let mut choices = Vec::new();
let finish_reason = openai_finish_reason_for_response(Some(&response), !tool_calls.is_empty());
if !tool_calls.is_empty() {
choices.push(json!({
"index": 0,
@@ -859,7 +954,7 @@ pub fn chat_completion_from_codex_sse(
"content": output_text_value,
"tool_calls": tool_calls,
},
"finish_reason": "tool_calls",
"finish_reason": finish_reason,
}));
} else {
choices.push(json!({
@@ -868,7 +963,7 @@ pub fn chat_completion_from_codex_sse(
"role": "assistant",
"content": output_text_value,
},
"finish_reason": "stop",
"finish_reason": finish_reason,
}));
}
@@ -1054,17 +1149,66 @@ fn output_text_from_item(item: &Value) -> Option<String> {
(!text.is_empty()).then_some(text)
}
fn openai_finish_reason_for_response(
response: Option<&Value>,
has_tool_calls: bool,
) -> &'static str {
let Some(response) = response else {
return openai_default_finish_reason(has_tool_calls);
};
if response.get("status").and_then(Value::as_str) != Some("incomplete") {
return openai_default_finish_reason(has_tool_calls);
}
match response_incomplete_reason(response) {
Some("max_output_tokens") => "length",
Some("content_filter") => "content_filter",
_ => openai_default_finish_reason(has_tool_calls),
}
}
fn response_incomplete_reason(response: &Value) -> Option<&str> {
response
.pointer("/incomplete_details/reason")
.or_else(|| response.get("reason"))
.and_then(Value::as_str)
}
fn openai_default_finish_reason(has_tool_calls: bool) -> &'static str {
if has_tool_calls {
"tool_calls"
} else {
"stop"
}
}
fn function_call_item(item: Option<&Value>) -> Option<(String, String, String)> {
let item = item?;
if item.get("type").and_then(Value::as_str) != Some("function_call") {
let item_type = item.get("type").and_then(Value::as_str)?;
if !matches!(item_type, "function_call" | "custom_tool_call") {
return None;
}
let item_id = item.get("id").and_then(Value::as_str)?;
let call_id = item.get("call_id").and_then(Value::as_str)?;
let name = item.get("name").and_then(Value::as_str)?;
let call_id = item
.get("call_id")
.and_then(Value::as_str)
.unwrap_or(item_id);
let name = item
.get("name")
.and_then(Value::as_str)
.filter(|name| !name.is_empty())
.unwrap_or(item_type);
Some((item_id.to_string(), call_id.to_string(), name.to_string()))
}
fn tool_arguments_from_value(value: &Value) -> &str {
value
.get("arguments")
.or_else(|| value.get("input"))
.and_then(Value::as_str)
.unwrap_or_default()
}
fn tool_call_from_done_event(
event: &Value,
function_call_items: &BTreeMap<String, FunctionCallInfo>,
@@ -1087,43 +1231,44 @@ fn tool_call_from_done_event(
.filter(|name| !name.is_empty())
.or_else(|| info.map(|info| info.name.as_str()))
.unwrap_or("unknown");
let arguments = event
.get("arguments")
.and_then(Value::as_str)
.unwrap_or_default();
let arguments = tool_arguments_from_value(event);
Some(openai_tool_call(&call_id, name, arguments))
}
fn tool_call_output_item(
item: Option<&Value>,
function_call_items: &BTreeMap<String, FunctionCallInfo>,
) -> Option<(String, String, String, String)> {
let item = item?;
let (item_id, call_id, name) = function_call_item(Some(item))?;
let info = function_call_items.get(&item_id);
let call_id = item
.get("call_id")
.and_then(Value::as_str)
.or_else(|| info.map(|info| info.call_id.as_str()))
.unwrap_or(&call_id)
.to_string();
let name = item
.get("name")
.and_then(Value::as_str)
.filter(|name| !name.is_empty())
.or_else(|| info.map(|info| info.name.as_str()))
.unwrap_or(&name)
.to_string();
let arguments = tool_arguments_from_value(item).to_string();
Some((item_id, call_id, name, arguments))
}
fn tool_call_from_output_item(
item: Option<&Value>,
function_call_items: &BTreeMap<String, FunctionCallInfo>,
finished_call_ids: &mut BTreeSet<String>,
) -> Option<Value> {
let item = item?;
if item.get("type").and_then(Value::as_str) != Some("function_call") {
let (_, call_id, name, arguments) = tool_call_output_item(item, function_call_items)?;
if !finished_call_ids.insert(call_id.clone()) {
return None;
}
let item_id = item.get("id").and_then(Value::as_str);
let info = item_id.and_then(|item_id| function_call_items.get(item_id));
let call_id = item
.get("call_id")
.and_then(Value::as_str)
.or_else(|| info.map(|info| info.call_id.as_str()))
.or(item_id)?;
if !finished_call_ids.insert(call_id.to_string()) {
return None;
}
let name = item
.get("name")
.and_then(Value::as_str)
.filter(|name| !name.is_empty())
.or_else(|| info.map(|info| info.name.as_str()))
.unwrap_or("unknown");
let arguments = item
.get("arguments")
.and_then(Value::as_str)
.unwrap_or_default();
Some(openai_tool_call(call_id, name, arguments))
Some(openai_tool_call(&call_id, &name, &arguments))
}
fn openai_tool_call(call_id: &str, name: &str, arguments: &str) -> Value {
backend/src/upstream/protocol/responses.rs
+34 -12
View File
@@ -219,11 +219,11 @@ fn complete_sse_body_prefix(body: &str) -> Option<&str> {
/// 从 Codex SSE 收集出的非流式 Responses 结果。
#[derive(Debug, Clone, PartialEq)]
pub enum CollectedResponse {
/// 收集到 `response.completed`。
/// 收集到 `response.completed` 或 `response.incomplete`。
Completed(Value),
/// 收集到 `response.failed` 或 `error`。
Failed(ResponsesSseFailure),
/// SSE 未包含 `response.completed`。
/// SSE 未包含成功终止响应。
MissingCompleted,
/// 完成响应为空。
Empty,
@@ -269,7 +269,7 @@ pub fn response_from_codex_sse(
let events = parse_sse_events(body)?;
let mut output_text = String::new();
let mut output_items = Vec::new();
let mut completed_response = None;
let mut terminal_response = None;
let mut failed_response = None;
for event in events {
@@ -287,8 +287,8 @@ pub fn response_from_codex_sse(
output_items.push(item.clone());
}
}
Some("response.completed") => {
completed_response = value.get("response").cloned();
Some("response.completed" | "response.incomplete") => {
terminal_response = value.get("response").cloned();
}
Some(event_name @ ("error" | "response.failed")) if failed_response.is_none() => {
failed_response = Some(ResponsesSseFailure::from_event(event_name, &value));
@@ -300,7 +300,7 @@ pub fn response_from_codex_sse(
if let Some(failure) = failed_response {
return Ok(CollectedResponse::Failed(failure));
}
let Some(mut response) = completed_response else {
let Some(mut response) = terminal_response else {
return Ok(CollectedResponse::MissingCompleted);
};
if is_empty_response(&response, &output_text, &output_items) {
@@ -335,7 +335,7 @@ pub fn completed_response_metadata(
function_call_ids.insert(call_id.to_string());
}
}
Some("response.completed") => {
Some("response.completed" | "response.incomplete") => {
response_id = value
.pointer("/response/id")
.and_then(Value::as_str)
@@ -364,7 +364,7 @@ pub fn reconvert_responses_sse_event_tuple_values(
mut data: Value,
tuple_schema: &Value,
) -> Value {
match responses_event_type(event_name, &data) {
match responses_event_type(event_name, Some(&data)) {
Some("response.output_text.delta") => {
reconvert_output_text_delta_tuple_values(&mut data, tuple_schema);
}
@@ -373,7 +373,7 @@ pub fn reconvert_responses_sse_event_tuple_values(
reconvert_output_item_tuple_values(item, tuple_schema);
}
}
Some("response.completed") => {
Some("response.completed" | "response.incomplete") => {
if let Some(response) = data.get_mut("response") {
reconvert_completed_response_tuple_values(response, Some(tuple_schema));
sync_output_text_from_output(response);
@@ -384,8 +384,23 @@ pub fn reconvert_responses_sse_event_tuple_values(
data
}
fn responses_event_type<'a>(event_name: Option<&'a str>, data: &'a Value) -> Option<&'a str> {
event_name.or_else(|| data.get("type").and_then(Value::as_str))
/// 判断 Responses SSE 事件是否为终止事件。
pub fn response_sse_event_is_terminal(event: &SseEvent) -> bool {
let value = serde_json::from_str::<Value>(&event.data).ok();
matches!(
responses_event_type(event.event.as_deref(), value.as_ref()),
Some("response.completed" | "response.incomplete" | "response.failed" | "error")
)
}
fn responses_event_type<'a>(
event_name: Option<&'a str>,
data: Option<&'a Value>,
) -> Option<&'a str> {
event_name.or_else(|| {
data.and_then(|data| data.get("type"))
.and_then(Value::as_str)
})
}
fn reconvert_output_text_delta_tuple_values(data: &mut Value, tuple_schema: &Value) {
@@ -403,6 +418,9 @@ fn is_empty_response(response: &Value, output_text: &str, output_items: &[Value]
if !output_text.trim().is_empty() || !output_items.is_empty() {
return false;
}
if response.get("status").and_then(Value::as_str) == Some("incomplete") {
return false;
}
response
.pointer("/usage/output_tokens")
@@ -432,9 +450,13 @@ fn ensure_completed_response_output(
return;
}
let item_status = response
.get("status")
.and_then(Value::as_str)
.unwrap_or("completed");
response["output"] = json!([{
"type": "message",
"status": "completed",
"status": item_status,
"role": "assistant",
"content": [{
"type": "output_text",
backend/tests/fixtures/responses/http_sse/chat_delta_incomplete_usage.sse
Vendored
+5
View File
@@ -0,0 +1,5 @@
event: response.output_text.delta
data: {"type":"response.output_text.delta","delta":"hello"}
event: response.incomplete
data: {"type":"response.incomplete","response":{"id":"resp_incomplete","object":"response","status":"incomplete","incomplete_details":{"reason":"max_output_tokens"},"usage":{"input_tokens":2,"output_tokens":3},"output":[{"type":"message","status":"incomplete","role":"assistant","content":[{"type":"output_text","text":"hello","annotations":[]}]}],"output_text":"hello"}}
backend/tests/proxy/dispatch/chat_upstream/mod.rs
+4
View File
@@ -168,6 +168,10 @@ const RESPONSES_STREAM_USAGE_SSE: &str = concat!(
include_str!("../../../fixtures/responses/http_sse/stream_usage.sse"),
"\n"
);
const RESPONSES_INCOMPLETE_USAGE_SSE: &str = concat!(
include_str!("../../../fixtures/responses/http_sse/chat_delta_incomplete_usage.sse"),
"\n"
);
const RESPONSES_STREAM_AFTER_429_SSE: &str = concat!(
include_str!("../../../fixtures/responses/http_sse/stream_after_429.sse"),
"\n"
backend/tests/proxy/dispatch/chat_upstream/responses_http.rs
+45
View File
@@ -99,6 +99,51 @@ async fn responses_should_honor_explicit_http_sse_transport() {
assert!(upstream_body.get("use_websocket").is_none());
}
#[tokio::test]
async fn responses_stream_should_treat_incomplete_as_terminal() {
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/codex/responses"))
.respond_with(
ResponseTemplate::new(200)
.insert_header("content-type", "text/event-stream")
.set_body_string(RESPONSES_INCOMPLETE_USAGE_SSE),
)
.expect(1)
.mount(&server)
.await;
let (app, api_key, _dir) = test_app_with_account(server.uri()).await;
let response = app
.oneshot(
Request::builder()
.method("POST")
.uri("/v1/responses")
.header("authorization", format!("Bearer {api_key}"))
.header("content-type", "application/json")
.body(Body::from(
json!({
"model": "gpt-5.5",
"input": [],
"stream": true,
"use_websocket": false
})
.to_string(),
))
.unwrap(),
)
.await
.unwrap();
let status = response.status();
let body = response_text(response).await;
assert_eq!(status, StatusCode::OK);
assert!(body.contains("event: response.incomplete"));
assert!(body.contains(r#""reason":"max_output_tokens""#));
assert!(!body.contains("event: response.failed"));
assert!(body.ends_with("data: [DONE]\n\n"));
}
#[tokio::test]
async fn responses_should_stagger_same_account_requests_before_sending_upstream() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
backend/tests/upstream/protocol/mod.rs
+2 -2
View File
@@ -9,8 +9,8 @@ use codex_proxy_rs::upstream::protocol::events::{
};
use codex_proxy_rs::upstream::protocol::responses::{
apply_response_model_options, http_sse_fallback_allowed, response_body_has_first_event,
response_from_codex_sse, transport_for_request, CodexResponsesRequest, CodexTransport,
CollectedResponse,
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, sse_body_has_done, DONE_SSE_FRAME,
backend/tests/upstream/protocol/openai_chat.rs
+243
View File
@@ -43,6 +43,91 @@ fn chat_completion_request_should_translate_to_codex_request() {
assert_eq!(codex.client_conversation_id.as_deref(), Some("client-123"));
}
#[test]
fn chat_completion_request_should_flatten_function_tools_for_codex_request() {
let mut request = basic_chat_request();
request.tools = Some(vec![
json!({
"type": "function",
"function": {
"name": "lookup",
"description": "Lookup docs",
"parameters": {"type": "object"},
"strict": true,
},
}),
json!({
"type": "web_search_preview",
"search_context_size": "low",
}),
]);
let codex = codex_proxy_rs::proxy::openai::chat::translate_chat_to_codex(request)
.expect("chat request should translate");
assert_eq!(
codex.tools,
Some(vec![
json!({
"type": "function",
"name": "lookup",
"description": "Lookup docs",
"parameters": {"type": "object"},
"strict": true,
}),
json!({
"type": "web_search_preview",
"search_context_size": "low",
}),
])
);
}
#[test]
fn chat_completion_request_should_flatten_legacy_functions_for_codex_request() {
let mut request = basic_chat_request();
request.functions = Some(vec![json!({
"name": "lookup",
"description": "Lookup docs",
"parameters": {"type": "object"},
"strict": false,
})]);
let codex = codex_proxy_rs::proxy::openai::chat::translate_chat_to_codex(request)
.expect("chat request should translate");
assert_eq!(
codex.tools,
Some(vec![json!({
"type": "function",
"name": "lookup",
"description": "Lookup docs",
"parameters": {"type": "object"},
"strict": false,
})])
);
}
#[test]
fn chat_completion_request_should_flatten_function_tool_choice_for_codex_request() {
let mut request = basic_chat_request();
request.tool_choice = Some(json!({
"type": "function",
"function": {"name": "lookup"},
}));
let codex = codex_proxy_rs::proxy::openai::chat::translate_chat_to_codex(request)
.expect("chat request should translate");
assert_eq!(
codex.tool_choice,
Some(json!({
"type": "function",
"name": "lookup",
}))
);
}
#[test]
fn sse_parser_should_combine_multiline_data_and_metadata() {
let events = codex_proxy_rs::upstream::protocol::sse::parse_sse_events(include_str!(
@@ -77,6 +162,20 @@ fn chat_completion_from_codex_sse_should_convert_completed_response() {
assert_eq!(response["choices"][0]["message"]["content"], "hello");
}
#[test]
fn chat_completion_from_codex_sse_should_convert_incomplete_response() {
let body = include_str!("../../fixtures/responses/http_sse/chat_delta_incomplete_usage.sse");
let response = codex_proxy_rs::proxy::openai::chat::chat_completion_from_codex_sse(
body, "gpt-5.5", false, None,
)
.expect("conversion should succeed")
.expect("terminal response");
assert_eq!(response["choices"][0]["finish_reason"], "length");
assert_eq!(response["choices"][0]["message"]["content"], "hello");
}
#[test]
fn chat_completion_stream_translator_should_emit_openai_chunks() {
let mut translator = codex_proxy_rs::proxy::openai::chat::ChatCompletionStreamTranslator::new(
@@ -109,3 +208,147 @@ fn chat_completion_stream_translator_should_emit_openai_chunks() {
],
);
}
#[test]
fn chat_completion_stream_translator_should_complete_incomplete_response() {
let mut translator = codex_proxy_rs::proxy::openai::chat::ChatCompletionStreamTranslator::new(
"gpt-5.5", false, None,
);
let body = include_str!("../../fixtures/responses/http_sse/chat_delta_incomplete_usage.sse");
let output = format!(
"{}{}",
translator.initial_frame(),
translator
.push_str(body)
.expect("stream conversion should succeed")
);
assert_substrings_appear_in_order(
&output,
&[
"\"delta\":{\"role\":\"assistant\"}",
"\"delta\":{\"content\":\"hello\"}",
"\"finish_reason\":\"length\"",
"\"usage\":{\"completion_tokens\":3,\"prompt_tokens\":2,\"total_tokens\":5}",
"data: [DONE]",
],
);
}
#[test]
fn chat_completion_stream_translator_should_emit_reasoning_text_delta() {
let mut translator = codex_proxy_rs::proxy::openai::chat::ChatCompletionStreamTranslator::new(
"gpt-5.5", true, None,
);
let output = translator
.push_str("event: response.reasoning_text.delta\ndata: {\"delta\":\"thinking\"}\n\n")
.expect("stream conversion should succeed");
assert!(output.contains("\"delta\":{\"reasoning_content\":\"thinking\"}"));
}
#[test]
fn chat_completion_stream_translator_should_emit_custom_tool_input_delta() {
let mut translator = codex_proxy_rs::proxy::openai::chat::ChatCompletionStreamTranslator::new(
"gpt-5.5", false, None,
);
let body = concat!(
"event: response.output_item.added\n",
"data: {\"item\":{\"type\":\"custom_tool_call\",\"id\":\"item_patch\",\"call_id\":\"call_patch\",\"name\":\"apply_patch\"}}\n\n",
"event: response.custom_tool_call_input.delta\n",
"data: {\"item_id\":\"item_patch\",\"delta\":\"*** Begin Patch\"}\n\n",
"event: response.completed\n",
"data: {\"response\":{\"id\":\"resp_1\",\"status\":\"completed\",\"usage\":{\"input_tokens\":1,\"output_tokens\":1}}}\n\n",
);
let output = format!(
"{}{}",
translator.initial_frame(),
translator
.push_str(body)
.expect("stream conversion should succeed")
);
assert_substrings_appear_in_order(
&output,
&[
"\"function\":{\"arguments\":\"\",\"name\":\"apply_patch\"}",
"\"id\":\"call_patch\"",
"\"function\":{\"arguments\":\"*** Begin Patch\"}",
"\"finish_reason\":\"tool_calls\"",
"data: [DONE]",
],
);
}
#[test]
fn chat_completion_from_codex_sse_should_convert_custom_tool_output_item() {
let body = concat!(
"event: response.output_item.done\n",
"data: {\"item\":{\"type\":\"custom_tool_call\",\"id\":\"item_patch\",\"call_id\":\"call_patch\",\"name\":\"apply_patch\",\"input\":\"*** Begin Patch\"}}\n\n",
"event: response.completed\n",
"data: {\"response\":{\"id\":\"resp_1\",\"status\":\"completed\",\"output\":[],\"usage\":{\"input_tokens\":1,\"output_tokens\":1}}}\n\n",
);
let response = codex_proxy_rs::proxy::openai::chat::chat_completion_from_codex_sse(
body, "gpt-5.5", false, None,
)
.expect("conversion should succeed")
.expect("completed response");
assert_eq!(
response["choices"][0]["message"]["tool_calls"][0],
json!({
"id": "call_patch",
"type": "function",
"function": {
"name": "apply_patch",
"arguments": "*** Begin Patch",
},
})
);
}
#[test]
fn chat_completion_stream_translator_should_emit_tool_arguments_from_output_item_done() {
let mut translator = codex_proxy_rs::proxy::openai::chat::ChatCompletionStreamTranslator::new(
"gpt-5.5", false, None,
);
let output = translator
.push_str(concat!(
"event: response.output_item.done\n",
"data: {\"item\":{\"type\":\"function_call\",\"id\":\"item_lookup\",\"call_id\":\"call_lookup\",\"name\":\"lookup\",\"arguments\":\"{\\\"q\\\":\\\"rust\\\"}\"}}\n\n",
))
.expect("stream conversion should succeed");
assert_substrings_appear_in_order(
&output,
&[
"\"function\":{\"arguments\":\"\",\"name\":\"lookup\"}",
"\"id\":\"call_lookup\"",
"\"function\":{\"arguments\":\"{\\\"q\\\":\\\"rust\\\"}\"}",
],
);
}
fn basic_chat_request() -> codex_proxy_rs::proxy::openai::chat::ChatCompletionRequest {
codex_proxy_rs::proxy::openai::chat::ChatCompletionRequest {
model: "gpt-5.5".to_string(),
stream: false,
messages: vec![codex_proxy_rs::proxy::openai::chat::ChatMessage {
role: "user".to_string(),
content: Some(json!("hello")),
name: None,
tool_calls: None,
tool_call_id: None,
function_call: None,
}],
reasoning_effort: None,
service_tier: None,
tools: None,
tool_choice: None,
parallel_tool_calls: None,
functions: None,
response_format: None,
user: None,
}
}
backend/tests/upstream/protocol/openai_responses.rs
+42
View File
@@ -97,6 +97,48 @@ fn response_from_codex_sse_should_collect_completed_response_with_deltas() {
);
}
#[test]
fn response_from_codex_sse_should_collect_incomplete_response() {
let body = include_str!("../../fixtures/responses/http_sse/chat_delta_incomplete_usage.sse");
let response = response_from_codex_sse(body, None).expect("conversion should succeed");
assert_eq!(
response,
CollectedResponse::Completed(json!({
"id": "resp_incomplete",
"object": "response",
"status": "incomplete",
"incomplete_details": {
"reason": "max_output_tokens"
},
"usage": {
"input_tokens": 2,
"output_tokens": 3
},
"output": [{
"type": "message",
"status": "incomplete",
"role": "assistant",
"content": [{
"type": "output_text",
"text": "hello",
"annotations": []
}]
}],
"output_text": "hello"
}))
);
}
#[test]
fn response_sse_event_is_terminal_should_match_incomplete_response() {
let body = include_str!("../../fixtures/responses/http_sse/chat_delta_incomplete_usage.sse");
let events = parse_sse_events(body).expect("sse should parse");
assert!(events.iter().any(response_sse_event_is_terminal));
}
#[test]
fn response_failed_sse_event_should_encode_openai_failure_shape() {
let frame = response_failed_sse_event("server_error", "stream_disconnected", "closed early");
docs/protocol-fidelity-audit-2026-07-03.md
+23 -18
View File
@@ -14,6 +14,7 @@
1. **desktop bundle 基准（强）**：从当日最新官方 `Codex.dmg`（`persistent.oaistatic.com/codex-app-prod/Codex.dmg`，构建时间 2026-07-03）解包出的 Electron `app.asar`，提取 `.vite/build/{main,src,worker}.js` 中的协议字符串。`worker.js` 是核心流处理器。
2. **OpenAI 官方契约基准（中）**：我知识内的 OpenAI API 契约，用于错误体、Models 端点等 bundle 未覆盖处。
3. **开源实现对照（辅助）**：本轮补充对照 `/home/zyy/桌面/Codes/sub2api` 与 `/home/zyy/桌面/Codes/CLIProxyAPI` 的 codegraph 结果。二者不能替代官方抓包，但可用于验证社区常见的 Chat Completions ⇄ Responses/Codex 转换形状。
### 基准的关键局限（务必先读）
@@ -53,27 +54,30 @@ response.output_item.done
- **`/v1/responses` 直通**（`TupleSseEventTransformer`）：无 schema 时逐字节透传，接近官方透传语义，保真度高。
- **`/v1/chat/completions` 转换**（`ChatCompletionStreamTranslator`）：主动把 Codex Responses SSE 转成 `chat.completion.chunk`，`match` 未命中的事件 `_ => {}` 丢弃。**偏差集中在此路径。**
官方 11 个核心事件中本项目识别 10 个。
官方 11 个核心事件中本项目当前已识别 11 个；`response.incomplete` 已在本轮补齐。
### 🔴 严重（已读代码核实）
**1. `response.incomplete` 全路径未处理** — 可信度最高（基准硬 + 已核实）
**1. `response.incomplete` 全路径终止处理（已修复）** — 可信度最高（基准硬 + 已核实）
- 基准：`response.incomplete` 在官方 `nh` 事件全集内，是合法终止事件（命中 `max_output_tokens`、内容策略等），官方按 `response.status` 采集 finishReasons 并原样透传。
- 项目自身 `upstream/protocol/websocket.rs:235,725` 已在 **WebSocket 路径**完整识别该事件（能取 `incomplete_details/reason`），但 **HTTP SSE 路径漏了它**：
- 项目自身 `upstream/protocol/websocket.rs:235,725` 已在 **WebSocket 路径**完整识别该事件（能取 `incomplete_details/reason`），修复前 **HTTP SSE 路径漏了它**：
- `proxy/openai/chat.rs` push_event 的 match 无 `response.incomplete` 分支 → 落入 `_ => {}` 丢弃。唯一发 `finish_reason` 和 `[DONE]` 的 `push_completed` 只在 `response.completed` 触发。
- `proxy/dispatch/responses.rs:2389` `sse_body_has_terminal_event` 只认 `completed|failed|error`。
- 触发场景：任意"未完成但正常终止"的响应（尤其 `max_output_tokens` 截断）。
- 后果：
- 修复前后果：
- chat 客户端：流无终止 chunk、无 `[DONE]`，流悬挂到超时。
- responses 客户端：在合法 `response.incomplete` 后被追加一个伪造的 `response.failed`(stream_disconnected)，出现双终止事件；后台把成功的截断响应误记为 502，污染用量/账号健康统计。
- 修复方向：SSE 两条路径把 `response.incomplete` 纳入终止事件集，复用 WebSocket 路径已有的 `websocket_incomplete_response_reason` 逻辑。
- 修复状态（2026-07-03）：已将 `response.incomplete` 纳入 Responses SSE 终止事件集；Chat Completions 流式/非流式转换会把 `max_output_tokens` 映射为 `finish_reason=length`，并正常输出终止 chunk / `[DONE]`；Responses live stream 不再为合法 incomplete 追加伪造 `response.failed`。
**2. Chat Completions 工具定义未从嵌套转扁平** — 方向可信，需实测确认
**2. Chat Completions 工具定义未从嵌套转扁平（已修复）** — 方向可信，已用开源实现辅助确认
- 基准：Codex 上游是 Responses API，函数工具应为扁平 `{type:"function", name, description, parameters}`；Chat 输入是嵌套 `{type:"function", function:{...}}`。二者不可互换。**注意**：工具 schema 的官方真实取值属"基准不确定"（不在 bundle 内），此条依据"Codex = Responses API"推断，方向可信但建议真实抓包确认后再改。
- 项目：`proxy/openai/chat.rs:676` `codex_tools()` 的 `tools` 分支 `return Some(tools)` 原样透传嵌套结构；legacy `functions` 还额外包一层 `{type:"function","function":function}`。
- 后果：若上游确实要扁平结构，`name`/`parameters` 在顶层缺失 → 上游报 400 或忽略工具 → `/v1/chat/completions` 工具调用整体失效。其余流式/非流式解析逻辑都写得完整，唯独请求方向的工具定义可能从一开始就错。
- 基准：Codex 上游是 Responses API，函数工具应为扁平 `{type:"function", name, description, parameters}`；Chat 输入是嵌套 `{type:"function", function:{...}}`。二者不可互换。**注意**：工具 schema 的官方真实取值属"基准不确定"（不在 bundle 内），此条依据"Codex = Responses API"和开源实现对照修复，后续仍可用真实抓包复核。
- 开源对照：
- `sub2api` `convertChatToolsToResponses()`：把 Chat `tools[].function` 和 legacy `functions[]` 转为 Responses 扁平 `{type:"function", name, description, parameters, strict}`。
- `CLIProxyAPI` `ConvertOpenAIRequestToCodex()`：把 Chat 函数工具扁平化，内置工具原样透传；函数型 `tool_choice` 从 `{type:"function", function:{name}}` 转为 `{type:"function", name}`。
- 项目修复前：`proxy/openai/chat.rs` `codex_tools()` 的 `tools` 分支原样透传嵌套结构；legacy `functions` 还额外包一层 `{type:"function","function":function}`。
- 修复状态（2026-07-03）：Chat `tools[].function`、legacy `functions[]` 已转为 Responses 扁平函数工具，非 function 内置工具继续透传；函数型 `tool_choice` 已转为 `{type:"function", name}`。
### 🟡 中等
@@ -88,11 +92,11 @@ response.output_item.done
SSE / chat 转换：
- chat `finish_reason` 恒为 `stop`/`tool_calls`（`chat.rs:398`），**永不产生 `length`/`content_filter`**；叠加 incomplete 未处理，截断响应既无 `length` 也无终止帧。
- chat 转换丢弃 `response.reasoning_text.delta`（只映射 `reasoning_summary_text.delta`）与 `custom_tool_call_input.delta`；若模型只发原始 reasoning_text 或 custom tool 增量，在 chat 路径被静默丢弃。两者在 `/v1/responses` 直通路径均正常透传。
- chat 路径 `tool_choice` 命名工具格式未转换（`chat.rs:499`）：Responses 期望 `{type:function, name}`，透传 Chat 的 `{type:function, function:{name}}` 会被上游忽略。
- chat `response.incomplete` 已映射 `length`/`content_filter`；其它终止状态仍只有 `stop`/`tool_calls`，若上游未来在 `response.completed` 中携带非 stop 终止语义，需要继续扩展。
- chat 转换 `response.reasoning_text.delta` 与 `response.custom_tool_call_input.delta` 已补齐；custom tool 增量按普通 function tool arguments delta 转为 Chat `tool_calls`。
- chat 路径函数型 `tool_choice` 命名工具格式已转换为 Responses 期望的 `{type:function, name}`。
- legacy `function_call` 回传 call_id 用 `fc_{name}` 伪造（`chat.rs:595`），与真实 call_id 对不上，多轮工具对话无法关联结果。
- 流式 `output_item.done` 无 function_call 兜底（`chat.rs:337` 仅处理 `image_generation_call`）；非流式有双来源兜底。边缘风险。
- 流式 `output_item.done` 已补 function/custom tool 兜底；若上游没有单独发送 arguments delta，也能从完整 output item 生成 Chat `tool_calls` 参数。
Models 端点（结构层面与 OpenAI 契约高度一致）：
@@ -106,17 +110,18 @@ Models 端点（结构层面与 OpenAI 契约高度一致）：
- `store` 默认 false 且始终序列化（`responses.rs:624`）。
- `include` 自动注入 `reasoning.encrypted_content`（`responses.rs:585`）— 符合公开 Codex 惯例但 bundle 无字面量命中。
- 不构造 `max_output_tokens`：客户端传入会被静默丢弃。
- 不构造 `max_output_tokens`：客户端传入会被静默丢弃。本轮对照结论分裂：`sub2api` 会把 Chat `max_tokens`/`max_completion_tokens` 映射到 Responses `max_output_tokens`；`CLIProxyAPI` 明确注释 Codex 不支持并禁用该映射。暂不改，仍需真实抓包或产品决策。
- `service_tier` 的 `fast`→`priority` 归一化（`responses.rs:597`）为本项目自定义。
- 两条 reasoning 构造路径（`proxy/openai/responses.rs:358` vs `responses.rs:544`）在 effort 缺失时行为不完全一致。
- `function_call_output` 无专门净化分支，靠 `_ => Some(object)` 原样透传。
## 优先级建议
1. **修 `response.incomplete`（🔴，基准硬）** — 两条 SSE 路径纳入终止事件集，复用 WebSocket 已有逻辑。收益最大、证据最硬、改动可控。
2. **确认 chat 工具定义转扁平（🔴，需实测）** — 先对官方 `codex` CLI 抓一次带 tools 的请求确认扁平结构，再改 `codex_tools()`。
3. **决策 quota→402 vs 429（🟡，产品权衡）** — 明确是要 OpenAI SDK 兼容还是语义化 HTTP 码。
4. 补 chat `finish_reason=length`、聚合错误 type 语义映射等中等项。
1. **修 `response.incomplete`（🔴，基准硬，已完成）** — 两条 SSE 路径已纳入终止事件集，Chat Completions 已补 `length` finish reason 和 `[DONE]`。
2. **chat 工具定义/`tool_choice` 转扁平（🔴，已完成）** — 已参考 sub2api 与 CLIProxyAPI 的 common path，并补充请求转换测试。
3. **chat reasoning/custom tool delta 与 `output_item.done` 兜底（🟡，已完成）** — 已补 Chat 流式/非流式转换测试。
4. **决策 quota→402 vs 429（🟡，产品权衡）** — 明确是要 OpenAI SDK 兼容还是语义化 HTTP 码。
5. 处理聚合错误 type 语义映射、Models 首拉为空/别名详情等中等项。
## 复现方法（解包基准）