mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-14 12:02:27 +08:00
Merge pull request #4591 from huangjunsen0406/fix/grok-claude-messages-prompt-cache
fix(grok): improve Claude Messages prompt-cache round-trip
This commit is contained in:
@@ -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) {
|
||||
|
||||
@@ -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 != "" {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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"`
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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_<uuid>.
|
||||
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())
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user