From c405398a48df32c7a8ceb831124300d9b52c6c5d Mon Sep 17 00:00:00 2001 From: Randi <55005611+rdself@users.noreply.github.com> Date: Mon, 27 Jul 2026 19:07:57 -0400 Subject: [PATCH] fix: align Claude headers with upstream streaming --- .../executor/claude_executor_execute.go | 16 +-- .../executor/claude_executor_request.go | 6 +- .../runtime/executor/claude_executor_test.go | 99 +++++++++++++++++++ 3 files changed, 112 insertions(+), 9 deletions(-) diff --git a/internal/runtime/executor/claude_executor_execute.go b/internal/runtime/executor/claude_executor_execute.go index da100536..0255a03a 100644 --- a/internal/runtime/executor/claude_executor_execute.go +++ b/internal/runtime/executor/claude_executor_execute.go @@ -32,15 +32,16 @@ func (e *ClaudeExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, r from := opts.SourceFormat responseFormat := cliproxyexecutor.ResponseFormatOrSource(opts) to := sdktranslator.FromString("claude") - // Use streaming translation to preserve function calling, except for claude. - stream := from != to + // Use an upstream stream whenever the downstream response needs translation + // from Claude events. Native Claude responses use the JSON response path. + upstreamStream := responseFormat != to originalPayloadSource := req.Payload if len(opts.OriginalRequest) > 0 { originalPayloadSource = opts.OriginalRequest } originalPayload := originalPayloadSource - originalTranslated := helps.TranslateRequestWithCodexMultiAgentV2(ctx, opts.Headers, e.cfg, from, to, baseModel, originalPayload, stream) - body := helps.TranslateRequestWithCodexMultiAgentV2(ctx, opts.Headers, e.cfg, from, to, baseModel, req.Payload, stream) + originalTranslated := helps.TranslateRequestWithCodexMultiAgentV2(ctx, opts.Headers, e.cfg, from, to, baseModel, originalPayload, upstreamStream) + body := helps.TranslateRequestWithCodexMultiAgentV2(ctx, opts.Headers, e.cfg, from, to, baseModel, req.Payload, upstreamStream) body = helps.SetStringIfDifferent(body, "model", upstreamModel) body, err = thinking.ApplyThinking(body, req.Model, from.String(), to.String(), e.Identifier()) @@ -83,6 +84,9 @@ func (e *ClaudeExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, r // Normalize TTL values to prevent ordering violations under prompt-caching-scope-2026-01-05. // A 1h-TTL block must not appear after a 5m-TTL block in evaluation order (tools→system→messages). body = normalizeCacheControlTTL(body) + // Payload rules and other request processing may rewrite stream. Keep the + // upstream body, transport headers, and response parser on one authority. + body = helps.SetBoolIfDifferent(body, "stream", upstreamStream) // Extract betas from body and convert to header var extraBetas []string @@ -107,7 +111,7 @@ func (e *ClaudeExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, r if err != nil { return resp, err } - if errHeaders := applyClaudeHeaders(httpReq, auth, apiKey, false, extraBetas, e.cfg, opts.Headers); errHeaders != nil { + if errHeaders := applyClaudeHeaders(httpReq, auth, apiKey, upstreamStream, extraBetas, e.cfg, opts.Headers); errHeaders != nil { return resp, errHeaders } var authID, authLabel, authType, authValue string @@ -181,7 +185,7 @@ func (e *ClaudeExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, r return resp, err } helps.AppendAPIResponseChunk(ctx, e.cfg, data) - if stream { + if upstreamStream { if errValidate := validateClaudeStreamingResponse(data); errValidate != nil { helps.RecordAPIResponseError(ctx, e.cfg, errValidate) return resp, errValidate diff --git a/internal/runtime/executor/claude_executor_request.go b/internal/runtime/executor/claude_executor_request.go index 847132e9..86874440 100644 --- a/internal/runtime/executor/claude_executor_request.go +++ b/internal/runtime/executor/claude_executor_request.go @@ -344,10 +344,10 @@ func applyClaudeHeaders(r *http.Request, auth *cliproxyauth.Auth, apiKey string, attrs = auth.Attributes } util.ApplyCustomHeadersFromAttrs(r, attrs) - // Re-enforce Accept-Encoding: identity after ApplyCustomHeadersFromAttrs, which - // may override it with a user-configured value. Compressed SSE breaks the line - // scanner regardless of user preference, so this is non-negotiable for streams. + // Re-enforce the SSE transport contract after custom headers. A custom Accept + // value can disable event negotiation, while compressed SSE breaks line parsing. if stream { + r.Header.Set("Accept", "text/event-stream") r.Header.Set("Accept-Encoding", "identity") } return nil diff --git a/internal/runtime/executor/claude_executor_test.go b/internal/runtime/executor/claude_executor_test.go index 9f6e9013..3e2946f7 100644 --- a/internal/runtime/executor/claude_executor_test.go +++ b/internal/runtime/executor/claude_executor_test.go @@ -1622,6 +1622,105 @@ func TestClaudeExecutor_ExecuteOpenAINonStreamConvertsValidClaudeStream(t *testi } } +func TestClaudeExecutor_ExecuteTransportMatchesResponseFormat(t *testing.T) { + const model = "claude-3-5-sonnet-20241022" + streamResponse := strings.Join([]string{ + `event: message_start`, + `data: {"type":"message_start","message":{"id":"msg_123","model":"claude-3-5-sonnet-20241022"}}`, + `event: content_block_delta`, + `data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"ok"}}`, + `event: message_delta`, + `data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"input_tokens":2,"output_tokens":1}}`, + `event: message_stop`, + `data: {"type":"message_stop"}`, + ``, + }, "\n") + jsonResponse := `{"id":"msg_123","type":"message","role":"assistant","model":"claude-3-5-sonnet-20241022","content":[{"type":"text","text":"ok"}],"stop_reason":"end_turn","stop_sequence":null,"usage":{"input_tokens":2,"output_tokens":1}}` + + tests := []struct { + name string + sourceFormat sdktranslator.Format + responseFormat sdktranslator.Format + wantStream bool + }{ + {name: "OpenAI to OpenAI uses SSE", sourceFormat: sdktranslator.FormatOpenAI, responseFormat: sdktranslator.FormatOpenAI, wantStream: true}, + {name: "OpenAI to Claude uses JSON", sourceFormat: sdktranslator.FormatOpenAI, responseFormat: sdktranslator.FormatClaude, wantStream: false}, + {name: "Claude to OpenAI uses SSE", sourceFormat: sdktranslator.FormatClaude, responseFormat: sdktranslator.FormatOpenAI, wantStream: true}, + {name: "Claude to Claude uses JSON", sourceFormat: sdktranslator.FormatClaude, responseFormat: sdktranslator.FormatClaude, wantStream: false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var seenBody []byte + var seenHeaders http.Header + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + seenBody, _ = io.ReadAll(r.Body) + seenHeaders = r.Header.Clone() + if tt.wantStream { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte(streamResponse)) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(jsonResponse)) + })) + defer server.Close() + + executor := NewClaudeExecutor(&config.Config{ + Payload: config.PayloadConfig{ + Override: []config.PayloadRule{{ + Models: []config.PayloadModelRule{{Name: model, Protocol: "claude"}}, + Params: map[string]any{"stream": !tt.wantStream}, + }}, + }, + }) + attributes := map[string]string{ + "api_key": "key-123", + "base_url": server.URL, + } + if tt.wantStream { + attributes["header:Accept"] = "application/json" + attributes["header:Accept-Encoding"] = "gzip, deflate, br, zstd" + } + auth := &cliproxyauth.Auth{Attributes: attributes} + payload := []byte(`{"model":"claude-3-5-sonnet-20241022","stream":false,"messages":[{"role":"user","content":"hi"}]}`) + + _, err := executor.Execute(context.Background(), auth, cliproxyexecutor.Request{ + Model: model, + Payload: payload, + }, cliproxyexecutor.Options{ + SourceFormat: tt.sourceFormat, + ResponseFormat: tt.responseFormat, + Headers: http.Header{ + "Anthropic-Beta": []string{"client-beta"}, + }, + }) + if err != nil { + t.Fatalf("Execute error: %v", err) + } + stream := gjson.GetBytes(seenBody, "stream") + if !stream.Exists() || stream.Bool() != tt.wantStream { + t.Fatalf("upstream stream = %s, want %t; body=%s", stream.Raw, tt.wantStream, string(seenBody)) + } + wantAccept := "application/json" + wantEncoding := "gzip, deflate, br, zstd" + if tt.wantStream { + wantAccept = "text/event-stream" + wantEncoding = "identity" + } + if got := seenHeaders.Get("Accept"); got != wantAccept { + t.Fatalf("Accept = %q, want %q", got, wantAccept) + } + if got := seenHeaders.Get("Accept-Encoding"); got != wantEncoding { + t.Fatalf("Accept-Encoding = %q, want %q", got, wantEncoding) + } + if got := seenHeaders.Get("Anthropic-Beta"); !strings.Contains(got, "client-beta") { + t.Fatalf("Anthropic-Beta = %q, want client beta preserved", got) + } + }) + } +} + func executeOpenAIChatCompletionThroughClaude(t *testing.T, upstreamBody string) (cliproxyexecutor.Response, error) { t.Helper() -- 2.51.2