fix(gateway): support compact SSE format in Anthropic /responses bridge

Fixes #4653 ## Problem The /v1/responses endpoint (Responses → Anthropic bridge) failed when upstream providers returned SSE events in compact format without spaces after colons (e.g. 'event:message_start' instead of 'event: message_start'). This affected Kimi API (api.kimi.com) and potentially other upstreams, causing 'Upstream stream ended without a response' errors. ## Root Cause The SSE parser in gateway_forward_as_responses.go used strict prefix matching with hardcoded spaces: if !strings.HasPrefix(line, "event: ") { ... } if !strings.HasPrefix(dataLine, "data: ") { ... } According to the SSE spec (https://html.spec.whatwg.org/multipage/server-sent-events.html), the space after the colon is optional. Both formats are valid: event: message_start ✓ event:message_start ✓ ## Solution Introduced parseAnthropicSSEField() helper that: - Accepts both 'field: value' and 'field:value' formats - Uses TrimSpace() to handle optional/multiple spaces - Returns (value, ok) for consistent error handling Applied to both streaming handlers: - handleResponsesBufferedStreamingResponse - handleResponsesStreamingResponse ## Testing Added comprehensive unit tests: - TestParseAnthropicSSEField: validates parser with 8 edge cases - TestHandleResponsesBufferedStreamingResponse_CompactSSEFormat - TestHandleResponsesStreamingResponse_CompactSSEFormat All existing tests pass, confirming backward compatibility.
2026-10-07 17:08:33 +08:00 · 2026-07-20 20:40:08 +08:00
parent e625ce3b3b
commit e5ad287843
2 changed files with 162 additions and 8 deletions
@@ -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 {
@@ -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`)
+}