diff --git a/backend/internal/service/gateway_forward_as_chat_completions.go b/backend/internal/service/gateway_forward_as_chat_completions.go index 5a542ae3a1..93da85a7b2 100644 --- a/backend/internal/service/gateway_forward_as_chat_completions.go +++ b/backend/internal/service/gateway_forward_as_chat_completions.go @@ -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 { diff --git a/backend/internal/service/gateway_forward_as_chat_completions_test.go b/backend/internal/service/gateway_forward_as_chat_completions_test.go index 2cd10b7914..d1047bdc9b 100644 --- a/backend/internal/service/gateway_forward_as_chat_completions_test.go +++ b/backend/internal/service/gateway_forward_as_chat_completions_test.go @@ -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)