diff --git a/backend/src/proxy/dispatch/responses/live_stream.rs b/backend/src/proxy/dispatch/responses/live_stream.rs index 6cc25e31d..fa0c4c78e 100644 --- a/backend/src/proxy/dispatch/responses/live_stream.rs +++ b/backend/src/proxy/dispatch/responses/live_stream.rs @@ -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> { diff --git a/backend/src/proxy/openai/chat.rs b/backend/src/proxy/openai/chat.rs index 944e9ea07..6b1a0b000 100644 --- a/backend/src/proxy/openai/chat.rs +++ b/backend/src/proxy/openai/chat.rs @@ -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 { } fn codex_tools(tools: Option>, functions: Option>) -> Option> { - 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) -> Option { + 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 { (!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, @@ -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, +) -> 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, finished_call_ids: &mut BTreeSet, ) -> Option { - 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 { diff --git a/backend/src/upstream/protocol/responses.rs b/backend/src/upstream/protocol/responses.rs index ac336afd6..3f638dcf2 100644 --- a/backend/src/upstream/protocol/responses.rs +++ b/backend/src/upstream/protocol/responses.rs @@ -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::(&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", diff --git a/backend/tests/fixtures/responses/http_sse/chat_delta_incomplete_usage.sse b/backend/tests/fixtures/responses/http_sse/chat_delta_incomplete_usage.sse new file mode 100644 index 000000000..0d2a0a49d --- /dev/null +++ b/backend/tests/fixtures/responses/http_sse/chat_delta_incomplete_usage.sse @@ -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"}} diff --git a/backend/tests/proxy/dispatch/chat_upstream/mod.rs b/backend/tests/proxy/dispatch/chat_upstream/mod.rs index ec27bcd74..60a9c8cea 100644 --- a/backend/tests/proxy/dispatch/chat_upstream/mod.rs +++ b/backend/tests/proxy/dispatch/chat_upstream/mod.rs @@ -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" diff --git a/backend/tests/proxy/dispatch/chat_upstream/responses_http.rs b/backend/tests/proxy/dispatch/chat_upstream/responses_http.rs index 82ba2fb96..7baed7143 100644 --- a/backend/tests/proxy/dispatch/chat_upstream/responses_http.rs +++ b/backend/tests/proxy/dispatch/chat_upstream/responses_http.rs @@ -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(); diff --git a/backend/tests/upstream/protocol/mod.rs b/backend/tests/upstream/protocol/mod.rs index 783a77554..dd0db209f 100644 --- a/backend/tests/upstream/protocol/mod.rs +++ b/backend/tests/upstream/protocol/mod.rs @@ -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, diff --git a/backend/tests/upstream/protocol/openai_chat.rs b/backend/tests/upstream/protocol/openai_chat.rs index cd133224f..b8b63ae1d 100644 --- a/backend/tests/upstream/protocol/openai_chat.rs +++ b/backend/tests/upstream/protocol/openai_chat.rs @@ -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, + } +} diff --git a/backend/tests/upstream/protocol/openai_responses.rs b/backend/tests/upstream/protocol/openai_responses.rs index 3ea9be541..f65438653 100644 --- a/backend/tests/upstream/protocol/openai_responses.rs +++ b/backend/tests/upstream/protocol/openai_responses.rs @@ -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"); diff --git a/docs/protocol-fidelity-audit-2026-07-03.md b/docs/protocol-fidelity-audit-2026-07-03.md index 508aae201..526b9bf87 100644 --- a/docs/protocol-fidelity-audit-2026-07-03.md +++ b/docs/protocol-fidelity-audit-2026-07-03.md @@ -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 首拉为空/别名详情等中等项。 ## 复现方法(解包基准)