fix: strip stream_options from request body before parsing

OpenAI-compatible SDKs add stream_options to streaming requests, but
providers like Anthropic reject it with 400. In FULL_DUPLEX_STREAMED
mode, body mutations via StreamedBodyResponse cannot reliably remove
fields from chunks already forwarded by envoy.

Strip stream_options from accumulated raw bytes before json.Unmarshal
using JSON parse/delete/re-serialize — avoids false matches when the
text "stream_options" appears inside message string values.

Signed-off-by: Noy Itzikowitz <nitzikow@redhat.com>

Author: noyitz

pkg/handlers/request.go

@@ -64,6 +64,8 @@ func (s *Server) HandleRequestHeaders(ctx context.Context, reqCtx *RequestContex
 func (s *Server) HandleRequestBody(ctx context.Context, reqCtx *RequestContext, requestBodyBytes []byte) ([]*eppb.ProcessingResponse, error) {
 	var ret []*eppb.ProcessingResponse
 
+	requestBodyBytes = stripStreamOptions(requestBodyBytes)
+
 	if err := json.Unmarshal(requestBodyBytes, &reqCtx.Request.Body); err != nil {
 		return nil, errcommon.Error{Code: errcommon.BadRequest, Msg: fmt.Sprintf("failed to parse request body: %v", err)}
 	}
@@ -161,6 +163,28 @@ func addStreamedBodyResponse(responses []*eppb.ProcessingResponse, requestBodyBy
 	return responses
 }
 
+// stripStreamOptions removes the top-level "stream_options" key from JSON
+// request bodies. OpenAI SDKs add this field for streaming requests, but
+// providers like Anthropic reject it with 400. Body mutations via
+// StreamedBodyResponse cannot reliably remove fields already forwarded in
+// FULL_DUPLEX_STREAMED mode, so the strip runs on accumulated raw bytes
+// before json.Unmarshal.
+func stripStreamOptions(body []byte) []byte {
+	var parsed map[string]interface{}
+	if err := json.Unmarshal(body, &parsed); err != nil {
+		return body
+	}
+	if _, exists := parsed["stream_options"]; !exists {
+		return body
+	}
+	delete(parsed, "stream_options")
+	result, err := json.Marshal(parsed)
+	if err != nil {
+		return body
+	}
+	return result
+}
+
 // HandleRequestTrailers handles request trailers.
 func (s *Server) HandleRequestTrailers(trailers *eppb.HttpTrailers) ([]*eppb.ProcessingResponse, error) {
 	return []*eppb.ProcessingResponse{

pkg/handlers/request_test.go

@@ -886,6 +886,164 @@ func buildStreamingResponse(bodyBytes []byte, setHeaders map[string]string, remo
 	}
 }
 
+// === stripStreamOptions Tests ===
+
+func TestStripStreamOptions(t *testing.T) {
+	tests := []struct {
+		name     string
+		input    string
+		wantBody map[string]any
+	}{
+		{
+			name:  "removes top-level stream_options key",
+			input: `{"model":"claude-opus-4-8","messages":[{"role":"user","content":"hi"}],"max_tokens":5,"stream":true,"stream_options":{"include_usage":true}}`,
+			wantBody: map[string]any{
+				"model":      "claude-opus-4-8",
+				"messages":   []any{map[string]any{"role": "user", "content": "hi"}},
+				"max_tokens": float64(5),
+				"stream":     true,
+			},
+		},
+		{
+			name:  "no stream_options — body unchanged",
+			input: `{"model":"claude-opus-4-8","messages":[{"role":"user","content":"hi"}],"max_tokens":5}`,
+			wantBody: map[string]any{
+				"model":      "claude-opus-4-8",
+				"messages":   []any{map[string]any{"role": "user", "content": "hi"}},
+				"max_tokens": float64(5),
+			},
+		},
+		{
+			name:  "stream_options text inside message string is preserved",
+			input: `{"model":"claude-opus-4-8","messages":[{"role":"user","content":"please fix stream_options handling"}],"max_tokens":5,"stream_options":{"include_usage":true}}`,
+			wantBody: map[string]any{
+				"model":      "claude-opus-4-8",
+				"messages":   []any{map[string]any{"role": "user", "content": "please fix stream_options handling"}},
+				"max_tokens": float64(5),
+			},
+		},
+		{
+			name:  "only stream_options text inside string — no top-level key to remove",
+			input: `{"model":"test","messages":[{"role":"user","content":"the stream_options field should not appear"}],"max_tokens":5}`,
+			wantBody: map[string]any{
+				"model":      "test",
+				"messages":   []any{map[string]any{"role": "user", "content": "the stream_options field should not appear"}},
+				"max_tokens": float64(5),
+			},
+		},
+		{
+			name:  "large body with stream_options",
+			input: `{"model":"claude-opus-4-8","messages":[{"role":"user","content":"` + strings.Repeat("A", 50000) + `"}],"max_tokens":5,"stream":true,"stream_options":{"include_usage":true}}`,
+			wantBody: map[string]any{
+				"model":      "claude-opus-4-8",
+				"messages":   []any{map[string]any{"role": "user", "content": strings.Repeat("A", 50000)}},
+				"max_tokens": float64(5),
+				"stream":     true,
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			result := stripStreamOptions([]byte(tc.input))
+
+			var got map[string]any
+			if err := json.Unmarshal(result, &got); err != nil {
+				t.Fatalf("failed to unmarshal result: %v", err)
+			}
+
+			if _, exists := got["stream_options"]; exists {
+				t.Error("stream_options key still present after stripping")
+			}
+
+			wantBytes, _ := json.Marshal(tc.wantBody)
+			gotBytes, _ := json.Marshal(got)
+			if string(wantBytes) != string(gotBytes) {
+				t.Errorf("body mismatch:\nwant: %s\ngot:  %s", wantBytes, gotBytes)
+			}
+		})
+	}
+
+	t.Run("invalid JSON returns original bytes", func(t *testing.T) {
+		input := []byte(`not json at all`)
+		result := stripStreamOptions(input)
+		if string(result) != string(input) {
+			t.Errorf("expected original bytes for invalid JSON, got: %s", result)
+		}
+	})
+
+	t.Run("empty body returns original bytes", func(t *testing.T) {
+		result := stripStreamOptions([]byte{})
+		if len(result) != 0 {
+			t.Errorf("expected empty result for empty input, got: %s", result)
+		}
+	})
+}
+
+func TestHandleRequestBody_StripsStreamOptions(t *testing.T) {
+	metrics.Register()
+	ctx := logutil.NewTestLoggerIntoContext(context.Background())
+
+	tests := []struct {
+		name string
+		body string
+	}{
+		{
+			name: "small body with stream_options is stripped before plugin processing",
+			body: `{"model":"claude-opus-4-8","messages":[{"role":"user","content":"hi"}],"max_tokens":5,"stream":true,"stream_options":{"include_usage":true}}`,
+		},
+		{
+			name: "large body with stream_options (simulates multi-chunk request)",
+			body: `{"model":"claude-opus-4-8","messages":[{"role":"user","content":"` + strings.Repeat("X", 100000) + `"}],"max_tokens":5,"stream":true,"stream_options":{"include_usage":true}}`,
+		},
+		{
+			name: "body without stream_options passes through unchanged",
+			body: `{"model":"claude-opus-4-8","messages":[{"role":"user","content":"hi"}],"max_tokens":5}`,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			profiles := newTestProfiles()
+			server := newServerForTest(profiles)
+			reqCtx := &RequestContext{
+				CycleState: plugin.NewCycleState(),
+				Request:    requesthandling.NewInferenceRequest(),
+			}
+
+			resp, err := server.HandleRequestBody(ctx, reqCtx, []byte(tc.body))
+			if err != nil {
+				t.Fatalf("HandleRequestBody returned unexpected error: %v", err)
+			}
+
+			if _, exists := reqCtx.Request.Body["stream_options"]; exists {
+				t.Error("stream_options still present in parsed request body after HandleRequestBody")
+			}
+
+			for _, r := range resp {
+				if rb, ok := r.Response.(*extProcPb.ProcessingResponse_RequestBody); ok {
+					if rb.RequestBody != nil && rb.RequestBody.Response != nil {
+						if bm := rb.RequestBody.Response.BodyMutation; bm != nil {
+							if sr, ok := bm.Mutation.(*extProcPb.BodyMutation_StreamedResponse); ok {
+								var bodyMap map[string]any
+								if err := json.Unmarshal(sr.StreamedResponse.Body, &bodyMap); err == nil {
+									if _, exists := bodyMap["stream_options"]; exists {
+										t.Errorf("stream_options found in StreamedBodyResponse body chunk")
+									}
+								}
+							}
+						}
+					}
+				}
+			}
+
+			if len(resp) < 2 {
+				t.Fatalf("expected at least 2 responses (headers + body), got %d", len(resp))
+			}
+		})
+	}
+}
+
 // === Request Body Tests (body mutations) ===
 
 type bodyMutatingPlugin struct {