diff --git a/backend/crates/gateway-api/src/openai/chat_completions/delivery.rs b/backend/crates/gateway-api/src/openai/chat_completions/delivery.rs new file mode 100644 index 000000000..17e11248e --- /dev/null +++ b/backend/crates/gateway-api/src/openai/chat_completions/delivery.rs @@ -0,0 +1,319 @@ +//! Chat 出口持有协议投影缓冲 提交与资源回收始终委托原受管正文 + +use std::collections::VecDeque; + +use axum::body::to_bytes; +use bytes::Bytes; +use futures::future::BoxFuture; +use gateway_core::{ + engine::middleware::{ + MiddlewareBody, MiddlewareError, MiddlewareFrame, MiddlewareFraming, MiddlewareHeader, + MiddlewareResponse, + }, + error::{GatewayError, GatewayErrorKind}, +}; +use gateway_protocol::openai::sse::{DONE_SSE_FRAME, SseEvent, SseEventDecoder, sse_frame_is_done}; +use serde_json::Value; + +use crate::openai::{middleware::error_response, responses::ProtocolErrorBody}; + +use super::response::{ChatIdentity, ChatStreamEncoder, project_response}; + +pub(super) fn project_middleware_response( + response: MiddlewareResponse, + identity: ChatIdentity, + streaming: bool, + include_usage: bool, +) -> MiddlewareResponse { + if !(200..300).contains(&response.status_code()) { + return response; + } + let (protocol, status, mut headers, body, envelope) = response.into_parts(); + // 投影改变正文长度和表示形式 不继承 Responses 的实体长度或编码 + headers.retain(|header| { + !["content-length", "content-type", "content-encoding"] + .iter() + .any(|name| header.name().eq_ignore_ascii_case(name)) + }); + headers.push(MiddlewareHeader::new( + "content-type", + Bytes::from_static(if streaming { + b"text/event-stream" + } else { + b"application/json" + }), + )); + let projection = if streaming { + Projection::Stream(Box::new(StreamProjection { + encoder: ChatStreamEncoder::new(identity, include_usage), + decoder: SseEventDecoder::default(), + done_received: false, + })) + } else { + Projection::Json(identity) + }; + let mut response = MiddlewareResponse::new( + protocol, + status, + headers, + Box::new(ChatBody { + inner: Some(body), + projection, + pending: VecDeque::new(), + committed: false, + finished: false, + }), + ); + if let Some(envelope) = envelope { + response = response.with_envelope(envelope); + } + response +} + +pub(super) fn chat_error_frame(data: &Value) -> Bytes { + let error = data + .pointer("/response/error") + .or_else(|| data.get("error")); + Bytes::from(format!("data: {}\n\n", serde_json::json!({"error": error}))) +} + +enum Projection { + Json(ChatIdentity), + Stream(Box), +} + +impl Projection { + fn terminal_seen(&self) -> bool { + matches!(self, Self::Stream(stream) if stream.terminal_seen()) + } +} + +struct StreamProjection { + encoder: ChatStreamEncoder, + decoder: SseEventDecoder, + done_received: bool, +} + +impl StreamProjection { + fn terminal_seen(&self) -> bool { + self.encoder.is_completed() || self.encoder.has_wire_failure() + } + + fn project(&mut self, events: Vec) -> Result, MiddlewareError> { + let mut output = Vec::new(); + for event in events { + if self.done_received { + return Err(invalid_response()); + } + let data = serde_json::from_str(&event.data).map_err(|_| invalid_response())?; + // codec 区分辅助事件与终态后的业务输出 交付层不再维护一份事件分类 + output.extend( + self.encoder + .push_event(event.event.as_deref(), &data) + .map_err(projection_error)?, + ); + } + Ok(output) + } +} + +struct ChatBody { + inner: Option>, + projection: Projection, + pending: VecDeque, + committed: bool, + finished: bool, +} + +impl ChatBody { + async fn read_one(&mut self) -> Result<(), MiddlewareError> { + let inner = self.inner.as_mut().ok_or(MiddlewareError::InvalidState)?; + let Some(frame) = inner.next_frame().await? else { + if let Projection::Stream(stream) = &mut self.projection { + let events = stream.decoder.finish().map_err(|_| invalid_response())?; + let frames = stream.project(events)?; + if !frames.is_empty() || !stream.done_received || !inner.is_finalized() { + return Err(invalid_response()); + } + self.pending.push_back(done_frame()); + } + self.finished = true; + return Ok(()); + }; + let transformed = frame.transformed(); + let (bytes, framing, terminal, envelope) = frame.into_parts(); + let output = match &mut self.projection { + Projection::Json(identity) => { + if framing != MiddlewareFraming::JsonDocument || !terminal { + return Err(invalid_response()); + } + let value = serde_json::from_slice(&bytes).map_err(|_| invalid_response())?; + let value = project_response(&value, identity).map_err(projection_error)?; + vec![MiddlewareFrame::new( + Bytes::from(serde_json::to_vec(&value).map_err(|_| invalid_response())?), + MiddlewareFraming::JsonDocument, + true, + )] + } + Projection::Stream(stream) => { + if framing != MiddlewareFraming::SseEvent { + return Err(invalid_response()); + } + let events = stream + .decoder + .push(&bytes) + .map_err(|_| invalid_response())?; + let output = stream.project(events)?; + // 共用 SSE 解码器只返回业务事件 DONE 由受管正文的控制帧单独确认 + if std::str::from_utf8(&bytes).is_ok_and(sse_frame_is_done) { + if stream.done_received || !stream.terminal_seen() { + return Err(invalid_response()); + } + stream.done_received = true; + } + if terminal && !stream.done_received { + return Err(invalid_response()); + } + output + .into_iter() + .map(|bytes| MiddlewareFrame::new(bytes, MiddlewareFraming::SseEvent, false)) + .collect() + } + }; + let mut output: VecDeque<_> = output + .into_iter() + .map(|frame| frame.with_transformed(transformed)) + .collect(); + if let Some(envelope) = envelope + && let Some(last) = output.pop_back() + { + output.push_back(last.with_envelope(envelope)); + } + self.pending.extend(output); + Ok(()) + } + + async fn verify_terminal(&mut self) -> Result<(), MiddlewareError> { + while !self.finished { + self.read_one().await?; + } + Ok(()) + } + + async fn fail_stream(&mut self, error: MiddlewareError) { + self.pending.clear(); + if let Some(inner) = self.inner.take() { + inner.close().await; + } + let response = error_response(error); + let bytes = to_bytes(response.into_body(), usize::MAX) + .await + .unwrap_or_default(); + let value = serde_json::from_slice(&bytes).unwrap_or_else(|_| { + serde_json::json!({"error": { + "type": "server_error", "code": "invalid_upstream_response", + "message": "The gateway could not deliver the Chat response." + }}) + }); + self.pending.push_back(MiddlewareFrame::new( + chat_error_frame(&value), + MiddlewareFraming::SseEvent, + false, + )); + self.pending.push_back(done_frame()); + self.finished = true; + } +} + +impl MiddlewareBody for ChatBody { + fn next_frame(&mut self) -> BoxFuture<'_, Result, MiddlewareError>> { + Box::pin(async move { + while self.pending.is_empty() && !self.finished { + let result = match self.read_one().await { + Ok(()) if self.committed && self.projection.terminal_seen() => { + self.verify_terminal().await + } + result => result, + }; + if let Err(error) = result { + if !self.committed { + return Err(error); + } + self.fail_stream(error).await; + } + } + Ok(self.pending.pop_front()) + }) + } + + fn commit_downstream( + &mut self, + client_status_code: Option, + ) -> BoxFuture<'_, Result<(), MiddlewareError>> { + Box::pin(async move { + self.inner + .as_mut() + .ok_or(MiddlewareError::InvalidState)? + .commit_downstream(client_status_code) + .await?; + self.committed = true; + // HTTP 尚未发送首帧 即使首批已经包含 finish 或 usage 也先确认原会话完成结算 + if self.projection.terminal_seen() + && let Err(error) = self.verify_terminal().await + { + // 原会话已提交 不能把已记录的 200 改成另一 HTTP 状态 + // 首个 Chat 帧为 role 或 error 成功终态仍在 pending 内可安全撤回 + self.fail_stream(error).await; + } + Ok(()) + }) + } + + fn record_client_status(&mut self, status: u16) -> BoxFuture<'_, Result<(), MiddlewareError>> { + match self.inner.as_mut() { + Some(inner) => inner.record_client_status(status), + None => Box::pin(async { Ok(()) }), + } + } + + fn is_finalized(&self) -> bool { + self.inner.as_ref().is_none_or(|inner| inner.is_finalized()) + } + + fn close(mut self: Box) -> BoxFuture<'static, ()> { + let inner = self.inner.take(); + Box::pin(async move { + if let Some(inner) = inner { + inner.close().await; + } + }) + } +} + +fn projection_error(error: ProtocolErrorBody) -> MiddlewareError { + MiddlewareError::Gateway( + GatewayError::new( + GatewayErrorKind::UpstreamUnavailable, + "The gateway could not project the upstream response to Chat Completions.", + ) + .with_client_code(error.error.code), + ) +} + +fn invalid_response() -> MiddlewareError { + MiddlewareError::Gateway( + GatewayError::new( + GatewayErrorKind::UpstreamUnavailable, + "The gateway could not deliver the Chat response.", + ) + .with_client_code("invalid_upstream_response"), + ) +} + +fn done_frame() -> MiddlewareFrame { + MiddlewareFrame::new( + Bytes::from_static(DONE_SSE_FRAME.as_bytes()), + MiddlewareFraming::SseEvent, + true, + ) +} diff --git a/backend/crates/gateway-api/src/openai/chat_completions/http.rs b/backend/crates/gateway-api/src/openai/chat_completions/http.rs new file mode 100644 index 000000000..b90f529f5 --- /dev/null +++ b/backend/crates/gateway-api/src/openai/chat_completions/http.rs @@ -0,0 +1,100 @@ +//! Chat HTTP 入口只规范化协议 共用 Responses 的执行与交付链路 + +use std::{net::SocketAddr, time::SystemTime}; + +use axum::{ + body::Bytes, + extract::{Extension, State, connect_info::ConnectInfo}, + http::{HeaderMap, StatusCode, header::CONTENT_ENCODING}, + response::Response, +}; + +use crate::ApiState; +use crate::openai::{ + auth::{authenticate_client, client_access_error_response}, + error::{gateway_error_response, protocol_error_response}, + responses::{ + DecodedResponsesRequest, HttpResponseAdapter, RequestDecodeError, ResponsesHttpRequest, + decode_request_with_headers, decompress_request_body, execute_prepared_response, + }, + router::CHAT_COMPLETIONS_PATH, +}; + +use super::{ + delivery::{chat_error_frame, project_middleware_response}, + request::convert_request, + response::ChatIdentity, +}; + +pub(in crate::openai) async fn chat_completions( + State(state): State, + connect_info: Option>>, + ingress_id: Option>, + headers: HeaderMap, + body: Bytes, +) -> Response { + let service = state.openai().clone(); + let client = match authenticate_client(&service, &headers).await { + Ok(client) => client, + Err(error) => return client_access_error_response(error), + }; + let (decoded, middleware_body, include_usage) = match decode_request( + &body, + &headers, + client.snapshot().responses_max_decompressed_body_bytes(), + ) { + Ok(request) => request, + Err(error) => { + return protocol_error_response(StatusCode::BAD_REQUEST, error.protocol_body()); + } + }; + let prepared = match service.execution().prepare_execution(client).await { + Ok(prepared) => prepared, + Err(error) => return gateway_error_response(&error), + }; + let identity = ChatIdentity { + id: format!("chatcmpl-{}", prepared.request_id().as_str()), + created: SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap_or_default() + .as_secs(), + model: decoded.metadata().requested_model().to_owned(), + }; + let streaming = decoded.metadata().stream(); + let adapter = HttpResponseAdapter { + endpoint: CHAT_COMPLETIONS_PATH, + wrap: Box::new(move |response| { + project_middleware_response(response, identity, streaming, include_usage) + }), + stream_error: chat_error_frame, + }; + execute_prepared_response( + service, + prepared, + ResponsesHttpRequest { + peer_address: connect_info.map(|Extension(ConnectInfo(address))| address), + ingress_id, + headers, + decoded, + middleware_body, + }, + None, + Some(adapter), + ) + .await +} + +fn decode_request( + body: &[u8], + headers: &HeaderMap, + maximum_body_bytes: usize, +) -> Result<(DecodedResponsesRequest, Bytes, bool), RequestDecodeError> { + let body = decompress_request_body(body, headers, maximum_body_bytes)?; + let converted = convert_request(&body)?; + let body = + serde_json::to_vec(&converted.body).map_err(|_| RequestDecodeError::MalformedJson)?; + let mut decoded_headers = headers.clone(); + decoded_headers.remove(CONTENT_ENCODING); + let decoded = decode_request_with_headers(&body, &decoded_headers, maximum_body_bytes)?; + Ok((decoded, Bytes::from(body), converted.include_usage)) +} diff --git a/backend/crates/gateway-api/src/openai/chat_completions/mod.rs b/backend/crates/gateway-api/src/openai/chat_completions/mod.rs new file mode 100644 index 000000000..780380aa8 --- /dev/null +++ b/backend/crates/gateway-api/src/openai/chat_completions/mod.rs @@ -0,0 +1,8 @@ +//! Chat Completions 与 Responses 之间的 API 协议适配 + +mod delivery; +mod http; +mod request; +mod response; + +pub(super) use http::chat_completions; diff --git a/backend/crates/gateway-api/src/openai/chat_completions/request.rs b/backend/crates/gateway-api/src/openai/chat_completions/request.rs new file mode 100644 index 000000000..e0b69a373 --- /dev/null +++ b/backend/crates/gateway-api/src/openai/chat_completions/request.rs @@ -0,0 +1,550 @@ +//! 将 Chat 请求转换为 Responses 正文,协议选项留在 API,业务 JSON 与上游兼容规则保持原有归属 + +use serde_json::{Map, Value, json}; + +use crate::openai::responses::RequestDecodeError; + +pub(in crate::openai) struct ConvertedChatRequest { + pub(in crate::openai) body: Map, + pub(in crate::openai) include_usage: bool, +} + +pub(in crate::openai) fn convert_request( + body: &[u8], +) -> Result { + let value = serde_json::from_slice(body).map_err(|_| RequestDecodeError::MalformedJson)?; + let Value::Object(mut chat) = value else { + return Err(RequestDecodeError::ExpectedObject); + }; + validate_options(&chat)?; + let model = required_string(&mut chat, "model", "model", false)?; + let stream = optional_bool(&mut chat, "stream", "stream")?.unwrap_or(false); + let include_usage = stream_options(optional(&mut chat, "stream_options"), stream)?; + let input = messages(required(&mut chat, "messages", "messages")?)?; + let mut body = Map::from_iter([ + ("model".into(), model.into()), + ("input".into(), input.into()), + ("stream".into(), stream.into()), + ("store".into(), false.into()), + ]); + if let Some(value) = optional(&mut chat, "tools") { + body.insert("tools".into(), function_tools(value)?.into()); + } + if let Some(value) = optional(&mut chat, "tool_choice") { + let choice = tool_choice(value, body.get("tools"))?; + body.insert("tool_choice".into(), choice); + } + if let Some(value) = optional_bool(&mut chat, "parallel_tool_calls", "parallel_tool_calls")? { + body.insert("parallel_tool_calls".into(), value.into()); + } + + let mut text = Map::new(); + if let Some(value) = optional(&mut chat, "response_format") { + text.insert("format".into(), response_format(value)?); + } + if let Some(value) = optional(&mut chat, "verbosity") { + string(&value, "verbosity", false)?; + text.insert("verbosity".into(), value); + } + if !text.is_empty() { + body.insert("text".into(), text.into()); + } + if let Some(value) = optional(&mut chat, "reasoning_effort") { + string(&value, "reasoning_effort", false)?; + body.insert("reasoning".into(), json!({"effort": value})); + } + for field in ["max_tokens", "max_completion_tokens"] { + if let Some(value) = optional(&mut chat, field) { + if !value.as_u64().is_some_and(|tokens| tokens > 0) { + return Err(invalid(field)); + } + // 新字段优先,具体上游是否接受长度控制仍由现有 Provider 决定 + body.insert("max_output_tokens".into(), value); + } + } + for (field, maximum) in [("temperature", 2.0), ("top_p", 1.0)] { + if let Some(value) = optional(&mut chat, field) { + if !value + .as_f64() + .is_some_and(|number| (0.0..=maximum).contains(&number)) + { + return Err(invalid(field)); + } + body.insert(field.into(), value); + } + } + for field in [ + "user", + "safety_identifier", + "service_tier", + "prompt_cache_key", + "prompt_cache_retention", + ] { + if let Some(value) = optional(&mut chat, field) { + string(&value, field, true)?; + body.insert(field.into(), value); + } + } + for field in ["metadata", "prompt_cache_options", "moderation"] { + if let Some(value) = optional(&mut chat, field) { + let object = object(value, field)?; + body.insert(field.into(), object.into()); + } + } + // 未识别的客户端顶层扩展不等于 Responses 参数,不进入上游正文 + Ok(ConvertedChatRequest { + body, + include_usage, + }) +} + +fn validate_options(chat: &Map) -> Result<(), RequestDecodeError> { + for (field, value) in chat.iter().filter(|(_, value)| !value.is_null()) { + let supported = match field.as_str() { + "n" => value.as_u64() == Some(1), + "store" | "logprobs" => value == false, + "frequency_penalty" | "presence_penalty" => value.as_f64() == Some(0.0), + "top_logprobs" => value.as_u64() == Some(0), + "logit_bias" => value.as_object().is_some_and(Map::is_empty), + "functions" | "stop" => value.as_array().is_some_and(Vec::is_empty), + "modalities" => value == &json!(["text"]), + "audio" + | "prediction" + | "seed" + | "function_call" + | "web_search_options" + | "reasoning_content" + | "input" + | "previous_response_id" + | "conversation" + | "instructions" + | "reasoning" + | "text" + | "include" + | "background" + | "max_output_tokens" => false, + _ => true, + }; + if !supported { + return Err(unsupported(field)); + } + } + Ok(()) +} + +fn stream_options(value: Option, stream: bool) -> Result { + let Some(value) = value else { + return Ok(false); + }; + if !stream { + return Err(invalid("stream_options")); + } + let mut options = object(value, "stream_options")?; + let include_usage = optional_bool( + &mut options, + "include_usage", + "stream_options.include_usage", + )? + .unwrap_or(false); + if optional_bool( + &mut options, + "include_obfuscation", + "stream_options.include_obfuscation", + )? == Some(true) + { + // Chat 重编码会改变帧长度,不能声称继承上游流的长度混淆保证 + return Err(unsupported("stream_options.include_obfuscation")); + } + Ok(include_usage) +} + +fn messages(value: Value) -> Result, RequestDecodeError> { + let messages = array(value, "messages")?; + if messages.is_empty() { + return Err(RequestDecodeError::EmptyField { + field: "messages".into(), + }); + } + let mut input = Vec::with_capacity(messages.len()); + for (index, value) in messages.into_iter().enumerate() { + let path = format!("messages[{index}]"); + let mut message = object(value, &path)?; + let role = required_string(&mut message, "role", &format!("{path}.role"), false)?; + if !matches!( + role.as_str(), + "system" | "developer" | "user" | "assistant" | "tool" + ) { + return Err(unsupported(&format!("{path}.role"))); + } + if let Some(name) = optional(&mut message, "name") + && !string(&name, &format!("{path}.name"), true)?.is_empty() + { + return Err(unsupported(&format!("{path}.name"))); + } + for field in ["function_call", "reasoning_content", "reasoning", "audio"] { + if optional(&mut message, field).is_some() { + return Err(unsupported(&format!("{path}.{field}"))); + } + } + for (field, allowed_role) in [ + ("tool_calls", "assistant"), + ("refusal", "assistant"), + ("tool_call_id", "tool"), + ] { + if role != allowed_role && message.get(field).is_some_and(|value| !value.is_null()) { + return Err(invalid(&format!("{path}.{field}"))); + } + } + if role == "tool" { + let call_id = required_string( + &mut message, + "tool_call_id", + &format!("{path}.tool_call_id"), + false, + )?; + let content_path = format!("{path}.content"); + let content = required(&mut message, "content", &content_path)?; + let output = if content.is_string() { + content + } else { + content_parts(content, &role, &content_path, false)?.into() + }; + input.push( + json!({"type": "function_call_output", "call_id": call_id, "output": output}), + ); + continue; + } + + let calls = optional(&mut message, "tool_calls") + .map(|value| tool_calls(value, &format!("{path}.tool_calls"))) + .transpose()? + .unwrap_or_default(); + let refusal = optional(&mut message, "refusal"); + if let Some(value) = &refusal { + string(value, &format!("{path}.refusal"), true)?; + } + let content = optional(&mut message, "content"); + if content.is_none() && refusal.is_none() && calls.is_empty() { + return Err(invalid(&format!("{path}.content"))); + } + let is_assistant = role == "assistant"; + let mut parts = content + .map(|value| content_parts(value, &role, &format!("{path}.content"), is_assistant)) + .transpose()?; + if let Some(refusal) = refusal { + parts + .get_or_insert_with(Vec::new) + .push(json!({"type": "refusal", "refusal": refusal})); + } + if let Some(parts) = parts { + let mut item = json!({"type": "message", "role": role, "content": parts}); + if is_assistant { + // assistant 历史采用输出消息合同,普通文本和 refusal 都不能编码成 input_text + item["id"] = format!("msg_chat_{index}").into(); + item["status"] = "completed".into(); + } + input.push(item); + } + input.extend(calls); + } + Ok(input) +} + +fn content_parts( + value: Value, + role: &str, + path: &str, + is_assistant: bool, +) -> Result, RequestDecodeError> { + let parts = match value { + Value::String(text) => return Ok(vec![text_part(text, is_assistant)]), + Value::Array(parts) => parts, + _ => return Err(invalid_type(path, "a string or an array")), + }; + let mut result = Vec::with_capacity(parts.len()); + for (index, value) in parts.into_iter().enumerate() { + let path = format!("{path}[{index}]"); + let mut part = object(value, &path)?; + let kind = required_string(&mut part, "type", &format!("{path}.type"), false)?; + let mut converted = match kind.as_str() { + "text" => text_part( + required_string(&mut part, "text", &format!("{path}.text"), true)?, + is_assistant, + ), + "image_url" if role == "user" => image_part( + required(&mut part, "image_url", &format!("{path}.image_url"))?, + &format!("{path}.image_url"), + )?, + "refusal" if role == "assistant" => json!({ + "type": "refusal", + "refusal": required_string(&mut part, "refusal", &format!("{path}.refusal"), true)?, + }), + _ => return Err(unsupported(&format!("{path}.type"))), + }; + if let Some(value) = optional(&mut part, "prompt_cache_breakpoint") { + if is_assistant { + return Err(unsupported(&format!("{path}.prompt_cache_breakpoint"))); + } + converted["prompt_cache_breakpoint"] = + object(value, &format!("{path}.prompt_cache_breakpoint"))?.into(); + } + result.push(converted); + } + Ok(result) +} + +fn text_part(text: String, is_assistant: bool) -> Value { + if is_assistant { + json!({"type": "output_text", "text": text, "annotations": []}) + } else { + json!({"type": "input_text", "text": text}) + } +} + +fn image_part(value: Value, path: &str) -> Result { + let mut image = object(value, path)?; + let url = required_string(&mut image, "url", &format!("{path}.url"), false)?; + let valid = url::Url::parse(&url).is_ok_and(|parsed| { + matches!(parsed.scheme(), "http" | "https") && parsed.host_str().is_some() + || parsed.scheme() == "data" + && url.starts_with("data:image/") + && url + .split_once(";base64,") + .is_some_and(|(_, data)| !data.trim().is_empty()) + }); + if !valid { + return Err(invalid(&format!("{path}.url"))); + } + let mut converted = json!({"type": "input_image", "image_url": url}); + if let Some(detail) = optional(&mut image, "detail") { + string(&detail, &format!("{path}.detail"), false)?; + converted["detail"] = detail; + } + Ok(converted) +} + +fn tool_calls(value: Value, path: &str) -> Result, RequestDecodeError> { + let calls = array(value, path)?; + let mut result = Vec::with_capacity(calls.len()); + for (index, value) in calls.into_iter().enumerate() { + let path = format!("{path}[{index}]"); + let mut call = object(value, &path)?; + let id = required_string(&mut call, "id", &format!("{path}.id"), false)?; + if required_string(&mut call, "type", &format!("{path}.type"), false)? != "function" { + return Err(unsupported(&format!("{path}.type"))); + } + let function_path = format!("{path}.function"); + let mut function = object( + required(&mut call, "function", &function_path)?, + &function_path, + )?; + let name = required_string( + &mut function, + "name", + &format!("{function_path}.name"), + false, + )?; + let arguments = required_string( + &mut function, + "arguments", + &format!("{function_path}.arguments"), + true, + )?; + result.push( + json!({"type": "function_call", "call_id": id, "name": name, "arguments": arguments}), + ); + } + Ok(result) +} + +fn function_tools(value: Value) -> Result, RequestDecodeError> { + array(value, "tools")? + .into_iter() + .enumerate() + .map(|(index, value)| { + let path = format!("tools[{index}]"); + let mut tool = object(value, &path)?; + if required_string(&mut tool, "type", &format!("{path}.type"), false)? != "function" { + return Err(unsupported(&format!("{path}.type"))); + } + let function_path = format!("{path}.function"); + let mut function = object( + required(&mut tool, "function", &function_path)?, + &function_path, + )?; + let name = required_string( + &mut function, + "name", + &format!("{function_path}.name"), + false, + )?; + // Chat 的缺省是宽松函数 schema,不能让 Responses 的缺省收紧原始参数合同 + let strict = + optional_bool(&mut function, "strict", &format!("{function_path}.strict"))? + .unwrap_or(false); + let mut converted = json!({"type": "function", "name": name, "strict": strict}); + if let Some(description) = optional(&mut function, "description") { + string(&description, &format!("{function_path}.description"), true)?; + converted["description"] = description; + } + if let Some(parameters) = optional(&mut function, "parameters") { + converted["parameters"] = parameters; + } + Ok(converted) + }) + .collect() +} + +fn tool_choice(value: Value, tools: Option<&Value>) -> Result { + let tools = tools.and_then(Value::as_array); + if let Value::String(choice) = &value { + return match choice.as_str() { + "auto" | "none" => Ok(value), + "required" if tools.is_some_and(|tools| !tools.is_empty()) => Ok(value), + "required" => Err(invalid("tool_choice")), + _ => Err(unsupported("tool_choice")), + }; + } + let mut choice = object(value, "tool_choice")?; + if required_string(&mut choice, "type", "tool_choice.type", false)? != "function" { + return Err(unsupported("tool_choice.type")); + } + let mut function = object( + required(&mut choice, "function", "tool_choice.function")?, + "tool_choice.function", + )?; + let name = required_string(&mut function, "name", "tool_choice.function.name", false)?; + if !tools.is_some_and(|tools| { + tools + .iter() + .any(|tool| tool.get("name").and_then(Value::as_str) == Some(name.as_str())) + }) { + return Err(invalid("tool_choice.function.name")); + } + Ok(json!({"type": "function", "name": name})) +} + +fn response_format(value: Value) -> Result { + let mut format = object(value, "response_format")?; + let kind = required_string(&mut format, "type", "response_format.type", false)?; + match kind.as_str() { + "text" | "json_object" => Ok(json!({"type": kind})), + "json_schema" => { + let mut schema = object( + required(&mut format, "json_schema", "response_format.json_schema")?, + "response_format.json_schema", + )?; + let name = required_string( + &mut schema, + "name", + "response_format.json_schema.name", + false, + )?; + let definition = required(&mut schema, "schema", "response_format.json_schema.schema")?; + if definition.is_null() { + return Err(invalid("response_format.json_schema.schema")); + } + let mut converted = json!({"type": "json_schema", "name": name, "schema": definition}); + if let Some(strict) = + optional_bool(&mut schema, "strict", "response_format.json_schema.strict")? + { + converted["strict"] = strict.into(); + } + if let Some(description) = optional(&mut schema, "description") { + string( + &description, + "response_format.json_schema.description", + true, + )?; + converted["description"] = description; + } + Ok(converted) + } + _ => Err(unsupported("response_format.type")), + } +} + +fn optional(object: &mut Map, key: &str) -> Option { + object.remove(key).filter(|value| !value.is_null()) +} + +fn required( + object: &mut Map, + key: &str, + path: &str, +) -> Result { + object + .remove(key) + .ok_or_else(|| RequestDecodeError::MissingField { field: path.into() }) +} + +fn object(value: Value, path: &str) -> Result, RequestDecodeError> { + match value { + Value::Object(object) => Ok(object), + _ => Err(invalid_type(path, "an object")), + } +} + +fn array(value: Value, path: &str) -> Result, RequestDecodeError> { + match value { + Value::Array(array) => Ok(array), + _ => Err(invalid_type(path, "an array")), + } +} + +fn string<'a>( + value: &'a Value, + path: &str, + allow_empty: bool, +) -> Result<&'a str, RequestDecodeError> { + let value = value + .as_str() + .ok_or_else(|| invalid_type(path, "a string"))?; + if !allow_empty && value.trim().is_empty() { + return Err(RequestDecodeError::EmptyField { field: path.into() }); + } + Ok(value) +} + +fn required_string( + object: &mut Map, + key: &str, + path: &str, + allow_empty: bool, +) -> Result { + let value = required(object, key, path)?; + string(&value, path, allow_empty)?; + match value { + Value::String(value) => Ok(value), + _ => Err(invalid_type(path, "a string")), + } +} + +fn optional_bool( + object: &mut Map, + key: &str, + path: &str, +) -> Result, RequestDecodeError> { + optional(object, key) + .map(|value| { + value + .as_bool() + .ok_or_else(|| invalid_type(path, "a boolean")) + }) + .transpose() +} + +fn invalid_type(path: &str, expected: &'static str) -> RequestDecodeError { + RequestDecodeError::InvalidType { + field: path.into(), + expected, + } +} + +fn invalid(path: &str) -> RequestDecodeError { + RequestDecodeError::InvalidValue { field: path.into() } +} + +fn unsupported(path: &str) -> RequestDecodeError { + RequestDecodeError::UnsupportedField { field: path.into() } +} diff --git a/backend/crates/gateway-api/src/openai/chat_completions/response.rs b/backend/crates/gateway-api/src/openai/chat_completions/response.rs new file mode 100644 index 000000000..489823c39 --- /dev/null +++ b/backend/crates/gateway-api/src/openai/chat_completions/response.rs @@ -0,0 +1,664 @@ +//! 将实际 Responses wire 投影为 Chat JSON 和 SSE,不拥有执行、计费或终结生命周期 + +use std::collections::{BTreeMap, BTreeSet}; + +use bytes::Bytes; +use gateway_protocol::openai::events::{billable_usage_is_complete, extract_usage}; +use gateway_protocol::openai::sse::encode_sse_event; +use serde::Serialize; +use serde_json::{Value, json}; + +use crate::openai::responses::{ProtocolError, ProtocolErrorBody}; + +#[derive(Debug, Clone, Serialize)] +pub(in crate::openai) struct ChatIdentity { + pub id: String, + pub created: u64, + pub model: String, +} + +#[derive(Serialize)] +struct Envelope<'a> { + #[serde(flatten)] + identity: &'a ChatIdentity, + object: &'static str, + choices: Value, + #[serde(skip_serializing_if = "Option::is_none")] + usage: Option, + #[serde(skip_serializing_if = "Option::is_none")] + metadata: Option<&'a Value>, + #[serde(skip_serializing_if = "Option::is_none")] + service_tier: Option<&'a Value>, + #[serde(skip_serializing_if = "Option::is_none")] + moderation: Option, +} + +/// 输入已由现有 Responses collector 聚合,投影不再恢复或重建缺失的 output +pub(in crate::openai) fn project_response( + response: &Value, + identity: &ChatIdentity, +) -> Result { + let output = response + .get("output") + .and_then(Value::as_array) + .ok_or_else(invalid_response)?; + validate_output_identities(output)?; + let mut content = None; + let mut refusal = None; + let mut tools = Vec::new(); + for item in output { + match required_str(item, "type")? { + "message" => { + for part in message_parts(item)? { + let (kind, text) = part_text(part)?; + let target = match kind { + PartKind::Text => &mut content, + PartKind::Refusal => &mut refusal, + }; + target.get_or_insert_with(String::new).push_str(text); + } + } + "function_call" => { + let (id, name, arguments) = function_fields(item)?; + tools.push(json!({ + "id": id, "type": "function", + "function": {"name": name, "arguments": arguments} + })); + } + "reasoning" => {} + _ => return Err(invalid_response()), + } + } + let finish_reason = finish_reason(response, !tools.is_empty())?; + let mut message = json!({"role": "assistant", "content": content, "refusal": refusal}); + if !tools.is_empty() { + message["tool_calls"] = Value::Array(tools); + } + serde_json::to_value(Envelope { + identity, + object: "chat.completion", + choices: json!([{"index": 0, "message": message, "finish_reason": finish_reason, "logprobs": null}]), + usage: chat_usage(response), + metadata: response.get("metadata"), + service_tier: response.get("service_tier"), + moderation: project_moderation(response)?, + }) + .map_err(|_| invalid_response()) +} + +#[derive(Clone, Copy, PartialEq, Eq)] +enum PartKind { + Text, + Refusal, +} + +impl PartKind { + const fn field(self) -> &'static str { + match self { + Self::Text => "content", + Self::Refusal => "refusal", + } + } +} + +struct Part { + kind: PartKind, + text: String, +} + +struct Tool { + index: usize, + call_id: String, + name: String, + arguments: String, +} + +enum Output { + Message(BTreeMap), + Function(Tool), + Reasoning, +} + +struct OutputItem { + id: Option, + output: Output, +} + +#[derive(Clone, Copy, PartialEq, Eq)] +enum Terminal { + Completed, + Failed, +} + +/// 只保存身份关联与快照去重所需内容;终态帧由外层受管 body 决定何时交付 +pub(in crate::openai) struct ChatStreamEncoder { + identity: ChatIdentity, + include_usage: bool, + role_sent: bool, + output: BTreeMap, + terminal: Option, +} + +impl ChatStreamEncoder { + pub(in crate::openai) fn new(identity: ChatIdentity, include_usage: bool) -> Self { + Self { + identity, + include_usage, + role_sent: false, + output: BTreeMap::new(), + terminal: None, + } + } + + pub(in crate::openai) fn is_completed(&self) -> bool { + self.terminal == Some(Terminal::Completed) + } + + pub(in crate::openai) fn has_wire_failure(&self) -> bool { + self.terminal == Some(Terminal::Failed) + } + + pub(in crate::openai) fn push_event( + &mut self, + event_type: Option<&str>, + data: &Value, + ) -> Result, ProtocolErrorBody> { + let event_type = event_type.or_else(|| data.get("type").and_then(Value::as_str)); + if self.has_wire_failure() { + return Ok(Vec::new()); + } + let mut frames = Vec::new(); + match event_type { + Some("response.created") => { + self.require_open()?; + if !self.role_sent { + frames.push(self.delta(json!({"role": "assistant"}))?); + self.role_sent = true; + } + } + Some("response.output_text.delta" | "response.refusal.delta") => { + self.require_open()?; + let kind = if event_type == Some("response.refusal.delta") { + PartKind::Refusal + } else { + PartKind::Text + }; + let delta = required_str(data, "delta")?; + let part = self.event_part(data, kind)?; + part.text.push_str(delta); + if !delta.is_empty() { + frames.push(self.delta(json!({kind.field(): delta}))?); + } + } + Some("response.output_text.done" | "response.refusal.done") => { + self.require_open()?; + let (kind, field) = if event_type == Some("response.refusal.done") { + (PartKind::Refusal, "refusal") + } else { + (PartKind::Text, "text") + }; + let text = required_str(data, field)?; + let part = self.event_part(data, kind)?; + let suffix = append_snapshot(&mut part.text, text)?; + if !suffix.is_empty() { + frames.push(self.delta(json!({kind.field(): suffix}))?); + } + } + Some("response.content_part.added" | "response.content_part.done") => { + self.require_open()?; + let part = data.get("part").ok_or_else(invalid_response)?; + let (kind, text) = part_text(part)?; + let part = self.event_part(data, kind)?; + let suffix = append_snapshot(&mut part.text, text)?; + if !suffix.is_empty() { + frames.push(self.delta(json!({kind.field(): suffix}))?); + } + } + Some("response.output_item.added" | "response.output_item.done") => { + self.require_open()?; + self.item_snapshot( + required_index(data, "output_index")?, + data.get("item").ok_or_else(invalid_response)?, + false, + &mut frames, + )?; + } + Some( + "response.function_call_arguments.delta" | "response.function_call_arguments.done", + ) => { + self.require_open()?; + let index = required_index(data, "output_index")?; + let item = self.output.get_mut(&index).ok_or_else(invalid_response)?; + bind_id(item, optional_id(data, "item_id")?)?; + let Output::Function(tool) = &mut item.output else { + return Err(invalid_response()); + }; + let delta = if event_type == Some("response.function_call_arguments.delta") { + let delta = required_str(data, "delta")?; + tool.arguments.push_str(delta); + delta + } else { + append_snapshot(&mut tool.arguments, required_str(data, "arguments")?)? + }; + let tool_index = tool.index; + if !delta.is_empty() { + frames.push(self.delta(json!({"tool_calls": [{"index": tool_index, "function": {"arguments": delta}}]}))?); + } + } + Some("response.completed" | "response.incomplete") => { + self.require_open()?; + let response = data.get("response").ok_or_else(invalid_response)?; + let status = required_str(response, "status")?; + if (event_type == Some("response.completed") && status != "completed") + || (event_type == Some("response.incomplete") && status != "incomplete") + { + return Err(invalid_response()); + } + if let Some(output) = response.get("output") { + let output = output.as_array().ok_or_else(invalid_response)?; + validate_output_identities(output)?; + for (index, item) in output.iter().enumerate() { + self.item_snapshot(index as u64, item, true, &mut frames)?; + } + } else if self.output.is_empty() { + return Err(invalid_response()); + } + let finish_reason = finish_reason(response, self.tool_count() > 0)?; + frames.push(self.chunk( + json!([{"index": 0, "delta": {}, "finish_reason": finish_reason, "logprobs": null}]), + self.include_usage.then_some(Value::Null), + Some(response), + )?); + if self.include_usage + && let Some(usage) = chat_usage(response) + { + frames.push(self.chunk(json!([]), Some(usage), None)?); + } + self.terminal = Some(Terminal::Completed); + } + Some("response.failed" | "error") => { + self.require_open()?; + let error = data + .pointer("/response/error") + .or_else(|| data.get("error")) + .unwrap_or(data); + required_str(error, "message")?; + // 错误恢复信号和扩展由既有出口拥有,Chat 仅转换外层包装 + let mut error = error.clone(); + if error + .get("type") + .and_then(Value::as_str) + .is_none_or(|kind| kind == "error") + { + error["type"] = json!("server_error"); + } + frames.push(sse_frame(&json!({"error": error}))); + self.terminal = Some(Terminal::Failed); + } + // 辅助事件可随上游演进;实际 output item 和 content part 仍逐项检查可表达性 + _ => {} + } + // done-only 流也要提供完整 Chat 消息身份,辅助事件或私有推理不触发角色帧 + if !self.role_sent && !frames.is_empty() && !self.has_wire_failure() { + frames.insert(0, self.delta(json!({"role": "assistant"}))?); + self.role_sent = true; + } + Ok(frames) + } + + fn require_open(&self) -> Result<(), ProtocolErrorBody> { + if self.terminal.is_some() { + Err(invalid_response()) + } else { + Ok(()) + } + } + + fn tool_count(&self) -> usize { + self.output + .values() + .filter(|item| matches!(item.output, Output::Function(_))) + .count() + } + + fn event_part(&mut self, data: &Value, kind: PartKind) -> Result<&mut Part, ProtocolErrorBody> { + let index = required_index(data, "output_index")?; + let content_index = required_index(data, "content_index")?; + let id = optional_id(data, "item_id")?; + let item = self.output.entry(index).or_insert_with(|| OutputItem { + id: None, + output: Output::Message(BTreeMap::new()), + }); + bind_id(item, id)?; + message_part(item, content_index, kind) + } + + fn item_snapshot( + &mut self, + index: u64, + snapshot: &Value, + terminal: bool, + frames: &mut Vec, + ) -> Result<(), ProtocolErrorBody> { + let id = optional_id(snapshot, "id")?; + let kind = required_str(snapshot, "type")?; + let call_id = if kind == "function_call" { + Some(required_nonempty_str(snapshot, "call_id")?) + } else { + None + }; + let known_index = self.find_item(id, call_id)?; + let index = match known_index { + Some(known) if terminal || known == index => known, + Some(_) => return Err(invalid_response()), + None => index, + }; + if !self.output.contains_key(&index) { + let output = match kind { + "message" => Output::Message(BTreeMap::new()), + "reasoning" => Output::Reasoning, + "function_call" => { + let (call_id, name, _) = function_fields(snapshot)?; + let tool_index = self.tool_count(); + frames.push(self.delta(json!({"tool_calls": [{"index": tool_index, "id": call_id, "type": "function", "function": {"name": name, "arguments": ""}}]}))?); + Output::Function(Tool { + index: tool_index, + call_id: call_id.to_owned(), + name: name.to_owned(), + arguments: String::new(), + }) + } + _ => return Err(invalid_response()), + }; + self.output.insert(index, OutputItem { id: None, output }); + } + let item = self.output.get_mut(&index).ok_or_else(invalid_response)?; + bind_id(item, id)?; + match (&mut item.output, kind) { + (Output::Message(parts), "message") => { + for (index, part) in message_parts(snapshot)?.iter().enumerate() { + let (kind, text) = part_text(part)?; + let part = part_entry(parts, index as u64, kind)?; + let suffix = append_snapshot(&mut part.text, text)?; + if !suffix.is_empty() { + frames.push(delta_frame( + &self.identity, + self.include_usage, + json!({kind.field(): suffix}), + )?); + } + } + } + (Output::Function(tool), "function_call") => { + let (id, name, arguments) = function_fields(snapshot)?; + if tool.call_id != id || tool.name != name { + return Err(invalid_response()); + } + let suffix = append_snapshot(&mut tool.arguments, arguments)?; + if !suffix.is_empty() { + frames.push(delta_frame(&self.identity, self.include_usage, json!({"tool_calls": [{"index": tool.index, "function": {"arguments": suffix}}]}))?); + } + } + (Output::Reasoning, "reasoning") => {} + _ => return Err(invalid_response()), + } + Ok(()) + } + + fn find_item( + &self, + id: Option<&str>, + call_id: Option<&str>, + ) -> Result, ProtocolErrorBody> { + let mut found = None; + for (&index, item) in &self.output { + let same_id = id.is_some() && item.id.as_deref() == id; + let same_call = matches!(&item.output, Output::Function(tool) if call_id == Some(tool.call_id.as_str())); + if same_id || same_call { + if found.is_some_and(|previous| previous != index) { + return Err(invalid_response()); + } + found = Some(index); + } + } + Ok(found) + } + + fn delta(&self, delta: Value) -> Result { + delta_frame(&self.identity, self.include_usage, delta) + } + + fn chunk( + &self, + choices: Value, + usage: Option, + response: Option<&Value>, + ) -> Result { + chunk_frame(&self.identity, choices, usage, response) + } +} + +fn message_part( + item: &mut OutputItem, + index: u64, + kind: PartKind, +) -> Result<&mut Part, ProtocolErrorBody> { + let Output::Message(parts) = &mut item.output else { + return Err(invalid_response()); + }; + part_entry(parts, index, kind) +} + +fn part_entry( + parts: &mut BTreeMap, + index: u64, + kind: PartKind, +) -> Result<&mut Part, ProtocolErrorBody> { + let part = parts.entry(index).or_insert_with(|| Part { + kind, + text: String::new(), + }); + if part.kind != kind { + return Err(invalid_response()); + } + Ok(part) +} + +fn bind_id(item: &mut OutputItem, id: Option<&str>) -> Result<(), ProtocolErrorBody> { + if let Some(id) = id { + if item.id.as_deref().is_some_and(|previous| previous != id) { + return Err(invalid_response()); + } + if item.id.is_none() { + item.id = Some(id.to_owned()); + } + } + Ok(()) +} + +fn append_snapshot<'a>(text: &mut String, snapshot: &'a str) -> Result<&'a str, ProtocolErrorBody> { + let suffix = snapshot + .strip_prefix(text.as_str()) + .ok_or_else(invalid_response)?; + text.push_str(suffix); + Ok(suffix) +} + +fn message_parts(item: &Value) -> Result<&[Value], ProtocolErrorBody> { + if required_str(item, "role")? != "assistant" { + return Err(invalid_response()); + } + item.get("content") + .and_then(Value::as_array) + .map(Vec::as_slice) + .ok_or_else(invalid_response) +} + +fn part_text(part: &Value) -> Result<(PartKind, &str), ProtocolErrorBody> { + match required_str(part, "type")? { + "output_text" => Ok((PartKind::Text, required_str(part, "text")?)), + "refusal" => Ok((PartKind::Refusal, required_str(part, "refusal")?)), + _ => Err(invalid_response()), + } +} + +fn function_fields(item: &Value) -> Result<(&str, &str, &str), ProtocolErrorBody> { + Ok(( + required_nonempty_str(item, "call_id")?, + required_nonempty_str(item, "name")?, + required_str(item, "arguments")?, + )) +} + +fn validate_output_identities(output: &[Value]) -> Result<(), ProtocolErrorBody> { + let mut item_ids = BTreeSet::new(); + let mut call_ids = BTreeSet::new(); + for item in output { + if let Some(id) = optional_id(item, "id")? + && !item_ids.insert(id) + { + return Err(invalid_response()); + } + if item.get("type").and_then(Value::as_str) == Some("function_call") + && !call_ids.insert(required_nonempty_str(item, "call_id")?) + { + return Err(invalid_response()); + } + } + Ok(()) +} + +fn finish_reason(response: &Value, has_tools: bool) -> Result<&'static str, ProtocolErrorBody> { + if response.get("error").is_some_and(|value| !value.is_null()) { + return Err(invalid_response()); + } + match required_str(response, "status")? { + "completed" => Ok(if has_tools { "tool_calls" } else { "stop" }), + "incomplete" => match response + .pointer("/incomplete_details/reason") + .and_then(Value::as_str) + { + Some("max_output_tokens") => Ok("length"), + Some("content_filter") => Ok("content_filter"), + _ => Err(invalid_response()), + }, + _ => Err(invalid_response()), + } +} + +fn chat_usage(response: &Value) -> Option { + let usage = extract_usage(response)?; + if !billable_usage_is_complete(response, usage) { + return None; + } + let mut result = json!({"prompt_tokens": usage.input_tokens, "completion_tokens": usage.output_tokens, "total_tokens": usage.total_tokens}); + if response + .pointer("/usage/input_tokens_details/cached_tokens") + .is_some() + { + result["prompt_tokens_details"] = json!({"cached_tokens": usage.cached_input_tokens}); + } + if response + .pointer("/usage/output_tokens_details/reasoning_tokens") + .and_then(Value::as_u64) + .is_some_and(|tokens| tokens <= usage.output_tokens) + { + result["completion_tokens_details"] = + json!({"reasoning_tokens": usage.reasoning_output_tokens}); + } + Some(result) +} + +fn project_moderation(response: &Value) -> Result, ProtocolErrorBody> { + let Some(moderation) = response.get("moderation").filter(|value| !value.is_null()) else { + return Ok(None); + }; + let mut result = json!({}); + for side in ["input", "output"] { + let value = moderation.get(side).ok_or_else(invalid_response)?; + result[side] = match required_str(value, "type")? { + "moderation_result" => { + json!({"type": "moderation_results", "model": required_str(value, "model")?, "results": [value]}) + } + "error" => value.clone(), + _ => return Err(invalid_response()), + }; + } + Ok(Some(result)) +} + +fn delta_frame( + identity: &ChatIdentity, + include_usage: bool, + delta: Value, +) -> Result { + chunk_frame( + identity, + json!([{"index": 0, "delta": delta, "finish_reason": null, "logprobs": null}]), + include_usage.then_some(Value::Null), + None, + ) +} + +fn chunk_frame( + identity: &ChatIdentity, + choices: Value, + usage: Option, + response: Option<&Value>, +) -> Result { + let value = serde_json::to_value(Envelope { + identity, + object: "chat.completion.chunk", + choices, + usage, + metadata: None, + service_tier: response.and_then(|value| value.get("service_tier")), + moderation: response.map(project_moderation).transpose()?.flatten(), + }) + .map_err(|_| invalid_response())?; + Ok(sse_frame(&value)) +} + +fn sse_frame(value: &Value) -> Bytes { + Bytes::from(encode_sse_event("", &value.to_string())) +} + +fn required_str<'a>(value: &'a Value, field: &str) -> Result<&'a str, ProtocolErrorBody> { + value + .get(field) + .and_then(Value::as_str) + .ok_or_else(invalid_response) +} + +fn required_nonempty_str<'a>(value: &'a Value, field: &str) -> Result<&'a str, ProtocolErrorBody> { + let value = required_str(value, field)?; + if value.is_empty() { + return Err(invalid_response()); + } + Ok(value) +} + +fn optional_id<'a>(value: &'a Value, field: &str) -> Result, ProtocolErrorBody> { + value + .get(field) + .map(|_| required_nonempty_str(value, field)) + .transpose() +} + +fn required_index(value: &Value, field: &str) -> Result { + value + .get(field) + .and_then(Value::as_u64) + .ok_or_else(invalid_response) +} + +fn invalid_response() -> ProtocolErrorBody { + ProtocolErrorBody { + error: ProtocolError { + kind: "server_error", + code: "invalid_upstream_response", + message: "The upstream response cannot be represented as a Chat completion.".to_owned(), + param: None, + }, + } +} diff --git a/backend/crates/gateway-api/src/openai/mod.rs b/backend/crates/gateway-api/src/openai/mod.rs index d2245a92a..eca2f90a7 100644 --- a/backend/crates/gateway-api/src/openai/mod.rs +++ b/backend/crates/gateway-api/src/openai/mod.rs @@ -1,6 +1,7 @@ //! OpenAI 客户端协议 adapter pub mod auth; +mod chat_completions; mod endpoint; pub mod error; pub mod images; diff --git a/backend/crates/gateway-api/src/openai/responses/http.rs b/backend/crates/gateway-api/src/openai/responses/http.rs index f6e42034a..2619d199b 100644 --- a/backend/crates/gateway-api/src/openai/responses/http.rs +++ b/backend/crates/gateway-api/src/openai/responses/http.rs @@ -31,6 +31,7 @@ use gateway_core::event::{ProviderEvent, ProviderResponseHeader}; use gateway_core::lifecycle::{CancellationToken, ConnectionGuard}; use gateway_core::operation::OperationKind; use gateway_protocol::openai::sse::{DONE_SSE_FRAME, response_failed_sse_event_with_id}; +use serde_json::Value; use tokio::time::Instant; use crate::ApiState; @@ -120,6 +121,23 @@ pub(crate) async fn execute_prepared_responses( prepared: PreparedRootExecution, request: ResponsesHttpRequest, authorization: Option>, +) -> Response { + execute_prepared_response(service, prepared, request, authorization, None).await +} + +/// 下游协议只投影出口字节 执行会话和计量仍归原 owner +pub(in crate::openai) struct HttpResponseAdapter { + pub endpoint: &'static str, + pub wrap: Box MiddlewareResponse + Send>, + pub stream_error: fn(&Value) -> Bytes, +} + +pub(in crate::openai) async fn execute_prepared_response( + service: OpenAiService, + prepared: PreparedRootExecution, + request: ResponsesHttpRequest, + authorization: Option>, + adapter: Option, ) -> Response { let ResponsesHttpRequest { peer_address, @@ -148,8 +166,14 @@ pub(crate) async fn execute_prepared_responses( // 中间件正文已经由入口按上限解压;传输编码和旧长度不能与改写后的正文混用 middleware_headers.remove(CONTENT_ENCODING); middleware_headers.remove(CONTENT_LENGTH); + let endpoint = adapter + .as_ref() + .map_or(crate::openai::router::RESPONSES_PATH, |adapter| { + adapter.endpoint + }); + let stream_error = adapter.as_ref().map(|adapter| adapter.stream_error); let input = RequestInput { - endpoint: crate::openai::router::RESPONSES_PATH.to_owned(), + endpoint: endpoint.to_owned(), protocol: OPENAI_PROTOCOL.to_owned(), operation: Some(OperationKind::Generate), transport, @@ -194,12 +218,7 @@ pub(crate) async fn execute_prepared_responses( None }; let started = terminal_service - .start_prepared_response( - prepared, - decoded, - transport, - crate::openai::router::RESPONSES_PATH, - ) + .start_prepared_response(prepared, decoded, transport, endpoint) .await .map_err(MiddlewareError::Gateway)?; let _ = terminal_request_id.set(started.request_id.clone()); @@ -242,17 +261,18 @@ pub(crate) async fn execute_prepared_responses( } else { response }; - into_http_response( - validated_middleware_response(response, validation, stream), - expected, - ) - .await + let response = validated_middleware_response(response, validation, stream); + let response = match adapter { + Some(adapter) => (adapter.wrap)(response), + None => response, + }; + into_http_response(response, expected).await } Err(error) => error_response(error), } }; let response = if stream { - super::keepalive::response(response, keepalive).await + super::keepalive::response_with_encoder(response, keepalive, stream_error).await } else { response.await }; diff --git a/backend/crates/gateway-api/src/openai/responses/keepalive.rs b/backend/crates/gateway-api/src/openai/responses/keepalive.rs index 6a66d6140..7994c501b 100644 --- a/backend/crates/gateway-api/src/openai/responses/keepalive.rs +++ b/backend/crates/gateway-api/src/openai/responses/keepalive.rs @@ -21,6 +21,14 @@ use crate::openai::middleware::{SSE_KEEPALIVE_FRAME, SSE_KEEPALIVE_INTERVAL}; pub(super) async fn response( response: impl Future + Send + 'static, admitted: oneshot::Receiver, +) -> Response { + response_with_encoder(response, admitted, None).await +} + +pub(super) async fn response_with_encoder( + response: impl Future + Send + 'static, + admitted: oneshot::Receiver, + error_encoder: Option Bytes>, ) -> Response { let mut response = Box::pin(response); // 鉴权、模型门禁和准入失败仍保留原 HTTP 状态,不提前建立 SSE @@ -40,7 +48,7 @@ pub(super) async fn response( if !commit.record(Some(StatusCode::OK.as_u16())) { return response.await; } - let pending = PendingResponse::Waiting(response); + let pending = PendingResponse::Waiting(response, error_encoder); let stream = stream::once(async { Ok::<_, axum::Error>(SSE_KEEPALIVE_FRAME) }) .chain(stream::try_unfold(pending, PendingResponse::next)); let mut response = Response::new(Body::from_stream(stream)); @@ -58,7 +66,7 @@ pub(super) async fn response( } enum PendingResponse { - Waiting(BoxFuture<'static, Response>), + Waiting(BoxFuture<'static, Response>, Option Bytes>), Streaming(BodyDataStream), } @@ -66,7 +74,7 @@ impl PendingResponse { async fn next(mut self) -> Result, axum::Error> { loop { match &mut self { - Self::Waiting(response) => { + Self::Waiting(response, error_encoder) => { // 心跳只轮询同一个 future,不取消重建首帧等待或插件调用 let response = tokio::select! { biased; @@ -85,7 +93,7 @@ impl PendingResponse { let body = if sse { response.into_body() } else { - stream_failure(response).await + stream_failure(response, *error_encoder).await }; // 正式响应的正文已经带有交付期心跳,后续只转交,不叠加定时器 self = Self::Streaming(body.into_data_stream()); @@ -102,7 +110,7 @@ impl PendingResponse { } } -async fn stream_failure(response: Response) -> Body { +async fn stream_failure(response: Response, error_encoder: Option Bytes>) -> Body { let status = response.status(); let retry_after = response .headers() @@ -145,7 +153,10 @@ async fn stream_failure(response: Response) -> Body { .entry("retry-after") .or_insert(Value::String(retry_after)); } - let mut event = encode_sse_event("response.failed", &data.to_string()); - event.push_str(DONE_SSE_FRAME); + let mut event = match error_encoder { + Some(encode) => encode(&data).to_vec(), + None => encode_sse_event("response.failed", &data.to_string()).into_bytes(), + }; + event.extend_from_slice(DONE_SSE_FRAME.as_bytes()); Body::from(event) } diff --git a/backend/crates/gateway-api/src/openai/responses/mod.rs b/backend/crates/gateway-api/src/openai/responses/mod.rs index 2fabdea76..ff34dcaa8 100644 --- a/backend/crates/gateway-api/src/openai/responses/mod.rs +++ b/backend/crates/gateway-api/src/openai/responses/mod.rs @@ -10,10 +10,12 @@ pub mod websocket; pub(crate) use crate::request_context::request_client_context; pub use error::{ProtocolError, ProtocolErrorBody, RequestDecodeError, ResponseEncodeError}; +pub(in crate::openai) use http::{HttpResponseAdapter, execute_prepared_response}; pub(crate) use http::{ ResponseAuthorization, ResponsesHttpRequest, execute_prepared_responses, responses, }; pub use http::{collect_execution_response, stream_execution_response}; +pub(in crate::openai) use request::decompress_request_body; pub use request::{ ContinuationIntent, DecodedResponsesRequest, OpenAiRequestHeaders, ResponsesRequestMetadata, decode_request_with_headers, diff --git a/backend/crates/gateway-api/src/openai/responses/request.rs b/backend/crates/gateway-api/src/openai/responses/request.rs index c8241ba65..887f1dca5 100644 --- a/backend/crates/gateway-api/src/openai/responses/request.rs +++ b/backend/crates/gateway-api/src/openai/responses/request.rs @@ -371,7 +371,7 @@ const MAX_ZSTD_WINDOW_LOG: u32 = 26; /// 未压缩与 `identity` 借用原始正文;压缩正文在读取过程中限制展开大小, /// 避免先完整分配再检查 /// 只接受单一编码,重复头和叠加编码不能只解释第一项 -fn decompress_request_body<'a>( +pub(in crate::openai) fn decompress_request_body<'a>( body: &'a [u8], headers: &HeaderMap, max_decompressed_bytes: usize, diff --git a/backend/crates/gateway-api/src/openai/router.rs b/backend/crates/gateway-api/src/openai/router.rs index d753851ff..f7afe3a38 100644 --- a/backend/crates/gateway-api/src/openai/router.rs +++ b/backend/crates/gateway-api/src/openai/router.rs @@ -7,6 +7,7 @@ use axum::{ }; use super::{ + chat_completions::chat_completions, images::{image_edits, image_generations}, live, models::{model_detail, models}, @@ -19,6 +20,7 @@ use crate::ApiState; // 公开协议路径由 API 层统一定义;Core 使用操作类型,不根据这些路径分派业务 pub(super) const RESPONSES_PATH: &str = "/v1/responses"; +pub(super) const CHAT_COMPLETIONS_PATH: &str = "/v1/chat/completions"; pub(super) const IMAGE_GENERATIONS_PATH: &str = "/v1/images/generations"; pub(super) const IMAGE_EDITS_PATH: &str = "/v1/images/edits"; pub(super) const SEARCH_PATH: &str = "/v1/alpha/search"; @@ -33,6 +35,7 @@ pub(crate) fn router() -> Router { .route(IMAGE_EDITS_PATH, post(image_edits)) .route(SEARCH_PATH, post(standalone_search)) .route(RESPONSES_PATH, get(responses_websocket).post(responses)) + .route(CHAT_COMPLETIONS_PATH, post(chat_completions)) .route(MODELS_PATH, get(models)) // 官方 OpenAI 模型详情合同使用 path ID;它不属于 Admin API 约束 .route(MODEL_DETAIL_PATH, get(model_detail)) diff --git a/backend/crates/gateway-api/tests/openai/chat_completions/http.rs b/backend/crates/gateway-api/tests/openai/chat_completions/http.rs new file mode 100644 index 000000000..c6273c9ae --- /dev/null +++ b/backend/crates/gateway-api/tests/openai/chat_completions/http.rs @@ -0,0 +1,624 @@ +//! Chat HTTP 夹具和共享执行、鉴权、插件及终结边界 + +use std::{ + collections::VecDeque, + io::Write as _, + sync::{ + Arc, Mutex, + atomic::{AtomicBool, AtomicUsize, Ordering}, + }, + time::Duration, +}; + +use axum::{ + body::{Body, to_bytes}, + http::{Request, StatusCode}, + response::Response, +}; +use bytes::Bytes; +use futures::future::BoxFuture; +use gateway_core::{ + engine::{ + CommitRequirement, CoordinatedEvent, EngineError, ModelRequestId, + execution::{ + AuthenticatedClient, ClientAuthenticationError, ExecutionService, ExecutionSession, + PreparedRootExecution, ResponseCommit, StartExecution, StartProviderExecution, + StartedExecution, + }, + middleware::{ + FrozenMiddlewarePlan, MiddlewareContext, MiddlewareError, MiddlewareNext, + MiddlewarePlan, MiddlewareRequest, MiddlewareResponse, + }, + }, + error::GatewayError, + event::{ProviderEvent, ProviderResponseHeader}, + operation::Operation, + routing::{ + PublicModelId, + extensions::{ExtensionSetId, ExtensionSetLease, ExtensionSetReference}, + }, +}; +use serde_json::{Value, json}; +use tower::ServiceExt as _; + +use super::{chat_request, completed, created, message}; +use crate::openai::{ + api_router, authenticated_client, authenticated_client_for_provider_with_limit, + middleware::{RequestMiddleware, ResponseFrameAction}, +}; + +pub(super) const PATH: &str = "/v1/chat/completions"; +const KEY: &str = "sk_chat_fixture"; + +#[derive(Default)] +pub(super) struct Trace { + pub commits: AtomicUsize, + pub discarded: AtomicUsize, + pub cancelled: AtomicBool, + pub finalized: AtomicBool, + pub detached: AtomicUsize, + response_commit: ResponseCommit, +} + +#[derive(Default)] +pub(super) enum Finish { + #[default] + Success, + Error, + Pending, +} + +pub(super) struct Session { + events: VecDeque>, + pub trace: Arc, + pub finish: Finish, + pub delay: Option, + pub fail_commit: bool, + pub response_headers: Vec, + committed: bool, + buffered: bool, +} + +impl Session { + pub fn new(events: Vec) -> Self { + Self { + events: events.into_iter().map(|event| vec![event]).collect(), + trace: Arc::default(), + finish: Finish::Success, + delay: None, + fail_commit: false, + response_headers: Vec::new(), + committed: false, + buffered: false, + } + } + + pub fn with_batch(mut self, events: Vec) -> Self { + self.events.push_back(events); + self + } +} + +impl ExecutionSession for Session { + fn response_commit(&self) -> ResponseCommit { + self.trace.response_commit.clone() + } + + fn next_event(&mut self) -> BoxFuture<'_, Result, EngineError>> { + Box::pin(async move { + if let Some(delay) = self.delay.take() { + tokio::time::sleep(delay).await; + } + if let Some(event) = self.events.pop_front() { + let requirement = if self.committed { + CommitRequirement::AlreadyCommitted + } else { + CommitRequirement::CommitBeforeDelivery + }; + return CoordinatedEvent::try_batch(event, requirement).map(Some); + } + match self.finish { + Finish::Success => { + self.trace.finalized.store(true, Ordering::SeqCst); + Ok(None) + } + Finish::Error => { + self.trace.finalized.store(true, Ordering::SeqCst); + Err(EngineError::ProviderMetadataMismatch) + } + Finish::Pending => futures::future::pending().await, + } + }) + } + + fn collect_uncommitted(&mut self) -> BoxFuture<'_, Result, EngineError>> { + self.buffered = true; + Box::pin(async move { Ok(self.events.drain(..).flatten().collect()) }) + } + + fn response_headers(&self) -> &[ProviderResponseHeader] { + &self.response_headers + } + + fn discard_pending_delivery(&mut self) -> Result<(), EngineError> { + self.trace.discarded.fetch_add(1, Ordering::SeqCst); + Ok(()) + } + + fn commit_downstream(&mut self, status: Option) -> BoxFuture<'_, Result<(), EngineError>> { + Box::pin(async move { + self.trace.commits.fetch_add(1, Ordering::SeqCst); + if self.fail_commit { + return Err(EngineError::ProviderMetadataMismatch); + } + self.trace.response_commit.record(status); + self.committed = true; + if self.buffered { + self.trace.finalized.store(true, Ordering::SeqCst); + } + Ok(()) + }) + } + + fn record_client_status(&mut self, status: u16) -> BoxFuture<'_, Result<(), EngineError>> { + Box::pin(async move { + self.trace.response_commit.record(Some(status)); + Ok(()) + }) + } + + fn is_finalized(&self) -> bool { + self.trace.finalized.load(Ordering::SeqCst) + } + fn cancel(&self) { + self.trace.cancelled.store(true, Ordering::SeqCst); + } + fn detach_finalize(self: Box) -> BoxFuture<'static, ()> { + Box::pin(async move { + self.trace.detached.fetch_add(1, Ordering::SeqCst); + self.trace.finalized.store(true, Ordering::SeqCst); + }) + } +} + +#[derive(Clone, Debug)] +pub(super) struct Captured { + pub body: Value, + pub context: Value, + pub endpoint: String, +} + +pub(super) struct Execution { + pub captured: Mutex>, + pub client: AuthenticatedClient, + pub middleware: Option, + session: Mutex>, +} + +impl Execution { + pub fn new(session: Session) -> Self { + Self { + captured: Mutex::default(), + client: authenticated_client(KEY), + middleware: None, + session: Mutex::new(Some(session)), + } + } +} + +impl ExecutionService for Execution { + fn authenticate(&self, key: &str) -> Result { + if key == KEY { + Ok(self.client.clone()) + } else { + Err(ClientAuthenticationError::InvalidKey) + } + } + fn public_models(&self, _: &AuthenticatedClient) -> Vec { + Vec::new() + } + fn contains_public_model(&self, _: &AuthenticatedClient, _: &PublicModelId) -> bool { + true + } + fn middleware_plan(&self, _: &PreparedRootExecution) -> Option { + self.middleware.clone() + } + fn start( + &self, + request: StartExecution, + ) -> BoxFuture<'_, Result> { + Box::pin(async move { + let Operation::Generate(generation) = request.operation else { + panic!("Chat must generate") + }; + *self.captured.lock().unwrap() = Some(Captured { + body: Value::Object(generation.protocol_payload().body().clone()), + context: Value::Object(generation.protocol_payload().context().clone()), + endpoint: request.metadata.endpoint, + }); + Ok(StartedExecution { + request_id: ModelRequestId::new("req_chat_fixture").unwrap(), + created_at: std::time::SystemTime::now(), + stream: request.metadata.stream, + session: Box::new(self.session.lock().unwrap().take().expect("one request")), + }) + }) + } + fn start_provider_endpoint( + &self, + _: StartProviderExecution, + ) -> BoxFuture<'_, Result> { + Box::pin(async { panic!("Chat must use model execution") }) + } +} + +pub(super) struct Fixture { + pub router: axum::Router, + pub execution: Arc, + pub trace: Arc, +} + +impl Fixture { + pub async fn new(session: Session) -> Self { + Self::with_execution(Execution::new(session)).await + } + pub async fn with_execution(execution: Execution) -> Self { + let trace = execution + .session + .lock() + .unwrap() + .as_ref() + .unwrap() + .trace + .clone(); + let execution = Arc::new(execution); + Self { + router: api_router(execution.clone()).await, + execution, + trace, + } + } + pub async fn post(&self, body: Value) -> Response { + self.router.clone().oneshot(request(body)).await.unwrap() + } + pub fn captured(&self) -> Captured { + self.execution + .captured + .lock() + .unwrap() + .clone() + .expect("started request") + } +} + +pub(super) fn request(body: Value) -> Request { + Request::post(PATH) + .header("authorization", format!("Bearer {KEY}")) + .header("content-type", "application/json") + .body(Body::from(body.to_string())) + .unwrap() +} + +pub(super) async fn json_body(response: Response) -> Value { + serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap()).unwrap() +} + +#[derive(Debug, Default)] +pub(super) struct InspectMiddleware { + pub inner: RequestMiddleware, + pub requests: Mutex>, + pub update: Option, +} + +impl InspectMiddleware { + pub fn frozen(self: &Arc) -> FrozenMiddlewarePlan { + FrozenMiddlewarePlan::new( + self.clone(), + ExtensionSetReference::new( + ExtensionSetId::new("chat-inspector".to_owned()).unwrap(), + Arc::new(ReadyLease), + ), + ) + } +} + +struct ReadyLease; +impl ExtensionSetLease for ReadyLease { + fn is_ready(&self) -> bool { + true + } +} + +impl MiddlewarePlan for InspectMiddleware { + fn handle( + &self, + context: MiddlewareContext, + mut request: MiddlewareRequest, + next: MiddlewareNext, + ) -> BoxFuture<'static, Result> { + let mut body: Value = serde_json::from_slice(request.body()).unwrap(); + self.requests.lock().unwrap().push(body.clone()); + if let Some(update) = self.update { + update(&mut body); + let headers = request.headers().to_vec(); + request = request + .replace_parts( + "openai".to_owned(), + headers, + Bytes::from(body.to_string()), + None, + ) + .unwrap(); + } + self.inner.handle(context, request, next) + } +} + +#[tokio::test] +async fn rejects_missing_authentication_before_decoding_or_execution() { + let fixture = Fixture::new(Session::new(Vec::new())).await; + for authorization in [None, Some("Bearer sk_invalid")] { + let mut request = Request::post(PATH); + if let Some(authorization) = authorization { + request = request.header("authorization", authorization); + } + let response = fixture + .router + .clone() + .oneshot(request.body(Body::from("broken json")).unwrap()) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + } + assert!(fixture.execution.captured.lock().unwrap().is_none()); +} + +#[tokio::test] +async fn request_middleware_sees_responses_and_preserves_the_chat_endpoint() { + let middleware = Arc::new(InspectMiddleware { + inner: RequestMiddleware { + expected_operation: Some(gateway_core::operation::OperationKind::Generate), + ..RequestMiddleware::default() + }, + update: Some(|body| { + body["input"] = json!("rewritten by request middleware"); + }), + ..InspectMiddleware::default() + }); + let mut execution = Execution::new(Session::new(vec![completed(json!([message("ok")]))])); + execution.middleware = Some(middleware.frozen()); + let fixture = Fixture::with_execution(execution).await; + let response = fixture.post(chat_request(false)).await; + assert_eq!(response.status(), StatusCode::OK); + let downstream = json_body(response).await; + assert_eq!(downstream["object"], "chat.completion"); + let observed = middleware.requests.lock().unwrap(); + assert!(observed[0].get("messages").is_none()); + assert_eq!(observed[0]["input"][0]["role"], "user"); + assert_eq!( + middleware.inner.endpoints.lock().unwrap().as_slice(), + [PATH] + ); + let captured = fixture.captured(); + assert_eq!(captured.endpoint, PATH); + assert_eq!(captured.body["input"], "rewritten by request middleware"); + assert_eq!(fixture.trace.commits.load(Ordering::SeqCst), 1); +} + +#[tokio::test] +async fn dropping_chat_body_cancels_and_finalizes_the_original_execution() { + let mut session = Session::new(vec![created()]); + session.finish = Finish::Pending; + let fixture = Fixture::new(session).await; + let response = fixture.post(chat_request(true)).await; + assert_eq!(response.status(), StatusCode::OK); + drop(response); + for _ in 0..8 { + tokio::task::yield_now().await; + } + assert!(fixture.trace.cancelled.load(Ordering::SeqCst)); + assert_eq!(fixture.trace.detached.load(Ordering::SeqCst), 1); +} + +#[tokio::test] +async fn gzip_and_headers_reuse_the_responses_ingress_contract() { + let mut session = Session::new(vec![completed(json!([message("ok")]))]); + session.response_headers = vec![ + ProviderResponseHeader::new("x-future-multi", Bytes::from_static(b"first")), + ProviderResponseHeader::new("x-future-multi", Bytes::from_static(b"second")), + ProviderResponseHeader::new("connection", Bytes::from_static(b"x-hop-only")), + ProviderResponseHeader::new("x-hop-only", Bytes::from_static(b"removed")), + ]; + let fixture = Fixture::new(session).await; + let body = chat_request(false).to_string(); + let mut encoder = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::fast()); + encoder.write_all(body.as_bytes()).unwrap(); + let compressed = encoder.finish().unwrap(); + let request = Request::post(PATH) + .header("authorization", format!("Bearer {KEY}")) + .header("content-type", "application/json") + .header("content-encoding", "gzip") + .header("content-length", compressed.len()) + .header("session_id", "chat-session") + .header("x-future-input", "first") + .header("x-future-input", "second") + .body(Body::from(compressed)) + .unwrap(); + let response = fixture.router.clone().oneshot(request).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!( + response + .headers() + .get_all("x-future-multi") + .iter() + .map(|value| value.to_str().unwrap()) + .collect::>(), + ["first", "second"] + ); + assert!(!response.headers().contains_key("x-hop-only")); + assert!(!response.headers().contains_key("content-encoding")); + assert_eq!( + json_body(response).await["choices"][0]["message"]["content"], + "ok" + ); + let captured = fixture.captured(); + assert_eq!(captured.context["session_id"], "chat-session"); + let headers = captured.context["opaque_request_headers"] + .as_array() + .unwrap(); + assert_eq!( + headers + .iter() + .filter(|header| header[0] == "x-future-input") + .count(), + 2 + ); + assert!(!headers.iter().any(|header| matches!( + header[0].as_str(), + Some("content-encoding" | "content-length" | "authorization") + ))); +} + +#[tokio::test] +async fn corrupt_or_oversized_compressed_chat_never_starts_execution() { + let mut encoder = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::fast()); + encoder + .write_all(chat_request(false).to_string().as_bytes()) + .unwrap(); + let valid_compressed = encoder.finish().unwrap(); + for bytes in [b"broken gzip".to_vec(), valid_compressed] { + let mut execution = Execution::new(Session::new(Vec::new())); + execution.client = authenticated_client_for_provider_with_limit(KEY, "openai", 16); + let fixture = Fixture::with_execution(execution).await; + let request = Request::post(PATH) + .header("authorization", format!("Bearer {KEY}")) + .header("content-encoding", "gzip") + .body(Body::from(bytes)) + .unwrap(); + let response = fixture.router.clone().oneshot(request).await.unwrap(); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + assert!(fixture.execution.captured.lock().unwrap().is_none()); + } +} + +#[tokio::test] +async fn response_middleware_is_validated_as_responses_before_chat_projection() { + for matching_identity in [true, false] { + let replacement = json!({"id":if matching_identity {"resp_chat_test"} else {"resp_other"}, + "status":"completed","output":[message("plugin text")]}); + let middleware = Arc::new( + RequestMiddleware { + expected_operation: Some(gateway_core::operation::OperationKind::Generate), + ..RequestMiddleware::default() + } + .with_response_actions(vec![ResponseFrameAction::Replace(Bytes::from( + replacement.to_string(), + ))]), + ); + let mut execution = + Execution::new(Session::new(vec![completed(json!([message("original")]))])); + execution.middleware = Some(middleware.frozen()); + let fixture = Fixture::with_execution(execution).await; + let response = fixture.post(chat_request(false)).await; + if matching_identity { + assert_eq!(response.status(), StatusCode::OK); + assert_eq!( + json_body(response).await["choices"][0]["message"]["content"], + "plugin text" + ); + } else { + assert_eq!(response.status(), StatusCode::BAD_GATEWAY); + assert_eq!( + json_body(response).await["error"]["code"], + "middleware_failed" + ); + assert_eq!(fixture.trace.commits.load(Ordering::SeqCst), 0); + } + } +} + +#[tokio::test] +async fn commit_failure_does_not_deliver_buffered_or_first_terminal_chat_success() { + for streaming in [false, true] { + let mut session = Session::new(vec![completed(json!([message("not delivered")]))]); + session.fail_commit = true; + let fixture = Fixture::new(session).await; + let response = fixture.post(chat_request(streaming)).await; + assert!(response.status().is_server_error()); + let body = json_body(response).await; + assert!(body.get("error").is_some()); + assert!(body.get("choices").is_none()); + assert!(fixture.trace.cancelled.load(Ordering::SeqCst)); + } +} + +#[tokio::test] +async fn the_same_execution_fixture_finishes_native_responses_with_one_commit() { + for already_started in [false, true] { + let mut events = Vec::new(); + if already_started { + events.push(created()); + } + events.push(completed(json!([message("native baseline")]))); + let fixture = Fixture::new(Session::new(events)).await; + let mut request = request(json!({"model":"model-a","input":"hello","stream":true})); + *request.uri_mut() = "/v1/responses".parse().unwrap(); + let response = fixture.router.clone().oneshot(request).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let text = String::from_utf8(bytes.to_vec()).unwrap(); + assert!(text.contains("event: response.completed"), "{text}"); + assert!(text.contains("native baseline"), "{text}"); + assert!(!text.contains("response.failed"), "{text}"); + assert_eq!(text.matches("data: [DONE]").count(), 1, "{text}"); + assert!(text.ends_with("data: [DONE]\n\n"), "{text}"); + assert!(fixture.trace.finalized.load(Ordering::SeqCst)); + assert_eq!(fixture.trace.commits.load(Ordering::SeqCst), 1); + assert_eq!(fixture.trace.response_commit.status_code(), Some(200)); + } +} + +#[tokio::test] +async fn finalization_failure_never_leaks_success_finish_or_usage() { + for already_started in [false, true] { + let mut events = Vec::new(); + if already_started { + events.push(created()); + } + events.push(completed(json!([]))); + let mut session = Session::new(events); + session.finish = Finish::Error; + let fixture = Fixture::new(session).await; + let mut body = chat_request(true); + body["stream_options"] = json!({"include_usage":true}); + let response = fixture.post(body).await; + // 原会话已经接受 200 提交,后续终结错误只能作为流内错误交付 + assert_eq!(response.status(), StatusCode::OK); + let (_, chunks) = super::response::sse_body(response).await; + assert!(chunks.iter().any(|chunk| chunk.get("error").is_some())); + assert!( + !chunks + .iter() + .any(|chunk| chunk["choices"][0]["finish_reason"].is_string()) + ); + assert!(!chunks.iter().any(|chunk| chunk["choices"] == json!([]))); + assert!(fixture.trace.finalized.load(Ordering::SeqCst)); + assert_eq!(fixture.trace.commits.load(Ordering::SeqCst), 1); + assert_eq!(fixture.trace.response_commit.status_code(), Some(200)); + } +} + +#[tokio::test(start_paused = true)] +async fn error_after_admission_keepalive_uses_chat_error_encoding() { + let mut session = Session::new(Vec::new()); + session.delay = Some(Duration::from_secs(31)); + session.finish = Finish::Error; + let fixture = Fixture::new(session).await; + let response = fixture.post(chat_request(true)).await; + let (body, chunks) = super::response::sse_body(response).await; + assert!(body.starts_with(": keep-alive\n\n")); + assert!(chunks.iter().any(|chunk| chunk.get("error").is_some())); + assert!(!body.contains("response.failed")); + assert_eq!(fixture.trace.commits.load(Ordering::SeqCst), 0); + assert_eq!(fixture.trace.response_commit.status_code(), Some(200)); +} diff --git a/backend/crates/gateway-api/tests/openai/chat_completions/mod.rs b/backend/crates/gateway-api/tests/openai/chat_completions/mod.rs new file mode 100644 index 000000000..49f3a632a --- /dev/null +++ b/backend/crates/gateway-api/tests/openai/chat_completions/mod.rs @@ -0,0 +1,45 @@ +//! Chat Completions 的请求转换、输出投影与 HTTP 生命周期测试 + +mod http; +mod request; +mod response; + +use gateway_core::event::{ProtocolWireEvent, ProviderEvent}; +use serde_json::{Value, json}; + +fn chat_request(stream: bool) -> Value { + json!({"model":"model-a","messages":[{"role":"user","content":"hello"}],"stream":stream}) +} + +fn wire(kind: &str, mut value: Value) -> ProviderEvent { + value["type"] = json!(kind); + ProviderEvent::wire(ProtocolWireEvent::json("openai", Some(kind.to_owned()), value).unwrap()) +} + +fn completed(output: Value) -> ProviderEvent { + wire( + "response.completed", + json!({"response":{ + "id":"resp_chat_test","object":"response","model":"upstream-model", + "status":"completed","output":output, + "usage":{"input_tokens":11,"output_tokens":7,"total_tokens":18, + "input_tokens_details":{"cached_tokens":3}, + "output_tokens_details":{"reasoning_tokens":2}} + }}), + ) +} + +fn message(text: &str) -> Value { + json!({"id":"msg_chat_test","type":"message","role":"assistant","status":"completed", + "content":[{"type":"output_text","text":text,"annotations":[]}]}) +} + +fn created() -> ProviderEvent { + wire( + "response.created", + json!({"response":{ + "id":"resp_chat_test","object":"response","model":"upstream-model", + "status":"in_progress","output":[] + }}), + ) +} diff --git a/backend/crates/gateway-api/tests/openai/chat_completions/request.rs b/backend/crates/gateway-api/tests/openai/chat_completions/request.rs new file mode 100644 index 000000000..47bcf89cd --- /dev/null +++ b/backend/crates/gateway-api/tests/openai/chat_completions/request.rs @@ -0,0 +1,243 @@ +//! Chat 请求映射保留消息、工具和不透明业务数据的语义 + +use axum::http::StatusCode; +use serde_json::{Value, json}; + +use super::{ + chat_request, completed, + http::{Fixture, Session, json_body}, + message, +}; + +async fn convert(body: Value) -> Value { + let fixture = Fixture::new(Session::new(vec![completed(json!([message("ok")]))])).await; + let response = fixture.post(body).await; + let status = response.status(); + let result = json_body(response).await; + assert_eq!(status, StatusCode::OK, "{result}"); + fixture.captured().body +} + +#[tokio::test] +async fn preserves_ordered_roles_images_and_modern_tool_round_trip() { + let arguments = "{ \"city\" : \"北京\", \"store\":true }"; + let converted = convert(json!({"model":"model-a","messages":[ + {"role":"system","content":"system text"}, + {"role":"developer","content":"developer text"}, + {"role":"user","content":[{"type":"text","text":"describe"}, + {"type":"image_url","image_url":{"url":"data:image/png;base64,aGVsbG8=","detail":"high"}}]}, + {"role":"assistant","content":"checking","tool_calls":[ + {"id":"call-weather","type":"function","function":{"name":"weather","arguments":arguments}}]}, + {"role":"tool","tool_call_id":"call-weather","content":"{\"temperature\":25}"}, + {"role":"user","content":"continue"} + ]})).await; + let input = converted["input"].as_array().unwrap(); + assert_eq!( + input + .iter() + .filter_map(|item| item["role"].as_str()) + .collect::>(), + ["system", "developer", "user", "assistant", "user"] + ); + assert_eq!( + input[2]["content"][1], + json!({"type":"input_image","image_url":"data:image/png;base64,aGVsbG8=","detail":"high"}) + ); + assert_eq!( + input[3]["content"], + json!([{"type":"output_text","text":"checking","annotations":[]}]) + ); + assert_eq!( + input[4], + json!({"type":"function_call","call_id":"call-weather","name":"weather","arguments":arguments}) + ); + assert_eq!( + input[5], + json!({"type":"function_call_output","call_id":"call-weather","output":"{\"temperature\":25}"}) + ); + assert!(converted.get("previous_response_id").is_none()); +} + +#[tokio::test] +async fn assistant_history_uses_output_content_for_text_and_refusals() { + for content in [ + json!("previous answer"), + json!([{"type":"text","text":"previous answer"}]), + json!([{"type":"text","text":"previous answer"},{"type":"refusal","refusal":"declined"}]), + ] { + let converted = convert(json!({"model":"model-a","messages":[ + {"role":"user","content":"first turn"}, + {"role":"assistant","content":content}, + {"role":"user","content":"continue from the previous answer"} + ]})) + .await; + let history = &converted["input"][1]; + assert_eq!(history["role"], "assistant"); + assert_eq!(history["status"], "completed"); + assert!(!history["id"].as_str().unwrap().is_empty()); + assert_eq!( + history["content"][0], + json!({"type":"output_text","text":"previous answer","annotations":[]}) + ); + if content.is_array() && content.as_array().unwrap().len() == 2 { + assert_eq!( + history["content"][1], + json!({"type":"refusal","refusal":"declined"}) + ); + } + assert_eq!(converted["input"][2]["content"][0]["type"], "input_text"); + } +} + +#[tokio::test] +async fn preserves_nested_schema_metadata_and_future_opaque_values() { + let schema = json!({"type":"object","properties":{ + "messages":{"type":"string"},"store":{"type":"boolean"}, + "function_call":{"type":"string"},"future":{"x-extension":[1,2]}}, + "required":["messages"],"additionalProperties":true}); + let mut body = chat_request(false); + body["model"] = json!("future-model-opaque:β"); + body["tools"] = json!([{"type":"function","function":{"name":"inspect","parameters":schema}}]); + body["tool_choice"] = json!({"type":"function","function":{"name":"inspect"}}); + body["response_format"] = json!({"type":"json_schema","json_schema":{ + "name":"result","description":"preserve","strict":true,"schema":schema}}); + body["metadata"] = json!({"messages":"business data","future":"kept"}); + body["reasoning_effort"] = json!("future-effort"); + body["verbosity"] = json!("future-verbosity"); + body["service_tier"] = json!("future-tier"); + body["future_client_extension"] = json!({"opaque":true}); + let converted = convert(body).await; + assert_eq!(converted["model"], "future-model-opaque:β"); + assert_eq!(converted["tools"][0]["parameters"], schema); + assert_eq!(converted["tools"][0]["strict"], false); + assert_eq!( + converted["tool_choice"], + json!({"type":"function","name":"inspect"}) + ); + assert_eq!(converted["text"]["format"]["schema"], schema); + assert_eq!(converted["text"]["format"]["strict"], true); + assert_eq!( + converted["metadata"], + json!({"messages":"business data","future":"kept"}) + ); + assert_eq!(converted["reasoning"]["effort"], "future-effort"); + assert_eq!(converted["text"]["verbosity"], "future-verbosity"); + assert_eq!(converted["service_tier"], "future-tier"); + assert!(converted.get("future_client_extension").is_none()); +} + +#[tokio::test] +async fn normalizes_neutral_defaults_without_dropping_shared_provider_parameters() { + let mut body = chat_request(false); + for name in [ + "stream", + "store", + "n", + "functions", + "function_call", + "logprobs", + "seed", + ] { + body[name] = Value::Null; + } + body["frequency_penalty"] = json!(0); + body["presence_penalty"] = json!(0); + body["stop"] = json!([]); + body["max_tokens"] = json!(12); + body["max_completion_tokens"] = json!(24); + body["temperature"] = json!(0.2); + body["top_p"] = json!(0.9); + body["prompt_cache_retention"] = json!("24h"); + let converted = convert(body).await; + assert_eq!(converted["stream"], false); + assert_eq!(converted["store"], false); + assert_eq!(converted["max_output_tokens"], 24); + assert_eq!(converted["temperature"], 0.2); + assert_eq!(converted["top_p"], 0.9); + assert_eq!(converted["prompt_cache_retention"], "24h"); +} + +#[tokio::test] +async fn refuses_known_unrepresentable_features_before_starting_execution() { + for (name, value) in [ + ("n", json!(2)), + ("store", json!(true)), + ("stop", json!(["stop"])), + ("seed", json!(7)), + ("frequency_penalty", json!(0.3)), + ("logprobs", json!(true)), + ("functions", json!([{"name":"old"}])), + ("function_call", json!("auto")), + ("input", json!("not chat")), + ("previous_response_id", json!("resp_previous")), + ("audio", json!({"voice":"alloy","format":"wav"})), + ( + "prediction", + json!({"type":"content","content":"predicted"}), + ), + ] { + let fixture = Fixture::new(Session::new(Vec::new())).await; + let mut body = chat_request(false); + body[name] = value; + let response = fixture.post(body).await; + assert_eq!(response.status(), StatusCode::BAD_REQUEST, "{name}"); + let body = json_body(response).await; + assert_eq!(body["error"]["param"], name, "{body}"); + assert!( + fixture.execution.captured.lock().unwrap().is_none(), + "{name}" + ); + } +} + +#[tokio::test] +async fn rejects_unmapped_message_content_and_identity_fields() { + for message in [ + json!({"role":"user","name":"speaker","content":"hello"}), + json!({"role":"user","content":[{"type":"file","file":{"file_id":"file-example"}}]}), + json!({"role":"assistant","content":"hello","reasoning_content":"private"}), + json!({"role":"assistant","content":[{"type":"text","text":"history","prompt_cache_breakpoint":{"type":"default"}}]}), + json!({"role":"tool","content":"unassociated output"}), + ] { + let fixture = Fixture::new(Session::new(Vec::new())).await; + let response = fixture + .post(json!({"model":"model-a","messages":[message]})) + .await; + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + let body = json_body(response).await; + assert!( + body["error"]["param"] + .as_str() + .unwrap() + .starts_with("messages") + ); + assert!(fixture.execution.captured.lock().unwrap().is_none()); + } +} + +#[tokio::test] +async fn assistant_tool_history_keeps_arguments_opaque_and_cache_annotations_structural() { + let cache = json!({"type":"default","future_setting":{"keep":true}}); + let arguments = "not valid JSON {\n \"raw\" : 🦀"; + let converted = convert(json!({"model":"model-a","messages":[ + {"role":"user","content":[{"type":"text","text":"cached","prompt_cache_breakpoint":cache}]}, + {"role":"assistant","content":null,"tool_calls":[{"id":"call-raw","type":"function", + "function":{"name":"raw","arguments":arguments}}]}, + {"role":"tool","tool_call_id":"call-raw","content":[{"type":"text","text":"result"}]} + ],"prompt_cache_options":{"future":{"keep":true}},"moderation":{"future":{"keep":true}}})) + .await; + assert_eq!( + converted["input"][0]["content"][0]["prompt_cache_breakpoint"], + cache + ); + assert_eq!(converted["input"][1]["arguments"], arguments); + assert_eq!( + converted["input"][2]["output"], + json!([{"type":"input_text","text":"result"}]) + ); + assert_eq!( + converted["prompt_cache_options"], + json!({"future":{"keep":true}}) + ); + assert_eq!(converted["moderation"], json!({"future":{"keep":true}})); +} diff --git a/backend/crates/gateway-api/tests/openai/chat_completions/response.rs b/backend/crates/gateway-api/tests/openai/chat_completions/response.rs new file mode 100644 index 000000000..86bcfe82c --- /dev/null +++ b/backend/crates/gateway-api/tests/openai/chat_completions/response.rs @@ -0,0 +1,596 @@ +//! Chat JSON 与 SSE 的真实 wire 投影、去重和协议扩展边界 + +use axum::{body::to_bytes, http::StatusCode, response::Response}; +use serde_json::{Value, json}; +use std::collections::BTreeMap; + +use super::{ + chat_request, completed, created, + http::{Fixture, Session, json_body}, + message, wire, +}; + +pub(super) async fn sse_body(response: Response) -> (String, Vec) { + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(response.headers()["content-type"], "text/event-stream"); + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let text = String::from_utf8(bytes.to_vec()).unwrap(); + assert_eq!(text.matches("data: [DONE]").count(), 1, "{text}"); + assert!(text.ends_with("data: [DONE]\n\n"), "{text}"); + assert!(!text.contains("event: response."), "{text}"); + let chunks = text + .lines() + .filter_map(|line| line.strip_prefix("data: ")) + .filter(|data| *data != "[DONE]") + .map(|data| serde_json::from_str(data).unwrap()) + .collect(); + (text, chunks) +} + +fn combined(chunks: &[Value], name: &str) -> String { + chunks + .iter() + .filter_map(|value| value["choices"][0]["delta"][name].as_str()) + .collect() +} + +fn tool(id: &str, call_id: &str, name: &str, arguments: &str) -> Value { + json!({"type":"function_call","id":id,"call_id":call_id,"name":name,"arguments":arguments,"status":"completed"}) +} + +#[tokio::test] +async fn json_projects_public_identity_text_and_actual_usage() { + let fixture = Fixture::new(Session::new(vec![completed(json!([message("你好 🌍")]))])).await; + let response = fixture.post(chat_request(false)).await; + assert_eq!(response.status(), StatusCode::OK); + assert_eq!( + response.headers()["x-gateway-request-id"], + "req_chat_fixture" + ); + let body = json_body(response).await; + assert_eq!(body["object"], "chat.completion"); + assert!(body["id"].as_str().unwrap().starts_with("chatcmpl-")); + assert!(body["created"].as_u64().is_some()); + assert_eq!(body["model"], "model-a"); + assert_eq!(body["choices"][0]["message"]["content"], "你好 🌍"); + assert_eq!(body["choices"][0]["finish_reason"], "stop"); + assert_eq!( + body["usage"], + json!({"prompt_tokens":11,"completion_tokens":7,"total_tokens":18, + "prompt_tokens_details":{"cached_tokens":3},"completion_tokens_details":{"reasoning_tokens":2}}) + ); +} + +#[tokio::test] +async fn json_tool_arguments_and_refusal_do_not_become_assistant_text() { + for output in [ + json!([tool("fc1", "call1", "weather", "{ \"city\" : \"北京\" }")]), + json!([{"type":"reasoning","id":"rs1","summary":[{"type":"summary_text","text":"private"}]}, + {"type":"message","id":"refusal1","role":"assistant","status":"completed", + "content":[{"type":"refusal","refusal":"cannot help"}]}]), + ] { + let fixture = Fixture::new(Session::new(vec![completed(output.clone())])).await; + let response = fixture.post(chat_request(false)).await; + assert_eq!(response.status(), StatusCode::OK); + let body = json_body(response).await; + let choice = &body["choices"][0]; + assert_eq!(choice["message"]["content"], Value::Null); + assert!(!body.to_string().contains("private")); + if output[0]["type"] == "function_call" { + assert_eq!(choice["finish_reason"], "tool_calls"); + assert_eq!( + choice["message"]["tool_calls"][0]["function"]["arguments"], + "{ \"city\" : \"北京\" }" + ); + assert_eq!(choice["message"]["tool_calls"][0]["id"], "call1"); + } else { + assert_eq!(choice["message"]["refusal"], "cannot help"); + } + } +} + +#[tokio::test] +async fn sse_deduplicates_text_snapshots_and_emits_usage_only_when_requested() { + for include_usage in [false, true] { + let fixture = Fixture::new(Session::new(vec![created(), + wire("response.output_text.delta",json!({"item_id":"msg_chat_test","output_index":0,"content_index":0,"delta":"你好 "})), + wire("response.output_text.done",json!({"item_id":"msg_chat_test","output_index":0,"content_index":0,"text":"你好 🌍"})), + wire("response.content_part.done",json!({"item_id":"msg_chat_test","output_index":0,"content_index":0,"part":{"type":"output_text","text":"你好 🌍"}})), + wire("response.output_item.done",json!({"output_index":0,"item":message("你好 🌍")})), + completed(json!([message("你好 🌍")]))])).await; + let mut body = chat_request(true); + body["stream_options"] = json!({"include_usage":include_usage}); + let (_, chunks) = sse_body(fixture.post(body).await).await; + assert_eq!(combined(&chunks, "content"), "你好 🌍"); + assert_eq!( + chunks + .iter() + .filter(|chunk| chunk["choices"][0]["delta"]["role"] == "assistant") + .count(), + 1 + ); + assert_eq!( + chunks + .iter() + .filter(|chunk| chunk["choices"][0]["finish_reason"] == "stop") + .count(), + 1 + ); + for chunk in &chunks { + assert_eq!(chunk["object"], "chat.completion.chunk"); + assert_eq!(chunk["id"], chunks[0]["id"]); + assert_eq!(chunk["created"], chunks[0]["created"]); + assert_eq!(chunk["model"], "model-a"); + } + let usage: Vec<_> = chunks + .iter() + .filter(|chunk| chunk["choices"] == json!([])) + .collect(); + assert_eq!(usage.len(), usize::from(include_usage)); + if include_usage { + assert_eq!(usage[0]["usage"]["total_tokens"], 18); + } + let captured = fixture.captured(); + assert!(captured.body.get("stream_options").is_none()); + } +} + +#[tokio::test] +async fn sse_preserves_interleaved_tools_and_reordered_terminal_snapshots() { + let first = tool( + "fc_first", + "call_first", + "weather", + "{ \"city\": \"北京\" }", + ); + let second = tool("fc_second", "call_second", "search", "{\"query\":\"Rust\"}"); + let fixture = Fixture::new(Session::new(vec![ + created(), + wire( + "response.output_item.added", + json!({"output_index":2,"item":tool("fc_first","call_first","weather","")}), + ), + wire( + "response.output_item.added", + json!({"output_index":5,"item":tool("fc_second","call_second","search","")}), + ), + wire( + "response.function_call_arguments.delta", + json!({"output_index":2,"item_id":"fc_first","delta":"{ \"city\": "}), + ), + wire( + "response.function_call_arguments.delta", + json!({"output_index":5,"item_id":"fc_second","delta":"{\"query\":"}), + ), + wire( + "response.function_call_arguments.delta", + json!({"output_index":2,"item_id":"fc_first","delta":"\"北京\" }"}), + ), + wire( + "response.function_call_arguments.done", + json!({"output_index":5,"item_id":"fc_second","arguments":second["arguments"]}), + ), + wire( + "response.output_item.done", + json!({"output_index":2,"item":first}), + ), + completed(json!([second, first])), + ])) + .await; + let (_, chunks) = sse_body(fixture.post(chat_request(true)).await).await; + let mut tools = BTreeMap::::new(); + for chunk in &chunks { + if let Some(calls) = chunk["choices"][0]["delta"]["tool_calls"].as_array() { + for call in calls { + let entry = tools.entry(call["index"].as_u64().unwrap()).or_default(); + if let Some(id) = call["id"].as_str() { + entry.0.push_str(id); + } + if let Some(name) = call["function"]["name"].as_str() { + entry.1.push_str(name); + } + if let Some(arguments) = call["function"]["arguments"].as_str() { + entry.2.push_str(arguments); + } + } + } + } + assert_eq!( + tools, + BTreeMap::from([ + ( + 0, + ( + "call_first".into(), + "weather".into(), + "{ \"city\": \"北京\" }".into() + ) + ), + ( + 1, + ( + "call_second".into(), + "search".into(), + "{\"query\":\"Rust\"}".into() + ) + ), + ]) + ); + assert_eq!( + chunks.last().unwrap()["choices"][0]["finish_reason"], + "tool_calls" + ); +} + +#[tokio::test] +async fn extra_attributes_and_auxiliary_events_preserve_existing_chat_results() { + let mut text = message("preserved"); + text["future_optional"] = json!({"opaque":true}); + text["content"][0]["future_attribute"] = json!([1, 2, 3]); + let fixture = Fixture::new(Session::new(vec![ + wire("response.future_auxiliary", json!({"future":true})), + wire( + "response.reasoning_summary_text.delta", + json!({"output_index":0,"summary_index":0,"delta":"private"}), + ), + created(), + completed(json!([text])), + ])) + .await; + let (_, chunks) = sse_body(fixture.post(chat_request(true)).await).await; + assert_eq!(combined(&chunks, "content"), "preserved"); + assert!( + !chunks + .iter() + .any(|chunk| chunk.to_string().contains("private")) + ); + assert_eq!( + fixture + .trace + .discarded + .load(std::sync::atomic::Ordering::SeqCst), + 2 + ); +} + +#[tokio::test] +async fn incomplete_projects_only_supported_finish_reasons() { + for (reason, expected) in [ + ("max_output_tokens", "length"), + ("content_filter", "content_filter"), + ] { + for streaming in [false, true] { + let event = wire( + "response.incomplete", + json!({"response":{ + "id":"resp_chat_test","status":"incomplete","output":[message("partial")], + "incomplete_details":{"reason":reason}}}), + ); + let fixture = Fixture::new(Session::new(vec![created(), event])).await; + let response = fixture.post(chat_request(streaming)).await; + if streaming { + let (_, chunks) = sse_body(response).await; + assert_eq!( + chunks.last().unwrap()["choices"][0]["finish_reason"], + expected + ); + } else { + assert_eq!(response.status(), StatusCode::OK); + assert_eq!( + json_body(response).await["choices"][0]["finish_reason"], + expected + ); + } + } + } +} + +#[tokio::test] +async fn encrypted_reasoning_stays_private_while_public_text_and_usage_are_projected() { + let reasoning = json!({ + "id":"rs_private","type":"reasoning", + "summary":[{"type":"summary_text","text":"PRIVATE_SUMMARY_MARKER"}], + "encrypted_content":"OPAQUE_ENCRYPTED_REASONING_MARKER" + }); + for streaming in [false, true] { + let fixture = Fixture::new(Session::new(vec![ + created(), + wire( + "response.output_item.added", + json!({"output_index":0,"item":{"id":"rs_private","type":"reasoning","summary":[]}}), + ), + wire( + "response.reasoning_summary_text.delta", + json!({"item_id":"rs_private","output_index":0,"summary_index":0,"delta":"PRIVATE_SUMMARY_MARKER"}), + ), + wire( + "response.output_item.done", + json!({"output_index":0,"item":reasoning}), + ), + wire( + "response.output_item.done", + json!({"output_index":1,"item":message("public answer")}), + ), + completed(json!([reasoning, message("public answer")])), + ])) + .await; + let mut request = chat_request(streaming); + if streaming { + request["stream_options"] = json!({"include_usage":true}); + } + let response = fixture.post(request).await; + let (body, content, usage) = if streaming { + let (body, chunks) = sse_body(response).await; + let usage = chunks + .iter() + .find(|chunk| chunk["choices"] == json!([])) + .expect("terminal usage chunk")["usage"] + .clone(); + (body, combined(&chunks, "content"), usage) + } else { + assert_eq!(response.status(), StatusCode::OK); + let body = json_body(response).await; + ( + body.to_string(), + body["choices"][0]["message"]["content"] + .as_str() + .unwrap() + .to_owned(), + body["usage"].clone(), + ) + }; + assert_eq!(content, "public answer"); + assert!(!body.contains("PRIVATE_SUMMARY_MARKER"), "{body}"); + assert!( + !body.contains("OPAQUE_ENCRYPTED_REASONING_MARKER"), + "{body}" + ); + assert_eq!(usage["completion_tokens_details"]["reasoning_tokens"], 2); + } +} + +#[tokio::test] +async fn unrepresentable_business_output_fails_without_successful_chat_completion() { + for output in [ + json!({"id":"future_output","type":"future_business_output","payload":"unrepresentable"}), + json!({"id":"ig_native","type":"image_generation_call","status":"completed","result":"OPAQUE_IMAGE_MARKER"}), + ] { + for streaming in [false, true] { + let fixture = + Fixture::new(Session::new(vec![created(), completed(json!([output]))])).await; + let mut request = chat_request(streaming); + if streaming { + request["stream_options"] = json!({"include_usage":true}); + } + let response = fixture.post(request).await; + if streaming { + let (body, chunks) = sse_body(response).await; + assert!(chunks.iter().any(|chunk| chunk.get("error").is_some())); + assert!( + !chunks + .iter() + .any(|chunk| chunk["choices"][0]["finish_reason"].is_string()) + ); + assert!(!chunks.iter().any(|chunk| chunk["choices"] == json!([]))); + assert!(!body.contains("OPAQUE_IMAGE_MARKER")); + } else { + assert!(response.status().is_server_error()); + let body = json_body(response).await; + assert!(body.get("error").is_some()); + assert!(body.get("choices").is_none()); + assert!(!body.to_string().contains("OPAQUE_IMAGE_MARKER")); + } + } + } +} + +#[tokio::test] +async fn stream_without_terminal_does_not_emit_a_successful_finish_or_usage() { + let fixture = Fixture::new(Session::new(vec![ + created(), + wire( + "response.output_text.delta", + json!({"item_id":"msg_chat_test","output_index":0,"content_index":0,"delta":"partial"}), + ), + ])) + .await; + let mut body = chat_request(true); + body["stream_options"] = json!({"include_usage":true}); + let (_, chunks) = sse_body(fixture.post(body).await).await; + assert!(chunks.iter().any(|chunk| chunk.get("error").is_some())); + assert!( + !chunks + .iter() + .any(|chunk| chunk["choices"][0]["finish_reason"].is_string()) + ); + assert!(!chunks.iter().any(|chunk| chunk["choices"] == json!([]))); +} + +#[tokio::test] +async fn first_batch_terminal_can_finish_without_creating_usage_facts() { + let fixture = Fixture::new(Session::new(vec![wire( + "response.completed", + json!({"response":{ + "id":"resp_chat_test","status":"completed","output":[message("done only")] + }}), + )])) + .await; + let mut body = chat_request(true); + body["stream_options"] = json!({"include_usage":true}); + let (_, chunks) = sse_body(fixture.post(body).await).await; + assert_eq!(combined(&chunks, "content"), "done only"); + assert_eq!( + chunks + .iter() + .filter(|chunk| chunk["choices"][0]["finish_reason"] == "stop") + .count(), + 1 + ); + assert!(!chunks.iter().any(|chunk| chunk["usage"].is_object())); + assert!( + fixture + .trace + .finalized + .load(std::sync::atomic::Ordering::SeqCst) + ); +} + +#[tokio::test] +async fn native_sse_byte_fragments_and_multiple_events_preserve_unicode() { + let first = gateway_protocol::openai::sse::encode_sse_event("response.output_text.delta",&json!({ + "type":"response.output_text.delta","item_id":"msg_chat_test","output_index":0,"content_index":0,"delta":"你" + }).to_string()); + let second = gateway_protocol::openai::sse::encode_sse_event("response.output_text.delta",&json!({ + "type":"response.output_text.delta","item_id":"msg_chat_test","output_index":0,"content_index":0,"delta":"好" + }).to_string()); + let split = first.find('你').unwrap() + 1; + let mut remainder = first.as_bytes()[split..].to_vec(); + remainder.extend_from_slice(second.as_bytes()); + let raw = |bytes| { + gateway_core::event::ProviderEvent::wire( + gateway_core::event::ProtocolWireEvent::raw_sse("openai", bytes).unwrap(), + ) + }; + let fixture = Fixture::new(Session::new(vec![ + created(), + raw(bytes::Bytes::copy_from_slice(&first.as_bytes()[..split])), + raw(bytes::Bytes::from(remainder)), + completed(json!([message("你好")])), + ])) + .await; + let (_, chunks) = sse_body(fixture.post(chat_request(true)).await).await; + assert_eq!(combined(&chunks, "content"), "你好"); + assert_eq!( + chunks.last().unwrap()["choices"][0]["finish_reason"], + "stop" + ); +} + +#[tokio::test] +async fn refusal_deltas_and_done_snapshots_produce_only_standard_refusal() { + let output = json!({"id":"msg_refusal","type":"message","role":"assistant","status":"completed", + "content":[{"type":"refusal","refusal":"cannot help"}]}); + let fixture = Fixture::new(Session::new(vec![created(), + wire("response.refusal.delta",json!({"item_id":"msg_refusal","output_index":0,"content_index":0,"delta":"cannot "})), + wire("response.refusal.done",json!({"item_id":"msg_refusal","output_index":0,"content_index":0,"refusal":"cannot help"})), + completed(json!([output]))])).await; + let (_, chunks) = sse_body(fixture.post(chat_request(true)).await).await; + assert_eq!(combined(&chunks, "refusal"), "cannot help"); + assert_eq!(combined(&chunks, "content"), ""); + assert_eq!( + chunks.last().unwrap()["choices"][0]["finish_reason"], + "stop" + ); +} + +#[tokio::test] +async fn auxiliary_events_after_completed_preserve_exactly_one_finish_and_usage() { + for with_created in [false, true] { + // 终态与辅助事件同属一批 Core 交付 不构造已终结会话继续产出事件的另一种异常 + let session = Session::new(if with_created { + vec![created()] + } else { + Vec::new() + }) + .with_batch(vec![ + completed(json!([message("completed text")])), + wire( + "response.future_metadata", + json!({"response_id":"resp_chat_test","future":{"kept":true}}), + ), + ]); + let fixture = Fixture::new(session).await; + let mut body = chat_request(true); + body["stream_options"] = json!({"include_usage":true}); + let (_, chunks) = sse_body(fixture.post(body).await).await; + assert_eq!(combined(&chunks, "content"), "completed text"); + assert_eq!( + chunks + .iter() + .filter(|chunk| chunk["choices"][0]["finish_reason"] == "stop") + .count(), + 1 + ); + let usage = chunks + .iter() + .filter(|chunk| chunk["choices"] == json!([])) + .collect::>(); + assert_eq!(usage.len(), 1); + assert_eq!(usage[0]["usage"]["total_tokens"], 18); + assert!(!chunks.iter().any(|chunk| chunk.get("error").is_some())); + assert_eq!( + fixture + .trace + .commits + .load(std::sync::atomic::Ordering::SeqCst), + 1 + ); + } +} + +#[tokio::test] +async fn wire_failures_preserve_existing_recovery_fields_without_success_chunks() { + for shape in ["response.failed", "error", "bare_error"] { + let mut expected = json!({ + "type":"rate_limit_error","code":"rate_limit_exceeded","param":null, + "message":"retry later","headers":{"retry-after":"2"}, + "recovery":{"action":"retry","future":[1,2]}, + "future_error_property":{"enabled":true} + }); + let failure = match shape { + "response.failed" => wire( + "response.failed", + json!({"response":{ + "id":"resp_chat_test","object":"response","model":"upstream-model", + "status":"failed","output":[],"error":expected.clone() + }}), + ), + "error" => wire("error", json!({"error":expected.clone()})), + _ => { + // 裸事件的 type 是 wire 名称,仅该包装身份归一化为 Chat 错误类型 + let failure = wire("error", expected.clone()); + expected["type"] = json!("server_error"); + failure + } + }; + let fixture = Fixture::new(Session::new(vec![created(), failure])).await; + let mut body = chat_request(true); + body["stream_options"] = json!({"include_usage":true}); + let (_, chunks) = sse_body(fixture.post(body).await).await; + let errors = chunks + .iter() + .filter_map(|chunk| chunk.get("error")) + .collect::>(); + assert_eq!(errors, vec![&expected], "{shape}"); + assert!( + !chunks + .iter() + .any(|chunk| chunk["choices"][0]["finish_reason"].is_string()), + "{shape}" + ); + assert!(!chunks.iter().any(|chunk| chunk["choices"] == json!([]))); + assert!(!chunks.iter().any(|chunk| chunk["usage"].is_object())); + assert_eq!( + fixture + .trace + .commits + .load(std::sync::atomic::Ordering::SeqCst), + 1 + ); + } +} + +#[tokio::test] +async fn conflicting_text_snapshot_fails_instead_of_appending_replacement_content() { + let fixture = Fixture::new(Session::new(vec![created(), + wire("response.output_text.delta",json!({"item_id":"msg_chat_test","output_index":0,"content_index":0,"delta":"already sent"})), + completed(json!([message("different result")]))])).await; + let (_, chunks) = sse_body(fixture.post(chat_request(true)).await).await; + assert_eq!(combined(&chunks, "content"), "already sent"); + assert!(chunks.iter().any(|chunk| chunk.get("error").is_some())); + assert!( + !chunks + .iter() + .any(|chunk| chunk["choices"][0]["finish_reason"].is_string()) + ); +} diff --git a/backend/crates/gateway-api/tests/openai/mod.rs b/backend/crates/gateway-api/tests/openai/mod.rs index 643f5ebb5..8b2889326 100644 --- a/backend/crates/gateway-api/tests/openai/mod.rs +++ b/backend/crates/gateway-api/tests/openai/mod.rs @@ -1,6 +1,7 @@ //! OpenAI 协议 HTTP 与 WebSocket 接口的测试入口 mod auth; +mod chat_completions; mod endpoint; mod error; mod image_policy; diff --git a/backend/crates/gateway-api/tests/openai/responses/http.rs b/backend/crates/gateway-api/tests/openai/responses/http.rs index 9f8c85cc4..de4b60b83 100644 --- a/backend/crates/gateway-api/tests/openai/responses/http.rs +++ b/backend/crates/gateway-api/tests/openai/responses/http.rs @@ -2550,6 +2550,110 @@ async fn buffered_response_collects_completed_output_items_without_rewriting_str assert!(!trace.is_cancelled()); } +#[tokio::test] +async fn native_responses_route_preserves_images_and_encrypted_reasoning_in_json_and_sse() { + let admin = crate::admin::AdminTestFixture::new().await.services; + let reasoning = json!({ + "id":"rs_native","type":"reasoning","encrypted_content":"opaque-reasoning-for-replay", + "summary":[{"type":"summary_text","text":"native summary"}], + "future_reasoning_field":{"keep":true} + }); + let image = json!({ + "id":"ig_native","type":"image_generation_call","status":"completed", + "result":"b3BhcXVlLWltYWdl","output_format":"png", + "future_image_field":{"keep":true} + }); + let terminal = json!({ + "id":"resp_test","object":"response","model":"public-model","status":"completed", + "output":[reasoning, image],"future_terminal_field":{"keep":true} + }); + for streaming in [false, true] { + let trace = Arc::new(Trace::default()); + let mut expected_sse = Vec::new(); + let mut events = Vec::new(); + for (index, (kind, mut data, facts)) in [ + ( + "response.created", + json!({"response":{"id":"resp_test","model":"public-model","status":"in_progress","output":[]}}), + vec![started()], + ), + ( + "response.output_item.done", + json!({"output_index":0,"item":reasoning}), + vec![], + ), + ( + "response.image_generation_call.partial_image", + json!({"item_id":"ig_native","output_index":1,"partial_image_index":0,"partial_image_b64":"b3BhcXVlLXBhcnRpYWw="}), + vec![], + ), + ( + "response.output_item.done", + json!({"output_index":1,"item":image}), + vec![], + ), + ( + "response.completed", + json!({"response":terminal}), + vec![completed()], + ), + ] + .into_iter() + .enumerate() + { + data["type"] = json!(kind); + let id = format!("native-boundary-{index}"); + // 保留行结束符、空白和 SSE 元数据,不能仅比较 JSON 后掩盖重编码 + let raw = Bytes::from(format!( + "id: {id}\r\nevent: {kind}\r\nretry: 1234\r\ndata: {data} \r\n\r\n" + )); + expected_sse.extend_from_slice(&raw); + events.push(ProviderEvent::canonical_with_wire( + facts, + ProtocolWireEvent::json_with_raw_sse_metadata( + "openai", + Some(kind.to_owned()), + data, + raw, + Some(id), + Some(1234), + ) + .unwrap(), + )); + } + let session = if streaming { + FakeSession::streaming( + Arc::clone(&trace), + vec![ + NextStep::Event( + CoordinatedEvent::try_batch( + events, + CommitRequirement::CommitBeforeDelivery, + ) + .unwrap(), + ), + NextStep::FinalizeSuccess, + ], + ) + } else { + FakeSession::buffered_provider(Arc::clone(&trace), events) + }; + // 使用真实 /v1/responses 入口,覆盖共享执行函数的可选适配器为空分支 + let response = response(&admin, "openai", session, streaming).await; + assert_eq!(response.status(), StatusCode::OK); + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + if streaming { + expected_sse.extend_from_slice(b"data: [DONE]\n\n"); + assert_eq!(body.as_ref(), expected_sse.as_slice()); + assert_eq!(trace.snapshot(), ["next_event", "commit", "next_end"]); + } else { + assert_eq!(serde_json::from_slice::(&body).unwrap(), terminal); + assert_eq!(trace.snapshot(), ["collect", "commit"]); + } + assert!(!trace.is_cancelled()); + } +} + #[tokio::test] async fn buffered_response_keeps_authoritative_terminal_and_supports_omitted_output() { let item = json!({"id": "rs_test", "type": "reasoning", "encrypted_content": "opaque"}); diff --git a/backend/crates/gateway-plugin/sdk/docs/capabilities.md b/backend/crates/gateway-plugin/sdk/docs/capabilities.md index a47fec5ca..c573eba02 100644 --- a/backend/crates/gateway-plugin/sdk/docs/capabilities.md +++ b/backend/crates/gateway-plugin/sdk/docs/capabilities.md @@ -344,6 +344,10 @@ let status = host.service::((account_id, true)).await? 插件改写后用于后续认证、版本检查、正文解压及执行;不新增设置业务的中间件入口。`timeout_ms` 使用宿主 HTTP 超时基线, 插件可改写或用 `None` 清除;终端按有效值执行,不恢复宿主默认值。宿主快照不可用时 `settings` 为 `null` +`POST /v1/chat/completions` 在 HTTP 挂载中保留 Chat 请求与 Chat JSON/SSE 响应视图, +转换为 Responses 发生在内层模型入口。自行按 URI 筛选、改写或分派模型请求的 HTTP 插件, +需要将该路径包含在自身匹配范围内;只匹配 `/v1/responses` 不会覆盖 Chat + `HttpCall.settings_sources` 返回 `config_revision`、宿主 `host` 基线、运行设置的 `overrides`、Key 作用域的 `execution` 与 `http_timeout`。 每条改写记录含 `instance_id`、`order` 和 `value`;来源由宿主记录,修改查询结果不会写入设置。 宿主比较提交前后的值,仅记录实际改变的设置项;同值赋值不新增改写记录。每个顶层设置项整体替换,`false`、`0`、 @@ -468,6 +472,13 @@ Responses 终端重新解码插件正文时读取改写后的正文上限;atte 更早的改写使用上述 HTTP 请求的 `settings`。`request.head.settings_sources` 使用相同的来源结构,`execution.input` 表示进入当前 Key 作用域时的有效值。 原生 Responses WebSocket 每轮请求读取当前宿主快照,重新认证并解析 Key 策略,同时保留握手阶段的显式改写;已开始的执行保持自己的冻结值 +Chat Completions 在 request 与 attempt 挂载中使用规范化的 Responses 请求和响应正文, +`request.head.endpoint` 仍为 `/v1/chat/completions`,`protocol` 仍为 `openai`。 +request 改写按 Responses 合同重新解码,响应先经过该合同的复核,再由 API 投影成 Chat 交给外层 HTTP 挂载。 +Chat 的下游流模式及 `include_usage` 由 API 持有,request 阶段不能切换已确定的下游流模式。 +同一边界的原样保留不要求 Chat 与 Responses 字节相同;需要读写 `messages` 或 `choices` 的插件应使用 HTTP 挂载, +字段范围见 [Chat 合同](../../../../../docs/api.md#chat-completions-请求) + 中间件可读取和改写完整请求、响应 headers、正文与响应帧,也不按字段名称过滤认证、Cookie、会话或连接头。 该能力遵循[完整信任](manifest.md#完整信任)边界;类型、HTTP 格式、大小及资源生命周期仍需符合合同 diff --git a/docs/api.md b/docs/api.md index e8a4db420..3d697f936 100644 --- a/docs/api.md +++ b/docs/api.md @@ -7,7 +7,7 @@ | 查找内容 | 入口 | | --- | --- | | 鉴权与响应 | [公共约定](#1-鉴权与公共约定) · [浏览器会话与 Key 自助页](#4-浏览器认证) | -| 客户端协议 | [Responses、Images、模型与 Provider 操作](#3-openai-数据面与模型目录) · [额度查询](#api-key-额度查询) | +| 客户端协议 | [Responses、Chat、Images、模型与 Provider 操作](#3-openai-数据面与模型目录) · [额度查询](#api-key-额度查询) | | 账号与授权范围 | [账号](#5-账号) · [代理](#独立代理管理--managed-proxies) · [分组](#6-账号分组) · [Client Key](#7-client-key) | | 系统与运维 | [健康检查](#2-健康检查) · [运行设置](#8-运行设置) · [备份](#9-备份) · [统计与诊断](#10-dashboard用量与错误) · [版本与更新](#11-版本更新与重启) | | 插件 | [安装包与来源](#安装包与来源) · [运行实例](#121-运行实例) · [管理扩展](#122-管理扩展页面与原始-api) | @@ -200,13 +200,14 @@ Dashboard 的 `lastUsedRelative` 和账号额度的 `refreshedAtRelative` 由后 ## 3. OpenAI 数据面与模型目录 -除下述 Responses 入站解压保护外,Responses、Images 和 standalone Search HTTP body、 +除下述入站解压保护外,Responses、Chat Completions、Images 和 standalone Search HTTP body、 WebSocket message 和 frame 不设置网关私有长度上限;协议可接受性由上游决定 | 方法 | 路由 | 说明 | | --- | --- | --- | | `POST` | `/v1/responses` | OpenAI Responses JSON;`stream=true` 返回 SSE,否则返回完整 JSON | | `GET` | `/v1/responses` | 通过 HTTP Upgrade 建立 Responses WebSocket | +| `POST` | `/v1/chat/completions` | Chat Completions 转 Responses 执行;返回单 choice 的 Chat JSON 或 SSE | | `POST` | `/v1/alpha/search` | Codex standalone web search;JSON 请求与响应正文原样转发 | | `POST` | `/v1/images/generations` | 通过 OpenAI Provider 发起图像生成;JSON 请求与响应正文原样转发 | | `POST` | `/v1/images/edits` | 通过 OpenAI Provider 发起图像编辑;JSON 原样转发,multipart 上传转为原生 Images JSON | @@ -216,6 +217,93 @@ WebSocket message 和 frame 不设置网关私有长度上限;协议可接受 | `GET` / `POST` | `/v1/providers/{provider}/http/{endpoint}` | 调用指定 Provider 预先声明的固定 HTTP 操作;不是任意 URL 代理 | | `GET` | `/v1/usage` | 查询当前 Client Key 的日与周额度,仅使用网关已结算的 USD 账本 | +### Chat Completions 请求 + +`POST /v1/chat/completions` 接受 Chat 消息历史,转换为 Responses 请求执行,再交付 Chat 格式结果。 +每次请求须提供非空字符串 `model` 和非空数组 `messages`,多轮对话由客户端携带完整历史; +没有 Chat 查询、删除或服务端历史续接接口。`stream` 省略、`null` 或 `false` 时返回 JSON,`true` 时返回 SSE + +官方 OpenAI SDK 可使用 Chat Completions 方法,将 `base_url` 指向网关的 `/v1`。 +下面的 `CPR_CLIENT_KEY` 为管理端创建的 Client Key,`CPR_MODEL` 为当前 Key 可用的模型 ID,可通过 `/v1/models` 查询: + +```python +import os +from openai import OpenAI + +client = OpenAI( + base_url="https://gateway.example.com/v1", + api_key=os.environ["CPR_CLIENT_KEY"], +) +completion = client.chat.completions.create( + model=os.environ["CPR_MODEL"], + messages=[{"role": "user", "content": "你好"}], +) +print(completion.choices[0].message.content) +``` + +#### 请求字段 + +| 字段 | 支持与映射 | +| --- | --- | +| `messages[].role` | `system`、`developer`、`user`、`assistant`、`tool`,保持消息顺序;assistant 文本历史使用 Responses 的 `output_text` | +| `messages[].content` | 字符串或文本块数组;`user` 还可包含 `image_url`,支持 HTTP(S) URL 和非空 `data:image/...;base64,...`,保留 `detail`,不代取图片 | +| `messages[].refusal`、assistant 的 `refusal` 内容块 | 保留拒绝历史;assistant 的 `content` 仅在同时有工具调用或拒绝内容时可省略或为 `null` | +| `messages[].tool_calls`、`messages[].tool_call_id` | assistant 的 function 调用转为 `function_call`,tool 结果转为 `function_call_output`,保留调用 ID 与 arguments 字符串,不解析或执行函数 | +| `tools` | 仅支持现代 `type: "function"`;将 `function` 定义展开为 Responses 工具,保留 parameters schema;`strict` 省略或 `null` 时显式使用 `false` | +| `tool_choice`、`parallel_tool_calls` | 支持 `auto`、`none`、`required` 和指定 function;`required` 须提供工具,指定名称须存在于 `tools`;并行选项保留布尔值 | +| `response_format` | `text`、`json_object`、`json_schema` 映射到 `text.format`,JSON Schema 保留原值 | +| `reasoning_effort`、`verbosity` | 分别映射到 `reasoning.effort`、`text.verbosity`;接受非空字符串,不固定枚举档位 | +| `max_completion_tokens`、`max_tokens` | 正整数映射到 `max_output_tokens`,同时提供时前者优先 | +| `temperature`、`top_p` | 分别接受 0 到 2、0 到 1 的数字,包含端点;是否生效见下文 | +| `user`、`safety_identifier`、`service_tier`、`prompt_cache_key`、`prompt_cache_retention` | 两侧同语义的字符串保留原值,档位与缓存策略由上游判断 | +| `metadata`、`prompt_cache_options`、`moderation` | 两侧同结构的对象原样携带,不逐层重建内部字段 | +| 文本、图片内容块的 `prompt_cache_breakpoint` | 同结构对象原样携带;assistant 输出消息不支持此选项 | + +可选映射参数为 `null` 时按未提供处理。字段类型、角色组合或工具选择无效时,在访问上游前返回 `400`, +错误中的 `param` 指向对应字段,不回显字段值。工具参数字符串和 schema 的业务有效性由客户端与上游按各自合同处理 + +以下参数只接受所列中性值;省略或 `null` 也接受,其他值返回 `400 unsupported_parameter`: + +| 字段 | 接受值 | +| --- | --- | +| `n` | `1` | +| `store`、`logprobs` | `false` | +| `top_logprobs`、`frequency_penalty`、`presence_penalty` | `0` | +| `logit_bias` | `{}` | +| `stop`、旧式 `functions` | `[]` | +| `modalities` | `["text"]` | + +`function_call`、`seed`、`prediction`、`audio`、`web_search_options`、`reasoning_content` 及消息级 `reasoning` 不支持非 `null` 值。 +非空消息 `name`、旧式 `function` 角色、文件内容、音频内容、custom 或 hosted 工具及未知内容类型也会明确拒绝。 +该入口不接收 Responses 请求形状;`input`、`previous_response_id`、`conversation`、`instructions`、`reasoning`、 +`text`、`include`、`background` 和 `max_output_tokens` 不支持非 `null` 值,需要这些原生能力时使用 `/v1/responses` + +未知客户端附加字段只在请求、消息及工具等协议包装对象中忽略,不据名称自动转交 Responses, +也不递归删除 schema、metadata 或工具参数内的业务字段。新枚举值和同结构对象内部的扩展交由上游判断; +新的工具类型、输出模态或续接语义需要明确映射,不承诺跨协议完全透传 + +#### 响应与共用规则 + +非流式响应包含一个 `choices` 项,支持文本、拒绝和 function 工具调用。 +流式响应交付对应的 Chat delta;`stream_options` 仅在 `stream: true` 时可提供, +`include_usage: true` 在完整上游用量可用时于结束前追加 `choices: []` 的用量块,其余块的 `usage` 为 `null`。 +未提供完整用量时不补造 Token 数;缺省不启用流长度混淆,`include_obfuscation: true` 不受支持,`false` 可接受 + +响应对象新增普通属性、非必要事件不影响已有内容映射;无法表达的业务输出、缺少必要终态或终态冲突会失败, +不以丢弃内容或补造成功结束掩盖错误。OpenAI 是此入口的兼容目标,混合分组仍沿用现有选路与 fallback 规则, +不能据此推导所有 Provider 都具备相同的 Chat 能力 + +Responses 的 reasoning 项(包括摘要与 `encrypted_content`)不投影到 Chat 消息, +实际推理用量仍保留。需要推理密文回放或图片输出时使用原生 Responses 接口;Chat 的图片输入支持不代表图片输出可用 + +Chat 沿用 [Client Key 鉴权](#openai-数据面客户端接口)、[请求期限](#请求期限)、[请求解压](#请求解压)、 +[请求头与身份](#请求头与身份)及[隐私策略](#隐私策略)。隐私策略的 `request_body` 面向转换后的 Responses 正文, +例如 `$.input`,不需要另外维护一套针对 `messages` 的规则 + +转换后的请求仍执行统一[正文兼容](#正文兼容):顶层 `temperature`、`max_output_tokens` 和 +`prompt_cache_retention` 会被移除。因此接受相应 Chat 参数不表示采样、输出长度或缓存保留设置必然生效, +不会通过本地截断模拟长度控制;`top_p` 等其他参数继续由上游判断 + ### Images 请求 Images 使用原生非流式 JSON 端点,JSON 请求与响应正文原样转发。 @@ -272,7 +360,7 @@ Codex 的 review 等子代理请求仍使用 `/v1/responses`,并通过 `x-open #### 请求解压 -`POST /v1/responses` 在鉴权后按 `Content-Encoding` 解压,再解析 JSON;支持单一 `gzip`、 +`POST /v1/responses` 与 `POST /v1/chat/completions` 在鉴权后按 `Content-Encoding` 解压,再解析 JSON;支持单一 `gzip`、 `deflate`(zlib 封装)和 `zstd`,缺省、空值或 `identity` 直接使用原始正文。gzip 多成员与 zstd 多帧连续解码,整体展开结果受运行设置 `responsesMaxDecompressedBodyBytes` 约束(默认 64 MiB), 超限在继续展开前返回 `400 request_too_large`,错误信息包含当前请求的上限字节数。zstd diff --git a/docs/architecture.md b/docs/architecture.md index f8483ca49..cfbb3613e 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -462,6 +462,13 @@ Client Key 鉴权完成后,API adapter 从有界请求头识别客户端,Cor - 可恢复观测写入失败不能替换已经确定的客户端协议结果 Responses 按模型目录编译候选;全局模型映射是精确映射,未命中时模型名原样交给候选 Provider。 +Chat Completions 由 API 将入站消息规范化为 Responses 正文,进入共用 Generate 执行链; +request 中间件回程经过 Responses 协议复核后,再由 API 投影成 Chat JSON 或 SSE。 +HTTP 中间件位于转换外侧,request 与 attempt 中间件处理 Responses,实际 endpoint 始终为 `/v1/chat/completions`。 +Chat 复用既有提交、取消与终结生命周期;Core 继续独占路由、重试、交付事实与结算, +Provider 继续独占上游身份、字段兼容、隐私策略及账号隔离,Chat 层只负责协议映射。 +字段合同见 [Chat Completions](api.md#chat-completions-请求),插件视图见 [SDK 请求中间件](../backend/crates/gateway-plugin/sdk/docs/capabilities.md#模型请求与-attempt) + Images 与 standalone Search 是 OpenAI Provider 自有端点:两者都不参与文本模型映射,只在 Client Key 的账号范围确实包含 Codex 账号时生成单一 OpenAI 候选。Images 的图片模型由 Provider 从正文旁路 提取,用于账号模型政策检查,并在中间件改写后复验当前租约账号;不要求图片模型进入文本目录。