diff --git a/README.md b/README.md index 16ebab0a32..5d0c4f433a 100644 --- a/README.md +++ b/README.md @@ -58,6 +58,11 @@ Please read the following carefully before using this project: Thanks to AIGoCode for sponsoring this project! AIGoCode is an all-in-one platform that integrates Claude Code, Codex, and the latest Gemini models, providing you with stable, efficient, and highly cost-effective AI coding services. The platform offers flexible subscription plans, zero risk of account suspension, direct access with no VPN required, and lightning-fast responses. AIGoCode has prepared a special benefit for sub2api users: if you register via this link, you'll receive an extra 10% bonus credit on your first top-up! + +CodexEverywhere +Real GPT-5.6 series at 3% of OpenAI pricing — CodexEverywhere is democratizing access to frontier models for developers worldwide. We believe in transparency and honesty, with model quality verified by active community oversight for months. USD and crypto friendly. Start with a free $20 trial at codex-everywhere.com. + + bmoplus Huge thanks to BmoPlus for sponsoring this project! BmoPlus is a highly reliable AI account provider built strictly for heavy AI users and developers. They offer rock-solid, ready-to-use accounts and official top-up services for ChatGPT Plus / ChatGPT Pro (Full Warranty) / Claude Pro / Super Grok / Gemini Pro. By registering and ordering through BmoPlus - Premium AI Accounts & Top-ups, users can unlock the mind-blowing rate of 10% of the official GPT subscription price (90% OFF) diff --git a/README_CN.md b/README_CN.md index e879045374..faa0200c58 100644 --- a/README_CN.md +++ b/README_CN.md @@ -59,6 +59,11 @@ 感谢 AIGoCode 赞助了本项目！AIGoCode 是一站式集成 Claude Code、Codex 以及最新 Gemini 模型的综合平台，为您提供稳定、高效、高性价比的 AI 编程服务。平台提供灵活的订阅方案，零封号风险，免 VPN 直连，响应极速。AIGoCode 为 sub2api 用户准备了专属福利：通过此链接注册，首次充值可额外获得 10% 赠送额度！ + +CodexEverywhere +Real GPT-5.6 series at 3% of OpenAI pricing — CodexEverywhere is democratizing access to frontier models for developers worldwide. We believe in transparency and honesty, with model quality verified by active community oversight for months. USD and crypto friendly. Start with a free $20 trial at codex-everywhere.com. + + bmoplus 感谢 BmoPlus 赞助了本项目！BmoPlus 是一家专为AI订阅重度用户打造的可靠 AI 账号代充服务商，提供稳定的 ChatGPT Plus / ChatGPT Pro(全程质保) / Claude Pro / Super Grok / Gemini Pro 的官方代充&成品账号。 通过BmoPlus AI成品号专卖/代充注册下单的用户，可享GPT 官网订阅一折 的震撼价格！ diff --git a/README_JA.md b/README_JA.md index 17890e4557..9bd3802a32 100644 --- a/README_JA.md +++ b/README_JA.md @@ -58,6 +58,11 @@ AIGoCode のご支援に感謝します！AIGoCode は Claude Code、Codex、最新の Gemini モデルを統合したオールインワンプラットフォームで、安定的かつ効率的でコストパフォーマンスに優れた AI コーディングサービスを提供します。柔軟なサブスクリプションプラン、アカウント停止リスクゼロ、VPN 不要の直接アクセス、超高速レスポンスが特長です。AIGoCode は sub2api ユーザー向けに特別特典を用意しています：こちらのリンクから登録すると、初回チャージ時に 10% のボーナスクレジットを追加プレゼント！ + +CodexEverywhere +OpenAI 公式価格のわずか 3% で本物の GPT-5.6 シリーズを提供 — CodexEverywhere は世界中の開発者にフロンティアモデルへのアクセスを民主化しています。私たちは透明性と誠実さを信条とし、モデル品質は数か月にわたるアクティブなコミュニティの監視によって検証されています。USD および暗号通貨に対応。codex-everywhere.com で $20 の無料トライアルから始めましょう。 + + bmoplus 本プロジェクトにご支援いただいた BmoPlus に感謝いたします！BmoPlusは、AIサブスクリプションのヘビーユーザー向けに特化した信頼性の高いAIアカウントサービスプロバイダーであり、安定した ChatGPT Plus / ChatGPT Pro (完全保証) / Claude Pro / Super Grok / Gemini Pro の公式代行チャージおよび即納アカウントを提供しています。こちらのBmoPlus AIアカウント専門店/代行チャージ経由でご登録・ご注文いただいたユーザー様は、GPTを 公式サイト価格の約1割（90% OFF） という驚異的な価格でご利用いただけます！ diff --git a/assets/partners/logos/codex-everywhere.jpg b/assets/partners/logos/codex-everywhere.jpg new file mode 100644 index 0000000000..a89679ada5 Binary files /dev/null and b/assets/partners/logos/codex-everywhere.jpg differ diff --git a/backend/cmd/server/VERSION b/backend/cmd/server/VERSION index ce08f35612..24f1cf39c7 100644 --- a/backend/cmd/server/VERSION +++ b/backend/cmd/server/VERSION @@ -1 +1 @@ -0.1.182 +0.1.183 diff --git a/backend/internal/handler/endpoint.go b/backend/internal/handler/endpoint.go index 6985392926..d13bbee47a 100644 --- a/backend/internal/handler/endpoint.go +++ b/backend/internal/handler/endpoint.go @@ -312,6 +312,14 @@ func GetInboundEndpoint(c *gin.Context) string { // and the account platform. Handlers call this after scheduling an // account, passing account.Platform. func GetUpstreamEndpoint(c *gin.Context, platform string) string { + // OpenAI 转发服务维护独立的运行时端点上下文，覆盖普通入站推导。 + // 这对 force_chat_completions 的错误路径尤为重要：此时可能没有 + // ForwardResult，不能把入站 /v1/responses 误报成上游端点。 + if platform == service.PlatformOpenAI || platform == service.PlatformGrok || service.IsCNProvider(platform) { + if endpoint := service.GetActualOpenAIUpstreamEndpoint(c); endpoint != "" { + return endpoint + } + } if c != nil { if value, ok := c.Get(ctxKeyActualUpstreamEndpoint); ok { if endpoint, ok := value.(string); ok && endpoint != "" { diff --git a/backend/internal/handler/endpoint_test.go b/backend/internal/handler/endpoint_test.go index 005c397d44..f87b27fcd1 100644 --- a/backend/internal/handler/endpoint_test.go +++ b/backend/internal/handler/endpoint_test.go @@ -184,6 +184,16 @@ func TestGetUpstreamEndpointPrefersRuntimeOverride(t *testing.T) { require.Equal(t, EndpointMessages, GetUpstreamEndpoint(c, service.PlatformAntigravity)) } +func TestGetUpstreamEndpointUsesOpenAIRuntimeOverride(t *testing.T) { + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, EndpointResponses, nil) + c.Set(ctxKeyInboundEndpoint, EndpointResponses) + + service.SetActualOpenAIUpstreamEndpoint(c, EndpointChatCompletions) + require.Equal(t, EndpointChatCompletions, GetUpstreamEndpoint(c, service.PlatformOpenAI)) +} + func TestResolveOpenAIUpstreamEndpointPrefersForwardResult(t *testing.T) { tests := []struct { name string diff --git a/backend/internal/pkg/apicompat/responses_client_tools.go b/backend/internal/pkg/apicompat/responses_client_tools.go index b1e2c412f1..5e831c1c52 100644 --- a/backend/internal/pkg/apicompat/responses_client_tools.go +++ b/backend/internal/pkg/apicompat/responses_client_tools.go @@ -235,12 +235,12 @@ func rewriteClientToolHistory(value any, adapter *ResponsesClientToolMapping) (b typed["type"] = "function_call" typed["arguments"] = customToolCallArguments(stringValue(typed["input"])) delete(typed, "input") - dropInvalidLoweredFunctionItemID(typed) + normalizeLoweredFunctionItemID(typed) changed = true } case "custom_tool_call_output": typed["type"] = "function_call_output" - dropInvalidLoweredFunctionItemID(typed) + normalizeLoweredFunctionItemID(typed) normalizeClientToolOutput(typed) changed = true case "tool_search_call": @@ -249,7 +249,7 @@ func rewriteClientToolHistory(value any, adapter *ResponsesClientToolMapping) (b typed["name"] = toolSearchProxyName typed["arguments"] = rawObjectString(typed["arguments"]) delete(typed, "execution") - dropInvalidLoweredFunctionItemID(typed) + normalizeLoweredFunctionItemID(typed) changed = true } case "tool_search_output": @@ -259,7 +259,7 @@ func rewriteClientToolHistory(value any, adapter *ResponsesClientToolMapping) (b return fmt.Errorf("tool_search_output requires a non-empty string call_id before it can be lowered to function_call_output") } typed["type"] = "function_call_output" - dropInvalidLoweredFunctionItemID(typed) + normalizeLoweredFunctionItemID(typed) if err := normalizeToolSearchOutput(typed); err != nil { return err } @@ -280,14 +280,73 @@ func rewriteClientToolHistory(value any, adapter *ResponsesClientToolMapping) (b return changed, nil } -// dropInvalidLoweredFunctionItemID removes Codex client-only item IDs such as +// normalizeLoweredFunctionItemID reconciles Codex client-only item IDs such as // ctc_*, ctco_*, tsc_*, and tso_* after their item type is lowered to the -// function protocol. Function upstreams validate these IDs with the fc prefix; -// call_id, which is preserved separately, is the tool call/output pairing key. -func dropInvalidLoweredFunctionItemID(item map[string]any) { +// function protocol, which validates these IDs with the fc prefix. A ctc_/tsc_ +// call ID maps back to the fc_ ID it was raised from; the remaining IDs have no +// function-protocol counterpart and are dropped. call_id, which is preserved +// separately, stays the tool call/output pairing key either way. +func normalizeLoweredFunctionItemID(item map[string]any) { id := strings.TrimSpace(stringValue(item["id"])) - if id != "" && !strings.HasPrefix(id, "fc") { - delete(item, "id") + if id == "" || strings.HasPrefix(id, "fc") { + return + } + // A ctc_/tsc_ call ID on a lowered call item is the one we minted from the + // upstream's own fc_ ID when the item was raised, so map it back instead of + // dropping it. Anything else has no known function-protocol counterpart. + if recovered := retypedResponsesToolCallItemID(id, "function_call"); recovered != id { + item["id"] = recovered + return + } + delete(item, "id") +} + +// responsesToolCallItemIDPrefixes lists the Responses item ID prefixes that are +// tied to a specific tool-call item type. +var responsesToolCallItemIDPrefixes = []string{"fc_", "ctc_", "tsc_"} + +// responsesToolCallItemIDPrefix reports the ID prefix the Responses API +// validates for itemType, or "" when the type constrains no prefix. +func responsesToolCallItemIDPrefix(itemType string) string { + switch itemType { + case "custom_tool_call": + return "ctc_" + case "tool_search_call": + return "tsc_" + case "function_call": + return "fc_" + default: + return "" + } +} + +// retypedResponsesToolCallItemID re-prefixes an upstream item ID so it agrees +// with the item type we raise it back to. A function-only upstream answers a +// lowered custom tool with an fc_ item ID; emitting that ID on the restored +// custom_tool_call poisons the client's history, because a later replay of the +// same item to an upstream that validates IDs fails with +// "Invalid 'input[N].id' ... Expected an ID that begins with 'ctc'". +// The suffix is preserved so the ID stays stable and unique per upstream item. +// IDs that carry no known tool-call prefix are left alone rather than guessed at. +func retypedResponsesToolCallItemID(id, itemType string) string { + want := responsesToolCallItemIDPrefix(itemType) + if want == "" || id == "" || strings.HasPrefix(id, want) { + return id + } + for _, known := range responsesToolCallItemIDPrefixes { + if known != want && strings.HasPrefix(id, known) { + return want + strings.TrimPrefix(id, known) + } + } + return id +} + +// retypeResponsesToolCallItemID applies retypedResponsesToolCallItemID to a +// decoded item map. +func retypeResponsesToolCallItemID(item map[string]any, itemType string) { + id := strings.TrimSpace(stringValue(item["id"])) + if retyped := retypedResponsesToolCallItemID(id, itemType); retyped != id { + item["id"] = retyped } } @@ -432,12 +491,14 @@ func restoreClientToolValue(value any, adapter *ResponsesClientToolMapping) bool name := strings.TrimSpace(stringValue(typed["name"])) if adapter.CustomTools[name] { typed["type"] = "custom_tool_call" + retypeResponsesToolCallItemID(typed, "custom_tool_call") typed["input"] = extractCustomToolCallInput(rawObjectString(typed["arguments"])) delete(typed, "arguments") delete(typed, "namespace") changed = true } else if adapter.ToolSearch && name == toolSearchProxyName { typed["type"] = "tool_search_call" + retypeResponsesToolCallItemID(typed, "tool_search_call") typed["execution"] = "client" typed["arguments"] = json.RawMessage(toolSearchCallArgumentsJSON(rawObjectString(typed["arguments"]))) delete(typed, "name") @@ -464,12 +525,15 @@ type ResponsesClientToolStreamRestorer struct { } type responsesClientToolStreamCall struct { - kind string - name string - callID string - itemID string - outputIdx int - arguments strings.Builder + kind string + name string + // callID and itemID stay as the upstream sent them so later upstream + // events keep matching this call; clientItemID is what we emit. + callID string + itemID string + clientItemID string + outputIdx int + arguments strings.Builder } func NewResponsesClientToolStreamRestorer(mapping ResponsesClientToolMapping) *ResponsesClientToolStreamRestorer { @@ -508,6 +572,9 @@ func (r *ResponsesClientToolStreamRestorer) Restore(event ResponsesStreamEvent) event.Item.Arguments = "{}" event.Item.Namespace = "" } + if call.clientItemID != "" { + event.Item.ID = call.clientItemID + } } emit(r.restoreNamespaceEvent(event)) case "response.function_call_arguments.delta": @@ -525,9 +592,9 @@ func (r *ResponsesClientToolStreamRestorer) Restore(event ResponsesStreamEvent) if call.kind == "custom" { input := extractCustomToolCallInput(call.arguments.String()) if input != "" { - emit(ResponsesStreamEvent{Type: "response.custom_tool_call_input.delta", OutputIndex: call.outputIdx, ItemID: call.itemID, Delta: input}) + emit(ResponsesStreamEvent{Type: "response.custom_tool_call_input.delta", OutputIndex: call.outputIdx, ItemID: call.clientItemID, Delta: input}) } - emit(ResponsesStreamEvent{Type: "response.custom_tool_call_input.done", OutputIndex: call.outputIdx, ItemID: call.itemID, CallID: call.callID, Name: call.name, Input: input}) + emit(ResponsesStreamEvent{Type: "response.custom_tool_call_input.done", OutputIndex: call.outputIdx, ItemID: call.clientItemID, CallID: call.callID, Name: call.name, Input: input}) } return out } @@ -548,6 +615,9 @@ func (r *ResponsesClientToolStreamRestorer) Restore(event ResponsesStreamEvent) } event.Item.Namespace = "" } + if call.clientItemID != "" { + event.Item.ID = call.clientItemID + } delete(r.calls, call.itemID) delete(r.calls, call.callID) delete(r.byOutput, call.outputIdx) @@ -684,6 +754,15 @@ func (r *ResponsesClientToolStreamRestorer) resequenceRaw(payload []byte, sequen return [][]byte{encoded}, true, nil } +// responsesClientToolItemType maps a restorer call kind to the item type the +// client sees. +func responsesClientToolItemType(kind string) string { + if kind == "custom" { + return "custom_tool_call" + } + return "tool_search_call" +} + func (r *ResponsesClientToolStreamRestorer) recordItem(event ResponsesStreamEvent) *responsesClientToolStreamCall { if event.Item == nil || event.Item.Type != "function_call" { return nil @@ -704,7 +783,14 @@ func (r *ResponsesClientToolStreamRestorer) recordItem(event ResponsesStreamEven } call := r.calls[key] if call == nil { - call = &responsesClientToolStreamCall{kind: kind, name: name, callID: event.Item.CallID, itemID: event.Item.ID, outputIdx: event.OutputIndex} + call = &responsesClientToolStreamCall{ + kind: kind, + name: name, + callID: event.Item.CallID, + itemID: event.Item.ID, + clientItemID: retypedResponsesToolCallItemID(event.Item.ID, responsesClientToolItemType(kind)), + outputIdx: event.OutputIndex, + } r.calls[key] = call if call.callID != "" { r.calls[call.callID] = call @@ -758,11 +844,13 @@ func restoreResponsesOutputClientTools(outputs []ResponsesOutput, adapter *Respo } if adapter.CustomTools[output.Name] { output.Type = "custom_tool_call" + output.ID = retypedResponsesToolCallItemID(output.ID, output.Type) output.Input = extractCustomToolCallInput(output.Arguments) output.Arguments = "" output.Namespace = "" } else if adapter.ToolSearch && output.Name == toolSearchProxyName { output.Type = "tool_search_call" + output.ID = retypedResponsesToolCallItemID(output.ID, output.Type) output.Name = "" output.Namespace = "" } diff --git a/backend/internal/pkg/apicompat/responses_client_tools_item_id_helper_test.go b/backend/internal/pkg/apicompat/responses_client_tools_item_id_helper_test.go new file mode 100644 index 0000000000..66448ed661 --- /dev/null +++ b/backend/internal/pkg/apicompat/responses_client_tools_item_id_helper_test.go @@ -0,0 +1,29 @@ +package apicompat + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestRetypedResponsesToolCallItemID(t *testing.T) { + for _, tc := range []struct { + name string + id string + itemType string + want string + }{ + {"function id raised to custom", "fc_abc", "custom_tool_call", "ctc_abc"}, + {"function id raised to tool search", "fc_abc", "tool_search_call", "tsc_abc"}, + {"already correct is untouched", "ctc_abc", "custom_tool_call", "ctc_abc"}, + {"custom id lowered to function", "ctc_abc", "function_call", "fc_abc"}, + {"unknown prefix is left alone", "item_abc", "custom_tool_call", "item_abc"}, + {"unprefixed id is left alone", "abc", "custom_tool_call", "abc"}, + {"empty id stays empty", "", "custom_tool_call", ""}, + {"unconstrained item type is left alone", "fc_abc", "message", "fc_abc"}, + } { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.want, retypedResponsesToolCallItemID(tc.id, tc.itemType)) + }) + } +} diff --git a/backend/internal/pkg/apicompat/responses_client_tools_item_id_test.go b/backend/internal/pkg/apicompat/responses_client_tools_item_id_test.go new file mode 100644 index 0000000000..724f77b75d --- /dev/null +++ b/backend/internal/pkg/apicompat/responses_client_tools_item_id_test.go @@ -0,0 +1,132 @@ +package apicompat + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +// A function-only upstream answers a lowered custom tool with an fc_ item ID. +// Emitting that ID on the restored custom_tool_call breaks the client the next +// time the item is replayed to an upstream that validates item IDs: +// "Invalid 'input[N].id': 'fc_...'. Expected an ID that begins with 'ctc'". +func TestRestoreResponsesClientToolPayload_RetypesToolCallItemIDs(t *testing.T) { + mapping := ResponsesClientToolMapping{ + CustomTools: map[string]bool{"exec": true}, ToolSearch: true, + NamespaceTools: map[string]ResponsesNamespaceName{"team__send": {Namespace: "team", Name: "send"}}, + } + payload := []byte(`{"id":"resp","output":[` + + `{"type":"function_call","id":"fc_abc123","call_id":"call_1","name":"exec","arguments":"{\"input\":\"dir\"}"},` + + `{"type":"function_call","id":"fc_def456","call_id":"call_2","name":"tool_search","arguments":"{\"query\":\"git\"}"},` + + `{"type":"function_call","id":"fc_ghi789","call_id":"call_3","name":"team__send","arguments":"{}"}]}`) + + restored, changed, err := RestoreResponsesClientToolPayload(payload, mapping) + require.NoError(t, err) + require.True(t, changed) + require.JSONEq(t, `{"id":"resp","output":[`+ + `{"type":"custom_tool_call","id":"ctc_abc123","call_id":"call_1","name":"exec","input":"dir"},`+ + `{"type":"tool_search_call","id":"tsc_def456","call_id":"call_2","execution":"client","arguments":{"query":"git"}},`+ + `{"type":"function_call","id":"fc_ghi789","call_id":"call_3","name":"send","namespace":"team","arguments":"{}"}]}`, + string(restored)) +} + +func TestRestoreResponsesOutputClientTools_RetypesToolCallItemIDs(t *testing.T) { + mapping := ResponsesClientToolMapping{CustomTools: map[string]bool{"exec": true}, ToolSearch: true} + outputs := []ResponsesOutput{ + {Type: "function_call", ID: "fc_abc123", CallID: "call_1", Name: "exec", Arguments: `{"input":"dir"}`}, + {Type: "function_call", ID: "fc_def456", CallID: "call_2", Name: toolSearchProxyName, Arguments: `{"query":"git"}`}, + } + + restoreResponsesOutputClientTools(outputs, &mapping) + + require.Equal(t, "custom_tool_call", outputs[0].Type) + require.Equal(t, "ctc_abc123", outputs[0].ID) + require.Equal(t, "call_1", outputs[0].CallID) + require.Equal(t, "tool_search_call", outputs[1].Type) + require.Equal(t, "tsc_def456", outputs[1].ID) + require.Equal(t, "call_2", outputs[1].CallID) +} + +func TestResponsesClientToolStreamRestorer_RetypesCustomToolCallItemID(t *testing.T) { + const upstreamID = "fc_09f77ac43cf7db36016a8920e7934487" + const clientID = "ctc_09f77ac43cf7db36016a8920e7934487" + + restorer := NewResponsesClientToolStreamRestorer(ResponsesClientToolMapping{CustomTools: map[string]bool{"exec": true}}) + + added := restorer.Restore(ResponsesStreamEvent{ + Type: "response.output_item.added", SequenceNumber: 0, OutputIndex: 0, + Item: &ResponsesOutput{Type: "function_call", ID: upstreamID, CallID: "call_1", Name: "exec", Status: "in_progress"}, + }) + require.Len(t, added, 1) + require.Equal(t, "custom_tool_call", added[0].Item.Type) + require.Equal(t, clientID, added[0].Item.ID) + + // Later upstream events still address the item by its upstream ID. + require.Empty(t, restorer.Restore(ResponsesStreamEvent{ + Type: "response.function_call_arguments.delta", SequenceNumber: 1, ItemID: upstreamID, Delta: `{"input":"di`, + })) + done := restorer.Restore(ResponsesStreamEvent{ + Type: "response.function_call_arguments.done", SequenceNumber: 2, ItemID: upstreamID, + CallID: "call_1", Name: "exec", Arguments: `{"input":"dir"}`, + }) + require.Len(t, done, 2) + require.Equal(t, "response.custom_tool_call_input.delta", done[0].Type) + require.Equal(t, clientID, done[0].ItemID) + require.Equal(t, "response.custom_tool_call_input.done", done[1].Type) + require.Equal(t, clientID, done[1].ItemID) + require.Equal(t, "call_1", done[1].CallID) + + closed := restorer.Restore(ResponsesStreamEvent{ + Type: "response.output_item.done", SequenceNumber: 3, OutputIndex: 0, + Item: &ResponsesOutput{Type: "function_call", ID: upstreamID, CallID: "call_1", Name: "exec", Arguments: `{"input":"dir"}`, Status: "completed"}, + }) + require.Len(t, closed, 1) + require.Equal(t, "custom_tool_call", closed[0].Item.Type) + require.Equal(t, clientID, closed[0].Item.ID) + require.Equal(t, "dir", closed[0].Item.Input) +} + +func TestResponsesClientToolStreamRestorer_RetypesToolSearchCallItemID(t *testing.T) { + restorer := NewResponsesClientToolStreamRestorer(ResponsesClientToolMapping{ToolSearch: true}) + + added := restorer.Restore(ResponsesStreamEvent{ + Type: "response.output_item.added", SequenceNumber: 0, OutputIndex: 0, + Item: &ResponsesOutput{Type: "function_call", ID: "fc_search1", CallID: "call_1", Name: toolSearchProxyName, Status: "in_progress"}, + }) + require.Len(t, added, 1) + require.Equal(t, "tool_search_call", added[0].Item.Type) + require.Equal(t, "tsc_search1", added[0].Item.ID) +} + +// The WS bridge replays restored items back to the upstream, so the ID we hand +// the client has to map back to the upstream's own fc_ ID on the way down. +func TestAdaptResponsesClientTools_RecoversRetypedToolCallItemID(t *testing.T) { + req := map[string]any{ + "tools": []any{ + map[string]any{"type": "custom", "name": "exec"}, + map[string]any{"type": "tool_search"}, + }, + "input": []any{ + map[string]any{"type": "custom_tool_call", "id": "ctc_upstream1", "call_id": "call_1", "name": "exec", "input": "dir"}, + map[string]any{"type": "tool_search_call", "id": "tsc_upstream2", "call_id": "call_2", "arguments": map[string]any{"query": "git"}}, + map[string]any{"type": "custom_tool_call_output", "id": "ctco_client", "call_id": "call_1", "output": "ok"}, + }, + } + + _, changed, err := AdaptResponsesClientTools(req) + require.NoError(t, err) + require.True(t, changed) + + input := requireResponsesClientToolValue[[]any](t, req["input"]) + require.Len(t, input, 3) + customCall := requireResponsesClientToolValue[map[string]any](t, input[0]) + require.Equal(t, "function_call", customCall["type"]) + require.Equal(t, "fc_upstream1", customCall["id"]) + searchCall := requireResponsesClientToolValue[map[string]any](t, input[1]) + require.Equal(t, "function_call", searchCall["type"]) + require.Equal(t, "fc_upstream2", searchCall["id"]) + // Output items have no function-protocol ID counterpart and stay dropped. + customOutput := requireResponsesClientToolValue[map[string]any](t, input[2]) + require.Equal(t, "function_call_output", customOutput["type"]) + require.NotContains(t, customOutput, "id") +} diff --git a/backend/internal/pkg/apicompat/responses_client_tools_test.go b/backend/internal/pkg/apicompat/responses_client_tools_test.go index 2b3460755f..bfe3fd2eb7 100644 --- a/backend/internal/pkg/apicompat/responses_client_tools_test.go +++ b/backend/internal/pkg/apicompat/responses_client_tools_test.go @@ -48,14 +48,14 @@ func TestAdaptResponsesClientTools_LowersDeclarationsHistoryChoiceAndNamespaces( input := requireResponsesClientToolValue[[]any](t, req["input"]) customCall := requireResponsesClientToolValue[map[string]any](t, input[0]) require.Equal(t, "function_call", customCall["type"]) - require.NotContains(t, customCall, "id") + require.Equal(t, "fc_client", customCall["id"]) require.JSONEq(t, `{"input":"dir"}`, requireResponsesClientToolValue[string](t, customCall["arguments"])) customOutput := requireResponsesClientToolValue[map[string]any](t, input[1]) require.Equal(t, "function_call_output", customOutput["type"]) require.NotContains(t, customOutput, "id") searchCall := requireResponsesClientToolValue[map[string]any](t, input[2]) require.Equal(t, "function_call", searchCall["type"]) - require.NotContains(t, searchCall, "id") + require.Equal(t, "fc_client", searchCall["id"]) require.Equal(t, toolSearchProxyName, searchCall["name"]) require.JSONEq(t, `{"query":"git"}`, requireResponsesClientToolValue[string](t, searchCall["arguments"])) searchOutput := requireResponsesClientToolValue[map[string]any](t, input[3]) diff --git a/backend/internal/repository/channel_monitor_v2_aggregation.go b/backend/internal/repository/channel_monitor_v2_aggregation.go index 6ad2bbf9b3..9bec5877b5 100644 --- a/backend/internal/repository/channel_monitor_v2_aggregation.go +++ b/backend/internal/repository/channel_monitor_v2_aggregation.go @@ -255,7 +255,7 @@ WITH dedup AS ( -- group errors aggregate under platform 'composite', which is never an -- enabled config platform, and are filtered out of every monitor v2 query. lower(CASE - WHEN g.platform = 'composite' THEN COALESCE(NULLIF(TRIM(a.platform)), NULLIF(NULLIF(lower(TRIM(current_error.platform)), ''), 'composite'), 'unknown') + WHEN g.platform = 'composite' THEN COALESCE(NULLIF(TRIM(a.platform), ''), NULLIF(NULLIF(lower(TRIM(current_error.platform)), ''), 'composite'), 'unknown') ELSE COALESCE(NULLIF(TRIM(current_error.platform), ''), 'unknown') END) AS platform, COALESCE(current_error.group_id, 0) AS group_id, diff --git a/backend/internal/repository/channel_monitor_v2_repo_test.go b/backend/internal/repository/channel_monitor_v2_repo_test.go index 55c9a96460..82d5e61130 100644 --- a/backend/internal/repository/channel_monitor_v2_repo_test.go +++ b/backend/internal/repository/channel_monitor_v2_repo_test.go @@ -116,6 +116,8 @@ func TestChannelMonitorV2ErrorAggregationResolvesCompositePlatform(t *testing.T) require.Contains(t, query, "left join groups g on g.id = current_error.group_id") require.Contains(t, query, "left join accounts a on a.id = current_error.account_id") require.Contains(t, query, "a.platform") + require.Contains(t, query, "nullif(trim(a.platform), '')") + require.NotContains(t, query, "nullif(trim(a.platform))") } func TestChannelMonitorV2UsageSuccessExcludesCyberBillingRows(t *testing.T) { diff --git a/backend/internal/repository/user_repo.go b/backend/internal/repository/user_repo.go index c072f09b27..3d02d6fbff 100644 --- a/backend/internal/repository/user_repo.go +++ b/backend/internal/repository/user_repo.go @@ -1145,12 +1145,17 @@ func (r *userRepository) ExistsByEmailAlias(ctx context.Context, email string) ( } func existsByEmailAliasWithClient(ctx context.Context, client *dbent.Client, email string) (bool, error) { + _, exists, err := emailAliasOwnerIDWithClient(ctx, client, email, 0) + return exists, err +} + +func emailAliasOwnerIDWithClient(ctx context.Context, client *dbent.Client, email string, currentUserID int64) (int64, bool, error) { if client == nil { - return false, nil + return 0, false, nil } probes := service.EmailAliasDedupProbes(email) if len(probes) == 0 { - return false, nil + return 0, false, nil } preds := make([]predicate.User, 0, 2*len(probes)) @@ -1164,20 +1169,82 @@ func existsByEmailAliasWithClient(ctx context.Context, client *dbent.Client, ema candidates, err := client.User.Query(). Where(dbuser.Or(preds...)). Limit(emailAliasCandidateLimit). - Select(dbuser.FieldEmail). - Strings(ctx) + Select(dbuser.FieldID, dbuser.FieldEmail). + All(ctx) if err != nil { - return false, err + return 0, false, err } // 探针会有过度匹配（点号只在 Gmail 家族无意义），最终判定必须回到完整归一化规则。 + // 返回“其他用户”优先于当前用户，避免历史重复数据让调用方误判为仅当前用户占用。 identity := service.NormalizeEmailForAliasDedup(email) + var selfID int64 + selfExists := false for _, candidate := range candidates { - if service.NormalizeEmailForAliasDedup(candidate) == identity { - return true, nil + if service.NormalizeEmailForAliasDedup(candidate.Email) != identity { + continue + } + if candidate.ID != 0 && candidate.ID != currentUserID { + return candidate.ID, true, nil + } + if candidate.ID == currentUserID { + selfID = candidate.ID + selfExists = true } } - return false, nil + return selfID, selfExists, nil +} + +// UpdateEmailWithAliasGuard 在调用方事务内更新主邮箱与密码哈希。 +// +// 邮箱换绑不能只依赖服务层前置查重：两个并发请求可能同时看到同一收件箱未被占用。 +// 这里先按“字面邮箱 + 收件箱身份”加锁，复查是否已被其他用户占用，再执行写入； +// PostgreSQL 使用事务级 advisory lock 跨实例互斥，测试内存库则由进程内锁兜底。 +func (r *userRepository) UpdateEmailWithAliasGuard( + ctx context.Context, + userID int64, + email string, + passwordHash string, +) error { + if userID <= 0 { + return service.ErrUserNotFound + } + if strings.TrimSpace(email) == "" || passwordHash == "" { + return fmt.Errorf("email identity update requires email and password hash") + } + tx := dbent.TxFromContext(ctx) + if tx == nil { + return fmt.Errorf("email identity update requires a transaction") + } + client := tx.Client() + + releaseEmailLock, err := lockRepositoryScopedKeys( + ctx, + client, + txAwareSQLExecutor(ctx, r.sql, r.client), + normalizedEmailUniquenessLockKey(email), + emailAliasUniquenessLockKey(email), + ) + if err != nil { + return err + } + defer releaseEmailLock() + + ownerID, exists, err := emailAliasOwnerIDWithClient(ctx, client, email, userID) + if err != nil { + return err + } + if exists && ownerID != userID { + return service.ErrEmailExists + } + + if _, err := client.User.UpdateOneID(userID). + SetEmail(email). + SetPasswordHash(passwordHash). + Save(ctx); err != nil { + return translatePersistenceError(err, service.ErrUserNotFound, service.ErrEmailExists) + } + return nil } // dotStrippedEmailExpr 渲染下面的表达式：去掉存量邮箱的大小写、首尾空白（与 diff --git a/backend/internal/service/antigravity_gateway_compat.go b/backend/internal/service/antigravity_gateway_compat.go index 49ca076884..1d6d39e1aa 100644 --- a/backend/internal/service/antigravity_gateway_compat.go +++ b/backend/internal/service/antigravity_gateway_compat.go @@ -29,6 +29,8 @@ const ( AntigravityCredentialRejectedReason GatewayFailureReason = "antigravity_oauth_credential_rejected" ) +const antigravityCompatMaxTokens = 64000 + type antigravityCompatRequest struct { protocol antigravityCompatProtocol originalBody []byte @@ -158,7 +160,7 @@ func preserveChatCompletionTokenLimit(request *apicompat.ChatCompletionsRequest, limit = request.MaxCompletionTokens } if limit != nil && *limit > 0 { - claudeRequest.MaxTokens = *limit + claudeRequest.MaxTokens = min(*limit, antigravityCompatMaxTokens) } } diff --git a/backend/internal/service/antigravity_gateway_compat_test.go b/backend/internal/service/antigravity_gateway_compat_test.go index 3df0b0081a..a92b1654c2 100644 --- a/backend/internal/service/antigravity_gateway_compat_test.go +++ b/backend/internal/service/antigravity_gateway_compat_test.go @@ -12,6 +12,7 @@ import ( "time" "github.com/Wei-Shaw/sub2api/internal/config" + "github.com/Wei-Shaw/sub2api/internal/pkg/apicompat" "github.com/gin-gonic/gin" "github.com/stretchr/testify/require" "github.com/tidwall/gjson" @@ -285,6 +286,21 @@ func TestAntigravityCompatPreservesChatTokenLimit(t *testing.T) { body: `{"model":"gemini-3.1-pro-high","messages":[{"role":"user","content":"ok"}],"max_tokens":8,"max_completion_tokens":13}`, want: 13, }, + { + name: "max_tokens at safe ceiling is preserved", + body: `{"model":"gemini-3.1-pro-high","messages":[{"role":"user","content":"ok"}],"max_tokens":64000}`, + want: 64000, + }, + { + name: "max_tokens above safe ceiling is clamped", + body: `{"model":"gemini-3.1-pro-high","messages":[{"role":"user","content":"ok"}],"max_tokens":64001}`, + want: 64000, + }, + { + name: "precedence applies before clamping", + body: `{"model":"gemini-3.1-pro-high","messages":[{"role":"user","content":"ok"}],"max_tokens":8,"max_completion_tokens":64001}`, + want: 64000, + }, } for _, tt := range tests { @@ -310,6 +326,27 @@ func TestAntigravityCompatPreservesChatTokenLimit(t *testing.T) { } } +func TestPreserveChatCompletionTokenLimitIgnoresAbsentAndNonPositiveValues(t *testing.T) { + tests := []struct { + name string + request apicompat.ChatCompletionsRequest + }{ + {name: "absent"}, + {name: "zero max_tokens", request: apicompat.ChatCompletionsRequest{MaxTokens: antigravityCompatIntPtr(0)}}, + {name: "negative max_completion_tokens takes precedence", request: apicompat.ChatCompletionsRequest{MaxTokens: antigravityCompatIntPtr(12), MaxCompletionTokens: antigravityCompatIntPtr(-1)}}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + claudeRequest := &apicompat.AnthropicRequest{MaxTokens: 99} + preserveChatCompletionTokenLimit(&tt.request, claudeRequest) + require.Equal(t, 99, claudeRequest.MaxTokens) + }) + } +} + +func antigravityCompatIntPtr(v int) *int { return &v } + func TestAntigravityCompatRoutesByMappedModelFamily(t *testing.T) { gin.SetMode(gin.TestMode) tests := []struct { diff --git a/backend/internal/service/auth_email_binding.go b/backend/internal/service/auth_email_binding.go index 4fe3c4e85b..733e76b7fd 100644 --- a/backend/internal/service/auth_email_binding.go +++ b/backend/internal/service/auth_email_binding.go @@ -56,12 +56,8 @@ func (s *AuthService) BindEmailIdentity( return nil, ErrPasswordIncorrect } - existingUser, err := s.userRepo.GetByEmail(ctx, normalizedEmail) - switch { - case err == nil && existingUser != nil && existingUser.ID != userID: - return nil, ErrEmailExists - case err != nil && !errors.Is(err, ErrUserNotFound): - return nil, ErrServiceUnavailable + if err := s.ensureEmailIdentityAvailableForUser(ctx, currentUser, normalizedEmail); err != nil { + return nil, err } hashedPassword, err := s.HashPassword(password) @@ -115,19 +111,16 @@ func (s *AuthService) SendEmailIdentityBindCode(ctx context.Context, userID int6 if s.emailService == nil { return ErrServiceUnavailable } - if _, err := s.userRepo.GetByID(ctx, userID); err != nil { + currentUser, err := s.userRepo.GetByID(ctx, userID) + if err != nil { if errors.Is(err, ErrUserNotFound) { return ErrUserNotFound } return ErrServiceUnavailable } - existingUser, err := s.userRepo.GetByEmail(ctx, normalizedEmail) - switch { - case err == nil && existingUser != nil && existingUser.ID != userID: - return ErrEmailExists - case err != nil && !errors.Is(err, ErrUserNotFound): - return ErrServiceUnavailable + if err := s.ensureEmailIdentityAvailableForUser(ctx, currentUser, normalizedEmail); err != nil { + return err } siteName := "Sub2API" @@ -137,6 +130,45 @@ func (s *AuthService) SendEmailIdentityBindCode(ctx context.Context, userID int6 return s.emailService.SendVerifyCode(ctx, normalizedEmail, siteName, firstEmailLocale(locale)) } +// ensureEmailIdentityAvailableForUser 在发码 / 提交换绑前做快速查重。 +// 精确地址或 provider alias 若已指向其他用户的收件箱则直接拒绝； +// 当前用户自己的收件箱允许继续，便于其更换自身的 alias 变体。 +func (s *AuthService) ensureEmailIdentityAvailableForUser( + ctx context.Context, + currentUser *User, + email string, +) error { + if currentUser == nil { + return ErrUserNotFound + } + + existingUser, err := s.userRepo.GetByEmail(ctx, email) + switch { + case err == nil: + if existingUser == nil || existingUser.ID == currentUser.ID { + break + } + return ErrEmailExists + case errors.Is(err, ErrUserNotFound): + // Continue to alias lookup below. + default: + return ErrServiceUnavailable + } + + if NormalizeEmailForAliasDedup(currentUser.Email) == NormalizeEmailForAliasDedup(email) { + return nil + } + + aliasExists, err := s.userRepo.ExistsByEmailAlias(ctx, email) + if err != nil { + return ErrServiceUnavailable + } + if aliasExists { + return ErrEmailExists + } + return nil +} + func normalizeEmailForIdentityBinding(email string) (string, error) { normalized := strings.ToLower(strings.TrimSpace(email)) if normalized == "" || len(normalized) > 255 { @@ -153,6 +185,12 @@ func hasBindableEmailIdentitySubject(email string) bool { return normalized != "" && !isReservedEmail(normalized) } +// emailIdentityAliasGuardRepository 是主邮箱替换所需的事务内原子仓储能力， +// 用于关闭服务层前置查重与实际写入之间的并发窗口。 +type emailIdentityAliasGuardRepository interface { + UpdateEmailWithAliasGuard(ctx context.Context, userID int64, email string, passwordHash string) error +} + func (s *AuthService) updateBoundEmailIdentityTx( ctx context.Context, currentUser *User, @@ -192,16 +230,15 @@ func (s *AuthService) updateBoundEmailIdentityWithClient( return ErrServiceUnavailable } - oldEmail := currentUser.Email - if _, err := client.User.UpdateOneID(currentUser.ID). - SetEmail(email). - SetPasswordHash(hashedPassword). - Save(ctx); err != nil { - if dbent.IsConstraintError(err) { - return ErrEmailExists - } + guard, ok := s.userRepo.(emailIdentityAliasGuardRepository) + if !ok { return ErrServiceUnavailable } + if err := guard.UpdateEmailWithAliasGuard(ctx, currentUser.ID, email, hashedPassword); err != nil { + return err + } + + oldEmail := currentUser.Email if err := replaceBoundEmailAuthIdentityWithClient(ctx, client, currentUser.ID, oldEmail, email, "auth_service_email_bind"); err != nil { if errors.Is(err, ErrEmailExists) { diff --git a/backend/internal/service/auth_service_email_bind_test.go b/backend/internal/service/auth_service_email_bind_test.go index 2d78862ef3..9bdca02f2d 100644 --- a/backend/internal/service/auth_service_email_bind_test.go +++ b/backend/internal/service/auth_service_email_bind_test.go @@ -6,6 +6,7 @@ import ( "context" "database/sql" "errors" + "fmt" "sync" "testing" "time" @@ -13,6 +14,7 @@ import ( dbent "github.com/Wei-Shaw/sub2api/ent" "github.com/Wei-Shaw/sub2api/ent/authidentity" "github.com/Wei-Shaw/sub2api/ent/enttest" + dbuser "github.com/Wei-Shaw/sub2api/ent/user" "github.com/Wei-Shaw/sub2api/internal/config" "github.com/Wei-Shaw/sub2api/internal/pkg/pagination" "github.com/Wei-Shaw/sub2api/internal/repository" @@ -69,7 +71,8 @@ func newAuthServiceForEmailBindWithRefreshCache( ) (*service.AuthService, service.UserRepository, *dbent.Client) { t.Helper() - db, err := sql.Open("sqlite", "file:auth_service_email_bind?mode=memory&cache=shared") + dbName := fmt.Sprintf("file:auth_service_email_bind_%d?mode=memory&cache=shared", time.Now().UnixNano()) + db, err := sql.Open("sqlite", dbName) require.NoError(t, err) t.Cleanup(func() { _ = db.Close() }) @@ -214,6 +217,143 @@ func TestAuthServiceBindEmailIdentity_RejectsExistingEmailOnAnotherUser(t *testi require.Equal(t, 0, countProviderGrantRecords(t, client, sourceUser.ID, "email", "first_bind")) } +func TestAuthServiceBindEmailIdentity_RejectsAliasOfExistingEmailOnAnotherUser(t *testing.T) { + cache := &emailBindCacheStub{ + data: &service.VerificationCodeData{ + Code: "123456", + CreatedAt: time.Now().UTC().Add(-10 * time.Minute), + ExpiresAt: time.Now().UTC().Add(10 * time.Minute), + }, + } + svc, _, client := newAuthServiceForEmailBind(t, nil, cache, nil) + + ctx := context.Background() + sourceUser := createEmailBindTestUser( + t, + client, + "source-user"+service.OIDCConnectSyntheticEmailDomain, + "source-user", + "old-hash", + ) + createEmailBindTestUser(t, client, "zck.ioio123@gmail.com", "inbox-owner", "hash") + + err := svc.SendEmailIdentityBindCode(ctx, sourceUser.ID, "zckioio123+new@gmail.com") + require.ErrorIs(t, err, service.ErrEmailExists) + require.Empty(t, cache.setEmails) + + updatedUser, err := svc.BindEmailIdentity( + ctx, + sourceUser.ID, + "zckioio123+new@gmail.com", + "123456", + "new-password", + ) + require.ErrorIs(t, err, service.ErrEmailExists) + require.Nil(t, updatedUser) + + storedUser, err := client.User.Get(ctx, sourceUser.ID) + require.NoError(t, err) + require.Equal(t, "source-user"+service.OIDCConnectSyntheticEmailDomain, storedUser.Email) + require.Equal(t, "old-hash", storedUser.PasswordHash) +} + +func TestAuthServiceBindEmailIdentity_AllowsOnlyOneConcurrentAliasVariant(t *testing.T) { + cache := &emailBindCacheStub{ + data: &service.VerificationCodeData{ + Code: "123456", + CreatedAt: time.Now().UTC(), + ExpiresAt: time.Now().UTC().Add(10 * time.Minute), + }, + } + svc, _, client := newAuthServiceForEmailBind(t, nil, cache, nil) + + ctx := context.Background() + unique := fmt.Sprintf("%d", time.Now().UnixNano()) + first := createEmailBindTestUser( + t, + client, + "first-"+unique+service.OIDCConnectSyntheticEmailDomain, + "first-"+unique, + "old-hash", + ) + second := createEmailBindTestUser( + t, + client, + "second-"+unique+service.OIDCConnectSyntheticEmailDomain, + "second-"+unique, + "old-hash", + ) + + start := make(chan struct{}) + results := make(chan error, 2) + go func() { + <-start + _, err := svc.BindEmailIdentity(ctx, first.ID, "inbox-"+unique+"+one@gmail.com", "123456", "new-password") + results <- err + }() + go func() { + <-start + _, err := svc.BindEmailIdentity(ctx, second.ID, "inbox-"+unique+"+two@gmail.com", "123456", "new-password") + results <- err + }() + close(start) + + var successes, conflicts int + for range 2 { + err := <-results + switch { + case err == nil: + successes++ + case errors.Is(err, service.ErrEmailExists): + conflicts++ + default: + t.Fatalf("unexpected bind error: %v", err) + } + } + require.Equal(t, 1, successes) + require.Equal(t, 1, conflicts) + + boundCount, err := client.User.Query(). + Where(dbuser.EmailIn( + "inbox-"+unique+"+one@gmail.com", + "inbox-"+unique+"+two@gmail.com", + )). + Count(ctx) + require.NoError(t, err) + require.Equal(t, 1, boundCount) +} + +func TestAuthServiceBindEmailIdentity_RejectsNewAliasWhenAnotherUserSharesCurrentUserInbox(t *testing.T) { + cache := &emailBindCacheStub{ + data: &service.VerificationCodeData{ + Code: "123456", + CreatedAt: time.Now().UTC(), + ExpiresAt: time.Now().UTC().Add(10 * time.Minute), + }, + } + svc, _, client := newAuthServiceForEmailBind(t, nil, cache, nil) + + ctx := context.Background() + hashedPassword, err := svc.HashPassword("current-password") + require.NoError(t, err) + currentUser := createEmailBindTestUser(t, client, "inbox+own@gmail.com", "current", hashedPassword) + createEmailBindTestUser(t, client, "inbox+legacy@gmail.com", "legacy", "hash") + + updatedUser, err := svc.BindEmailIdentity( + ctx, + currentUser.ID, + "inbox+new@gmail.com", + "123456", + "current-password", + ) + require.ErrorIs(t, err, service.ErrEmailExists) + require.Nil(t, updatedUser) + + storedUser, err := client.User.Get(ctx, currentUser.ID) + require.NoError(t, err) + require.Equal(t, "inbox+own@gmail.com", storedUser.Email) +} + func TestAuthServiceBindEmailIdentity_RollsBackWhenFirstBindDefaultsFail(t *testing.T) { assigner := &flakyEmailBindDefaultSubAssignerStub{err: errors.New("temporary assign failure")} cache := &emailBindCacheStub{ diff --git a/backend/internal/service/openai_account_runtime_block_fastpath.go b/backend/internal/service/openai_account_runtime_block_fastpath.go index 1d8da95913..1f402202ca 100644 --- a/backend/internal/service/openai_account_runtime_block_fastpath.go +++ b/backend/internal/service/openai_account_runtime_block_fastpath.go @@ -28,6 +28,49 @@ type OpenAIOAuth429FailoverState struct { grokOAuth429FollowupPending bool } +type openAIOAuth429Disposition uint8 + +const ( + openAIOAuth429Transient openAIOAuth429Disposition = iota + openAIOAuth429Quota5h + openAIOAuth429Quota7d + openAIOAuth429QuotaReset +) + +// classifyOpenAIOAuth429 区分账号配额耗尽信号与普通瞬时 429。明确窗口达到 +// 100% 时以该窗口为准；没有 100% 标记但包含重置头时，沿用 v179 的兼容语义， +// 仍视为配额限流信号。 +func classifyOpenAIOAuth429(headers http.Header, responseBody []byte) (openAIOAuth429Disposition, *time.Time) { + if snapshot := ParseCodexRateLimitHeaders(headers); snapshot != nil { + if normalized := snapshot.Normalize(); normalized != nil { + if normalized.Used7dPercent != nil && *normalized.Used7dPercent >= 100 { + if normalized.Reset7dSeconds != nil { + now := time.Now() + resetAt := now.Add(time.Duration(*normalized.Reset7dSeconds) * time.Second) + return openAIOAuth429Quota7d, &resetAt + } + return openAIOAuth429Quota7d, nil + } + if normalized.Used5hPercent != nil && *normalized.Used5hPercent >= 100 { + if normalized.Reset5hSeconds != nil { + now := time.Now() + resetAt := now.Add(time.Duration(*normalized.Reset5hSeconds) * time.Second) + return openAIOAuth429Quota5h, &resetAt + } + return openAIOAuth429Quota5h, nil + } + } + } + if resetAt := calculateOpenAI429ResetTime(headers); resetAt != nil { + return openAIOAuth429QuotaReset, resetAt + } + if resetUnix := parseOpenAIRateLimitResetTime(responseBody); resetUnix != nil { + resetAt := time.Unix(*resetUnix, 0) + return openAIOAuth429QuotaReset, &resetAt + } + return openAIOAuth429Transient, nil +} + func openAIAccountStateContext(ctx context.Context) (context.Context, context.CancelFunc) { base := context.Background() if ctx != nil { @@ -165,19 +208,16 @@ func (s *OpenAIGatewayService) markOpenAIOAuth429RateLimited(ctx context.Context return } s.recordOpenAIOAuth429() - if s.openAIOAuth429RetryWindowActive(account) { + disposition, resetAt := classifyOpenAIOAuth429(headers, responseBody) + if disposition == openAIOAuth429Transient && s.openAIOAuth429RetryWindowActive(account) { return } cooldownUntil := time.Now().Add(openAIOAuth429FallbackCooldown) - if s.rateLimitService != nil { - if resetAt := s.rateLimitService.calculateOpenAI429ResetTime(headers); resetAt != nil && resetAt.After(time.Now()) { - cooldownUntil = *resetAt - } else if resetUnix := parseOpenAIRateLimitResetTime(responseBody); resetUnix != nil { - if resetAt := time.Unix(*resetUnix, 0); resetAt.After(time.Now()) { - cooldownUntil = resetAt - } - } else if cooldown, ok := s.rateLimitService.get429FallbackCooldown(ctx, account); ok && cooldown > 0 { + if resetAt != nil && resetAt.After(time.Now()) { + cooldownUntil = *resetAt + } else if s.rateLimitService != nil { + if cooldown, ok := s.rateLimitService.get429FallbackCooldown(ctx, account); ok && cooldown > 0 { cooldownUntil = time.Now().Add(cooldown) } } @@ -186,9 +226,17 @@ func (s *OpenAIGatewayService) markOpenAIOAuth429RateLimited(ctx context.Context } func (s *OpenAIGatewayService) shouldRetryOpenAIOAuth429OnSameAccount(account *Account, statusCode int, shouldDisable bool) bool { + return s.shouldRetryOpenAIOAuth429OnSameAccountWithResponse(account, statusCode, shouldDisable, nil, nil) +} + +func (s *OpenAIGatewayService) shouldRetryOpenAIOAuth429OnSameAccountWithResponse(account *Account, statusCode int, shouldDisable bool, headers http.Header, responseBody []byte) bool { if shouldDisable || statusCode != http.StatusTooManyRequests || !isOpenAIOAuthAccount(account) || account.IsShadow() { return false } + disposition, _ := classifyOpenAIOAuth429(headers, responseBody) + if disposition != openAIOAuth429Transient { + return false + } // markOpenAIOAuth429RateLimited parks the account once the window expires. // Do not accidentally create a fresh window after that transition. if s.isOpenAIAccountRuntimeBlocked(account) { @@ -199,10 +247,14 @@ func (s *OpenAIGatewayService) shouldRetryOpenAIOAuth429OnSameAccount(account *A // ShouldRetryOpenAIOAuth429 lets RateLimitService defer persistent account // cooldown until the gateway's same-account retry window is exhausted. -func (s *OpenAIGatewayService) ShouldRetryOpenAIOAuth429(account *Account, _ http.Header, _ []byte) bool { +func (s *OpenAIGatewayService) ShouldRetryOpenAIOAuth429(account *Account, headers http.Header, responseBody []byte) bool { if s == nil || !isOpenAIOAuthAccount(account) || account.IsShadow() || s.isOpenAIAccountRuntimeBlocked(account) { return false } + disposition, _ := classifyOpenAIOAuth429(headers, responseBody) + if disposition != openAIOAuth429Transient { + return false + } return s.openAIOAuth429RetryWindowActive(account) } diff --git a/backend/internal/service/openai_account_runtime_block_fastpath_test.go b/backend/internal/service/openai_account_runtime_block_fastpath_test.go index fcda6d6a05..89d8c5b5ec 100644 --- a/backend/internal/service/openai_account_runtime_block_fastpath_test.go +++ b/backend/internal/service/openai_account_runtime_block_fastpath_test.go @@ -14,7 +14,7 @@ import ( ) type oauth429RateLimitRepo struct { - AccountRepository + mockAccountRepoForGemini setRateLimitedCalls int lastRateLimitedUntil time.Time } @@ -62,6 +62,38 @@ func TestOpenAI429FastPath_BlocksOAuthOnlyAfterRetryWindow(t *testing.T) { require.False(t, svc.shouldRetryOpenAIOAuth429OnSameAccount(account, http.StatusTooManyRequests, false)) } +func TestOpenAI429FastPath_BlocksOAuthImmediatelyWhenSevenDayQuotaIsExhausted(t *testing.T) { + repo := &oauth429RateLimitRepo{} + rateLimits := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + svc := &OpenAIGatewayService{rateLimitService: rateLimits} + rateLimits.SetAccountRuntimeBlocker(svc) + account := &Account{ID: 423, Platform: PlatformOpenAI, Type: AccountTypeOAuth} + headers := http.Header{} + headers.Set("x-codex-primary-used-percent", "100") + headers.Set("x-codex-primary-reset-after-seconds", "604800") + headers.Set("x-codex-primary-window-minutes", "10080") + headers.Set("x-codex-secondary-used-percent", "20") + headers.Set("x-codex-secondary-reset-after-seconds", "3600") + headers.Set("x-codex-secondary-window-minutes", "300") + + shouldDisable := svc.handleOpenAIAccountUpstreamError(context.Background(), account, http.StatusTooManyRequests, headers, []byte(`{"error":{"type":"rate_limit_error","code":"rate_limit_exceeded"}}`)) + + require.False(t, shouldDisable) + require.True(t, svc.isOpenAIAccountRuntimeBlocked(account)) + require.Equal(t, 1, repo.setRateLimitedCalls) + require.Greater(t, time.Until(repo.lastRateLimitedUntil), 6*24*time.Hour) + require.False(t, svc.ShouldRetryOpenAIOAuth429(account, headers, nil)) +} + +func TestOpenAI429FastPath_RetriesOAuthWhenNoQuotaSignalExists(t *testing.T) { + svc := &OpenAIGatewayService{} + account := &Account{ID: 424, Platform: PlatformOpenAI, Type: AccountTypeOAuth} + headers := http.Header{"Retry-After": []string{"1"}} + + require.True(t, svc.ShouldRetryOpenAIOAuth429(account, headers, []byte(`{"error":{"type":"rate_limit_error","message":"try again"}}`))) + require.False(t, svc.isOpenAIAccountRuntimeBlocked(account)) +} + func TestOpenAIStream429IgnoresSuccessfulQuotaSnapshotHeaders(t *testing.T) { repo := &oauth429RateLimitRepo{} rateLimits := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) @@ -164,7 +196,7 @@ func TestOpenAI429FastPath_SkipsSparkShadow(t *testing.T) { svc.markOpenAIOAuth429RateLimited(context.Background(), normal, headers, nil) require.False(t, svc.isOpenAIAccountRuntimeBlocked(shadow), "spark shadow must not be runtime-blocked by /responses global 429") - require.False(t, svc.isOpenAIAccountRuntimeBlocked(normal), "normal OpenAI OAuth account stays schedulable during its retry window") + require.True(t, svc.isOpenAIAccountRuntimeBlocked(normal), "normal OpenAI OAuth account with an exhausted 5h window must be paused") } func TestOpenAIRuntimeBlock_AppliesToOpenAIAPIKeyWhenRateLimitServiceStopsScheduling(t *testing.T) { diff --git a/backend/internal/service/openai_gateway_cc_pipeline.go b/backend/internal/service/openai_gateway_cc_pipeline.go index b0e98e5926..a22288b70a 100644 --- a/backend/internal/service/openai_gateway_cc_pipeline.go +++ b/backend/internal/service/openai_gateway_cc_pipeline.go @@ -186,6 +186,10 @@ func (s *OpenAIGatewayService) sendCCUpstreamRequest( if err != nil { return nil, fmt.Errorf("build upstream request: %w", err) } + // 记录本次实际选择的协议端点，供错误日志和用量日志在没有 + // OpenAIForwardResult（例如 503/传输失败）时使用。每次发送都覆盖， + // 避免 Gin context 在账号 failover 尝试之间残留旧端点。 + SetActualOpenAIUpstreamEndpoint(c, "/v1/chat/completions") upstreamReq = upstreamReq.WithContext(WithHTTPUpstreamProfile(upstreamReq.Context(), HTTPUpstreamProfileOpenAI)) upstreamReq.Header.Set("Content-Type", "application/json") upstreamReq.Header.Set("Authorization", "Bearer "+bearerToken) diff --git a/backend/internal/service/openai_gateway_chat_completions.go b/backend/internal/service/openai_gateway_chat_completions.go index 69c1458fdc..a3a992175c 100644 --- a/backend/internal/service/openai_gateway_chat_completions.go +++ b/backend/internal/service/openai_gateway_chat_completions.go @@ -60,6 +60,10 @@ func (s *OpenAIGatewayService) ForwardAsChatCompletions( defaultMappedModel string, ) (*OpenAIForwardResult, error) { beginUpstreamResponseModelObservation(c) + ClearActualOpenAIUpstreamEndpoint(c) + if shouldForwardOpenAIResponsesViaRawChatCompletions(account) { + SetActualOpenAIUpstreamEndpoint(c, "/v1/chat/completions") + } setCodexToolNameReverse(c, nil) if _, err := s.prepareCodexAccountIdentitySource(ctx, c, account); err != nil { return nil, err diff --git a/backend/internal/service/openai_gateway_cn_fixes_test.go b/backend/internal/service/openai_gateway_cn_fixes_test.go index f5cd28ee39..576f8bfd42 100644 --- a/backend/internal/service/openai_gateway_cn_fixes_test.go +++ b/backend/internal/service/openai_gateway_cn_fixes_test.go @@ -11,6 +11,7 @@ package service import ( "context" + "errors" "net/http" "net/http/httptest" "testing" @@ -142,3 +143,112 @@ func TestHandle403_CNProviderStructured403TempUnschedulableFirstHit(t *testing.T require.Equal(t, 1, repo.tempCalls) require.Contains(t, repo.lastTempReason, "(1/3)") } + +func TestIsCNProviderConcurrencyLimit403_ExactClassification(t *testing.T) { + kimi := &Account{Platform: PlatformKimi} + + require.True(t, isCNProviderConcurrencyLimit403(kimi, kimiConcurrentRequestLimitMessage)) + require.True(t, isCNProviderConcurrencyLimit403(kimi, " "+kimiConcurrentRequestLimitMessage+"\n")) + + for name, tc := range map[string]struct { + account *Account + message string + }{ + "permission denied": {kimi, "You do not have permission to access this resource."}, + "generic concurrency wording": {kimi, "concurrent request limit reached"}, + "near match missing punctuation": {kimi, "You've reached your concurrent request limit. Please wait for your ongoing requests to finish and try again"}, + "other CN provider": {&Account{Platform: PlatformZhipu}, kimiConcurrentRequestLimitMessage}, + "non CN provider": {&Account{Platform: PlatformOpenAI}, kimiConcurrentRequestLimitMessage}, + "nil account": {nil, kimiConcurrentRequestLimitMessage}, + } { + t.Run(name, func(t *testing.T) { + require.False(t, isCNProviderConcurrencyLimit403(tc.account, tc.message)) + }) + } +} + +func TestHandle403_OtherCNProviderWithKimiConcurrencyMessageUsesNormalPolicy(t *testing.T) { + repo := &rateLimitAccountRepoStub{} + counter := &openAI403CounterCacheStub{counts: []int64{openAI403DisableThreshold}} + blocker := &runtimeBlockRecorder{} + service := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + service.SetOpenAI403CounterCache(counter) + service.SetAccountRuntimeBlocker(blocker) + account := &Account{ID: 405, Platform: PlatformZhipu, Type: AccountTypeAPIKey} + + shouldDisable := service.HandleUpstreamError( + context.Background(), account, http.StatusForbidden, http.Header{}, + []byte(`{"error":{"message":"You've reached your concurrent request limit. Please wait for your ongoing requests to finish and try again."}}`), + ) + + require.True(t, shouldDisable) + require.Equal(t, 1, repo.setErrorCalls, "non-Kimi CN provider must retain the normal permanent-error policy") + require.Equal(t, 0, repo.tempCalls) + require.Empty(t, counter.counts, "normal CN 403 policy must consume the counter result") + require.Equal(t, []string{"auth_error"}, blocker.reasons, "the Kimi-specific runtime block must not apply") +} + +func TestHandle403_CNProviderConcurrencyLimitAlwaysUsesTemporaryCooldown(t *testing.T) { + repo := &rateLimitAccountRepoStub{} + counter := &openAI403CounterCacheStub{counts: []int64{openAI403DisableThreshold}} + blocker := &runtimeBlockRecorder{} + service := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + service.SetOpenAI403CounterCache(counter) + service.SetAccountRuntimeBlocker(blocker) + account := &Account{ID: 403, Platform: PlatformKimi, Type: AccountTypeAPIKey} + + shouldDisable := service.HandleUpstreamError( + context.Background(), account, http.StatusForbidden, http.Header{}, + []byte(`{"error":{"message":"You've reached your concurrent request limit. Please wait for your ongoing requests to finish and try again."}}`), + ) + + require.True(t, shouldDisable, "the request must still fail over to another account") + require.Equal(t, 0, repo.setErrorCalls) + require.Equal(t, 1, repo.tempCalls) + require.Contains(t, repo.lastTempReason, cnConcurrencyLimitReasonPrefix) + require.Equal(t, []int64{openAI403DisableThreshold}, counter.counts, "transient concurrency 403 must bypass the permanent-error counter") + require.Len(t, blocker.accounts, 1) + require.Equal(t, cnConcurrencyLimitReasonPrefix, blocker.reasons[0]) + require.True(t, blocker.until[0].After(time.Now())) +} + +func TestHandle403_KimiConcurrencyLimitRepositoryFailureKeepsRuntimeBlock(t *testing.T) { + repo := &rateLimitAccountRepoStub{tempErr: errors.New("repository unavailable")} + counter := &openAI403CounterCacheStub{counts: []int64{openAI403DisableThreshold}} + blocker := &runtimeBlockRecorder{} + service := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + service.SetOpenAI403CounterCache(counter) + service.SetAccountRuntimeBlocker(blocker) + account := &Account{ID: 406, Platform: PlatformKimi, Type: AccountTypeAPIKey} + + shouldDisable := service.HandleUpstreamError( + context.Background(), account, http.StatusForbidden, http.Header{}, + []byte(`{"error":{"message":"You've reached your concurrent request limit. Please wait for your ongoing requests to finish and try again."}}`), + ) + + require.True(t, shouldDisable, "the current request must fail over even when persistence fails") + require.Equal(t, 1, repo.tempCalls, "the temporary cooldown should still be persisted when possible") + require.Equal(t, 0, repo.setErrorCalls, "persistence failure must not fall back to permanent account error") + require.Equal(t, []int64{openAI403DisableThreshold}, counter.counts, "persistence failure must not enter the permanent-error counter path") + require.Len(t, blocker.accounts, 1, "the in-memory runtime block must survive repository failure") + require.Same(t, account, blocker.accounts[0]) + require.Equal(t, cnConcurrencyLimitReasonPrefix, blocker.reasons[0]) + require.True(t, blocker.until[0].After(time.Now())) +} + +func TestHandle403_CNProviderNearMatchRetainsNormalPermanentErrorPolicy(t *testing.T) { + repo := &rateLimitAccountRepoStub{} + counter := &openAI403CounterCacheStub{counts: []int64{openAI403DisableThreshold}} + service := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + service.SetOpenAI403CounterCache(counter) + account := &Account{ID: 404, Platform: PlatformKimi, Type: AccountTypeAPIKey} + + shouldDisable := service.HandleUpstreamError( + context.Background(), account, http.StatusForbidden, http.Header{}, + []byte(`{"error":{"message":"You've reached your concurrent request limit. Please contact support."}}`), + ) + + require.True(t, shouldDisable) + require.Equal(t, 1, repo.setErrorCalls, "non-exact 403 must retain existing permission/auth protection") + require.Equal(t, 0, repo.tempCalls) +} diff --git a/backend/internal/service/openai_gateway_forward.go b/backend/internal/service/openai_gateway_forward.go index 8dfecea276..9a4c2389e3 100644 --- a/backend/internal/service/openai_gateway_forward.go +++ b/backend/internal/service/openai_gateway_forward.go @@ -20,6 +20,10 @@ import ( // Forward forwards request to OpenAI API func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, account *Account, body []byte) (*OpenAIForwardResult, error) { beginUpstreamResponseModelObservation(c) + ClearActualOpenAIUpstreamEndpoint(c) + if shouldForwardOpenAIResponsesViaRawChatCompletions(account) { + SetActualOpenAIUpstreamEndpoint(c, "/v1/chat/completions") + } filteredBody, filterErr := filterOpenAIResponsesNoneReasoningEffortForAccount(account, body) if filterErr != nil { return nil, filterErr diff --git a/backend/internal/service/openai_gateway_messages.go b/backend/internal/service/openai_gateway_messages.go index 17c4d633d1..a494be8adf 100644 --- a/backend/internal/service/openai_gateway_messages.go +++ b/backend/internal/service/openai_gateway_messages.go @@ -34,6 +34,10 @@ func (s *OpenAIGatewayService) ForwardAsAnthropic( defaultMappedModel string, ) (*OpenAIForwardResult, error) { beginUpstreamResponseModelObservation(c) + ClearActualOpenAIUpstreamEndpoint(c) + if shouldForwardOpenAIResponsesViaRawChatCompletions(account) { + SetActualOpenAIUpstreamEndpoint(c, "/v1/chat/completions") + } setCodexToolNameReverse(c, nil) if _, err := s.prepareCodexAccountIdentitySource(ctx, c, account); err != nil { return nil, err diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index b87ce9df87..e3f19f335e 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -1678,7 +1678,14 @@ func (s *OpenAIGatewayService) newOpenAIStreamFailoverError( }, }) retryableOnSameAccount := openAIStreamFailedEventRetryableOnSameAccount(account, payload, message) - failoverErr := s.newOpenAIAccountFailoverError(account, statusCode, headers, payload, message, shouldDisable, retryableOnSameAccount) + // 流终止事件承载在 HTTP 200 内，外层响应头描述的是成功流状态，而不是语义上的 + // 429 事件。仅在配额分类时忽略这些头；故障转移错误仍保留它们，使 Retry-After + // 和请求 ID 能继续传递给后续处理。 + classificationHeaders := headers + if statusCode == http.StatusTooManyRequests { + classificationHeaders = nil + } + failoverErr := s.newOpenAIAccountFailoverErrorWithClassificationHeaders(account, statusCode, headers, classificationHeaders, payload, message, shouldDisable, retryableOnSameAccount) if failoverErr.IsCredentialFailure() || failoverErr.RequestScopedTransient { return failoverErr } diff --git a/backend/internal/service/openai_gateway_responses_chat_fallback_test.go b/backend/internal/service/openai_gateway_responses_chat_fallback_test.go index 2989209202..bb994b1aa4 100644 --- a/backend/internal/service/openai_gateway_responses_chat_fallback_test.go +++ b/backend/internal/service/openai_gateway_responses_chat_fallback_test.go @@ -39,11 +39,13 @@ func TestForwardResponses_ForceChatCompletionsRoutesNonStreamingToChatCompletion cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream, } + SetActualOpenAIUpstreamEndpoint(c, "/v1/responses") result, err := svc.Forward(context.Background(), c, forceChatResponsesFallbackAccount(), body) require.NoError(t, err) require.NotNil(t, result) require.Equal(t, "http://upstream.example/v1/chat/completions", upstream.lastReq.URL.String()) + require.Equal(t, "/v1/chat/completions", GetActualOpenAIUpstreamEndpoint(c)) require.Equal(t, HTTPUpstreamProfileOpenAI, HTTPUpstreamProfileFromContext(upstream.lastReq.Context())) require.Equal(t, "hello", gjson.GetBytes(upstream.lastBody, "messages.0.content").String()) require.False(t, gjson.GetBytes(upstream.lastBody, "input").Exists()) diff --git a/backend/internal/service/openai_gateway_scheduling.go b/backend/internal/service/openai_gateway_scheduling.go index 2a4a4f3d09..a90bad3e25 100644 --- a/backend/internal/service/openai_gateway_scheduling.go +++ b/backend/internal/service/openai_gateway_scheduling.go @@ -28,6 +28,7 @@ const ( ) var explicitOpenAIHeaderSessionNames = []string{ + "session-id", "session_id", "conversation_id", openCodeSessionAffinityHeader, @@ -145,7 +146,7 @@ func (s *OpenAIGatewayService) GenerateExplicitSessionHash(c *gin.Context, body // GenerateSessionHash generates a sticky-session hash for OpenAI requests. // // Priority: -// 1. Header: session_id +// 1. Header: session-id / session_id // 2. Header: conversation_id // 3. Header: x-session-affinity / x-session-id / x-opencode-session (OpenCode) // 4. Header: x-conversation-id (CodeBuddy) @@ -1173,6 +1174,11 @@ func (s *OpenAIGatewayService) selectAccountWithLoadAwareness(ctx context.Contex } // ============ Layer 1: Sticky session ============ + // A healthy sticky account whose bounded wait queue is full may be used as a + // one-request capacity spillover in Layer 2. Keep that spillover temporary: + // rewriting the durable binding here would make a short burst migrate the + // whole conversation to a cache-cold account. + stickySpillover := false if sessionHash != "" { accountID := stickyAccountID if accountID > 0 && !isExcluded(accountID) { @@ -1214,6 +1220,7 @@ func (s *OpenAIGatewayService) selectAccountWithLoadAwareness(ctx context.Contex MaxWaiting: cfg.StickySessionMaxWaiting, }) } + stickySpillover = true } } } @@ -1365,7 +1372,7 @@ func (s *OpenAIGatewayService) selectAccountWithLoadAwareness(ctx context.Contex if selectErr != nil { return nil, true, selectErr } - if sessionHash != "" && !gatewayProfitControlGateActive(ctx) { + if sessionHash != "" && !stickySpillover && !gatewayProfitControlGateActive(ctx) { _ = s.setStickySessionAccountID(ctx, groupID, sessionHash, fresh.ID, openaiStickySessionTTL) } return selection, true, nil @@ -1404,7 +1411,7 @@ func (s *OpenAIGatewayService) selectAccountWithLoadAwareness(ctx context.Contex if selectErr != nil { return nil, selectErr } - if sessionHash != "" && !gatewayProfitControlGateActive(ctx) { + if sessionHash != "" && !stickySpillover && !gatewayProfitControlGateActive(ctx) { _ = s.setStickySessionAccountID(ctx, groupID, sessionHash, fresh.ID, openaiStickySessionTTL) } return selection, nil diff --git a/backend/internal/service/openai_gateway_service.go b/backend/internal/service/openai_gateway_service.go index c4d085b024..5ce98fba0f 100644 --- a/backend/internal/service/openai_gateway_service.go +++ b/backend/internal/service/openai_gateway_service.go @@ -319,6 +319,16 @@ func SetActualOpenAIUpstreamEndpoint(c *gin.Context, endpoint string) { } } +// ClearActualOpenAIUpstreamEndpoint 清理当前转发尝试记录的端点。 +// Handler 会在账号 failover 尝试间复用同一个 Gin context，因此每次尝试 +// 都必须从无残留状态开始。 +func ClearActualOpenAIUpstreamEndpoint(c *gin.Context) { + if c == nil { + return + } + c.Set(openAIUpstreamEndpointContextKey, "") +} + // GetActualOpenAIUpstreamEndpoint returns the endpoint recorded by the latest // forwarding attempt in this request. func GetActualOpenAIUpstreamEndpoint(c *gin.Context) string { diff --git a/backend/internal/service/openai_gateway_service_test.go b/backend/internal/service/openai_gateway_service_test.go index 1e41298cc1..568c948627 100644 --- a/backend/internal/service/openai_gateway_service_test.go +++ b/backend/internal/service/openai_gateway_service_test.go @@ -391,6 +391,7 @@ func TestOpenAIGatewayService_ClientSessionHeaderPriority(t *testing.T) { name string value string }{ + {name: "session-id", value: "codex-session"}, {name: "session_id", value: "generic-session"}, {name: "conversation_id", value: "generic-conversation"}, {name: openCodeSessionAffinityHeader, value: "opencode-affinity"}, @@ -416,6 +417,31 @@ func TestOpenAIGatewayService_ClientSessionHeaderPriority(t *testing.T) { require.Equal(t, "body-session", svc.ExtractSessionID(c, body)) } +func TestOpenAIGatewayService_CodexSessionIDKeepsReconnectHashStable(t *testing.T) { + gin.SetMode(gin.TestMode) + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodGet, "/v1/responses", nil) + c.Request.Header.Set("session-id", "codex-reconnect-session") + + svc := &OpenAIGatewayService{} + warmup := []byte(`{ + "type":"response.create", + "model":"gpt-5.6-sol", + "generate":false, + "tools":[{"type":"custom","name":"exec"}], + "input":[{"role":"user","content":"warmup"}] + }`) + business := []byte(`{ + "type":"response.create", + "model":"gpt-5.6-sol", + "input":[{"role":"user","content":"install codex"}] + }`) + + require.Equal(t, svc.GenerateSessionHash(c, warmup), svc.GenerateSessionHash(c, business)) + require.Equal(t, "codex-reconnect-session", svc.ExtractSessionID(c, business)) +} + func TestOpenAIGatewayService_ClientSessionHeadersIgnorePerRequestIDs(t *testing.T) { gin.SetMode(gin.TestMode) recorder := httptest.NewRecorder() @@ -1181,6 +1207,49 @@ func TestOpenAISelectAccountWithLoadAwareness_StickyWaitPlan(t *testing.T) { } } +func TestOpenAISelectAccountWithLoadAwareness_StickyCapacitySpilloverKeepsBinding(t *testing.T) { + sessionHash := "sticky-spillover" + groupID := int64(1) + repo := stubOpenAIAccountRepo{ + accounts: []Account{ + {ID: 1, Platform: PlatformOpenAI, Status: StatusActive, Schedulable: true, Concurrency: 6, Priority: 1, GroupIDs: []int64{groupID}}, + {ID: 2, Platform: PlatformOpenAI, Status: StatusActive, Schedulable: true, Concurrency: 6, Priority: 1, GroupIDs: []int64{groupID}}, + }, + } + cache := &stubGatewayCache{ + sessionBindings: map[string]int64{"openai:" + sessionHash: 1}, + } + concurrencyCache := stubConcurrencyCache{ + acquireResults: map[int64]bool{1: false, 2: true}, + waitCounts: map[int64]int{1: 1}, + loadMap: map[int64]*AccountLoadInfo{ + 1: {AccountID: 1, LoadRate: 100}, + 2: {AccountID: 2, LoadRate: 10}, + }, + } + cfg := &config.Config{RunMode: config.RunModeStandard} + cfg.Gateway.Scheduling.LoadBatchEnabled = true + cfg.Gateway.Scheduling.StickySessionMaxWaiting = 1 + + svc := &OpenAIGatewayService{ + accountRepo: repo, + cache: cache, + cfg: cfg, + concurrencyService: NewConcurrencyService(concurrencyCache), + } + + selection, err := svc.SelectAccountWithLoadAwareness(context.Background(), &groupID, sessionHash, "gpt-4", nil) + require.NoError(t, err) + require.NotNil(t, selection) + require.NotNil(t, selection.Account) + require.Equal(t, int64(2), selection.Account.ID, "capacity spillover should use the other account for this request") + require.True(t, selection.Acquired) + require.Equal(t, int64(1), cache.sessionBindings["openai:"+sessionHash], "capacity spillover must not migrate the durable sticky binding") + if selection.ReleaseFunc != nil { + selection.ReleaseFunc() + } +} + func TestOpenAISelectAccountWithLoadAwareness_PrefersLowerLoad(t *testing.T) { groupID := int64(1) repo := stubOpenAIAccountRepo{ diff --git a/backend/internal/service/openai_gateway_upstream_errors.go b/backend/internal/service/openai_gateway_upstream_errors.go index 4a4e9ab7fc..70c000e321 100644 --- a/backend/internal/service/openai_gateway_upstream_errors.go +++ b/backend/internal/service/openai_gateway_upstream_errors.go @@ -333,7 +333,20 @@ func (s *OpenAIGatewayService) newOpenAIAccountFailoverError( shouldDisable bool, retryableOnSameAccount bool, ) *UpstreamFailoverError { - oauth429Retry := s.shouldRetryOpenAIOAuth429OnSameAccount(account, statusCode, shouldDisable) + return s.newOpenAIAccountFailoverErrorWithClassificationHeaders(account, statusCode, responseHeaders, responseHeaders, responseBody, upstreamMsg, shouldDisable, retryableOnSameAccount) +} + +func (s *OpenAIGatewayService) newOpenAIAccountFailoverErrorWithClassificationHeaders( + account *Account, + statusCode int, + responseHeaders http.Header, + classificationHeaders http.Header, + responseBody []byte, + upstreamMsg string, + shouldDisable bool, + retryableOnSameAccount bool, +) *UpstreamFailoverError { + oauth429Retry := s.shouldRetryOpenAIOAuth429OnSameAccountWithResponse(account, statusCode, shouldDisable, classificationHeaders, responseBody) failoverErr := newOpenAIUpstreamFailoverError( statusCode, responseHeaders, diff --git a/backend/internal/service/openai_ws_forwarder_logutil.go b/backend/internal/service/openai_ws_forwarder_logutil.go index ac5a3720b1..ddb37d8380 100644 --- a/backend/internal/service/openai_ws_forwarder_logutil.go +++ b/backend/internal/service/openai_ws_forwarder_logutil.go @@ -65,7 +65,10 @@ func resolveOpenAIWSSessionHeaders(c *gin.Context, promptCacheKey string) openAI ConversationSource: "none", } if c != nil && c.Request != nil { - if sessionID := strings.TrimSpace(c.Request.Header.Get("session_id")); sessionID != "" { + if sessionID := strings.TrimSpace(c.Request.Header.Get("session-id")); sessionID != "" { + resolution.SessionID = sessionID + resolution.SessionSource = "header_session-id" + } else if sessionID := strings.TrimSpace(c.Request.Header.Get("session_id")); sessionID != "" { resolution.SessionID = sessionID resolution.SessionSource = "header_session_id" } diff --git a/backend/internal/service/openai_ws_forwarder_logutil_test.go b/backend/internal/service/openai_ws_forwarder_logutil_test.go new file mode 100644 index 0000000000..eaf8b5f22b --- /dev/null +++ b/backend/internal/service/openai_ws_forwarder_logutil_test.go @@ -0,0 +1,37 @@ +package service + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +func TestResolveOpenAIWSSessionHeadersPrefersCodexHyphenHeader(t *testing.T) { + gin.SetMode(gin.TestMode) + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodGet, "/v1/responses", nil) + c.Request.Header.Set("session-id", "codex-session") + c.Request.Header.Set("session_id", "legacy-session") + + resolution := resolveOpenAIWSSessionHeaders(c, "prompt-cache") + + require.Equal(t, "codex-session", resolution.SessionID) + require.Equal(t, "header_session-id", resolution.SessionSource) +} + +func TestResolveOpenAIWSSessionHeadersFallsBackToLegacyHeader(t *testing.T) { + gin.SetMode(gin.TestMode) + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodGet, "/v1/responses", nil) + c.Request.Header.Set("session_id", "legacy-session") + + resolution := resolveOpenAIWSSessionHeaders(c, "prompt-cache") + + require.Equal(t, "legacy-session", resolution.SessionID) + require.Equal(t, "header_session_id", resolution.SessionSource) +} diff --git a/backend/internal/service/openai_ws_forwarder_success_test.go b/backend/internal/service/openai_ws_forwarder_success_test.go index e775104fa2..6923352d7d 100644 --- a/backend/internal/service/openai_ws_forwarder_success_test.go +++ b/backend/internal/service/openai_ws_forwarder_success_test.go @@ -804,6 +804,73 @@ func TestOpenAIGatewayService_Forward_WSv2_OAuthStoreFalseByDefault(t *testing.T require.Equal(t, isolateOpenAIUpstreamSessionID(0, account, "conv-oauth-1"), captureDialer.lastHeaders.Get("conversation_id")) } +func TestOpenAIGatewayService_Forward_WSv2_OAuthSanitizesInvalidNativeToolItemID(t *testing.T) { + gin.SetMode(gin.TestMode) + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/openai/v1/responses", nil) + c.Request.Header.Set("User-Agent", "codex_cli_rs/0.144.1") + groupID := int64(5662) + c.Set("api_key", &APIKey{GroupID: &groupID}) + + cfg := newOpenAIWSV2TestConfig() + cfg.Security.URLAllowlist.Enabled = false + cfg.Security.URLAllowlist.AllowInsecureHTTP = true + cfg.Gateway.OpenAIWS.MaxConnsPerAccount = 1 + cfg.Gateway.OpenAIWS.MinIdlePerAccount = 0 + cfg.Gateway.OpenAIWS.MaxIdlePerAccount = 1 + cfg.Gateway.OpenAIWS.QueueLimitPerConn = 8 + cfg.Gateway.OpenAIWS.DialTimeoutSeconds = 3 + cfg.Gateway.OpenAIWS.ReadTimeoutSeconds = 3 + cfg.Gateway.OpenAIWS.WriteTimeoutSeconds = 3 + + captureConn := &openAIWSCaptureConn{events: [][]byte{ + []byte(`{"type":"response.completed","response":{"id":"resp_oauth_tool_history","model":"gpt-5.6-sol","usage":{"input_tokens":1,"output_tokens":1}}}`), + }} + captureDialer := &openAIWSCaptureDialer{conn: captureConn} + pool := newOpenAIWSConnPool(cfg) + pool.setClientDialerForTest(captureDialer) + + svc := &OpenAIGatewayService{ + cfg: cfg, + httpUpstream: &httpUpstreamRecorder{}, + cache: &stubGatewayCache{}, + openaiWSResolver: NewOpenAIWSProtocolResolver(cfg), + toolCorrector: NewCodexToolCorrector(), + openaiWSPool: pool, + } + + account := &Account{ + ID: 5662, + Name: "openai-oauth-ws-tool-history", + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + Status: StatusActive, + Schedulable: true, + Concurrency: 1, + Credentials: map[string]any{"access_token": "test-oauth-token"}, + Extra: map[string]any{ + "responses_websockets_v2_enabled": true, + }, + } + + body := []byte(`{"model":"gpt-5.6-sol","stream":false,"instructions":"Continue the task.","input":[{"type":"custom_tool_call","id":"fc_hotfix_probe","call_id":"fc_hotfix","name":"exec","input":"pwd","status":"completed"},{"type":"custom_tool_call_output","call_id":"fc_hotfix","output":"done"}]}`) + result, err := svc.Forward(context.Background(), c, account, body) + require.NoError(t, err) + require.NotNil(t, result) + + captureConn.mu.Lock() + requestPayload := cloneMapStringAny(captureConn.lastWrite) + captureConn.mu.Unlock() + requestJSON := requestToJSONString(requestPayload) + require.Equal(t, "response.create", gjson.Get(requestJSON, "type").String()) + require.False(t, gjson.Get(requestJSON, "input.0.id").Exists(), "stale fc_* ID must not be replayed as a native custom_tool_call ID") + require.Equal(t, "ctc_hotfix", gjson.Get(requestJSON, "input.0.call_id").String()) + require.Equal(t, "custom_tool_call_output", gjson.Get(requestJSON, "input.1.type").String()) + require.Equal(t, "ctc_hotfix", gjson.Get(requestJSON, "input.1.call_id").String()) +} + func TestOpenAIGatewayService_Forward_WSv2_OAuthOriginatorCompatibility(t *testing.T) { gin.SetMode(gin.TestMode) diff --git a/backend/internal/service/ratelimit_cn_providers.go b/backend/internal/service/ratelimit_cn_providers.go index 4aab5a8751..17239e4bbc 100644 --- a/backend/internal/service/ratelimit_cn_providers.go +++ b/backend/internal/service/ratelimit_cn_providers.go @@ -26,6 +26,33 @@ const cnBalanceExtraSuffixLow = "balance_low" // 其他子系统（阈值/限流/401）写入的临时停调。 const cnBalanceLowReasonPrefix = "cn_balance_low" +const kimiConcurrentRequestLimitMessage = "You've reached your concurrent request limit. Please wait for your ongoing requests to finish and try again." + +const cnConcurrencyLimitReasonPrefix = "cn_concurrency_limit" + +func isCNProviderConcurrencyLimit403(account *Account, upstreamMsg string) bool { + return account != nil && account.Platform == PlatformKimi && + strings.TrimSpace(upstreamMsg) == kimiConcurrentRequestLimitMessage +} + +func (s *RateLimitService) handleCNProviderConcurrencyLimit403( + ctx context.Context, + account *Account, +) { + until := time.Now().Add(time.Duration(openAI403CooldownMinutesDefault) * time.Minute) + reason := cnConcurrencyLimitReasonPrefix + ": " + kimiConcurrentRequestLimitMessage + s.notifyAccountSchedulingBlocked(account, until, cnConcurrencyLimitReasonPrefix) + if err := s.accountRepo.SetTempUnschedulable(ctx, account.ID, until, reason); err != nil { + slog.Warn("cn_concurrency_limit_set_temp_unschedulable_failed", "account_id", account.ID, "error", err) + return + } + slog.Info("cn_provider_concurrency_limited", + "account_id", account.ID, + "platform", account.Platform, + "until", until.UTC(), + ) +} + // cnBalanceLowReason 构造余额不足临时停调的 reason（带稳定前缀）。 func cnBalanceLowReason(upstreamMsg string) string { if upstreamMsg = strings.TrimSpace(upstreamMsg); upstreamMsg != "" { diff --git a/backend/internal/service/ratelimit_service.go b/backend/internal/service/ratelimit_service.go index 0a52925cd9..ff8bfac268 100644 --- a/backend/internal/service/ratelimit_service.go +++ b/backend/internal/service/ratelimit_service.go @@ -936,6 +936,13 @@ func (s *RateLimitService) handle403(ctx context.Context, account *Account, upst if account.Platform == PlatformAntigravity { return s.handleAntigravity403(ctx, account, upstreamMsg, responseBody) } + // Kimi reports its transient per-account concurrency/business limit as a 403. + // Keep the normal 403 failover signal (true), but never feed this exact message + // into the escalating 403 counter that can permanently mark the account error. + if isCNProviderConcurrencyLimit403(account, upstreamMsg) { + s.handleCNProviderConcurrencyLimit403(ctx, account) + return true + } // 国产供应商与 openai 同口径:HTML 403(CDN/代理拦截页)不构成账号失效证据, // 且 403 在 failover 状态集里会被逐账号重放——直接 SetError 会让一个坏请求/ // 一层坏代理连环永久禁用整组账号。走 HTML 豁免 + N 次累计 + 临时冷却。 diff --git a/backend/internal/service/ratelimit_service_401_test.go b/backend/internal/service/ratelimit_service_401_test.go index 48e6a41def..c1f9f06954 100644 --- a/backend/internal/service/ratelimit_service_401_test.go +++ b/backend/internal/service/ratelimit_service_401_test.go @@ -25,6 +25,7 @@ type rateLimitAccountRepoStub struct { lastTempReason string lastErrorID int64 lastTempID int64 + tempErr error } func (r *rateLimitAccountRepoStub) SetError(ctx context.Context, id int64, errorMsg string) error { @@ -38,7 +39,7 @@ func (r *rateLimitAccountRepoStub) SetTempUnschedulable(ctx context.Context, id r.tempCalls++ r.lastTempID = id r.lastTempReason = reason - return nil + return r.tempErr } func (r *rateLimitAccountRepoStub) UpdateCredentials(ctx context.Context, id int64, credentials map[string]any) error {