fix: gcp anthropic streaming usage (envoyproxy#1187)

**Description**
Usage event were parsed as raw json instead of event type + json.
Example of stream chunks:

```
event: message_start
data: {"type":"message_start","message":{"id":"msg_vrtx_01V4HwqjNGwvFRqBVuhQNCXW","type":"message","role":"assistant","model":"claude-sonnet-4-20250514","content":[],"stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":20,"cache_creation_input_tokens":0,"cache_read_input_tokens":0,"output_tokens":1}}  }

event: ping
data: {"type": "ping"}

event: content_block_start
data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"There"}}

```

---------

Signed-off-by: Suren Vartanian <[email protected]>
Signed-off-by: Dan Sun <[email protected]>
Co-authored-by: Suren Vartanian <[email protected]>
Co-authored-by: Dan Sun <[email protected]>
Co-authored-by: Takeshi Yoneda <[email protected]>

Author: 4 people

internal/extproc/translator/anthropic_gcpanthropic.go

@@ -6,6 +6,7 @@
 package translator
 
 import (
+	"bytes"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -95,18 +96,44 @@ func (a *anthropicToGCPAnthropicTranslator) ResponseBody(_ map[string]string, bo
 		return nil, nil, LLMTokenUsage{}, fmt.Errorf("failed to read response body: %w", err)
 	}
 
-	// For streaming chunks, try to extract token usage from message_delta events.
+	// For streaming chunks, parse SSE format to extract token usage.
 	if !endOfStream {
-		// Try to parse as a message_delta event to extract usage.
-		var eventData map[string]any
-		if unmarshalErr := json.Unmarshal(bodyBytes, &eventData); unmarshalErr == nil {
-			if eventType, ok := eventData["type"].(string); ok && eventType == "message_delta" {
-				if usageData, ok := eventData["usage"].(map[string]any); ok {
-					// Extract token usage from the message_delta event.
-					if outputTokens, ok := usageData["output_tokens"].(float64); ok {
-						tokenUsage = LLMTokenUsage{
-							OutputTokens: uint32(outputTokens), //nolint:gosec
-							TotalTokens:  uint32(outputTokens), //nolint:gosec // Only output tokens available in streaming
+		// Parse SSE format - split by lines and look for data: lines.
+		for line := range bytes.Lines(bodyBytes) {
+			line = bytes.TrimSpace(line)
+			dataPrefix := []byte("data: ")
+			if bytes.HasPrefix(line, dataPrefix) {
+				jsonData := bytes.TrimPrefix(line, dataPrefix)
+
+				var eventData map[string]any
+				if unmarshalErr := json.Unmarshal(jsonData, &eventData); unmarshalErr != nil {
+					// Skip lines with invalid JSON (like ping events or malformed data).
+					continue
+				}
+				if eventType, ok := eventData["type"].(string); ok {
+					switch eventType {
+					case "message_start":
+						// Extract input tokens from message.usage.
+						if messageData, ok := eventData["message"].(map[string]any); ok {
+							if usageData, ok := messageData["usage"].(map[string]any); ok {
+								if inputTokens, ok := usageData["input_tokens"].(float64); ok {
+									tokenUsage.InputTokens = uint32(inputTokens) //nolint:gosec
+								}
+								// Some message_start events may include initial output tokens.
+								if outputTokens, ok := usageData["output_tokens"].(float64); ok && outputTokens > 0 {
+									tokenUsage.OutputTokens = uint32(outputTokens) //nolint:gosec
+								}
+								tokenUsage.TotalTokens = tokenUsage.InputTokens + tokenUsage.OutputTokens
+							}
+						}
+
+					case "message_delta":
+						if usageData, ok := eventData["usage"].(map[string]any); ok {
+							if outputTokens, ok := usageData["output_tokens"].(float64); ok {
+								// Add to existing output tokens (in case message_start had some initial ones).
+								tokenUsage.OutputTokens += uint32(outputTokens) //nolint:gosec
+								tokenUsage.TotalTokens = tokenUsage.InputTokens + tokenUsage.OutputTokens
+							}
 						}
 					}
 				}

internal/extproc/translator/anthropic_gcpanthropic_test.go

@@ -509,58 +509,58 @@ func TestAnthropicToGCPAnthropicTranslator_ResponseBody_StreamingTokenUsage(t *t
 	}{
 		{
 			name:        "regular streaming chunk without usage",
-			chunk:       `{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":" to me."}}`,
+			chunk:       "event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\" to me.\"}}\n\n",
 			endOfStream: false,
 			expectedUsage: LLMTokenUsage{
 				InputTokens:  0,
 				OutputTokens: 0,
 				TotalTokens:  0,
 			},
-			expectedBody: `{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":" to me."}}`,
+			expectedBody: "event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\" to me.\"}}\n\n",
 		},
 		{
 			name:        "message_delta chunk with token usage",
-			chunk:       `{"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":84}}`,
+			chunk:       "event: message_delta\ndata: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\",\"stop_sequence\":null},\"usage\":{\"output_tokens\":84}}\n\n",
 			endOfStream: false,
 			expectedUsage: LLMTokenUsage{
 				InputTokens:  0,
 				OutputTokens: 84,
 				TotalTokens:  84,
 			},
-			expectedBody: `{"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":84}}`,
+			expectedBody: "event: message_delta\ndata: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\",\"stop_sequence\":null},\"usage\":{\"output_tokens\":84}}\n\n",
 		},
 		{
 			name:        "message_stop chunk without usage",
-			chunk:       `{"type":"message_stop"}`,
+			chunk:       "event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n",
 			endOfStream: false,
 			expectedUsage: LLMTokenUsage{
 				InputTokens:  0,
 				OutputTokens: 0,
 				TotalTokens:  0,
 			},
-			expectedBody: `{"type":"message_stop"}`,
+			expectedBody: "event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n",
 		},
 		{
 			name:        "invalid json chunk",
-			chunk:       `{"invalid": "json"}`,
+			chunk:       "event: invalid\ndata: {\"invalid\": \"json\"}\n\n",
 			endOfStream: false,
 			expectedUsage: LLMTokenUsage{
 				InputTokens:  0,
 				OutputTokens: 0,
 				TotalTokens:  0,
 			},
-			expectedBody: `{"invalid": "json"}`,
+			expectedBody: "event: invalid\ndata: {\"invalid\": \"json\"}\n\n",
 		},
 		{
 			name:        "message_delta with decimal output_tokens",
-			chunk:       `{"type":"message_delta","delta":{"stop_reason":"tool_use"},"usage":{"output_tokens":42.0}}`,
+			chunk:       "event: message_delta\ndata: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"tool_use\"},\"usage\":{\"output_tokens\":42.0}}\n\n",
 			endOfStream: false,
 			expectedUsage: LLMTokenUsage{
 				InputTokens:  0,
 				OutputTokens: 42,
 				TotalTokens:  42,
 			},
-			expectedBody: `{"type":"message_delta","delta":{"stop_reason":"tool_use"},"usage":{"output_tokens":42.0}}`,
+			expectedBody: "event: message_delta\ndata: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"tool_use\"},\"usage\":{\"output_tokens\":42.0}}\n\n",
 		},
 	}
 
@@ -574,7 +574,87 @@ func TestAnthropicToGCPAnthropicTranslator_ResponseBody_StreamingTokenUsage(t *t
 			require.NoError(t, err)
 			require.Nil(t, headerMutation)
 			require.NotNil(t, bodyMutation)
-			require.JSONEq(t, tt.expectedBody, string(bodyMutation.GetBody()))
+			require.Equal(t, tt.expectedBody, string(bodyMutation.GetBody()))
+			require.Equal(t, tt.expectedUsage, tokenUsage)
+		})
+	}
+}
+
+func TestAnthropicToGCPAnthropicTranslator_ResponseBody_StreamingEdgeCases(t *testing.T) {
+	translator := NewAnthropicToGCPAnthropicTranslator("2023-06-01", "")
+
+	tests := []struct {
+		name          string
+		chunk         string
+		expectedUsage LLMTokenUsage
+	}{
+		{
+			name:  "message_start without message field",
+			chunk: "event: message_start\ndata: {\"type\":\"message_start\"}\n\n",
+			expectedUsage: LLMTokenUsage{
+				InputTokens:  0,
+				OutputTokens: 0,
+				TotalTokens:  0,
+			},
+		},
+		{
+			name:  "message_start without usage field",
+			chunk: "event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_123\"}}\n\n",
+			expectedUsage: LLMTokenUsage{
+				InputTokens:  0,
+				OutputTokens: 0,
+				TotalTokens:  0,
+			},
+		},
+		{
+			name:  "message_start with output_tokens = 0",
+			chunk: "event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"usage\":{\"input_tokens\":20,\"output_tokens\":0}}}\n\n",
+			expectedUsage: LLMTokenUsage{
+				InputTokens:  20,
+				OutputTokens: 0,
+				TotalTokens:  20,
+			},
+		},
+		{
+			name:  "message_start with output_tokens > 0",
+			chunk: "event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"usage\":{\"input_tokens\":15,\"output_tokens\":5}}}\n\n",
+			expectedUsage: LLMTokenUsage{
+				InputTokens:  15,
+				OutputTokens: 5,
+				TotalTokens:  20,
+			},
+		},
+		{
+			name:  "message_delta without usage field",
+			chunk: "event: message_delta\ndata: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\"}}\n\n",
+			expectedUsage: LLMTokenUsage{
+				InputTokens:  0,
+				OutputTokens: 0,
+				TotalTokens:  0,
+			},
+		},
+		{
+			name:  "invalid json in data",
+			chunk: "event: message_start\ndata: {invalid json}\n\n",
+			expectedUsage: LLMTokenUsage{
+				InputTokens:  0,
+				OutputTokens: 0,
+				TotalTokens:  0,
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			bodyReader := bytes.NewReader([]byte(tt.chunk))
+			respHeaders := map[string]string{"content-type": "application/json"}
+
+			headerMutation, bodyMutation, tokenUsage, err := translator.ResponseBody(respHeaders, bodyReader, false)
+
+			require.NoError(t, err)
+			require.Nil(t, headerMutation)
+			require.NotNil(t, bodyMutation)
+			require.Equal(t, tt.chunk, string(bodyMutation.GetBody()))
 			require.Equal(t, tt.expectedUsage, tokenUsage)
 		})
 	}