diff --git a/backend/internal/pkg/apicompat/anthropic_responses_test.go b/backend/internal/pkg/apicompat/anthropic_responses_test.go index c13e9985a..539ccda0d 100644 --- a/backend/internal/pkg/apicompat/anthropic_responses_test.go +++ b/backend/internal/pkg/apicompat/anthropic_responses_test.go @@ -146,7 +146,7 @@ func TestAnthropicToResponses_ToolUse(t *testing.T) { assert.Equal(t, "Sunny, 72°F", items[3].Output) } -func TestAnthropicToResponses_ThinkingIgnored(t *testing.T) { +func TestAnthropicToResponses_ThinkingWithoutSignatureIgnored(t *testing.T) { req := &AnthropicRequest{ Model: "gpt-5.2", MaxTokens: 1024, @@ -162,7 +162,7 @@ func TestAnthropicToResponses_ThinkingIgnored(t *testing.T) { var items []ResponsesInputItem require.NoError(t, json.Unmarshal(resp.Input, &items)) - // user + assistant(text only, thinking ignored) + user = 3 + // user + assistant(text only, thinking without signature ignored) + user = 3 require.Len(t, items, 3) assert.Equal(t, "assistant", items[1].Role) // Assistant content should only have text, not thinking. @@ -173,6 +173,30 @@ func TestAnthropicToResponses_ThinkingIgnored(t *testing.T) { assert.Equal(t, "Hi!", parts[0].Text) } +func TestAnthropicToResponses_ThinkingSignatureBecomesReasoning(t *testing.T) { + req := &AnthropicRequest{ + Model: "grok-4.5", + MaxTokens: 1024, + Messages: []AnthropicMessage{ + {Role: "user", Content: json.RawMessage(`"Hello"`)}, + {Role: "assistant", Content: json.RawMessage(`[{"type":"thinking","thinking":"plan","signature":"enc-rs-1"},{"type":"text","text":"Hi!"},{"type":"tool_use","id":"toolu_1","name":"Bash","input":{"command":"ls"}}]`)}, + {Role: "user", Content: json.RawMessage(`[{"type":"tool_result","tool_use_id":"toolu_1","content":"ok"}]`)}, + }, + } + + resp, err := AnthropicToResponses(req) + require.NoError(t, err) + + var items []ResponsesInputItem + require.NoError(t, json.Unmarshal(resp.Input, &items)) + // user + reasoning + assistant text + function_call + function_call_output + require.GreaterOrEqual(t, len(items), 4) + assert.Equal(t, "reasoning", items[1].Type) + assert.Equal(t, "enc-rs-1", items[1].EncryptedContent) + assert.Equal(t, "assistant", items[2].Role) + assert.Equal(t, "function_call", items[3].Type) +} + func TestAnthropicToResponses_MaxTokensFloor(t *testing.T) { req := &AnthropicRequest{ Model: "gpt-5.2", @@ -372,7 +396,8 @@ func TestResponsesToAnthropic_Reasoning(t *testing.T) { Status: "completed", Output: []ResponsesOutput{ { - Type: "reasoning", + Type: "reasoning", + EncryptedContent: "enc-rs-roundtrip", Summary: []ResponsesSummary{ {Type: "summary_text", Text: "Thinking about the answer..."}, }, @@ -390,10 +415,57 @@ func TestResponsesToAnthropic_Reasoning(t *testing.T) { require.Len(t, anth.Content, 2) assert.Equal(t, "thinking", anth.Content[0].Type) assert.Equal(t, "Thinking about the answer...", anth.Content[0].Thinking) + assert.Equal(t, "enc-rs-roundtrip", anth.Content[0].Signature) assert.Equal(t, "text", anth.Content[1].Type) assert.Equal(t, "42", anth.Content[1].Text) } +func TestResponsesToAnthropic_StreamEmitsThinkingSignature(t *testing.T) { + state := NewResponsesEventToAnthropicState() + var all []AnthropicStreamEvent + + appendAll := func(events []AnthropicStreamEvent) { + all = append(all, events...) + } + + appendAll(ResponsesEventToAnthropicEvents(&ResponsesStreamEvent{ + Type: "response.output_item.added", + OutputIndex: 0, + Item: &ResponsesOutput{Type: "reasoning", ID: "rs_1"}, + }, state)) + appendAll(ResponsesEventToAnthropicEvents(&ResponsesStreamEvent{ + Type: "response.reasoning_summary_text.delta", + OutputIndex: 0, + Delta: "thinking...", + SummaryIndex: 0, + }, state)) + // summary.done must not close the thinking block before encrypted_content arrives + appendAll(ResponsesEventToAnthropicEvents(&ResponsesStreamEvent{ + Type: "response.reasoning_summary_text.done", + OutputIndex: 0, + SummaryIndex: 0, + }, state)) + appendAll(ResponsesEventToAnthropicEvents(&ResponsesStreamEvent{ + Type: "response.output_item.done", + OutputIndex: 0, + Item: &ResponsesOutput{ + Type: "reasoning", + ID: "rs_1", + EncryptedContent: "enc-stream-1", + Status: "completed", + }, + }, state)) + + var sawSignature bool + for _, ev := range all { + if ev.Type == "content_block_delta" && ev.Delta != nil && ev.Delta.Type == "signature_delta" { + assert.Equal(t, "enc-stream-1", ev.Delta.Signature) + sawSignature = true + } + } + require.True(t, sawSignature, "expected signature_delta with encrypted_content") +} + func TestResponsesToAnthropic_Incomplete(t *testing.T) { resp := &ResponsesResponse{ ID: "resp_inc", @@ -785,12 +857,26 @@ func TestStreamingReasoning(t *testing.T) { assert.Equal(t, "thinking_delta", events[0].Delta.Type) assert.Equal(t, "Let me think...", events[0].Delta.Thinking) - // reasoning done + // summary.done keeps thinking open until output_item.done (for signature) events = ResponsesEventToAnthropicEvents(&ResponsesStreamEvent{ Type: "response.reasoning_summary_text.done", }, state) - require.Len(t, events, 1) - assert.Equal(t, "content_block_stop", events[0].Type) + require.Len(t, events, 0) + + events = ResponsesEventToAnthropicEvents(&ResponsesStreamEvent{ + Type: "response.output_item.done", + OutputIndex: 0, + Item: &ResponsesOutput{ + Type: "reasoning", + EncryptedContent: "enc-rs-stream", + Status: "completed", + }, + }, state) + require.Len(t, events, 2) + assert.Equal(t, "content_block_delta", events[0].Type) + assert.Equal(t, "signature_delta", events[0].Delta.Type) + assert.Equal(t, "enc-rs-stream", events[0].Delta.Signature) + assert.Equal(t, "content_block_stop", events[1].Type) } func TestStreamingIncomplete(t *testing.T) { diff --git a/backend/internal/pkg/apicompat/anthropic_to_responses.go b/backend/internal/pkg/apicompat/anthropic_to_responses.go index e2011bee0..dfbe7b5e0 100644 --- a/backend/internal/pkg/apicompat/anthropic_to_responses.go +++ b/backend/internal/pkg/apicompat/anthropic_to_responses.go @@ -257,7 +257,9 @@ func anthropicUserToResponses(raw json.RawMessage) ([]ResponsesInputItem, error) // anthropicAssistantToResponses handles an Anthropic assistant message. // Text content → assistant message with output_text parts. // tool_use blocks → function_call items. -// thinking blocks → ignored (OpenAI doesn't accept them as input). +// thinking blocks with signature → reasoning items (encrypted_content) so +// multi-turn Grok/Codex prompt cache can reuse prior reasoning prefixes. +// thinking without signature remains ignored (not accepted as plain text input). func anthropicAssistantToResponses(raw json.RawMessage) ([]ResponsesInputItem, error) { // Try plain string. var s string @@ -277,6 +279,25 @@ func anthropicAssistantToResponses(raw json.RawMessage) ([]ResponsesInputItem, e var items []ResponsesInputItem + // Preserve turn order: reasoning → assistant text → tool calls. xAI/Codex + // multi-turn cache and tool continuations expect reasoning before the + // assistant message that followed it. + for _, b := range blocks { + if b.Type != "thinking" { + continue + } + sig := strings.TrimSpace(b.Signature) + // Only replay provider ciphertext. Skip GPT/Codex-style gAAAA blobs and + // empty placeholders — xAI returns 400 on decrypt for foreign signatures. + if sig == "" || strings.HasPrefix(sig, "gAAAA") { + continue + } + items = append(items, ResponsesInputItem{ + Type: "reasoning", + EncryptedContent: sig, + }) + } + // Text content → assistant message with output_text content parts. text := extractAnthropicTextFromBlocks(blocks) if text != "" { diff --git a/backend/internal/pkg/apicompat/responses_to_anthropic.go b/backend/internal/pkg/apicompat/responses_to_anthropic.go index 20b8ac029..129fff08d 100644 --- a/backend/internal/pkg/apicompat/responses_to_anthropic.go +++ b/backend/internal/pkg/apicompat/responses_to_anthropic.go @@ -3,6 +3,7 @@ package apicompat import ( "encoding/json" "fmt" + "strings" "time" ) @@ -32,10 +33,14 @@ func ResponsesToAnthropic(resp *ResponsesResponse, model string) *AnthropicRespo summaryText += s.Text } } - if summaryText != "" { + // Always surface encrypted_content as thinking.signature so Claude + // Code / multi-turn clients can send it back. Signature-only + // thinking blocks are valid when the model omits a visible summary. + if summaryText != "" || strings.TrimSpace(item.EncryptedContent) != "" { blocks = append(blocks, AnthropicContentBlock{ - Type: "thinking", - Thinking: summaryText, + Type: "thinking", + Thinking: summaryText, + Signature: item.EncryptedContent, }) } case "message": @@ -177,7 +182,10 @@ type ResponsesEventToAnthropicState struct { CurrentToolName string CurrentToolArgs string CurrentToolHadDelta bool - HasToolCall bool + // PendingThinkingSignature is filled from reasoning.encrypted_content and + // emitted as signature_delta before the thinking block is closed. + PendingThinkingSignature string + HasToolCall bool // OutputIndexToBlockIdx maps Responses output_index → Anthropic content block index. OutputIndexToBlockIdx map[int]int @@ -228,7 +236,10 @@ func ResponsesEventToAnthropicEvents( "response.reasoning_text.delta": return resToAnthHandleReasoningDelta(evt, state) case "response.reasoning_summary_text.done": - return resToAnthHandleBlockDone(state) + // Keep the thinking block open until response.output_item.done. + // Grok/Codex attach encrypted_content on the finished reasoning item; + // closing early would drop signature_delta and break multi-turn cache. + return nil // response.done 是 Realtime/WS 与项目透传路径使用的终止别名; // 普通 Responses HTTP SSE 的公开终止事件仍以 response.completed 为主。 case "response.completed", "response.done", "response.incomplete", "response.failed": @@ -359,6 +370,7 @@ func resToAnthHandleOutputItemAdded(evt *ResponsesStreamEvent, state *ResponsesE state.OutputIndexToBlockIdx[evt.OutputIndex] = idx state.ContentBlockOpen = true state.CurrentBlockType = "thinking" + state.PendingThinkingSignature = strings.TrimSpace(evt.Item.EncryptedContent) events = append(events, AnthropicStreamEvent{ Type: "content_block_start", @@ -542,6 +554,13 @@ func resToAnthHandleOutputItemDone(evt *ResponsesStreamEvent, state *ResponsesEv return resToAnthHandleWebSearchDone(evt, state) } + // Capture encrypted_content on reasoning item done (often only present here). + if evt.Item.Type == "reasoning" { + if sig := strings.TrimSpace(evt.Item.EncryptedContent); sig != "" { + state.PendingThinkingSignature = sig + } + } + if state.ContentBlockOpen { return closeCurrentBlock(state) } @@ -663,13 +682,30 @@ func closeCurrentBlock(state *ResponsesEventToAnthropicState) []AnthropicStreamE return nil } idx := state.ContentBlockIndex + var events []AnthropicStreamEvent + // Emit signature_delta before stop so Claude clients retain encrypted + // reasoning for the next turn (required for Grok multi-turn cache). + if state.CurrentBlockType == "thinking" { + if sig := strings.TrimSpace(state.PendingThinkingSignature); sig != "" { + events = append(events, AnthropicStreamEvent{ + Type: "content_block_delta", + Index: &idx, + Delta: &AnthropicDelta{ + Type: "signature_delta", + Signature: sig, + }, + }) + } + state.PendingThinkingSignature = "" + } state.ContentBlockOpen = false state.ContentBlockIndex++ state.CurrentToolName = "" state.CurrentToolArgs = "" state.CurrentToolHadDelta = false - return []AnthropicStreamEvent{{ + events = append(events, AnthropicStreamEvent{ Type: "content_block_stop", Index: &idx, - }} + }) + return events } diff --git a/backend/internal/pkg/apicompat/types.go b/backend/internal/pkg/apicompat/types.go index 54b2adf3d..6031eb41d 100644 --- a/backend/internal/pkg/apicompat/types.go +++ b/backend/internal/pkg/apicompat/types.go @@ -60,6 +60,9 @@ type AnthropicContentBlock struct { // type=thinking Thinking string `json:"thinking,omitempty"` + // Signature carries provider encrypted reasoning (e.g. xAI encrypted_content) + // so multi-turn Claude clients can round-trip it back on subsequent turns. + Signature string `json:"signature,omitempty"` // type=image Source *AnthropicImageSource `json:"source,omitempty"` @@ -247,6 +250,9 @@ type ResponsesInputItem struct { Role string `json:"role,omitempty"` Content json.RawMessage `json:"content,omitempty"` // string or []ResponsesContentPart + // type=reasoning (multi-turn replay of encrypted reasoning) + EncryptedContent string `json:"encrypted_content,omitempty"` + // type=function_call CallID string `json:"call_id,omitempty"` Name string `json:"name,omitempty"` diff --git a/backend/internal/service/openai_gateway_grok.go b/backend/internal/service/openai_gateway_grok.go index dec30c3bd..680b34bbe 100644 --- a/backend/internal/service/openai_gateway_grok.go +++ b/backend/internal/service/openai_gateway_grok.go @@ -201,15 +201,125 @@ func isGrokInvalidEncryptedContentResponse(statusCode int, body []byte) bool { return false } - code := gjson.GetBytes(body, "code") - message := gjson.GetBytes(body, "error") - if code.Type != gjson.String || message.Type != gjson.String || - !strings.EqualFold(strings.TrimSpace(code.String()), "invalid-argument") { + // xAI has used both flat and nested error envelopes: + // {"code":"invalid-argument","error":"Could not decrypt the provided encrypted_content."} + // {"error":{"message":"Could not decrypt the provided encrypted_content."}} + code := strings.TrimSpace(gjson.GetBytes(body, "code").String()) + message := "" + errNode := gjson.GetBytes(body, "error") + switch { + case errNode.Type == gjson.String: + message = errNode.String() + case errNode.IsObject(): + message = firstNonEmpty(errNode.Get("message").String(), errNode.Get("error").String()) + if code == "" { + code = strings.TrimSpace(errNode.Get("code").String()) + } + default: + message = gjson.GetBytes(body, "message").String() + } + normalizedMessage := strings.ToLower(strings.TrimSpace(message)) + if normalizedMessage == "" { return false } - normalizedMessage := strings.ToLower(message.String()) - return strings.Contains(normalizedMessage, "decrypt") && strings.Contains(normalizedMessage, "encrypted_content") + if strings.EqualFold(code, "invalid_encrypted_content") { + return true + } + // Keep the official xAI flat-code gate so unrelated 400s are not retried. + if !strings.EqualFold(code, "invalid-argument") && code != "" { + return false + } + // Nested OpenAI-style envelopes may omit top-level code; require decrypt text. + if code == "" && !strings.Contains(normalizedMessage, "decrypt") { + return false + } + return strings.Contains(normalizedMessage, "encrypted_content") && + (strings.Contains(normalizedMessage, "decrypt") || + strings.Contains(normalizedMessage, "unmodified")) +} + +// requestHasGrokEncryptedReasoning reports whether the outbound Responses body +// still carries reasoning.encrypted_content that can be stripped for retry. +func requestHasGrokEncryptedReasoning(body []byte) bool { + input := gjson.GetBytes(body, "input") + if !input.Exists() { + return false + } + items := input.Array() + if input.IsObject() { + items = []gjson.Result{input} + } + for _, item := range items { + if strings.TrimSpace(item.Get("type").String()) != "reasoning" { + continue + } + enc := item.Get("encrypted_content") + if enc.Exists() && enc.Type != gjson.Null && strings.TrimSpace(enc.String()) != "" { + return true + } + } + return false +} + +type grokEncryptedContentStripRetriedKey struct{} + +func markGrokEncryptedContentStripRetried(ctx context.Context) context.Context { + return context.WithValue(ctx, grokEncryptedContentStripRetriedKey{}, true) +} + +func grokEncryptedContentStripRetried(ctx context.Context) bool { + v, _ := ctx.Value(grokEncryptedContentStripRetriedKey{}).(bool) + return v +} + +// stripAnthropicThinkingSignatures removes thinking.signature from Claude +// history so a different Grok OAuth account can accept multi-turn tool +// continuations after decrypt failures. Returns ok=false when nothing changed. +func stripAnthropicThinkingSignatures(body []byte) ([]byte, bool) { + if len(body) == 0 || !bytes.Contains(body, []byte(`"signature"`)) { + return body, false + } + var req map[string]any + if err := json.Unmarshal(body, &req); err != nil { + return body, false + } + messages, ok := req["messages"].([]any) + if !ok || len(messages) == 0 { + return body, false + } + changed := false + for _, rawMsg := range messages { + msg, ok := rawMsg.(map[string]any) + if !ok { + continue + } + content, ok := msg["content"].([]any) + if !ok { + continue + } + for _, rawBlock := range content { + block, ok := rawBlock.(map[string]any) + if !ok { + continue + } + if typ, _ := block["type"].(string); typ != "thinking" { + continue + } + if _, has := block["signature"]; has { + delete(block, "signature") + changed = true + } + } + } + if !changed { + return body, false + } + out, err := json.Marshal(req) + if err != nil { + return body, false + } + return out, true } func trimGrokInvalidEncryptedContentRetryBody(body []byte) ([]byte, bool, error) { diff --git a/backend/internal/service/openai_gateway_grok_cache.go b/backend/internal/service/openai_gateway_grok_cache.go index de52115f1..a7bb29f8e 100644 --- a/backend/internal/service/openai_gateway_grok_cache.go +++ b/backend/internal/service/openai_gateway_grok_cache.go @@ -4,6 +4,7 @@ import ( "encoding/json" "fmt" "net/http" + "regexp" "strings" "github.com/Wei-Shaw/sub2api/internal/pkg/xai" @@ -14,12 +15,47 @@ import ( const ( grokConversationIDHeader = "X-Grok-Conv-Id" + claudeCodeSessionHeader = "X-Claude-Code-Session-Id" grokClientToolCacheOptInHeader = "X-Sub2API-Grok-Client-Tool-Cache" grokFreeCacheNativeToolsJSON = `[{"type":"web_search"},{"type":"x_search"}]` grokFreeCacheDisabledToolChoice = "none" grokClientToolCacheOptInExtraKey = "grok_client_tool_cache_enabled" ) +// Claude Code metadata.user_id often ends with _session_. +var claudeCodeSessionSuffixPattern = regexp.MustCompile(`_session_([a-f0-9-]+)$`) + +// extractClaudeCodeSessionID resolves the Claude Code conversation id from +// headers or Anthropic/OpenAI-compatible payload metadata. +func extractClaudeCodeSessionID(c *gin.Context, body []byte) string { + if c != nil { + if seed := strings.TrimSpace(c.GetHeader(claudeCodeSessionHeader)); seed != "" { + return seed + } + } + return extractClaudeCodeSessionIDFromPayload(body) +} + +func extractClaudeCodeSessionIDFromPayload(body []byte) string { + if len(body) == 0 { + return "" + } + userID := strings.TrimSpace(gjson.GetBytes(body, "metadata.user_id").String()) + if userID == "" { + return "" + } + if matches := claudeCodeSessionSuffixPattern.FindStringSubmatch(userID); len(matches) >= 2 { + return matches[1] + } + // Claude Code may embed JSON: {"session_id":"..."} + if len(userID) > 0 && userID[0] == '{' { + if sid := strings.TrimSpace(gjson.Get(userID, "session_id").String()); sid != "" { + return sid + } + } + return "" +} + // resolveGrokCacheIdentity derives one stable, tenant-isolated routing identity // for xAI's server-side prompt cache. The returned value is safe to expose to // the upstream: it never contains the client's raw session identifier. @@ -68,7 +104,13 @@ func resolveGrokCacheIdentity(c *gin.Context, body []byte, explicitKey, upstream func explicitGrokCacheSeed(c *gin.Context, body []byte, explicitKey string) string { seed := "" if c != nil { - seed = strings.TrimSpace(c.GetHeader("session_id")) + // Claude Code session is the most stable multi-turn identity for + // /v1/messages → Grok bridges. Prefer it over generic session headers + // so prompt cache routing matches CPA behavior. + seed = extractClaudeCodeSessionID(c, body) + if seed == "" { + seed = strings.TrimSpace(c.GetHeader("session_id")) + } if seed == "" { seed = strings.TrimSpace(c.GetHeader("conversation_id")) } @@ -76,6 +118,9 @@ func explicitGrokCacheSeed(c *gin.Context, body []byte, explicitKey string) stri seed = strings.TrimSpace(c.GetHeader(grokConversationIDHeader)) } } + if seed == "" && len(body) > 0 { + seed = extractClaudeCodeSessionIDFromPayload(body) + } if seed == "" && len(body) > 0 { seed = strings.TrimSpace(gjson.GetBytes(body, "prompt_cache_key").String()) } diff --git a/backend/internal/service/openai_gateway_grok_cache_test.go b/backend/internal/service/openai_gateway_grok_cache_test.go index 034ba4843..2255f4a2b 100644 --- a/backend/internal/service/openai_gateway_grok_cache_test.go +++ b/backend/internal/service/openai_gateway_grok_cache_test.go @@ -140,6 +140,28 @@ func TestResolveGrokCacheIdentityExplicitHeaderPriority(t *testing.T) { require.Equal(t, want, got) } +func TestResolveGrokCacheIdentityPrefersClaudeCodeSession(t *testing.T) { + gin.SetMode(gin.TestMode) + c := newGrokCacheTestContext(701) + c.Request.Header.Set(claudeCodeSessionHeader, "cc-session-abc") + c.Request.Header.Set("session_id", "session-key") + body1 := []byte(`{"model":"grok","input":"turn-1"}`) + body2 := []byte(`{"model":"grok","input":"turn-2-different-user-text"}`) + + first := resolveGrokCacheIdentity(c, body1, "", "grok-4.5") + second := resolveGrokCacheIdentity(c, body2, "unrelated-explicit", "grok-4.5") + require.NotEmpty(t, first) + require.Equal(t, first, second, "same Claude Code session must keep stable Grok cache identity across turns") + + // metadata.user_id JSON form used by Claude Code clients + metaBody := []byte(`{"model":"grok","metadata":{"user_id":"{\"session_id\":\"meta-session-xyz\"}"},"input":"hi"}`) + metaOnly := newGrokCacheTestContext(702) + metaID := resolveGrokCacheIdentity(metaOnly, metaBody, "", "grok-4.5") + require.NotEmpty(t, metaID) + metaBody2 := []byte(`{"model":"grok","metadata":{"user_id":"{\"session_id\":\"meta-session-xyz\"}"},"input":"later turn"}`) + require.Equal(t, metaID, resolveGrokCacheIdentity(metaOnly, metaBody2, "", "grok-4.5")) +} + func TestResolveGrokCacheIdentityFailsClosedWithoutAPIKeyContext(t *testing.T) { gin.SetMode(gin.TestMode) c := newGrokCacheTestContext(0) diff --git a/backend/internal/service/openai_gateway_grok_test.go b/backend/internal/service/openai_gateway_grok_test.go index 4bc6eb9b0..235098d12 100644 --- a/backend/internal/service/openai_gateway_grok_test.go +++ b/backend/internal/service/openai_gateway_grok_test.go @@ -1654,11 +1654,6 @@ func TestForwardGrokResponsesInvalidEncryptedContentRecoveryDoesNotOvermatch(t * requestBody: `{"model":"grok","input":[{"type":"reasoning","encrypted_content":"cipher"}],"stream":false}`, responseBody: `{"code":"invalid-argument","error":"The provided encrypted_content is invalid."}`, }, - { - name: "nested OpenAI error shape", - requestBody: `{"model":"grok","input":[{"type":"reasoning","encrypted_content":"cipher"}],"stream":false}`, - responseBody: `{"code":"invalid-argument","error":{"message":"Could not decrypt the provided encrypted_content."}}`, - }, { name: "request has no encrypted reasoning", requestBody: `{"model":"grok","input":[{"type":"message","role":"user","content":"hi"}],"stream":false}`, @@ -1697,6 +1692,44 @@ func TestForwardGrokResponsesInvalidEncryptedContentRecoveryDoesNotOvermatch(t * } } +func TestForwardGrokResponsesInvalidEncryptedContentRecoveryNestedErrorShape(t *testing.T) { + gin.SetMode(gin.TestMode) + + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + body := []byte(`{"model":"grok","input":[{"type":"reasoning","encrypted_content":"cipher"},{"type":"message","role":"user","content":"hi"}],"stream":false}`) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(body)) + + account := &Account{ + ID: 4538, + Name: "grok-api-key", + Platform: PlatformGrok, + Type: AccountTypeAPIKey, + Concurrency: 1, + Credentials: map[string]any{"api_key": "token", "base_url": "https://api.x.ai/v1"}, + } + upstream := &httpUpstreamRecorder{responses: []*http.Response{ + { + StatusCode: http.StatusBadRequest, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader(`{"code":"invalid-argument","error":{"message":"Could not decrypt the provided encrypted_content."}}`)), + }, + { + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader(`{"id":"resp_ok","output":[],"usage":{"input_tokens":1,"output_tokens":1}}`)), + }, + }} + svc := &OpenAIGatewayService{httpUpstream: upstream} + + result, err := svc.forwardGrokResponses(context.Background(), c, account, body, "grok", false, time.Now()) + require.NoError(t, err) + require.NotNil(t, result) + require.Len(t, upstream.requests, 2) + require.True(t, gjson.GetBytes(upstream.bodies[0], "input.0.encrypted_content").Exists()) + require.False(t, gjson.GetBytes(upstream.bodies[1], "input.0.encrypted_content").Exists()) +} + func TestForwardGrokResponsesInvalidEncryptedContentRetryFailureIsTerminal(t *testing.T) { gin.SetMode(gin.TestMode) diff --git a/backend/internal/service/openai_gateway_messages.go b/backend/internal/service/openai_gateway_messages.go index db9e95a8f..d21cba5c5 100644 --- a/backend/internal/service/openai_gateway_messages.go +++ b/backend/internal/service/openai_gateway_messages.go @@ -1,10 +1,12 @@ package service import ( + "bytes" "context" "encoding/json" "errors" "fmt" + "io" "net/http" "strings" "sync/atomic" @@ -60,6 +62,18 @@ func (s *OpenAIGatewayService) ForwardAsAnthropic( anthropicDigestChain := "" anthropicMatchedDigestChain := "" compatPromptCacheInjected := false + // Grok is outside the gpt-5/codex compat injector, but Claude Code still + // carries a stable session id. Prefer that as the Grok prompt-cache seed so + // multi-turn /v1/messages traffic can hit xAI's server-side cache. + if promptCacheKey == "" && account.Platform == PlatformGrok { + if sessionSeed := extractClaudeCodeSessionID(c, body); sessionSeed != "" { + promptCacheKey = sessionSeed + compatPromptCacheInjected = true + } else if sessionSeed := promptCacheKeyFromAnthropicMetadataSession(&anthropicReq); sessionSeed != "" { + promptCacheKey = sessionSeed + compatPromptCacheInjected = true + } + } if promptCacheKey == "" && shouldAutoInjectPromptCacheKeyForCompat(upstreamModel) { promptCacheKey = promptCacheKeyFromAnthropicMetadataSession(&anthropicReq) if promptCacheKey == "" { @@ -319,9 +333,56 @@ func (s *OpenAIGatewayService) ForwardAsAnthropic( if account.Proxy != nil { proxyURL = account.Proxy.URL() } - resp, err := s.httpUpstream.Do(upstreamReq, proxyURL, account.ID, account.Concurrency) - if err != nil { - return nil, s.handleOpenAIUpstreamTransportError(ctx, c, account, err, false) + // Grok may reject encrypted reasoning replayed under a different OAuth + // account/cache identity. Match forwardGrokResponses: one strip+retry before + // treating the 400 as a hard failure / failover trigger. + var resp *http.Response + for attempt := 0; ; attempt++ { + if attempt > 0 { + if account.Platform != PlatformGrok { + break + } + upstreamCtxRetry, releaseRetry := detachUpstreamContext(ctx) + upstreamReq, err = buildGrokResponsesRequest(upstreamCtxRetry, c, account, responsesBody, token, grokCacheIdentity, s.cfg) + releaseRetry() + if err != nil { + return nil, fmt.Errorf("build grok retry request: %w", err) + } + } + resp, err = s.httpUpstream.Do(upstreamReq, proxyURL, account.ID, account.Concurrency) + if err != nil { + return nil, s.handleOpenAIUpstreamTransportError(ctx, c, account, err, false) + } + if account.Platform != PlatformGrok || attempt > 0 || resp.StatusCode != http.StatusBadRequest { + break + } + respBody := s.readUpstreamErrorBody(resp) + if resp.Body != nil { + _ = resp.Body.Close() + } + // Prefer explicit decrypt errors; also strip once on any 400 when the + // outbound body still carries reasoning.encrypted_content (account + // switch often returns opaque "Upstream error: 400"). + shouldStrip := isGrokInvalidEncryptedContentResponse(resp.StatusCode, respBody) || + requestHasGrokEncryptedReasoning(responsesBody) + if !shouldStrip { + resp.Body = io.NopCloser(bytes.NewReader(respBody)) + break + } + retryBody, changed, trimErr := trimGrokInvalidEncryptedContentRetryBody(responsesBody) + if trimErr != nil { + return nil, fmt.Errorf("prepare Grok invalid encrypted_content retry: %w", trimErr) + } + if !changed { + resp.Body = io.NopCloser(bytes.NewReader(respBody)) + break + } + responsesBody = retryBody + logger.L().Info("openai messages: retrying after stripping invalid Grok encrypted_content", + zap.Int64("account_id", account.ID), + zap.Bool("cache_identity_present", strings.TrimSpace(grokCacheIdentity) != ""), + zap.String("upstream_error_preview", truncateOpenAIWSLogValue(string(respBody), 240)), + ) } defer func() { _ = resp.Body.Close() }() @@ -348,6 +409,19 @@ func (s *OpenAIGatewayService) ForwardAsAnthropic( ) return s.ForwardAsAnthropic(ctx, c, account, body, promptCacheKey, defaultMappedModel) } + // Grok account-switched history often fails decrypt; strip encrypted + // reasoning once at the client-body level so failover accounts can accept + // the multi-turn tool continuation instead of cascading 400s. + if account.Platform == PlatformGrok && + isGrokInvalidEncryptedContentResponse(resp.StatusCode, respBody) && + !grokEncryptedContentStripRetried(ctx) { + if strippedBody, ok := stripAnthropicThinkingSignatures(body); ok { + logger.L().Info("openai messages: stripping thinking signatures for Grok failover retry", + zap.Int64("account_id", account.ID), + ) + return s.ForwardAsAnthropic(markGrokEncryptedContentStripRetried(ctx), c, account, strippedBody, promptCacheKey, defaultMappedModel) + } + } if foErr := s.failoverOpenAIUpstreamHTTPError(ctx, c, account, resp, respBody, upstreamMsg, upstreamModel); foErr != nil { return nil, foErr }