diff --git a/backend/crates/gateway-api/src/openai/responses/request.rs b/backend/crates/gateway-api/src/openai/responses/request.rs index f69819601..4e91aa56c 100644 --- a/backend/crates/gateway-api/src/openai/responses/request.rs +++ b/backend/crates/gateway-api/src/openai/responses/request.rs @@ -310,9 +310,18 @@ pub(super) fn decode_request_inner( ) -> Result { let value = serde_json::from_slice::(body).map_err(|_| RequestDecodeError::MalformedJson)?; - let Value::Object(mut object) = value else { + let Value::Object(object) = value else { return Err(RequestDecodeError::ExpectedObject); }; + decode_request_object(object, review, request_headers) +} + +/// 解码已解析的顶层 object；WebSocket 帧解码复用此内核，避免序列化往返。 +pub(super) fn decode_request_object( + mut object: Map, + review: bool, + request_headers: &OpenAiRequestHeaders, +) -> Result { // `use_websocket` 只影响本地 transport。无论值是否可识别，都不得进入上游 body。 let use_websocket = object .remove("use_websocket") diff --git a/backend/crates/gateway-api/src/openai/responses/websocket.rs b/backend/crates/gateway-api/src/openai/responses/websocket.rs index fe4409095..7d85824b7 100644 --- a/backend/crates/gateway-api/src/openai/responses/websocket.rs +++ b/backend/crates/gateway-api/src/openai/responses/websocket.rs @@ -549,9 +549,7 @@ fn decode_response_create_inner( if matches!(body.get("stream"), Some(value) if value.as_bool() != Some(true)) { return Err(ResponseCreateFrameError::StreamingRequired); } - let encoded = serde_json::to_vec(&Value::Object(body)) - .map_err(|_| ResponseCreateFrameError::InvalidJson)?; - super::request::decode_request_inner(&encoded, false, request_headers) + super::request::decode_request_object(body, false, request_headers) .map_err(ResponseCreateFrameError::Request) } diff --git a/backend/crates/gateway-protocol/src/openai/sse.rs b/backend/crates/gateway-protocol/src/openai/sse.rs index d9f88db11..174665685 100644 --- a/backend/crates/gateway-protocol/src/openai/sse.rs +++ b/backend/crates/gateway-protocol/src/openai/sse.rs @@ -72,26 +72,50 @@ pub const DONE_SSE_FRAME: &str = "data: [DONE]\n\n"; #[derive(Debug, Default)] pub struct SseEventDecoder { pending: Vec, + /// `pending` 中已确认不含帧分隔符的前缀长度。新数据到达时从该边界 + /// 回退分隔符最大跨界宽度继续扫描，大帧跨多个 chunk 不会被反复重扫。 + scanned: usize, } impl SseEventDecoder { + /// 返回从 `consumed` 起的下一个完整帧结束位置（含分隔符），并推进扫描边界。 + fn next_frame_end(&mut self, consumed: usize) -> Option { + // `\r\n\r\n` 分隔符可能跨 chunk 边界，扫描起点回退 3 字节。 + let from = self.scanned.saturating_sub(3).max(consumed); + match sse_frame_separator_bytes(&self.pending[from..]) { + Some((position, separator_len)) => { + let end = from + position + separator_len; + self.scanned = end; + Some(end) + } + None => { + self.scanned = self.pending.len(); + None + } + } + } + + fn commit_consumed(&mut self, consumed: usize) { + if consumed != 0 { + self.pending.drain(..consumed); + self.scanned = self.scanned.saturating_sub(consumed); + } + } + /// 追加一个字节块并返回其中已经完整的事件。 pub fn push(&mut self, chunk: &[u8]) -> Result, SseError> { self.pending.extend_from_slice(chunk); let mut events = Vec::new(); let mut consumed = 0usize; - while let Some(frame_len) = sse_frame_end(&self.pending[consumed..]) { - let end = consumed.saturating_add(frame_len); + while let Some(end) = self.next_frame_end(consumed) { let frame = std::str::from_utf8(&self.pending[consumed..end]) .map_err(|error| SseError::ParseError(error.to_string()))?; events.extend(parse_sse_events(frame)?); consumed = end; } - if consumed != 0 { - self.pending.drain(..consumed); - } + self.commit_consumed(consumed); if self.pending.len() > MAX_SSE_EVENT_BUFFER_BYTES { return Err(SseError::BufferExceeded { max_bytes: MAX_SSE_EVENT_BUFFER_BYTES, @@ -110,8 +134,7 @@ impl SseEventDecoder { let mut frames = Vec::new(); let mut consumed = 0usize; - while let Some(frame_len) = sse_frame_end(&self.pending[consumed..]) { - let end = consumed.saturating_add(frame_len); + while let Some(end) = self.next_frame_end(consumed) { let raw = self.pending[consumed..end].to_vec(); let events = std::str::from_utf8(&raw) .ok() @@ -121,9 +144,7 @@ impl SseEventDecoder { consumed = end; } - if consumed != 0 { - self.pending.drain(..consumed); - } + self.commit_consumed(consumed); if self.pending.len() > MAX_SSE_EVENT_BUFFER_BYTES { return Err(SseError::BufferExceeded { max_bytes: MAX_SSE_EVENT_BUFFER_BYTES, @@ -134,6 +155,7 @@ impl SseEventDecoder { /// 流结束时解析尚未带空行分隔符的最后一帧。 pub fn finish(&mut self) -> Result, SseError> { + self.scanned = 0; if self.pending.is_empty() { return Ok(Vec::new()); } @@ -147,6 +169,7 @@ impl SseEventDecoder { /// /// 与 [`Self::push_frames`] 一致，旁路解析失败不会丢弃原始帧。 pub fn finish_frames(&mut self) -> Result, SseError> { + self.scanned = 0; if self.pending.is_empty() { return Ok(Vec::new()); } diff --git a/backend/crates/gateway-protocol/tests/openai/sse.rs b/backend/crates/gateway-protocol/tests/openai/sse.rs index bee85ae24..1c6d5984a 100644 --- a/backend/crates/gateway-protocol/tests/openai/sse.rs +++ b/backend/crates/gateway-protocol/tests/openai/sse.rs @@ -107,6 +107,29 @@ fn incremental_decoder_should_decode_across_arbitrary_chunk_boundaries() { ); } +#[test] +fn incremental_decoder_should_find_separators_split_at_every_byte_boundary() { + // CRLF 分隔符按单字节到达时必须仍被识别；帧内容跨任意多次 push 不丢失。 + let body = b"event: one\r\ndata: first\r\n\r\nevent: two\ndata: second\n\ndata: third\r\n\r\n"; + let mut decoder = SseEventDecoder::default(); + let mut events = Vec::new(); + for byte in body { + events.extend(decoder.push(&[*byte]).expect("single-byte chunk")); + } + + assert_eq!( + events + .iter() + .map(|event| (event.event.as_deref(), event.data.as_str())) + .collect::>(), + vec![ + (Some("one"), "first"), + (Some("two"), "second"), + (None, "third"), + ] + ); +} + #[test] fn raw_frame_decoder_should_preserve_exact_frame_across_chunk_boundaries() { let raw = b": keep-alive\r\nid: evt_1\r\nevent: response.future\r\nretry: 250\r\ndata: { \"type\": \"response.future\", \"opaque\": true }\r\n\r\n";