diff --git a/backend/internal/service/gateway_forward_as_responses.go b/backend/internal/service/gateway_forward_as_responses.go index 774c903a5..e5f0ddc3c 100644 --- a/backend/internal/service/gateway_forward_as_responses.go +++ b/backend/internal/service/gateway_forward_as_responses.go @@ -222,6 +222,17 @@ func mergeAnthropicUsage(dst *ClaudeUsage, src apicompat.AnthropicUsage) { } } +// parseAnthropicSSEField parses an SSE field line in the form "field:value" or "field: value". +// According to the SSE spec (https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation), +// the space after the colon is optional. This function handles both formats. +func parseAnthropicSSEField(line, field string) (string, bool) { + prefix := field + ":" + if !strings.HasPrefix(line, prefix) { + return "", false + } + return strings.TrimSpace(strings.TrimPrefix(line, prefix)), true +} + // handleResponsesBufferedStreamingResponse reads all Anthropic SSE events from // the upstream streaming response, assembles them into a complete Anthropic // response, converts to Responses API JSON format, and writes it to the client. @@ -248,20 +259,20 @@ func (s *GatewayService) handleResponsesBufferedStreamingResponse( for scanner.Scan() { line := scanner.Text() - if !strings.HasPrefix(line, "event: ") { + eventType, ok := parseAnthropicSSEField(line, "event") + if !ok { continue } - eventType := strings.TrimPrefix(line, "event: ") // Read the data line if !scanner.Scan() { break } dataLine := scanner.Text() - if !strings.HasPrefix(dataLine, "data: ") { + payload, ok := parseAnthropicSSEField(dataLine, "data") + if !ok { continue } - payload := dataLine[6:] var event apicompat.AnthropicStreamEvent if err := json.Unmarshal([]byte(payload), &event); err != nil { @@ -470,20 +481,20 @@ func (s *GatewayService) handleResponsesStreamingResponse( // Read Anthropic SSE events for scanner.Scan() { line := scanner.Text() - if !strings.HasPrefix(line, "event: ") { + eventType, ok := parseAnthropicSSEField(line, "event") + if !ok { continue } - eventType := strings.TrimPrefix(line, "event: ") // Read data line if !scanner.Scan() { break } dataLine := scanner.Text() - if !strings.HasPrefix(dataLine, "data: ") { + payload, ok := parseAnthropicSSEField(dataLine, "data") + 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_responses_test.go b/backend/internal/service/gateway_forward_as_responses_test.go index 03b18a935..baaf8b954 100644 --- a/backend/internal/service/gateway_forward_as_responses_test.go +++ b/backend/internal/service/gateway_forward_as_responses_test.go @@ -96,3 +96,146 @@ func TestHandleResponsesStreamingResponse_PreservesMessageStartCacheUsage(t *tes require.Equal(t, 4, result.Usage.CacheCreationInputTokens) require.Contains(t, rec.Body.String(), `response.completed`) } + +func TestParseAnthropicSSEField(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + line string + field string + wantValue string + wantOK bool + }{ + { + name: "standard format with space", + line: "event: message_start", + field: "event", + wantValue: "message_start", + wantOK: true, + }, + { + name: "compact format without space", + line: "event:message_start", + field: "event", + wantValue: "message_start", + wantOK: true, + }, + { + name: "data field with space", + line: "data: {\"type\":\"message_start\"}", + field: "data", + wantValue: "{\"type\":\"message_start\"}", + wantOK: true, + }, + { + name: "data field without space", + line: "data:{\"type\":\"message_start\"}", + field: "data", + wantValue: "{\"type\":\"message_start\"}", + wantOK: true, + }, + { + name: "field with multiple spaces after colon", + line: "event: message_delta", + field: "event", + wantValue: "message_delta", + wantOK: true, + }, + { + name: "wrong field name", + line: "event: message_start", + field: "data", + wantValue: "", + wantOK: false, + }, + { + name: "empty line", + line: "", + field: "event", + wantValue: "", + wantOK: false, + }, + { + name: "line without colon", + line: "invalid line", + field: "event", + wantValue: "", + wantOK: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + gotValue, gotOK := parseAnthropicSSEField(tt.line, tt.field) + require.Equal(t, tt.wantOK, gotOK, "parseAnthropicSSEField() ok") + require.Equal(t, tt.wantValue, gotValue, "parseAnthropicSSEField() value") + }) + } +} + +func TestHandleResponsesBufferedStreamingResponse_CompactSSEFormat(t *testing.T) { + t.Parallel() + gin.SetMode(gin.TestMode) + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + + // Simulate compact SSE format without spaces after colons (e.g. Kimi API) + resp := &http.Response{ + Header: http.Header{"x-request-id": []string{"rid_compact"}}, + Body: io.NopCloser(strings.NewReader(strings.Join([]string{ + `event:message_start`, + `data:{"type":"message_start","message":{"id":"msg_compact","type":"message","role":"assistant","content":[],"model":"claude-sonnet-4.5","stop_reason":"","usage":{"input_tokens":10}}}`, + ``, + `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":5}}`, + ``, + }, "\n"))), + } + + svc := &GatewayService{} + result, err := svc.handleResponsesBufferedStreamingResponse(resp, c, "claude-sonnet-4.5", "claude-sonnet-4.5", nil, time.Now()) + require.NoError(t, err) + require.NotNil(t, result) + require.Equal(t, 10, result.Usage.InputTokens) + require.Equal(t, 5, result.Usage.OutputTokens) +} + +func TestHandleResponsesStreamingResponse_CompactSSEFormat(t *testing.T) { + t.Parallel() + gin.SetMode(gin.TestMode) + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + + // Simulate compact SSE format without spaces after colons (e.g. Kimi API) + resp := &http.Response{ + Header: http.Header{"x-request-id": []string{"rid_compact_stream"}}, + Body: io.NopCloser(strings.NewReader(strings.Join([]string{ + `event:message_start`, + `data:{"type":"message_start","message":{"id":"msg_compact_stream","type":"message","role":"assistant","content":[],"model":"claude-sonnet-4.5","stop_reason":"","usage":{"input_tokens":15}}}`, + ``, + `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":6}}`, + ``, + `event:message_stop`, + `data:{"type":"message_stop"}`, + ``, + }, "\n"))), + } + + svc := &GatewayService{} + result, err := svc.handleResponsesStreamingResponse(resp, c, "claude-sonnet-4.5", "claude-sonnet-4.5", nil, time.Now()) + require.NoError(t, err) + require.NotNil(t, result) + require.Equal(t, 15, result.Usage.InputTokens) + require.Equal(t, 6, result.Usage.OutputTokens) + require.Contains(t, rec.Body.String(), `response.completed`) +}