fix: use Content-Type header to select response parser

Address review feedback from @shmuelk: instead of trying JSON parse
and falling back to SSE on failure, check the Content-Type response
header upfront to select the correct parser.

- text/event-stream → SSE parser (parseSSEResponseBody)
- anything else → JSON parser (json.Unmarshal)

Also fix streaming tests:
- JSON body tests now use content-type: application/json (not text/event-stream)
- Tests receive response header ack before sending body chunks
- Tests receive chunk acks for non-final streaming chunks

Author: noyitz

pkg/handlers/response.go

@@ -22,6 +22,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"strconv"
+	"strings"
 	"time"
 
 	eppb "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
@@ -76,14 +77,20 @@ func (s *Server) HandleResponseBody(ctx context.Context, reqCtx *RequestContext,
 		return s.generateEmptyResponseBodyResponse(responseBodyBytes), nil
 	}
 
-	if err := json.Unmarshal(responseBodyBytes, &reqCtx.Response.Body); err != nil {
-		// Try parsing as SSE (Server-Sent Events) — streaming responses from providers
-		// like Anthropic use SSE format which isn't valid JSON.
-		if sseBody, sseErr := parseSSEResponseBody(responseBodyBytes); sseErr == nil && sseBody != nil {
-			reqCtx.Response.Body = sseBody
-			logger.V(logutil.VERBOSE).Info("parsed SSE response body for response plugins")
-		} else {
-			logger.Error(err, "Failed to parse response body as JSON or SSE, skipping response plugins")
+	contentType := reqCtx.Response.Headers["content-type"]
+	isSSE := strings.Contains(contentType, "text/event-stream")
+
+	if isSSE {
+		sseBody, err := parseSSEResponseBody(responseBodyBytes)
+		if err != nil || sseBody == nil {
+			logger.Error(err, "failed to parse SSE response body, skipping response plugins")
+			return s.generateEmptyResponseBodyResponse(responseBodyBytes), nil
+		}
+		reqCtx.Response.Body = sseBody
+		logger.V(logutil.VERBOSE).Info("parsed SSE response body for response plugins")
+	} else {
+		if err := json.Unmarshal(responseBodyBytes, &reqCtx.Response.Body); err != nil {
+			logger.Error(err, "failed to parse response body as JSON, skipping response plugins")
 			return s.generateEmptyResponseBodyResponse(responseBodyBytes), nil
 		}
 	}

pkg/handlers/server_test.go

@@ -210,7 +210,7 @@ func TestHandleResponseBody_Streaming(t *testing.T) {
 			respHeaders := utils.BuildEnvoyGRPCHeaders(map[string]string{
 				"x-test":       "body",
 				":method":      "POST",
-				"content-type": "text/event-stream",
+				"content-type": "application/json",
 			}, true)
 			request = &extProcPb.ProcessingRequest{
 				Request: &extProcPb.ProcessingRequest_ResponseHeaders{
@@ -221,6 +221,12 @@ func TestHandleResponseBody_Streaming(t *testing.T) {
 				t.Fatalf("send response headers: %v", err)
 			}
 
+			// Receive the response header ack before sending body chunks.
+			// HandleResponseHeaders always responds immediately so Envoy proceeds.
+			if _, err := process.Recv(); err != nil {
+				t.Fatalf("recv response headers ack: %v", err)
+			}
+
 			for _, c := range tc.chunks {
 				request = &extProcPb.ProcessingRequest{
 					Request: &extProcPb.ProcessingRequest_ResponseBody{
@@ -233,6 +239,12 @@ func TestHandleResponseBody_Streaming(t *testing.T) {
 				if err := process.Send(request); err != nil {
 					t.Fatalf("send response body chunk: %v", err)
 				}
+				// For non-final chunks, receive the streaming ack
+				if !c.endOfStream {
+					if _, err := process.Recv(); err != nil {
+						t.Fatalf("recv chunk ack: %v", err)
+					}
+				}
 			}
 
 			got := make([]*extProcPb.ProcessingResponse, 0, len(want))