feat(api): support Chat Completions through Responses (#542)

Co-authored-by: Zyy <zyycn@users.noreply.github.com>
Co-authored-by: Codex <noreply@openai.com>

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<StreamProjection>),
+}
+
+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<SseEvent>) -> Result<Vec<Bytes>, 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<Box<dyn MiddlewareBody>>,
+    projection: Projection,
+    pending: VecDeque<MiddlewareFrame>,
+    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<Option<MiddlewareFrame>, 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<u16>,
+    ) -> 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<Self>) -> 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,
+    )
+}

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<ApiState>,
+    connect_info: Option<Extension<ConnectInfo<SocketAddr>>>,
+    ingress_id: Option<Extension<tower_http::request_id::RequestId>>,
+    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))
+}

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;

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<String, Value>,
+    pub(in crate::openai) include_usage: bool,
+}
+
+pub(in crate::openai) fn convert_request(
+    body: &[u8],
+) -> Result<ConvertedChatRequest, RequestDecodeError> {
+    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<String, Value>) -> 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<Value>, stream: bool) -> Result<bool, RequestDecodeError> {
+    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<Vec<Value>, 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<Vec<Value>, 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<Value, RequestDecodeError> {
+    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<Vec<Value>, 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<Vec<Value>, 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<Value, RequestDecodeError> {
+    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<Value, RequestDecodeError> {
+    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<String, Value>, key: &str) -> Option<Value> {
+    object.remove(key).filter(|value| !value.is_null())
+}
+
+fn required(
+    object: &mut Map<String, Value>,
+    key: &str,
+    path: &str,
+) -> Result<Value, RequestDecodeError> {
+    object
+        .remove(key)
+        .ok_or_else(|| RequestDecodeError::MissingField { field: path.into() })
+}
+
+fn object(value: Value, path: &str) -> Result<Map<String, Value>, RequestDecodeError> {
+    match value {
+        Value::Object(object) => Ok(object),
+        _ => Err(invalid_type(path, "an object")),
+    }
+}
+
+fn array(value: Value, path: &str) -> Result<Vec<Value>, 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<String, Value>,
+    key: &str,
+    path: &str,
+    allow_empty: bool,
+) -> Result<String, RequestDecodeError> {
+    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<String, Value>,
+    key: &str,
+    path: &str,
+) -> Result<Option<bool>, 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() }
+}

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<Value>,
+    #[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<Value>,
+}
+
+/// 输入已由现有 Responses collector 聚合，投影不再恢复或重建缺失的 output
+pub(in crate::openai) fn project_response(
+    response: &Value,
+    identity: &ChatIdentity,
+) -> Result<Value, ProtocolErrorBody> {
+    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<u64, Part>),
+    Function(Tool),
+    Reasoning,
+}
+
+struct OutputItem {
+    id: Option<String>,
+    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<u64, OutputItem>,
+    terminal: Option<Terminal>,
+}
+
+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<Vec<Bytes>, 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<Bytes>,
+    ) -> 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<Option<u64>, 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<Bytes, ProtocolErrorBody> {
+        delta_frame(&self.identity, self.include_usage, delta)
+    }
+
+    fn chunk(
+        &self,
+        choices: Value,
+        usage: Option<Value>,
+        response: Option<&Value>,
+    ) -> Result<Bytes, ProtocolErrorBody> {
+        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<u64, Part>,
+    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<Value> {
+    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<Option<Value>, 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<Bytes, ProtocolErrorBody> {
+    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<Value>,
+    response: Option<&Value>,
+) -> Result<Bytes, ProtocolErrorBody> {
+    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<Option<&'a str>, ProtocolErrorBody> {
+    value
+        .get(field)
+        .map(|_| required_nonempty_str(value, field))
+        .transpose()
+}
+
+fn required_index(value: &Value, field: &str) -> Result<u64, ProtocolErrorBody> {
+    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,
+        },
+    }
+}

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;

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<Arc<dyn ResponseAuthorization>>,
+) -> Response {
+    execute_prepared_response(service, prepared, request, authorization, None).await
+}
+
+/// 下游协议只投影出口字节 执行会话和计量仍归原 owner
+pub(in crate::openai) struct HttpResponseAdapter {
+    pub endpoint: &'static str,
+    pub wrap: Box<dyn FnOnce(MiddlewareResponse) -> 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<Arc<dyn ResponseAuthorization>>,
+    adapter: Option<HttpResponseAdapter>,
 ) -> 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
     };

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<Output = Response> + Send + 'static,
     admitted: oneshot::Receiver<ResponseCommit>,
