refactor(gateway): remove parsed request object graphs

Keep large gateway payloads as raw body ranges and bind OpenAI parsed-body caches to the body bytes so failover and mapping do not reuse stale mutable state.
This commit is contained in:
name committed 2026-05-30 20:01:38 +08:00
1 parent d8cbf9ab5c
commit b1c4be4ac8
16 files changed
+693 -1222

No files matched your search

+2 -2
View File
@@ -747,10 +747,10 @@ func (h *GatewayHandler) Messages(c *gin.Context) {
// 应用渠道模型映射到请求 // 应用渠道模型映射到请求
if channelMapping.Mapped { if channelMapping.Mapped {
parsedReq.Model = channelMapping.MappedModel parsedReq.Model = channelMapping.MappedModel
parsedReq.Body.Replace(h.gatewayService.ReplaceModelInBody(parsedReq.Body.Bytes(), channelMapping.MappedModel)) parsedReq.ReplaceBody(h.gatewayService.ReplaceModelInBody(parsedReq.Body.Bytes(), channelMapping.MappedModel))
} }
// Bedrock CC 兼容:渠道模型映射后,清理 Anthropic API 专有字段、注入 Bedrock 必需字段 // Bedrock CC 兼容:渠道模型映射后,清理 Anthropic API 专有字段、注入 Bedrock 必需字段
parsedReq.Body.Replace(h.gatewayService.ApplyBedrockCCCompat(c.Request.Context(), parsedReq.Body.Bytes(), parsedReq.Model, account, apiKey.GroupID)) parsedReq.ReplaceBody(h.gatewayService.ApplyBedrockCCCompat(c.Request.Context(), parsedReq.Body.Bytes(), parsedReq.Model, account, apiKey.GroupID))
body = parsedReq.Body.Bytes() body = parsedReq.Body.Bytes()
// 转发请求 - 根据账号平台分流 // 转发请求 - 根据账号平台分流
+7 -5
View File
@@ -74,8 +74,12 @@ func claudeCodeBodyMapFromParsedRequest(parsedReq *service.ParsedRequest) map[st
bodyMap := map[string]any{ bodyMap := map[string]any{
"model": parsedReq.Model, "model": parsedReq.Model,
} }
if parsedReq.System != nil || parsedReq.HasSystem { if parsedReq.HasSystem {
bodyMap["system"] = parsedReq.System if system, ok := parsedReq.SystemValue(); ok {
bodyMap["system"] = system
} else {
bodyMap["system"] = nil
}
} }
if parsedReq.MetadataUserID != "" { if parsedReq.MetadataUserID != "" {
bodyMap["metadata"] = map[string]any{"user_id": parsedReq.MetadataUserID} bodyMap["metadata"] = map[string]any{"user_id": parsedReq.MetadataUserID}
@@ -87,11 +91,9 @@ func claudeCodeBodyMapFromContextCache(c *gin.Context) map[string]any {
if c == nil { if c == nil {
return nil return nil
} }
if cached, ok := c.Get(service.OpenAIParsedRequestBodyKey); ok { if bodyMap := service.CachedOpenAIParsedRequestBody(c); bodyMap != nil {
if bodyMap, ok := cached.(map[string]any); ok {
return bodyMap return bodyMap
} }
}
if cached, ok := c.Get(claudeCodeParsedRequestContextKey); ok { if cached, ok := c.Get(claudeCodeParsedRequestContextKey); ok {
switch v := cached.(type) { switch v := cached.(type) {
case *service.ParsedRequest: case *service.ParsedRequest:
@@ -185,13 +185,8 @@ func TestSetClaudeCodeClientContext_ReuseParsedRequestAndContextCache(t *testing
c.Request.Header.Set("anthropic-beta", "message-batches-2024-09-24") c.Request.Header.Set("anthropic-beta", "message-batches-2024-09-24")
c.Request.Header.Set("anthropic-version", "2023-06-01") c.Request.Header.Set("anthropic-version", "2023-06-01")
parsedReq := &service.ParsedRequest{ parsedReq, err := service.ParseGatewayRequest(service.NewRequestBodyRef(validClaudeCodeBodyJSON()), "")
Model: "claude-3-5-sonnet-20241022", require.NoError(t, err)
System: []any{
map[string]any{"text": "You are Claude Code, Anthropic's official CLI for Claude."},
},
MetadataUserID: "user_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa_account__session_aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
}
// body 非法 JSON,如果函数复用 parsedReq 成功则仍应判定为 Claude Code。 // body 非法 JSON,如果函数复用 parsedReq 成功则仍应判定为 Claude Code。
SetClaudeCodeClientContext(c, []byte(`{invalid`), parsedReq) SetClaudeCodeClientContext(c, []byte(`{invalid`), parsedReq)
@@ -204,7 +199,7 @@ func TestSetClaudeCodeClientContext_ReuseParsedRequestAndContextCache(t *testing
c.Request.Header.Set("X-App", "claude-code") c.Request.Header.Set("X-App", "claude-code")
c.Request.Header.Set("anthropic-beta", "message-batches-2024-09-24") c.Request.Header.Set("anthropic-beta", "message-batches-2024-09-24")
c.Request.Header.Set("anthropic-version", "2023-06-01") c.Request.Header.Set("anthropic-version", "2023-06-01")
c.Set(service.OpenAIParsedRequestBodyKey, map[string]any{ service.CacheOpenAIParsedRequestBody(c, []byte(`{invalid`), map[string]any{
"model": "claude-3-5-sonnet-20241022", "model": "claude-3-5-sonnet-20241022",
"system": []any{ "system": []any{
map[string]any{"text": "You are Claude Code, Anthropic's official CLI for Claude."}, map[string]any{"text": "You are Claude Code, Anthropic's official CLI for Claude."},
@@ -954,7 +954,7 @@ func (h *OpenAIGatewayHandler) validateFunctionCallOutputRequest(c *gin.Context,
return true return true
} }
c.Set(service.OpenAIParsedRequestBodyKey, reqBody) service.CacheOpenAIParsedRequestBody(c, body, reqBody)
validation := service.ValidateFunctionCallOutputContext(reqBody) validation := service.ValidateFunctionCallOutputContext(reqBody)
if !validation.HasFunctionCallOutput { if !validation.HasFunctionCallOutput {
return true return true
+28 -17
View File
@@ -4,6 +4,8 @@ import (
"encoding/json" "encoding/json"
"strings" "strings"
"time" "time"
"github.com/tidwall/gjson"
) )
// Anthropic 会话 Fallback 相关常量 // Anthropic 会话 Fallback 相关常量
@@ -30,30 +32,39 @@ func BuildAnthropicDigestChain(parsed *ParsedRequest) string {
var parts []string var parts []string
// 1. system prompt if systemRaw := parsed.SystemRaw(); len(systemRaw) > 0 && string(systemRaw) != "null" {
if parsed.System != nil { parts = append(parts, "s:"+shortHash(canonicalAnthropicDigestJSON(systemRaw)))
systemData, _ := json.Marshal(parsed.System)
if len(systemData) > 0 && string(systemData) != "null" {
parts = append(parts, "s:"+shortHash(systemData))
}
} }
// 2. messages messages := parsed.MessagesRaw()
for _, msg := range parsed.Messages { if len(messages) > 0 {
msgMap, ok := msg.(map[string]any) gjson.ParseBytes(messages).ForEach(func(_, msg gjson.Result) bool {
if !ok { prefix := rolePrefix(msg.Get("role").String())
continue content := msg.Get("content")
} parts = append(parts, prefix+":"+shortHash(canonicalAnthropicDigestJSON([]byte(content.Raw))))
role, _ := msgMap["role"].(string) return true
prefix := rolePrefix(role) })
content := msgMap["content"]
contentData, _ := json.Marshal(content)
parts = append(parts, prefix+":"+shortHash(contentData))
} }
return strings.Join(parts, "-") return strings.Join(parts, "-")
} }
// canonicalAnthropicDigestJSON 保持 digest 对 JSON key 顺序和空白不敏感。
func canonicalAnthropicDigestJSON(raw []byte) []byte {
if len(raw) == 0 {
return raw
}
var value any
if err := json.Unmarshal(raw, &value); err != nil {
return raw
}
canonical, err := json.Marshal(value)
if err != nil {
return raw
}
return canonical
}
// rolePrefix 将 Anthropic 的 role 映射为单字符前缀 // rolePrefix 将 Anthropic 的 role 映射为单字符前缀
func rolePrefix(role string) string { func rolePrefix(role string) string {
switch role { switch role {
@@ -1,3 +1,5 @@
//go:build unit
package service package service
import ( import (
@@ -5,6 +7,15 @@ import (
"testing" "testing"
) )
func mustParseAnthropicDigestRequest(t *testing.T, body string) *ParsedRequest {
t.Helper()
parsed, err := ParseGatewayRequest(NewRequestBodyRef([]byte(body)), "")
if err != nil {
t.Fatalf("ParseGatewayRequest failed: %v", err)
}
return parsed
}
func TestBuildAnthropicDigestChain_NilRequest(t *testing.T) { func TestBuildAnthropicDigestChain_NilRequest(t *testing.T) {
result := BuildAnthropicDigestChain(nil) result := BuildAnthropicDigestChain(nil)
if result != "" { if result != "" {
@@ -13,9 +24,7 @@ func TestBuildAnthropicDigestChain_NilRequest(t *testing.T) {
} }
func TestBuildAnthropicDigestChain_EmptyMessages(t *testing.T) { func TestBuildAnthropicDigestChain_EmptyMessages(t *testing.T) {
parsed := &ParsedRequest{ parsed := mustParseAnthropicDigestRequest(t, `{"messages":[]}`)
Messages: []any{},
}
result := BuildAnthropicDigestChain(parsed) result := BuildAnthropicDigestChain(parsed)
if result != "" { if result != "" {
t.Errorf("expected empty string for empty messages, got: %s", result) t.Errorf("expected empty string for empty messages, got: %s", result)
@@ -23,11 +32,7 @@ func TestBuildAnthropicDigestChain_EmptyMessages(t *testing.T) {
} }
func TestBuildAnthropicDigestChain_SingleUserMessage(t *testing.T) { func TestBuildAnthropicDigestChain_SingleUserMessage(t *testing.T) {
parsed := &ParsedRequest{ parsed := mustParseAnthropicDigestRequest(t, `{"messages":[{"role":"user","content":"hello"}]}`)
Messages: []any{
map[string]any{"role": "user", "content": "hello"},
},
}
result := BuildAnthropicDigestChain(parsed) result := BuildAnthropicDigestChain(parsed)
parts := splitChain(result) parts := splitChain(result)
if len(parts) != 1 { if len(parts) != 1 {
@@ -39,12 +44,7 @@ func TestBuildAnthropicDigestChain_SingleUserMessage(t *testing.T) {
} }
func TestBuildAnthropicDigestChain_UserAndAssistant(t *testing.T) { func TestBuildAnthropicDigestChain_UserAndAssistant(t *testing.T) {
parsed := &ParsedRequest{ parsed := mustParseAnthropicDigestRequest(t, `{"messages":[{"role":"user","content":"hello"},{"role":"assistant","content":"hi there"}]}`)
Messages: []any{
map[string]any{"role": "user", "content": "hello"},
map[string]any{"role": "assistant", "content": "hi there"},
},
}
result := BuildAnthropicDigestChain(parsed) result := BuildAnthropicDigestChain(parsed)
parts := splitChain(result) parts := splitChain(result)
if len(parts) != 2 { if len(parts) != 2 {
@@ -59,12 +59,7 @@ func TestBuildAnthropicDigestChain_UserAndAssistant(t *testing.T) {
} }
func TestBuildAnthropicDigestChain_WithSystemString(t *testing.T) { func TestBuildAnthropicDigestChain_WithSystemString(t *testing.T) {
parsed := &ParsedRequest{ parsed := mustParseAnthropicDigestRequest(t, `{"system":"You are a helpful assistant","messages":[{"role":"user","content":"hello"}]}`)
System: "You are a helpful assistant",
Messages: []any{
map[string]any{"role": "user", "content": "hello"},
},
}
result := BuildAnthropicDigestChain(parsed) result := BuildAnthropicDigestChain(parsed)
parts := splitChain(result) parts := splitChain(result)
if len(parts) != 2 { if len(parts) != 2 {
@@ -79,14 +74,7 @@ func TestBuildAnthropicDigestChain_WithSystemString(t *testing.T) {
} }
func TestBuildAnthropicDigestChain_WithSystemContentBlocks(t *testing.T) { func TestBuildAnthropicDigestChain_WithSystemContentBlocks(t *testing.T) {
parsed := &ParsedRequest{ parsed := mustParseAnthropicDigestRequest(t, `{"system":[{"type":"text","text":"You are a helpful assistant"}],"messages":[{"role":"user","content":"hello"}]}`)
System: []any{
map[string]any{"type": "text", "text": "You are a helpful assistant"},
},
Messages: []any{
map[string]any{"role": "user", "content": "hello"},
},
}
result := BuildAnthropicDigestChain(parsed) result := BuildAnthropicDigestChain(parsed)
parts := splitChain(result) parts := splitChain(result)
if len(parts) != 2 { if len(parts) != 2 {
@@ -100,74 +88,33 @@ func TestBuildAnthropicDigestChain_WithSystemContentBlocks(t *testing.T) {
func TestBuildAnthropicDigestChain_ConversationPrefixRelationship(t *testing.T) { func TestBuildAnthropicDigestChain_ConversationPrefixRelationship(t *testing.T) {
// 核心测试:验证对话增长时链的前缀关系 // 核心测试:验证对话增长时链的前缀关系
// 上一轮的完整链一定是下一轮链的前缀 // 上一轮的完整链一定是下一轮链的前缀
system := "You are a helpful assistant" round1 := mustParseAnthropicDigestRequest(t, `{"system":"You are a helpful assistant","messages":[{"role":"user","content":"hello"}]}`)
// 第 1 轮: system + user
round1 := &ParsedRequest{
System: system,
Messages: []any{
map[string]any{"role": "user", "content": "hello"},
},
}
chain1 := BuildAnthropicDigestChain(round1) chain1 := BuildAnthropicDigestChain(round1)
// 第 2 轮: system + user + assistant + user round2 := mustParseAnthropicDigestRequest(t, `{"system":"You are a helpful assistant","messages":[{"role":"user","content":"hello"},{"role":"assistant","content":"hi there"},{"role":"user","content":"how are you?"}]}`)
round2 := &ParsedRequest{
System: system,
Messages: []any{
map[string]any{"role": "user", "content": "hello"},
map[string]any{"role": "assistant", "content": "hi there"},
map[string]any{"role": "user", "content": "how are you?"},
},
}
chain2 := BuildAnthropicDigestChain(round2) chain2 := BuildAnthropicDigestChain(round2)
// 第 3 轮: system + user + assistant + user + assistant + user round3 := mustParseAnthropicDigestRequest(t, `{"system":"You are a helpful assistant","messages":[{"role":"user","content":"hello"},{"role":"assistant","content":"hi there"},{"role":"user","content":"how are you?"},{"role":"assistant","content":"I'm doing well"},{"role":"user","content":"great"}]}`)
round3 := &ParsedRequest{
System: system,
Messages: []any{
map[string]any{"role": "user", "content": "hello"},
map[string]any{"role": "assistant", "content": "hi there"},
map[string]any{"role": "user", "content": "how are you?"},
map[string]any{"role": "assistant", "content": "I'm doing well"},
map[string]any{"role": "user", "content": "great"},
},
}
chain3 := BuildAnthropicDigestChain(round3) chain3 := BuildAnthropicDigestChain(round3)
t.Logf("Chain1: %s", chain1) t.Logf("Chain1: %s", chain1)
t.Logf("Chain2: %s", chain2) t.Logf("Chain2: %s", chain2)
t.Logf("Chain3: %s", chain3) t.Logf("Chain3: %s", chain3)
// chain1 是 chain2 的前缀
if !strings.HasPrefix(chain2, chain1) { if !strings.HasPrefix(chain2, chain1) {
t.Errorf("chain1 should be prefix of chain2:\n chain1: %s\n chain2: %s", chain1, chain2) t.Errorf("chain1 should be prefix of chain2:\n chain1: %s\n chain2: %s", chain1, chain2)
} }
// chain2 是 chain3 的前缀
if !strings.HasPrefix(chain3, chain2) { if !strings.HasPrefix(chain3, chain2) {
t.Errorf("chain2 should be prefix of chain3:\n chain2: %s\n chain3: %s", chain2, chain3) t.Errorf("chain2 should be prefix of chain3:\n chain2: %s\n chain3: %s", chain2, chain3)
} }
// chain1 也是 chain3 的前缀(传递性)
if !strings.HasPrefix(chain3, chain1) { if !strings.HasPrefix(chain3, chain1) {
t.Errorf("chain1 should be prefix of chain3:\n chain1: %s\n chain3: %s", chain1, chain3) t.Errorf("chain1 should be prefix of chain3:\n chain1: %s\n chain3: %s", chain1, chain3)
} }
} }
func TestBuildAnthropicDigestChain_DifferentSystemProducesDifferentChain(t *testing.T) { func TestBuildAnthropicDigestChain_DifferentSystemProducesDifferentChain(t *testing.T) {
parsed1 := &ParsedRequest{ parsed1 := mustParseAnthropicDigestRequest(t, `{"system":"System A","messages":[{"role":"user","content":"hello"}]}`)
System: "System A", parsed2 := mustParseAnthropicDigestRequest(t, `{"system":"System B","messages":[{"role":"user","content":"hello"}]}`)
Messages: []any{
map[string]any{"role": "user", "content": "hello"},
},
}
parsed2 := &ParsedRequest{
System: "System B",
Messages: []any{
map[string]any{"role": "user", "content": "hello"},
},
}
chain1 := BuildAnthropicDigestChain(parsed1) chain1 := BuildAnthropicDigestChain(parsed1)
chain2 := BuildAnthropicDigestChain(parsed2) chain2 := BuildAnthropicDigestChain(parsed2)
@@ -176,7 +123,6 @@ func TestBuildAnthropicDigestChain_DifferentSystemProducesDifferentChain(t *test
t.Error("Different system prompts should produce different chains") t.Error("Different system prompts should produce different chains")
} }
// 但 user 部分的 hash 应该相同
parts1 := splitChain(chain1) parts1 := splitChain(chain1)
parts2 := splitChain(chain2) parts2 := splitChain(chain2)
if parts1[1] != parts2[1] { if parts1[1] != parts2[1] {
@@ -185,20 +131,8 @@ func TestBuildAnthropicDigestChain_DifferentSystemProducesDifferentChain(t *test
} }
func TestBuildAnthropicDigestChain_DifferentContentProducesDifferentChain(t *testing.T) { func TestBuildAnthropicDigestChain_DifferentContentProducesDifferentChain(t *testing.T) {
parsed1 := &ParsedRequest{ parsed1 := mustParseAnthropicDigestRequest(t, `{"messages":[{"role":"user","content":"hello"},{"role":"assistant","content":"ORIGINAL reply"},{"role":"user","content":"next"}]}`)
Messages: []any{ parsed2 := mustParseAnthropicDigestRequest(t, `{"messages":[{"role":"user","content":"hello"},{"role":"assistant","content":"TAMPERED reply"},{"role":"user","content":"next"}]}`)
map[string]any{"role": "user", "content": "hello"},
map[string]any{"role": "assistant", "content": "ORIGINAL reply"},
map[string]any{"role": "user", "content": "next"},
},
}
parsed2 := &ParsedRequest{
Messages: []any{
map[string]any{"role": "user", "content": "hello"},
map[string]any{"role": "assistant", "content": "TAMPERED reply"},
map[string]any{"role": "user", "content": "next"},
},
}
chain1 := BuildAnthropicDigestChain(parsed1) chain1 := BuildAnthropicDigestChain(parsed1)
chain2 := BuildAnthropicDigestChain(parsed2) chain2 := BuildAnthropicDigestChain(parsed2)
@@ -209,24 +143,16 @@ func TestBuildAnthropicDigestChain_DifferentContentProducesDifferentChain(t *tes
parts1 := splitChain(chain1) parts1 := splitChain(chain1)
parts2 := splitChain(chain2) parts2 := splitChain(chain2)
// 第一个 user message hash 应该相同
if parts1[0] != parts2[0] { if parts1[0] != parts2[0] {
t.Error("First user message hash should be the same") t.Error("First user message hash should be the same")
} }
// assistant reply hash 应该不同
if parts1[1] == parts2[1] { if parts1[1] == parts2[1] {
t.Error("Assistant reply hash should differ") t.Error("Assistant reply hash should differ")
} }
} }
func TestBuildAnthropicDigestChain_Deterministic(t *testing.T) { func TestBuildAnthropicDigestChain_Deterministic(t *testing.T) {
parsed := &ParsedRequest{ parsed := mustParseAnthropicDigestRequest(t, `{"system":"test system","messages":[{"role":"user","content":"hello"},{"role":"assistant","content":"hi"}]}`)
System: "test system",
Messages: []any{
map[string]any{"role": "user", "content": "hello"},
map[string]any{"role": "assistant", "content": "hi"},
},
}
chain1 := BuildAnthropicDigestChain(parsed) chain1 := BuildAnthropicDigestChain(parsed)
chain2 := BuildAnthropicDigestChain(parsed) chain2 := BuildAnthropicDigestChain(parsed)
@@ -236,6 +162,18 @@ func TestBuildAnthropicDigestChain_Deterministic(t *testing.T) {
} }
} }
func TestBuildAnthropicDigestChain_CanonicalJSON(t *testing.T) {
parsed1 := mustParseAnthropicDigestRequest(t, `{"system":[{"type":"text","text":"system"}],"messages":[{"role":"user","content":{"type":"text","text":"hello"}}]}`)
parsed2 := mustParseAnthropicDigestRequest(t, `{"system":[{"text":"system","type":"text"}],"messages":[{"role":"user","content":{"text":"hello","type":"text"}}]}`)
chain1 := BuildAnthropicDigestChain(parsed1)
chain2 := BuildAnthropicDigestChain(parsed2)
if chain1 != chain2 {
t.Errorf("semantically equivalent JSON should produce same chain: %s vs %s", chain1, chain2)
}
}
func TestGenerateAnthropicDigestSessionKey(t *testing.T) { func TestGenerateAnthropicDigestSessionKey(t *testing.T) {
tests := []struct { tests := []struct {
name string name string
@@ -278,7 +216,6 @@ func TestGenerateAnthropicDigestSessionKey(t *testing.T) {
}) })
} }
// 验证不同 uuid 产生不同 sessionKey
t.Run("different uuid different key", func(t *testing.T) { t.Run("different uuid different key", func(t *testing.T) {
hash := "sameprefix123456" hash := "sameprefix123456"
result1 := GenerateAnthropicDigestSessionKey(hash, "uuid0001-session-a") result1 := GenerateAnthropicDigestSessionKey(hash, "uuid0001-session-a")
@@ -297,18 +234,7 @@ func TestAnthropicSessionTTL(t *testing.T) {
} }
func TestBuildAnthropicDigestChain_ContentBlocks(t *testing.T) { func TestBuildAnthropicDigestChain_ContentBlocks(t *testing.T) {
// 测试 content 为 content blocks 数组的情况 parsed := mustParseAnthropicDigestRequest(t, `{"messages":[{"role":"user","content":[{"type":"text","text":"describe this image"},{"type":"image","source":{"type":"base64"}}]}]}`)
parsed := &ParsedRequest{
Messages: []any{
map[string]any{
"role": "user",
"content": []any{
map[string]any{"type": "text", "text": "describe this image"},
map[string]any{"type": "image", "source": map[string]any{"type": "base64"}},
},
},
},
}
result := BuildAnthropicDigestChain(parsed) result := BuildAnthropicDigestChain(parsed)
parts := splitChain(result) parts := splitChain(result)
if len(parts) != 1 { if len(parts) != 1 {
@@ -14,8 +14,6 @@ func TestBuildOAuthMetadataUserID_FallbackWithoutAccountUUID(t *testing.T) {
Model: "claude-sonnet-4-5", Model: "claude-sonnet-4-5",
Stream: true, Stream: true,
MetadataUserID: "", MetadataUserID: "",
System: nil,
Messages: nil,
} }
account := &Account{ account := &Account{
+147 -53
View File
@@ -51,6 +51,12 @@ type SessionContext struct {
APIKeyID int64 APIKeyID int64
} }
type jsonRange struct {
start int // 原始请求体中的起始偏移(闭区间)
end int // 原始请求体中的结束偏移(开区间)
kind gjson.Type // JSON 值类型,用于调用方做轻量分支
}
type RequestBodyRef struct { type RequestBodyRef struct {
data []byte data []byte
} }
@@ -80,6 +86,76 @@ func (b *RequestBodyRef) Replace(data []byte) {
b.data = data b.data = data
} }
func missingJSONRange() jsonRange {
return jsonRange{start: -1, end: -1}
}
func rangeFromResult(r gjson.Result) jsonRange {
if r.Raw == "" || r.Index <= 0 {
return missingJSONRange()
}
end := r.Index + len(r.Raw)
if end < r.Index {
return missingJSONRange()
}
return jsonRange{start: r.Index, end: end, kind: r.Type}
}
func (r jsonRange) exists() bool {
return r.start >= 0 && r.end >= r.start
}
func clearGatewayRequestRanges(parsed *ParsedRequest) {
if parsed == nil {
return
}
parsed.HasSystem = false
parsed.systemRange = missingJSONRange()
parsed.messagesRange = missingJSONRange()
}
func setGatewayRequestRanges(parsed *ParsedRequest, protocol string, jsonStr string) {
if parsed == nil {
return
}
switch protocol {
case domain.PlatformGemini:
if sysParts := gjson.Get(jsonStr, "systemInstruction.parts"); sysParts.Exists() && sysParts.IsArray() {
parsed.systemRange = rangeFromResult(sysParts)
}
if contents := gjson.Get(jsonStr, "contents"); contents.Exists() && contents.IsArray() {
parsed.messagesRange = rangeFromResult(contents)
}
default:
if sys := gjson.Get(jsonStr, "system"); sys.Exists() {
parsed.HasSystem = true
parsed.systemRange = rangeFromResult(sys)
}
if msgs := gjson.Get(jsonStr, "messages"); msgs.Exists() && msgs.IsArray() {
parsed.messagesRange = rangeFromResult(msgs)
}
}
}
func refreshGatewayRequestRanges(parsed *ParsedRequest, protocol string) error {
if parsed == nil {
return fmt.Errorf("empty request body")
}
clearGatewayRequestRanges(parsed)
if parsed.Body == nil {
return fmt.Errorf("empty request body")
}
bodyBytes := parsed.Body.Bytes()
if !gjson.ValidBytes(bodyBytes) {
return fmt.Errorf("invalid json")
}
jsonStr := *(*string)(unsafe.Pointer(&bodyBytes))
setGatewayRequestRanges(parsed, protocol, jsonStr)
return nil
}
// ParsedRequest 保存网关请求的预解析结果 // ParsedRequest 保存网关请求的预解析结果
// //
// 性能优化说明: // 性能优化说明:
@@ -93,18 +169,20 @@ func (b *RequestBodyRef) Replace(data []byte) {
// 2. 将解析结果 ParsedRequest 传递给 Service 层 // 2. 将解析结果 ParsedRequest 传递给 Service 层
// 3. 避免重复 json.Unmarshal,减少 CPU 和内存开销 // 3. 避免重复 json.Unmarshal,减少 CPU 和内存开销
type ParsedRequest struct { type ParsedRequest struct {
Body *RequestBodyRef // 原始请求体引用(保留用于转发) Body *RequestBodyRef // 原始请求体引用(保留用于转发);替换内容请走 ReplaceBody
Model string // 请求的模型名称 Model string // 请求的模型名称
Stream bool // 是否为流式请求 Stream bool // 是否为流式请求
MetadataUserID string // metadata.user_id(用于会话亲和) MetadataUserID string // metadata.user_id(用于会话亲和)
System any // system 字段内容
Messages []any // messages 数组
HasSystem bool // 是否包含 system 字段(包含 null 也视为显式传入) HasSystem bool // 是否包含 system 字段(包含 null 也视为显式传入)
ThinkingEnabled bool // 是否开启 thinking(部分平台会影响最终模型名) ThinkingEnabled bool // 是否开启 thinking(部分平台会影响最终模型名)
OutputEffort string // output_config.effort(Claude API 的推理强度控制) OutputEffort string // output_config.effort(Claude API 的推理强度控制)
MaxTokens int // max_tokens 值(用于探测请求拦截) MaxTokens int // max_tokens 值(用于探测请求拦截)
SessionContext *SessionContext // 可选:请求上下文区分因子(nil 时行为不变) SessionContext *SessionContext // 可选:请求上下文区分因子(nil 时行为不变)
protocol string // 当前 Body 的协议格式,用于 Body 替换后刷新 raw range
systemRange jsonRange // system/systemInstruction.parts 的 raw JSON 范围,绑定 Body 当前内容
messagesRange jsonRange // messages/contents 的 raw JSON 范围,绑定 Body 当前内容
// GroupID 请求所属分组 ID(来自 API Key) // GroupID 请求所属分组 ID(来自 API Key)
GroupID *int64 GroupID *int64
@@ -174,6 +252,9 @@ func ParseGatewayRequest(body *RequestBodyRef, protocol string) (*ParsedRequest,
parsed := &ParsedRequest{ parsed := &ParsedRequest{
Body: body, Body: body,
protocol: protocol,
systemRange: missingJSONRange(),
messagesRange: missingJSONRange(),
} }
// --- gjson 提取简单字段(避免完整 Unmarshal) --- // --- gjson 提取简单字段(避免完整 Unmarshal) ---
@@ -219,60 +300,73 @@ func ParseGatewayRequest(body *RequestBodyRef, protocol string) (*ParsedRequest,
} }
// --- system/messages 提取 --- // --- system/messages 提取 ---
// 避免把整个 body Unmarshal 到 map(会产生大量 map/接口分配)。 // 只保存大字段 raw range,不默认反序列化成 []any/map[string]any 对象图。
// 使用 gjson 抽取目标字段的 Raw,再对该子树进行 Unmarshal。 setGatewayRequestRanges(parsed, protocol, jsonStr)
switch protocol {
case domain.PlatformGemini:
// Gemini 原生格式: systemInstruction.parts / contents
if sysParts := gjson.Get(jsonStr, "systemInstruction.parts"); sysParts.Exists() && sysParts.IsArray() {
var parts []any
if err := json.Unmarshal(sliceRawFromBody(bodyBytes, sysParts), &parts); err != nil {
return nil, err
}
parsed.System = parts
}
if contents := gjson.Get(jsonStr, "contents"); contents.Exists() && contents.IsArray() {
var msgs []any
if err := json.Unmarshal(sliceRawFromBody(bodyBytes, contents), &msgs); err != nil {
return nil, err
}
parsed.Messages = msgs
}
default:
// Anthropic / OpenAI 格式: system / messages
// system 字段只要存在就视为显式提供(即使为 null),
// 以避免客户端传 null 时被默认 system 误注入。
if sys := gjson.Get(jsonStr, "system"); sys.Exists() {
parsed.HasSystem = true
switch sys.Type {
case gjson.Null:
parsed.System = nil
case gjson.String:
// 与 encoding/json 的 Unmarshal 行为一致:返回解码后的字符串。
parsed.System = sys.String()
default:
var system any
if err := json.Unmarshal(sliceRawFromBody(bodyBytes, sys), &system); err != nil {
return nil, err
}
parsed.System = system
}
}
if msgs := gjson.Get(jsonStr, "messages"); msgs.Exists() && msgs.IsArray() {
var messages []any
if err := json.Unmarshal(sliceRawFromBody(bodyBytes, msgs), &messages); err != nil {
return nil, err
}
parsed.Messages = messages
}
}
return parsed, nil return parsed, nil
} }
func (p *ParsedRequest) raw(r jsonRange) []byte {
if p == nil || p.Body == nil || !r.exists() {
return nil
}
body := p.Body.Bytes()
if r.end > len(body) {
return nil
}
return body[r.start:r.end]
}
func (p *ParsedRequest) SystemRaw() []byte {
return p.raw(p.systemRange)
}
func (p *ParsedRequest) MessagesRaw() []byte {
return p.raw(p.messagesRange)
}
func (p *ParsedRequest) DecodeSystem(dst any) error {
raw := p.SystemRaw()
if len(raw) == 0 {
return nil
}
return json.Unmarshal(raw, dst)
}
func (p *ParsedRequest) DecodeMessages(dst any) error {
raw := p.MessagesRaw()
if len(raw) == 0 {
return nil
}
return json.Unmarshal(raw, dst)
}
func (p *ParsedRequest) SystemValue() (any, bool) {
raw := p.SystemRaw()
if len(raw) == 0 {
return nil, false
}
var system any
if err := json.Unmarshal(raw, &system); err != nil {
return nil, false
}
return system, true
}
func (p *ParsedRequest) ReplaceBody(data []byte) {
if p == nil {
return
}
if p.Body == nil {
p.Body = NewRequestBodyRef(data)
} else {
p.Body.Replace(data)
}
if err := refreshGatewayRequestRanges(p, p.protocol); err != nil {
clearGatewayRequestRanges(p)
}
}
// sliceRawFromBody 返回 Result.Raw 对应的原始字节切片。 // sliceRawFromBody 返回 Result.Raw 对应的原始字节切片。
// 优先使用 Result.Index 直接从 body 切片,避免对大字段(如 messages)产生额外拷贝。 // 优先使用 Result.Index 直接从 body 切片,避免对大字段(如 messages)产生额外拷贝。
// 当 Index 不可用时,退化为复制(理论上极少发生)。 // 当 Index 不可用时,退化为复制(理论上极少发生)。
@@ -10,6 +10,7 @@ import (
"github.com/Wei-Shaw/sub2api/internal/domain" "github.com/Wei-Shaw/sub2api/internal/domain"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
"github.com/tidwall/gjson"
) )
func TestParseGatewayRequest(t *testing.T) { func TestParseGatewayRequest(t *testing.T) {
@@ -20,8 +21,8 @@ func TestParseGatewayRequest(t *testing.T) {
require.True(t, parsed.Stream) require.True(t, parsed.Stream)
require.Equal(t, "session_123e4567-e89b-12d3-a456-426614174000", parsed.MetadataUserID) require.Equal(t, "session_123e4567-e89b-12d3-a456-426614174000", parsed.MetadataUserID)
require.True(t, parsed.HasSystem) require.True(t, parsed.HasSystem)
require.NotNil(t, parsed.System) require.NotEmpty(t, parsed.SystemRaw())
require.Len(t, parsed.Messages, 1) require.NotEmpty(t, parsed.MessagesRaw())
require.False(t, parsed.ThinkingEnabled) require.False(t, parsed.ThinkingEnabled)
} }
@@ -61,7 +62,7 @@ func TestParseGatewayRequest_SystemNull(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
// 显式传入 system:null 也应视为“字段已存在”,避免默认 system 被注入。 // 显式传入 system:null 也应视为“字段已存在”,避免默认 system 被注入。
require.True(t, parsed.HasSystem) require.True(t, parsed.HasSystem)
require.Nil(t, parsed.System) require.Equal(t, []byte("null"), parsed.SystemRaw())
} }
func TestParseGatewayRequest_InvalidModelType(t *testing.T) { func TestParseGatewayRequest_InvalidModelType(t *testing.T) {
@@ -88,9 +89,9 @@ func TestParseGatewayRequest_GeminiContents(t *testing.T) {
}`) }`)
parsed, err := ParseGatewayRequest(NewRequestBodyRef(body), domain.PlatformGemini) parsed, err := ParseGatewayRequest(NewRequestBodyRef(body), domain.PlatformGemini)
require.NoError(t, err) require.NoError(t, err)
require.Len(t, parsed.Messages, 3, "should parse contents as Messages") require.Len(t, gjson.ParseBytes(parsed.MessagesRaw()).Array(), 3, "should parse contents as Messages")
require.False(t, parsed.HasSystem, "Gemini format should not set HasSystem") require.False(t, parsed.HasSystem, "Gemini format should not set HasSystem")
require.Nil(t, parsed.System, "no systemInstruction means nil System") require.Nil(t, parsed.SystemRaw(), "no systemInstruction means nil System")
} }
func TestParseGatewayRequest_GeminiSystemInstruction(t *testing.T) { func TestParseGatewayRequest_GeminiSystemInstruction(t *testing.T) {
@@ -104,14 +105,11 @@ func TestParseGatewayRequest_GeminiSystemInstruction(t *testing.T) {
}`) }`)
parsed, err := ParseGatewayRequest(NewRequestBodyRef(body), domain.PlatformGemini) parsed, err := ParseGatewayRequest(NewRequestBodyRef(body), domain.PlatformGemini)
require.NoError(t, err) require.NoError(t, err)
require.NotNil(t, parsed.System, "should parse systemInstruction.parts as System") system := gjson.ParseBytes(parsed.SystemRaw())
parts, ok := parsed.System.([]any) require.True(t, system.IsArray(), "should parse systemInstruction.parts as System")
require.True(t, ok) require.Len(t, system.Array(), 1)
require.Len(t, parts, 1) require.Equal(t, "You are a helpful assistant.", system.Get("0.text").String())
partMap, ok := parts[0].(map[string]any) require.Len(t, gjson.ParseBytes(parsed.MessagesRaw()).Array(), 1)
require.True(t, ok)
require.Equal(t, "You are a helpful assistant.", partMap["text"])
require.Len(t, parsed.Messages, 1)
} }
func TestParseGatewayRequest_GeminiWithModel(t *testing.T) { func TestParseGatewayRequest_GeminiWithModel(t *testing.T) {
@@ -122,7 +120,7 @@ func TestParseGatewayRequest_GeminiWithModel(t *testing.T) {
parsed, err := ParseGatewayRequest(NewRequestBodyRef(body), domain.PlatformGemini) parsed, err := ParseGatewayRequest(NewRequestBodyRef(body), domain.PlatformGemini)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, "gemini-2.5-pro", parsed.Model) require.Equal(t, "gemini-2.5-pro", parsed.Model)
require.Len(t, parsed.Messages, 1) require.Len(t, gjson.ParseBytes(parsed.MessagesRaw()).Array(), 1)
} }
func TestParseGatewayRequest_GeminiIgnoresAnthropicFields(t *testing.T) { func TestParseGatewayRequest_GeminiIgnoresAnthropicFields(t *testing.T) {
@@ -135,22 +133,22 @@ func TestParseGatewayRequest_GeminiIgnoresAnthropicFields(t *testing.T) {
parsed, err := ParseGatewayRequest(NewRequestBodyRef(body), domain.PlatformGemini) parsed, err := ParseGatewayRequest(NewRequestBodyRef(body), domain.PlatformGemini)
require.NoError(t, err) require.NoError(t, err)
require.False(t, parsed.HasSystem, "Gemini protocol should not parse Anthropic system field") require.False(t, parsed.HasSystem, "Gemini protocol should not parse Anthropic system field")
require.Nil(t, parsed.System, "no systemInstruction = nil System") require.Nil(t, parsed.SystemRaw(), "no systemInstruction = nil System")
require.Len(t, parsed.Messages, 1, "should use contents, not messages") require.Len(t, gjson.ParseBytes(parsed.MessagesRaw()).Array(), 1, "should use contents, not messages")
} }
func TestParseGatewayRequest_GeminiEmptyContents(t *testing.T) { func TestParseGatewayRequest_GeminiEmptyContents(t *testing.T) {
body := []byte(`{"contents": []}`) body := []byte(`{"contents": []}`)
parsed, err := ParseGatewayRequest(NewRequestBodyRef(body), domain.PlatformGemini) parsed, err := ParseGatewayRequest(NewRequestBodyRef(body), domain.PlatformGemini)
require.NoError(t, err) require.NoError(t, err)
require.Empty(t, parsed.Messages) require.Empty(t, gjson.ParseBytes(parsed.MessagesRaw()).Array())
} }
func TestParseGatewayRequest_GeminiNoContents(t *testing.T) { func TestParseGatewayRequest_GeminiNoContents(t *testing.T) {
body := []byte(`{"model": "gemini-2.5-flash"}`) body := []byte(`{"model": "gemini-2.5-flash"}`)
parsed, err := ParseGatewayRequest(NewRequestBodyRef(body), domain.PlatformGemini) parsed, err := ParseGatewayRequest(NewRequestBodyRef(body), domain.PlatformGemini)
require.NoError(t, err) require.NoError(t, err)
require.Nil(t, parsed.Messages) require.Nil(t, parsed.MessagesRaw())
require.Equal(t, "gemini-2.5-flash", parsed.Model) require.Equal(t, "gemini-2.5-flash", parsed.Model)
} }
@@ -165,11 +163,10 @@ func TestParseGatewayRequest_AnthropicIgnoresGeminiFields(t *testing.T) {
parsed, err := ParseGatewayRequest(NewRequestBodyRef(body), domain.PlatformAnthropic) parsed, err := ParseGatewayRequest(NewRequestBodyRef(body), domain.PlatformAnthropic)
require.NoError(t, err) require.NoError(t, err)
require.True(t, parsed.HasSystem) require.True(t, parsed.HasSystem)
require.Equal(t, "real system", parsed.System) require.Equal(t, "real system", gjson.ParseBytes(parsed.SystemRaw()).String())
require.Len(t, parsed.Messages, 1) messages := gjson.ParseBytes(parsed.MessagesRaw()).Array()
msg, ok := parsed.Messages[0].(map[string]any) require.Len(t, messages, 1)
require.True(t, ok) require.Equal(t, "real content", messages[0].Get("content").String())
require.Equal(t, "real content", msg["content"])
} }
func TestFilterThinkingBlocks(t *testing.T) { func TestFilterThinkingBlocks(t *testing.T) {
@@ -970,10 +967,10 @@ func TestParseGatewayRequest_OptionalFieldsMissing(t *testing.T) {
require.Equal(t, tt.wantMaxTokens, parsed.MaxTokens) require.Equal(t, tt.wantMaxTokens, parsed.MaxTokens)
if tt.wantMessagesNil { if tt.wantMessagesNil {
require.Nil(t, parsed.Messages) require.Nil(t, parsed.MessagesRaw())
} }
if tt.wantMessagesLen > 0 { if tt.wantMessagesLen > 0 {
require.Len(t, parsed.Messages, tt.wantMessagesLen) require.Len(t, gjson.ParseBytes(parsed.MessagesRaw()).Array(), tt.wantMessagesLen)
} }
}) })
} }
@@ -1087,25 +1084,8 @@ func parseGatewayRequestOld(body []byte, protocol string) (*ParsedRequest, error
} }
} }
// system / messages(按协议分支) if err := refreshGatewayRequestRanges(parsed, protocol); err != nil {
switch protocol { return nil, err
case domain.PlatformGemini:
if sysInst, ok := req["systemInstruction"].(map[string]any); ok {
if parts, ok := sysInst["parts"].([]any); ok {
parsed.System = parts
}
}
if contents, ok := req["contents"].([]any); ok {
parsed.Messages = contents
}
default:
if system, ok := req["system"]; ok {
parsed.HasSystem = true
parsed.System = system
}
if messages, ok := req["messages"].([]any); ok {
parsed.Messages = messages
}
} }
return parsed, nil return parsed, nil
+110 -85
View File
@@ -748,31 +748,10 @@ func (s *GatewayService) GenerateSessionHash(parsed *ParsedRequest) string {
_, _ = combined.WriteString(strconv.FormatInt(parsed.SessionContext.APIKeyID, 10)) _, _ = combined.WriteString(strconv.FormatInt(parsed.SessionContext.APIKeyID, 10))
_, _ = combined.WriteString("|") _, _ = combined.WriteString("|")
} }
if parsed.System != nil { if systemText := extractTextFromSystemRaw(parsed.SystemRaw()); systemText != "" {
systemText := s.extractTextFromSystem(parsed.System)
if systemText != "" {
_, _ = combined.WriteString(systemText) _, _ = combined.WriteString(systemText)
} }
} appendMessageTextsFromRaw(&combined, parsed.MessagesRaw())
for _, msg := range parsed.Messages {
if m, ok := msg.(map[string]any); ok {
if content, exists := m["content"]; exists {
// Anthropic: messages[].content
if msgText := s.extractTextFromContent(content); msgText != "" {
_, _ = combined.WriteString(msgText)
}
} else if parts, ok := m["parts"].([]any); ok {
// Gemini: contents[].parts[].text
for _, part := range parts {
if partMap, ok := part.(map[string]any); ok {
if text, ok := partMap["text"].(string); ok {
_, _ = combined.WriteString(text)
}
}
}
}
}
}
if combined.Len() > 0 { if combined.Len() > 0 {
hash := s.hashContent(combined.String()) hash := s.hashContent(combined.String())
slog.Info("sticky.hash_source", slog.Info("sticky.hash_source",
@@ -847,82 +826,127 @@ func (s *GatewayService) extractCacheableContent(parsed *ParsedRequest) string {
return "" return ""
} }
var builder strings.Builder systemText := extractCacheableTextFromSystemRaw(parsed.SystemRaw())
if messageText := extractCacheableTextFromMessagesRaw(parsed.MessagesRaw()); messageText != "" {
// 检查 system 中的 cacheable 内容 return messageText
if system, ok := parsed.System.([]any); ok {
for _, part := range system {
if partMap, ok := part.(map[string]any); ok {
if cc, ok := partMap["cache_control"].(map[string]any); ok {
if cc["type"] == "ephemeral" {
if text, ok := partMap["text"].(string); ok {
_, _ = builder.WriteString(text)
} }
}
}
}
}
}
systemText := builder.String()
// 检查 messages 中的 cacheable 内容
for _, msg := range parsed.Messages {
if msgMap, ok := msg.(map[string]any); ok {
if msgContent, ok := msgMap["content"].([]any); ok {
for _, part := range msgContent {
if partMap, ok := part.(map[string]any); ok {
if cc, ok := partMap["cache_control"].(map[string]any); ok {
if cc["type"] == "ephemeral" {
return s.extractTextFromContent(msgMap["content"])
}
}
}
}
}
}
}
return systemText return systemText
} }
func (s *GatewayService) extractTextFromSystem(system any) string { func extractTextFromSystemRaw(raw []byte) string {
switch v := system.(type) { system := gjson.ParseBytes(raw)
case string: switch system.Type {
return v case gjson.String:
case []any: return system.String()
var texts []string case gjson.JSON:
for _, part := range v { if !system.IsArray() {
if partMap, ok := part.(map[string]any); ok { return ""
if text, ok := partMap["text"].(string); ok {
texts = append(texts, text)
} }
var builder strings.Builder
system.ForEach(func(_, part gjson.Result) bool {
if text := part.Get("text").String(); text != "" {
_, _ = builder.WriteString(text)
} }
} return true
return strings.Join(texts, "") })
return builder.String()
} }
return "" return ""
} }
func (s *GatewayService) extractTextFromContent(content any) string { func extractTextFromContentRaw(content gjson.Result) string {
switch v := content.(type) { switch content.Type {
case string: case gjson.String:
return v return content.String()
case []any: case gjson.JSON:
var texts []string if !content.IsArray() {
for _, part := range v { return ""
if partMap, ok := part.(map[string]any); ok { }
if partMap["type"] == "text" { var builder strings.Builder
if text, ok := partMap["text"].(string); ok { content.ForEach(func(_, part gjson.Result) bool {
texts = append(texts, text) if part.Get("type").String() == "text" {
if text := part.Get("text").String(); text != "" {
_, _ = builder.WriteString(text)
} }
} }
} return true
} })
return strings.Join(texts, "") return builder.String()
} }
return "" return ""
} }
func appendMessageTextsFromRaw(builder *strings.Builder, raw []byte) {
if builder == nil || len(raw) == 0 {
return
}
messages := gjson.ParseBytes(raw)
if !messages.IsArray() {
return
}
messages.ForEach(func(_, msg gjson.Result) bool {
if content := msg.Get("content"); content.Exists() {
_, _ = builder.WriteString(extractTextFromContentRaw(content))
return true
}
parts := msg.Get("parts")
if parts.IsArray() {
parts.ForEach(func(_, part gjson.Result) bool {
if text := part.Get("text").String(); text != "" {
_, _ = builder.WriteString(text)
}
return true
})
}
return true
})
}
func extractCacheableTextFromSystemRaw(raw []byte) string {
system := gjson.ParseBytes(raw)
if !system.IsArray() {
return ""
}
var builder strings.Builder
system.ForEach(func(_, part gjson.Result) bool {
if part.Get("cache_control.type").String() == "ephemeral" {
if text := part.Get("text").String(); text != "" {
_, _ = builder.WriteString(text)
}
}
return true
})
return builder.String()
}
func extractCacheableTextFromMessagesRaw(raw []byte) string {
messages := gjson.ParseBytes(raw)
if !messages.IsArray() {
return ""
}
var text string
messages.ForEach(func(_, msg gjson.Result) bool {
content := msg.Get("content")
if !content.IsArray() {
return true
}
found := false
content.ForEach(func(_, part gjson.Result) bool {
if part.Get("cache_control.type").String() == "ephemeral" {
found = true
return false
}
return true
})
if found {
text = extractTextFromContentRaw(content)
return false
}
return true
})
return text
}
func (s *GatewayService) hashContent(content string) string { func (s *GatewayService) hashContent(content string) string {
h := xxhash.Sum64String(content) h := xxhash.Sum64String(content)
return strconv.FormatUint(h, 36) return strconv.FormatUint(h, 36)
@@ -1284,7 +1308,7 @@ func (s *GatewayService) applyClaudeCodeOAuthMimicryToBody(
systemRewritten := false systemRewritten := false
if !strings.Contains(strings.ToLower(model), "haiku") { if !strings.Contains(strings.ToLower(model), "haiku") {
body = rewriteSystemForNonClaudeCode(body, systemRaw) body = rewriteSystemForNonClaudeCode(body, normalizeSystemParam(systemRaw))
systemRewritten = true systemRewritten = true
} }
@@ -4474,7 +4498,8 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A
// Parrot 的 transform_request 从不检查客户端 system 内容,直接覆盖。 // Parrot 的 transform_request 从不检查客户端 system 内容,直接覆盖。
systemRewritten := false systemRewritten := false
if !strings.Contains(strings.ToLower(reqModel), "haiku") { if !strings.Contains(strings.ToLower(reqModel), "haiku") {
body = rewriteSystemForNonClaudeCode(body, parsed.System) systemRaw, _ := parsed.SystemValue()
body = rewriteSystemForNonClaudeCode(body, systemRaw)
systemRewritten = true systemRewritten = true
} }
@@ -2,6 +2,7 @@ package service
import ( import (
"strconv" "strconv"
"strings"
"testing" "testing"
) )
@@ -34,17 +35,20 @@ func BenchmarkExtractCacheableContent_System(b *testing.B) {
} }
func buildSystemCacheableRequest(parts int) *ParsedRequest { func buildSystemCacheableRequest(parts int) *ParsedRequest {
systemParts := make([]any, 0, parts) var builder strings.Builder
builder.WriteString(`{"system":[`)
for i := 0; i < parts; i++ { for i := 0; i < parts; i++ {
systemParts = append(systemParts, map[string]any{ if i > 0 {
"text": "system_part_" + strconv.Itoa(i), builder.WriteByte(',')
"cache_control": map[string]any{
"type": "ephemeral",
},
})
} }
return &ParsedRequest{ builder.WriteString(`{"text":"system_part_`)
System: systemParts, builder.WriteString(strconv.Itoa(i))
HasSystem: true, builder.WriteString(`","cache_control":{"type":"ephemeral"}}`)
} }
builder.WriteString(`]}`)
parsed, err := ParseGatewayRequest(NewRequestBodyRef([]byte(builder.String())), "")
if err != nil {
panic(err)
}
return parsed
} }
File diff suppressed because it is too large. Load diff
@@ -36,6 +36,13 @@ import (
"go.uber.org/zap" "go.uber.org/zap"
) )
// openAIParsedRequestBodyCache 绑定 body 指纹,避免 handler 预解析的旧 body 污染后续 forwardBody。
type openAIParsedRequestBodyCache struct {
bodyHash uint64
bodyLen int
reqBody map[string]any
}
const ( const (
// ChatGPT internal API for OAuth accounts // ChatGPT internal API for OAuth accounts
chatgptCodexURL = "https://chatgpt.com/backend-api/codex/responses" chatgptCodexURL = "https://chatgpt.com/backend-api/codex/responses"
@@ -2284,6 +2291,20 @@ func (s *OpenAIGatewayService) shouldFailoverOpenAIUpstreamResponse(statusCode i
return isOpenAITransientProcessingError(statusCode, upstreamMsg, upstreamBody) return isOpenAITransientProcessingError(statusCode, upstreamMsg, upstreamBody)
} }
func marshalOpenAIUpstreamJSON(v any) ([]byte, error) {
var buf bytes.Buffer
enc := json.NewEncoder(&buf)
enc.SetEscapeHTML(false)
if err := enc.Encode(v); err != nil {
return nil, err
}
out := buf.Bytes()
if len(out) > 0 && out[len(out)-1] == '\n' {
out = out[:len(out)-1]
}
return out, nil
}
func (s *OpenAIGatewayService) handleFailoverSideEffects(ctx context.Context, resp *http.Response, account *Account, requestedModel ...string) { func (s *OpenAIGatewayService) handleFailoverSideEffects(ctx context.Context, resp *http.Response, account *Account, requestedModel ...string) {
body, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20)) body, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
if len(requestedModel) > 0 { if len(requestedModel) > 0 {
@@ -2561,7 +2582,11 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
// 确保高版本模型向低版本模型映射不报错 // 确保高版本模型向低版本模型映射不报错
if !SupportsVerbosity(upstreamModel) { if !SupportsVerbosity(upstreamModel) {
if text, ok := reqBody["text"].(map[string]any); ok { if text, ok := reqBody["text"].(map[string]any); ok {
if _, exists := text["verbosity"]; exists {
delete(text, "verbosity") delete(text, "verbosity")
bodyModified = true
markPatchDelete("text.verbosity")
}
} }
} }
} }
@@ -2762,7 +2787,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
} }
if !serializedByPatch { if !serializedByPatch {
var marshalErr error var marshalErr error
body, marshalErr = json.Marshal(reqBody) body, marshalErr = marshalOpenAIUpstreamJSON(reqBody)
if marshalErr != nil { if marshalErr != nil {
return nil, fmt.Errorf("serialize request body: %w", marshalErr) return nil, fmt.Errorf("serialize request body: %w", marshalErr)
} }
@@ -3043,7 +3068,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
upstreamCode := extractUpstreamErrorCode(respBody) upstreamCode := extractUpstreamErrorCode(respBody)
if !httpInvalidEncryptedContentRetryTried && resp.StatusCode == http.StatusBadRequest && upstreamCode == "invalid_encrypted_content" { if !httpInvalidEncryptedContentRetryTried && resp.StatusCode == http.StatusBadRequest && upstreamCode == "invalid_encrypted_content" {
if trimOpenAIEncryptedReasoningItems(reqBody) { if trimOpenAIEncryptedReasoningItems(reqBody) {
body, err = json.Marshal(reqBody) body, err = marshalOpenAIUpstreamJSON(reqBody)
if err != nil { if err != nil {
return nil, fmt.Errorf("serialize invalid_encrypted_content retry body: %w", err) return nil, fmt.Errorf("serialize invalid_encrypted_content retry body: %w", err)
} }
@@ -3074,6 +3099,8 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
}) })
s.handleFailoverSideEffects(ctx, resp, account, upstreamModel) s.handleFailoverSideEffects(ctx, resp, account, upstreamModel)
// reqBody 会被本次账号尝试原地修改,failover 前必须释放,避免下一账号复用脏 map。
releaseOpenAIParsedRequestBody(c)
return nil, &UpstreamFailoverError{ return nil, &UpstreamFailoverError{
StatusCode: resp.StatusCode, StatusCode: resp.StatusCode,
ResponseBody: respBody, ResponseBody: respBody,
@@ -6711,7 +6738,7 @@ func sanitizeEmptyBase64InputImagesInOpenAIBody(body []byte) ([]byte, bool, erro
if !sanitizeEmptyBase64InputImagesInOpenAIRequestBodyMap(reqBody) { if !sanitizeEmptyBase64InputImagesInOpenAIRequestBodyMap(reqBody) {
return body, false, nil return body, false, nil
} }
normalized, err := json.Marshal(reqBody) normalized, err := marshalOpenAIUpstreamJSON(reqBody)
if err != nil { if err != nil {
return body, false, fmt.Errorf("serialize sanitized request body: %w", err) return body, false, fmt.Errorf("serialize sanitized request body: %w", err)
} }
@@ -6817,10 +6844,13 @@ func isEmptyBase64DataURI(raw string) bool {
} }
func getOpenAIRequestBodyMap(c *gin.Context, body []byte) (map[string]any, error) { func getOpenAIRequestBodyMap(c *gin.Context, body []byte) (map[string]any, error) {
// 同一个 gin.Context 内 failover/渠道映射可能传入新 body,缓存必须先校验 body 指纹。
bodyHash := xxhash.Sum64(body)
bodyLen := len(body)
if c != nil { if c != nil {
if cached, ok := c.Get(OpenAIParsedRequestBodyKey); ok { if cached, ok := c.Get(OpenAIParsedRequestBodyKey); ok {
if reqBody, ok := cached.(map[string]any); ok && reqBody != nil { if cache, ok := cached.(openAIParsedRequestBodyCache); ok && cache.reqBody != nil && cache.bodyLen == bodyLen && cache.bodyHash == bodyHash {
return reqBody, nil return cache.reqBody, nil
} }
} }
} }
@@ -6830,11 +6860,36 @@ func getOpenAIRequestBodyMap(c *gin.Context, body []byte) (map[string]any, error
return nil, fmt.Errorf("parse request: %w", err) return nil, fmt.Errorf("parse request: %w", err)
} }
if c != nil { if c != nil {
c.Set(OpenAIParsedRequestBodyKey, reqBody) c.Set(OpenAIParsedRequestBodyKey, openAIParsedRequestBodyCache{bodyHash: bodyHash, bodyLen: bodyLen, reqBody: reqBody})
} }
return reqBody, nil return reqBody, nil
} }
// CacheOpenAIParsedRequestBody 仅缓存与当前 body 绑定的解析结果。
func CacheOpenAIParsedRequestBody(c *gin.Context, body []byte, reqBody map[string]any) {
if c == nil || reqBody == nil {
return
}
c.Set(OpenAIParsedRequestBodyKey, openAIParsedRequestBodyCache{
bodyHash: xxhash.Sum64(body),
bodyLen: len(body),
reqBody: reqBody,
})
}
// CachedOpenAIParsedRequestBody 只给同请求内不关心 body 参数的轻量识别逻辑使用。
func CachedOpenAIParsedRequestBody(c *gin.Context) map[string]any {
if c == nil {
return nil
}
if cached, ok := c.Get(OpenAIParsedRequestBodyKey); ok {
if cache, ok := cached.(openAIParsedRequestBodyCache); ok {
return cache.reqBody
}
}
return nil
}
func releaseOpenAIParsedRequestBody(c *gin.Context) { func releaseOpenAIParsedRequestBody(c *gin.Context) {
if c == nil { if c == nil {
return return
@@ -112,7 +112,7 @@ func TestGetOpenAIRequestBodyMap_UsesContextCache(t *testing.T) {
c, _ := gin.CreateTestContext(rec) c, _ := gin.CreateTestContext(rec)
cached := map[string]any{"model": "cached-model", "stream": true} cached := map[string]any{"model": "cached-model", "stream": true}
c.Set(OpenAIParsedRequestBodyKey, cached) CacheOpenAIParsedRequestBody(c, []byte(`{invalid-json`), cached)
got, err := getOpenAIRequestBodyMap(c, []byte(`{invalid-json`)) got, err := getOpenAIRequestBodyMap(c, []byte(`{invalid-json`))
require.NoError(t, err) require.NoError(t, err)
@@ -134,11 +134,19 @@ func TestGetOpenAIRequestBodyMap_WriteBackContextCache(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, "gpt-5", got["model"]) require.Equal(t, "gpt-5", got["model"])
cached, ok := c.Get(OpenAIParsedRequestBodyKey) require.Equal(t, got, CachedOpenAIParsedRequestBody(c))
require.True(t, ok) }
cachedMap, ok := cached.(map[string]any)
require.True(t, ok) func TestGetOpenAIRequestBodyMap_IgnoresCacheForDifferentBody(t *testing.T) {
require.Equal(t, got, cachedMap) gin.SetMode(gin.TestMode)
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
CacheOpenAIParsedRequestBody(c, []byte(`{"model":"cached-model"}`), map[string]any{"model": "cached-model"})
got, err := getOpenAIRequestBodyMap(c, []byte(`{"model":"forward-model"}`))
require.NoError(t, err)
require.Equal(t, "forward-model", got["model"])
} }
func TestSanitizeEmptyBase64InputImagesInOpenAIRequestBodyMap(t *testing.T) { func TestSanitizeEmptyBase64InputImagesInOpenAIRequestBodyMap(t *testing.T) {
@@ -14,6 +14,7 @@ import (
"github.com/Wei-Shaw/sub2api/internal/config" "github.com/Wei-Shaw/sub2api/internal/config"
"github.com/Wei-Shaw/sub2api/internal/pkg/logger" "github.com/Wei-Shaw/sub2api/internal/pkg/logger"
"github.com/Wei-Shaw/sub2api/internal/pkg/openai_compat"
"github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint" "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint"
"github.com/gin-gonic/gin" "github.com/gin-gonic/gin"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
@@ -101,6 +102,57 @@ func TestOpenAIGatewayService_ResponsesUnknownModelDoesNotFallbackToGPT54(t *tes
require.True(t, rec.Code >= http.StatusBadRequest) require.True(t, rec.Code >= http.StatusBadRequest)
} }
func TestOpenAIGatewayService_NativeResponsesBodyModificationPreservesHTMLChars(t *testing.T) {
gin.SetMode(gin.TestMode)
payloadText := strings.Repeat(`<tag>&value</tag>`, 128)
originalBody := []byte(fmt.Sprintf(`{"model":"gpt-5.5","stream":false,"max_output_tokens":100,"previous_response_id":"resp_prev","input":[{"type":"message","role":"user","content":[{"type":"input_text","text":%q}]}]}`, payloadText))
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(originalBody))
c.Request.Header.Set("Content-Type", "application/json")
upstream := &httpUpstreamRecorder{resp: &http.Response{
StatusCode: http.StatusBadRequest,
Header: http.Header{"Content-Type": []string{"application/json"}, "x-request-id": []string{"rid_native_reencode"}},
Body: io.NopCloser(strings.NewReader(`{"error":{"type":"invalid_request_error","message":"stop after capture"}}`)),
}}
svc := &OpenAIGatewayService{
cfg: &config.Config{Security: config.SecurityConfig{URLAllowlist: config.URLAllowlistConfig{
Enabled: false,
AllowInsecureHTTP: true,
}}},
httpUpstream: upstream,
}
account := &Account{
ID: 456,
Name: "openai-apikey",
Platform: PlatformOpenAI,
Type: AccountTypeAPIKey,
Concurrency: 1,
Credentials: map[string]any{
"api_key": "sk-test",
"base_url": "http://upstream.example",
},
Extra: map[string]any{
openai_compat.ExtraKeyResponsesMode: string(openai_compat.ResponsesSupportModeAuto),
openai_compat.ExtraKeyResponsesSupported: true,
},
Status: StatusActive,
Schedulable: true,
}
result, err := svc.Forward(context.Background(), c, account, originalBody)
require.Error(t, err)
require.Nil(t, result)
require.NotNil(t, upstream.lastReq)
require.Equal(t, "http://upstream.example/v1/responses", upstream.lastReq.URL.String())
require.Contains(t, string(upstream.lastBody), payloadText)
require.NotContains(t, string(upstream.lastBody), `\\u003c`)
require.NotContains(t, string(upstream.lastBody), `\\u003e`)
require.NotContains(t, string(upstream.lastBody), `\\u0026`)
}
func TestOpenAIGatewayService_OAuthMessagesBridgeDoesNotInjectDefaultInstructions(t *testing.T) { func TestOpenAIGatewayService_OAuthMessagesBridgeDoesNotInjectDefaultInstructions(t *testing.T) {
gin.SetMode(gin.TestMode) gin.SetMode(gin.TestMode)
@@ -12,6 +12,7 @@ import (
"github.com/Wei-Shaw/sub2api/internal/config" "github.com/Wei-Shaw/sub2api/internal/config"
"github.com/Wei-Shaw/sub2api/internal/pkg/logger" "github.com/Wei-Shaw/sub2api/internal/pkg/logger"
"github.com/tidwall/gjson"
) )
// UserMsgQueueCache 用户消息串行队列 Redis 缓存接口 // UserMsgQueueCache 用户消息串行队列 Redis 缓存接口
@@ -62,43 +63,48 @@ func NewUserMessageQueueService(cache UserMsgQueueCache, rpmCache RPMCache, cfg
// 2. 最后一条消息 role == "user" // 2. 最后一条消息 role == "user"
// 3. 最后一条消息 content(如果是数组)中不含 type:"tool_result" / "tool_use_result" // 3. 最后一条消息 content(如果是数组)中不含 type:"tool_result" / "tool_use_result"
func IsRealUserMessage(parsed *ParsedRequest) bool { func IsRealUserMessage(parsed *ParsedRequest) bool {
if parsed == nil || len(parsed.Messages) == 0 { if parsed == nil {
return false
}
messagesRaw := parsed.MessagesRaw()
if len(messagesRaw) == 0 {
return false return false
} }
lastMsg := parsed.Messages[len(parsed.Messages)-1] messages := gjson.ParseBytes(messagesRaw)
msgMap, ok := lastMsg.(map[string]any) if !messages.IsArray() {
if !ok { return false
}
lastMsg := gjson.Result{}
messages.ForEach(func(_, msg gjson.Result) bool {
lastMsg = msg
return true
})
if !lastMsg.Exists() || !lastMsg.IsObject() {
return false
}
if lastMsg.Get("role").String() != "user" {
return false return false
} }
role, _ := msgMap["role"].(string) content := lastMsg.Get("content")
if role != "user" { if !content.Exists() {
return false return true
}
if !content.IsArray() {
return true
} }
// 检查 content 是否包含 tool_result 类型 isReal := true
content, ok := msgMap["content"] content.ForEach(func(_, item gjson.Result) bool {
if !ok { itemType := item.Get("type").String()
return true // 没有 content 字段,视为普通用户消息
}
contentArr, ok := content.([]any)
if !ok {
return true // content 不是数组(可能是 string),视为普通用户消息
}
for _, item := range contentArr {
itemMap, ok := item.(map[string]any)
if !ok {
continue
}
itemType, _ := itemMap["type"].(string)
if itemType == "tool_result" || itemType == "tool_use_result" { if itemType == "tool_result" || itemType == "tool_use_result" {
isReal = false
return false return false
} }
}
return true return true
})
return isReal
} }
// TryAcquire 尝试立即获取串行锁 // TryAcquire 尝试立即获取串行锁