fix(gateway): CC 桥接受 Anthropic 上游的 SSE 紧凑格式

SSE 规范允许字段冒号后省略空格(event:message_start / data:{...})。
Kimi 等 Anthropic 兼容上游返回紧凑格式时,Anthropic → Chat Completions
桥按 "event: " / "data: " 严格匹配会丢弃全部事件,且 dataLine[6:] 按带
空格前缀硬切载荷,最终流式/缓冲两条路径都报 "Upstream stream ended
without a response"(与 #4653 同根因;#4657 只修了 /v1/responses 桥,
本补丁补齐对称的 /v1/chat/completions 桥)。

复用 OpenAI 链路既有的 extractOpenAISSEEventLine / extractOpenAISSEDataLine
(正确处理可选空格/tab),流式与缓冲两处统一替换;补充紧凑格式的
缓冲/流式回归测试,原带空格格式用例保持通过。
This commit is contained in:
superman2003 committed 2026-07-20 21:00:29 +08:00
1 parent e625ce3b3b
commit 2707f2793d
2 files changed
+81 -8

No files matched your search

@@ -240,18 +240,19 @@ func (s *GatewayService) handleCCBufferedFromAnthropic(
for scanner.Scan() {
line := scanner.Text()
if !strings.HasPrefix(line, "event: ") {
// SSE 规范允许 `event:xxx`(冒号后无空格):Kimi 等 Anthropic 兼容上游
// 返回紧凑格式,严格匹配 "event: " 会丢弃全部事件(#4653 同根因)。
if _, ok := extractOpenAISSEEventLine(line); !ok {
continue
}
if !scanner.Scan() {
break
}
dataLine := scanner.Text()
if !strings.HasPrefix(dataLine, "data: ") {
payload, ok := extractOpenAISSEDataLine(scanner.Text())
if !ok {
continue
}
payload := dataLine[6:]
var event apicompat.AnthropicStreamEvent
if err := json.Unmarshal([]byte(payload), &event); err != nil {
@@ -447,18 +448,18 @@ func (s *GatewayService) handleCCStreamingFromAnthropic(
for scanner.Scan() {
line := scanner.Text()
if !strings.HasPrefix(line, "event: ") {
// 与缓冲路径一致:接受 SSE 紧凑格式(冒号后无空格,#4653 同根因)。
if _, ok := extractOpenAISSEEventLine(line); !ok {
continue
}
if !scanner.Scan() {
break
}
dataLine := scanner.Text()
if !strings.HasPrefix(dataLine, "data: ") {
payload, ok := extractOpenAISSEDataLine(scanner.Text())
if !ok {
continue
}
payload := dataLine[6:]
var event apicompat.AnthropicStreamEvent
if err := json.Unmarshal([]byte(payload), &event); err != nil {
@@ -75,6 +75,78 @@ func TestHandleCCBufferedFromAnthropic_PreservesMessageStartCacheUsageAndReasoni
require.Equal(t, "high", *result.ReasoningEffort)
}
// Kimi 等 Anthropic 兼容上游返回 SSE 紧凑格式(冒号后无空格),CC 桥此前按
// "event: " / "data: " 严格匹配会丢弃全部事件,最终报 "Upstream stream ended
// without a response"(#4653 同根因;#4657 只修了 /v1/responses 桥)。
func TestHandleCCBufferedFromAnthropic_CompactSSEFormat(t *testing.T) {
t.Parallel()
gin.SetMode(gin.TestMode)
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
resp := &http.Response{
Header: http.Header{"x-request-id": []string{"rid_cc_buffered_compact"}},
Body: io.NopCloser(strings.NewReader(strings.Join([]string{
`event:message_start`,
`data:{"type":"message_start","message":{"id":"msg_c1","type":"message","role":"assistant","content":[],"model":"k3","stop_reason":"","usage":{"input_tokens":15,"cache_read_input_tokens":5,"cache_creation_input_tokens":2}}}`,
``,
`event:content_block_start`,
`data:{"type":"content_block_start","index":0,"content_block":{"type":"text","text":"OK"}}`,
``,
`event:message_delta`,
`data:{"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":3}}`,
``,
}, "\n"))),
}
svc := &GatewayService{}
result, err := svc.handleCCBufferedFromAnthropic(resp, c, "k3", "k3", nil, time.Now())
require.NoError(t, err)
require.NotNil(t, result)
require.Equal(t, 15, result.Usage.InputTokens)
require.Equal(t, 3, result.Usage.OutputTokens)
require.Equal(t, 5, result.Usage.CacheReadInputTokens)
require.Equal(t, 2, result.Usage.CacheCreationInputTokens)
require.Contains(t, rec.Body.String(), `"OK"`, "紧凑格式事件必须被解析并产出响应内容")
}
func TestHandleCCStreamingFromAnthropic_CompactSSEFormat(t *testing.T) {
t.Parallel()
gin.SetMode(gin.TestMode)
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
resp := &http.Response{
Header: http.Header{"x-request-id": []string{"rid_cc_stream_compact"}},
Body: io.NopCloser(strings.NewReader(strings.Join([]string{
`event:message_start`,
`data:{"type":"message_start","message":{"id":"msg_c2","type":"message","role":"assistant","content":[],"model":"k3","stop_reason":"","usage":{"input_tokens":21,"cache_read_input_tokens":6,"cache_creation_input_tokens":1}}}`,
``,
`event:content_block_start`,
`data:{"type":"content_block_start","index":0,"content_block":{"type":"text","text":"OK"}}`,
``,
`event:message_delta`,
`data:{"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":4}}`,
``,
`event:message_stop`,
`data:{"type":"message_stop"}`,
``,
}, "\n"))),
}
svc := &GatewayService{}
result, err := svc.handleCCStreamingFromAnthropic(resp, c, "k3", "k3", nil, time.Now(), true)
require.NoError(t, err)
require.NotNil(t, result)
require.Equal(t, 21, result.Usage.InputTokens)
require.Equal(t, 4, result.Usage.OutputTokens)
require.Equal(t, 6, result.Usage.CacheReadInputTokens)
require.Equal(t, 1, result.Usage.CacheCreationInputTokens)
require.Contains(t, rec.Body.String(), `[DONE]`)
}
func TestHandleCCStreamingFromAnthropic_PreservesMessageStartCacheUsageAndReasoning(t *testing.T) {
t.Parallel()
gin.SetMode(gin.TestMode)