+) -> Response {
+    response_with_encoder(response, admitted, None).await
+}
+
+pub(super) async fn response_with_encoder(
+    response: impl Future<Output = Response> + Send + 'static,
+    admitted: oneshot::Receiver<ResponseCommit>,
+    error_encoder: Option<fn(&Value) -> 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<fn(&Value) -> Bytes>),
     Streaming(BodyDataStream),
 }
 
@@ -66,7 +74,7 @@ impl PendingResponse {
     async fn next(mut self) -> Result<Option<(Bytes, Self)>, 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<fn(&Value) -> 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)
 }

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,

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,

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<ApiState> {
         .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))

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<Vec<ProviderEvent>>,
+    pub trace: Arc<Trace>,
+    pub finish: Finish,
+    pub delay: Option<Duration>,
+    pub fail_commit: bool,
+    pub response_headers: Vec<ProviderResponseHeader>,
+    committed: bool,
+    buffered: bool,
+}
+
+impl Session {
+    pub fn new(events: Vec<ProviderEvent>) -> 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<ProviderEvent>) -> 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<Option<CoordinatedEvent>, 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<Vec<ProviderEvent>, 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<u16>) -> 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<Self>) -> 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<Option<Captured>>,
+    pub client: AuthenticatedClient,
+    pub middleware: Option<FrozenMiddlewarePlan>,
+    session: Mutex<Option<Session>>,
+}
+
+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<AuthenticatedClient, ClientAuthenticationError> {
+        if key == KEY {
+            Ok(self.client.clone())
+        } else {
+            Err(ClientAuthenticationError::InvalidKey)
+        }
+    }
+    fn public_models(&self, _: &AuthenticatedClient) -> Vec<PublicModelId> {
+        Vec::new()
+    }
+    fn contains_public_model(&self, _: &AuthenticatedClient, _: &PublicModelId) -> bool {
+        true
+    }
+    fn middleware_plan(&self, _: &PreparedRootExecution) -> Option<FrozenMiddlewarePlan> {
+        self.middleware.clone()
+    }
+    fn start(
+        &self,
+        request: StartExecution,
+    ) -> BoxFuture<'_, Result<StartedExecution, GatewayError>> {
+        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<StartedExecution, GatewayError>> {
+        Box::pin(async { panic!("Chat must use model execution") })
+    }
+}
+
+pub(super) struct Fixture {
+    pub router: axum::Router,
+    pub execution: Arc<Execution>,
+    pub trace: Arc<Trace>,
+}
+
+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<Body> {
+    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<Vec<Value>>,
+    pub update: Option<fn(&mut Value)>,
+}
+
+impl InspectMiddleware {
+    pub fn frozen(self: &Arc<Self>) -> 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<MiddlewareResponse, MiddlewareError>> {
+        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::<Vec<_>>(),
+        ["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));
+}

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":[]
+        }}),
+    )
+}

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::<Vec<_>>(),
+        ["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}}));
+}

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<Value>) {
+    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::<u64, (String, String, String)>::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::<Vec<_>>();
+        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::<Vec<_>>();
+        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())
+    );
+}

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;

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::<Value>(&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"});

backend/crates/gateway-plugin/sdk/docs/capabilities.md

@@ -344,6 +344,10 @@ let status = host.service::<SetResetCreditAutomation>((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 格式、大小及资源生命周期仍需符合合同
 

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

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 从正文旁路
 提取，用于账号模型政策检查，并在中间件改写后复验当前租约账号；不要求图片模型进入文本目录。