perf(protocol,api): incremental SSE scanning and map-based frame decode

Track a scanned-prefix cursor in SseEventDecoder so a large frame arriving across many chunks is scanned incrementally instead of rescanned from the start on every push (separator matching backs off three bytes to catch CRLF pairs split across chunks). Decode WebSocket response.create frames through a Map-accepting request-decode core, dropping the serialize-and-reparse round trip on the request start path.

Co-authored-by: Claude <noreply@anthropic.com>

backend/crates/gateway-api/src/openai/responses/request.rs

@@ -310,9 +310,18 @@ pub(super) fn decode_request_inner(
 ) -> Result<DecodedResponsesRequest, RequestDecodeError> {
     let value =
         serde_json::from_slice::<Value>(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<String, Value>,
+    review: bool,
+    request_headers: &OpenAiRequestHeaders,
+) -> Result<DecodedResponsesRequest, RequestDecodeError> {
     // `use_websocket` 只影响本地 transport。无论值是否可识别，都不得进入上游 body。
     let use_websocket = object
         .remove("use_websocket")

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)
 }
 

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<u8>,
+    /// `pending` 中已确认不含帧分隔符的前缀长度。新数据到达时从该边界
+    /// 回退分隔符最大跨界宽度继续扫描，大帧跨多个 chunk 不会被反复重扫。
+    scanned: usize,
 }
 
 impl SseEventDecoder {
+    /// 返回从 `consumed` 起的下一个完整帧结束位置（含分隔符），并推进扫描边界。
+    fn next_frame_end(&mut self, consumed: usize) -> Option<usize> {
+        // `\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<Vec<SseEvent>, 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<Vec<SseEvent>, 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<Vec<SseFrame>, SseError> {
+        self.scanned = 0;
         if self.pending.is_empty() {
             return Ok(Vec::new());
         }

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<_>>(),
+        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";