diff --git a/backend/internal/handler/openai_gateway_handler.go b/backend/internal/handler/openai_gateway_handler.go index 2ed648c49b..c5491f2f29 100644 --- a/backend/internal/handler/openai_gateway_handler.go +++ b/backend/internal/handler/openai_gateway_handler.go @@ -430,6 +430,7 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { sameAccountRetryCount := make(map[int64]int) var lastFailoverErr *service.UpstreamFailoverError var oauth429FailoverState service.OpenAIOAuth429FailoverState + var passthroughFailoverState openAIPassthroughFailoverState // 生图意图的 /v1/responses 请求必须调度到确实支持 Responses API 的账号，否则 // 会在 forward 阶段被静默降级为无法生图的 Chat Completions 直转（#4417）。 @@ -526,13 +527,17 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { // 用扣除 compact 心跳字节的口径快照：心跳注释不构成语义响应， // 不能因心跳字节变化而放弃 failover 换号（#3887）。 writerSizeBeforeForward := service.OpenAICompactKeepaliveAdjustedWrittenSize(c) + // 跨 passthrough 边界的 failover：从 Kiro 等透传账号切到 Bedrock 等非透传账号前， + // 从不可变的 canonical forwardBody 派生本次尝试 body 并整块剔除上游私有的加密 + // reasoning item（含耦合的 id/summary），避免非透传上游 400 拒绝 Kiro reasoning 形态。 + attemptBody := h.deriveOpenAIForwardAttemptBody(reqLog, forwardBody, account, &passthroughFailoverState) result, err := func() (*service.OpenAIForwardResult, error) { defer func() { if accountReleaseFunc != nil { accountReleaseFunc() } }() - return h.gatewayService.Forward(c.Request.Context(), c, account, forwardBody) + return h.gatewayService.Forward(c.Request.Context(), c, account, attemptBody) }() cyberBlockKeyHTTP := "" if service.GetOpsCyberPolicy(c) != nil { diff --git a/backend/internal/handler/openai_gateway_reasoning_failover.go b/backend/internal/handler/openai_gateway_reasoning_failover.go new file mode 100644 index 0000000000..fecba80330 --- /dev/null +++ b/backend/internal/handler/openai_gateway_reasoning_failover.go @@ -0,0 +1,62 @@ +package handler + +import ( + "github.com/Wei-Shaw/sub2api/internal/service" + "go.uber.org/zap" +) + +// openAIPassthroughFailoverState tracks whether this forwarding loop has attempted +// an OpenAI passthrough account. Once it has, every subsequent non-passthrough +// attempt must use a sanitized body because the immutable canonical body can carry +// provider-specific encrypted reasoning produced by that passthrough upstream. +type openAIPassthroughFailoverState struct { + passthroughSeen bool +} + +// deriveOpenAIForwardAttemptBody returns the request body for the upcoming forward +// attempt against account. It always derives from the immutable canonical body and +// only removes encrypted reasoning input item(s) when the loop has already attempted +// a passthrough account and the upcoming account is non-passthrough. This remains +// sticky across retries and additional non-passthrough accounts. Attempts before +// any passthrough account, and all passthrough attempts, forward the canonical body +// unchanged. The canonical slice is never mutated. +// +// This method is invoked exactly once per forward attempt, immediately before the +// Forward call, and advances the failover state as a side effect. +func (h *OpenAIGatewayHandler) deriveOpenAIForwardAttemptBody( + reqLog *zap.Logger, + canonicalBody []byte, + account *service.Account, + state *openAIPassthroughFailoverState, +) []byte { + currentPassthrough := account.IsOpenAIPassthroughEnabled() + if currentPassthrough { + state.passthroughSeen = true + return canonicalBody + } + if !state.passthroughSeen { + return canonicalBody + } + + sanitized, changed, err := service.SanitizeOpenAICrossModeFailoverReasoning(canonicalBody) + if err != nil { + if reqLog != nil { + reqLog.Warn("openai.failover_cross_mode_reasoning_sanitize_failed", + zap.Int64("account_id", account.ID), + zap.Error(err), + ) + } + return canonicalBody + } + if !changed { + return canonicalBody + } + if reqLog != nil { + reqLog.Info("openai.failover_cross_mode_reasoning_stripped", + zap.Int64("account_id", account.ID), + zap.Bool("account_passthrough", currentPassthrough), + zap.Bool("passthrough_seen", true), + ) + } + return sanitized +} diff --git a/backend/internal/handler/openai_gateway_reasoning_failover_test.go b/backend/internal/handler/openai_gateway_reasoning_failover_test.go new file mode 100644 index 0000000000..321364cd4e --- /dev/null +++ b/backend/internal/handler/openai_gateway_reasoning_failover_test.go @@ -0,0 +1,146 @@ +package handler + +import ( + "testing" + + "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/stretchr/testify/require" + "github.com/tidwall/gjson" +) + +// kiroReasoningCanonicalBody mirrors a native OpenAI Responses request whose input +// carries a provider-specific encrypted reasoning item (Kiro shape: encrypted_content +// coupled with an id and summary), plus surrounding message items. +const kiroReasoningCanonicalBody = `{"model":"gpt-5.1","stream":false,"input":[` + + `{"type":"message","role":"user","content":"hello"},` + + `{"type":"reasoning","id":"rs_kiro_abc123","encrypted_content":"ENC_BLOB","summary":[{"type":"summary_text","text":"thinking"}]},` + + `{"type":"message","role":"assistant","content":"hi"}` + + `]}` + +func newOpenAIPassthroughAccount(id int64, passthrough bool) *service.Account { + extra := map[string]any{"openai_passthrough": passthrough} + return &service.Account{ + ID: id, + Platform: service.PlatformOpenAI, + Type: service.AccountTypeAPIKey, + Extra: extra, + } +} + +func reasoningItemCount(t *testing.T, body []byte) int { + t.Helper() + count := 0 + gjson.GetBytes(body, "input").ForEach(func(_, item gjson.Result) bool { + if item.Get("type").String() == "reasoning" { + count++ + } + return true + }) + return count +} + +// TestDeriveOpenAIForwardAttemptBody_CrossModeStripsKiroReasoning drives the real +// account-switch/request-body seam used inside OpenAIGatewayHandler.Responses: the +// same deriveOpenAIForwardAttemptBody call the failover loop makes, fed real +// *service.Account values in sequence. It proves the first passthrough attempt gets +// the untouched reasoning item, and after a failover to a non-passthrough account +// the second attempt body has the entire reasoning item removed — while the +// immutable canonical body is never mutated. +func TestDeriveOpenAIForwardAttemptBody_CrossModeStripsKiroReasoning(t *testing.T) { + h := &OpenAIGatewayHandler{} + canonical := []byte(kiroReasoningCanonicalBody) + + kiro := newOpenAIPassthroughAccount(1, true) // Kiro: openai_passthrough=true + bedrock := newOpenAIPassthroughAccount(2, false) // Bedrock Mantle: passthrough disabled + + state := &openAIPassthroughFailoverState{} + + // Attempt 1 — passthrough (Kiro): original reasoning item must survive intact. + firstBody := h.deriveOpenAIForwardAttemptBody(nil, canonical, kiro, state) + require.Equal(t, 1, reasoningItemCount(t, firstBody), "first passthrough attempt must keep the reasoning item") + require.Equal(t, "ENC_BLOB", gjson.GetBytes(firstBody, "input.1.encrypted_content").String()) + require.Equal(t, "rs_kiro_abc123", gjson.GetBytes(firstBody, "input.1.id").String()) + require.JSONEq(t, kiroReasoningCanonicalBody, string(firstBody), "first attempt body must equal the canonical body") + + // Attempt 2 — failover switches to non-passthrough (Bedrock): the whole + // provider-specific encrypted reasoning item (id/summary/encrypted_content) is dropped. + secondBody := h.deriveOpenAIForwardAttemptBody(nil, canonical, bedrock, state) + require.Equal(t, 0, reasoningItemCount(t, secondBody), "cross-mode attempt must drop the reasoning item entirely") + require.False(t, gjson.GetBytes(secondBody, "input.#(encrypted_content)").Exists(), "no encrypted_content may remain") + require.NotContains(t, string(secondBody), "rs_kiro_abc123", "coupled reasoning id must be gone") + require.NotContains(t, string(secondBody), "ENC_BLOB") + + // Surviving message items are preserved. + require.Equal(t, 2, int(gjson.GetBytes(secondBody, "input.#").Int())) + require.Equal(t, "hello", gjson.GetBytes(secondBody, "input.0.content").String()) + require.Equal(t, "hi", gjson.GetBytes(secondBody, "input.1.content").String()) + + // The canonical body is immutable across attempts. + require.JSONEq(t, kiroReasoningCanonicalBody, string(canonical), "canonical forwardBody must never be mutated") +} + +// TestDeriveOpenAIForwardAttemptBody_SameModePreservesReasoning proves that +// non-passthrough attempts before any passthrough attempt, passthrough-family +// failovers, and switching into passthrough forward the canonical reasoning item +// unchanged. +func TestDeriveOpenAIForwardAttemptBody_SameModePreservesReasoning(t *testing.T) { + h := &OpenAIGatewayHandler{} + canonical := []byte(kiroReasoningCanonicalBody) + + t.Run("non_passthrough_to_non_passthrough", func(t *testing.T) { + state := &openAIPassthroughFailoverState{} + first := h.deriveOpenAIForwardAttemptBody(nil, canonical, newOpenAIPassthroughAccount(10, false), state) + second := h.deriveOpenAIForwardAttemptBody(nil, canonical, newOpenAIPassthroughAccount(11, false), state) + require.Equal(t, 1, reasoningItemCount(t, first)) + require.Equal(t, 1, reasoningItemCount(t, second), "Bedrock-family failover must preserve reasoning") + require.JSONEq(t, kiroReasoningCanonicalBody, string(second)) + }) + + t.Run("passthrough_to_passthrough", func(t *testing.T) { + state := &openAIPassthroughFailoverState{} + first := h.deriveOpenAIForwardAttemptBody(nil, canonical, newOpenAIPassthroughAccount(20, true), state) + second := h.deriveOpenAIForwardAttemptBody(nil, canonical, newOpenAIPassthroughAccount(21, true), state) + require.Equal(t, 1, reasoningItemCount(t, first)) + require.Equal(t, 1, reasoningItemCount(t, second), "Kiro-family failover must preserve reasoning") + }) + + t.Run("non_passthrough_to_passthrough", func(t *testing.T) { + state := &openAIPassthroughFailoverState{} + _ = h.deriveOpenAIForwardAttemptBody(nil, canonical, newOpenAIPassthroughAccount(30, false), state) + second := h.deriveOpenAIForwardAttemptBody(nil, canonical, newOpenAIPassthroughAccount(31, true), state) + require.Equal(t, 1, reasoningItemCount(t, second), "switching into passthrough must preserve reasoning") + }) + + t.Run("same_account_pool_retry", func(t *testing.T) { + state := &openAIPassthroughFailoverState{} + kiro := newOpenAIPassthroughAccount(40, true) + _ = h.deriveOpenAIForwardAttemptBody(nil, canonical, kiro, state) + retry := h.deriveOpenAIForwardAttemptBody(nil, canonical, kiro, state) + require.Equal(t, 1, reasoningItemCount(t, retry), "same-account retry must preserve reasoning") + }) +} + +// TestDeriveOpenAIForwardAttemptBody_SanitizationSticksAcrossBedrockRetries proves +// that once a passthrough account has been attempted, all later non-passthrough +// retries and failovers remain sanitized instead of restoring the canonical Kiro +// reasoning item after the first Bedrock failure. +func TestDeriveOpenAIForwardAttemptBody_SanitizationSticksAcrossBedrockRetries(t *testing.T) { + h := &OpenAIGatewayHandler{} + canonical := []byte(kiroReasoningCanonicalBody) + state := &openAIPassthroughFailoverState{} + + kiro := newOpenAIPassthroughAccount(50, true) + bedrockA := newOpenAIPassthroughAccount(51, false) + bedrockB := newOpenAIPassthroughAccount(52, false) + + first := h.deriveOpenAIForwardAttemptBody(nil, canonical, kiro, state) + second := h.deriveOpenAIForwardAttemptBody(nil, canonical, bedrockA, state) + retry := h.deriveOpenAIForwardAttemptBody(nil, canonical, bedrockA, state) + nextAccount := h.deriveOpenAIForwardAttemptBody(nil, canonical, bedrockB, state) + + require.Equal(t, 1, reasoningItemCount(t, first)) + require.Equal(t, 0, reasoningItemCount(t, second)) + require.Equal(t, 0, reasoningItemCount(t, retry), "same Bedrock account retry must remain sanitized") + require.Equal(t, 0, reasoningItemCount(t, nextAccount), "later Bedrock account must remain sanitized") + require.JSONEq(t, kiroReasoningCanonicalBody, string(canonical), "canonical forwardBody must never be mutated") +} diff --git a/backend/internal/service/openai_gateway_request_body.go b/backend/internal/service/openai_gateway_request_body.go index 14520958f2..fdd4019119 100644 --- a/backend/internal/service/openai_gateway_request_body.go +++ b/backend/internal/service/openai_gateway_request_body.go @@ -156,6 +156,119 @@ func sanitizeEncryptedReasoningInputItem(item any) (next any, changed bool, keep return inputItem, true, true } +// SanitizeOpenAICrossModeFailoverReasoning derives a failover attempt body from +// the canonical request body by dropping provider-specific encrypted reasoning +// input items in full (encrypted_content plus the coupled id/summary shape). +// +// This is the proactive counterpart to the reactive same-account +// invalid_encrypted_content recovery in Forward: when a failover switches from an +// OpenAI passthrough account (which forwards upstream-native encrypted reasoning, +// e.g. Kiro) to a non-passthrough account (e.g. Bedrock Mantle) that rejects the +// provider-specific reasoning IDs/shape, the whole reasoning item must go before +// the request reaches the new upstream. Unlike trimOpenAIEncryptedReasoningItems, +// which only strips the encrypted_content / null-content fields while preserving +// the reasoning item's id and summary, this drops the entire item. +// +// The input slice is treated as immutable and is never mutated; a distinct slice +// is returned only when changed is true. +func SanitizeOpenAICrossModeFailoverReasoning(body []byte) (sanitized []byte, changed bool, err error) { + if len(body) == 0 { + return body, false, nil + } + if !gjson.GetBytes(body, "input").Exists() { + return body, false, nil + } + var decoded map[string]any + decoder := json.NewDecoder(bytes.NewReader(body)) + decoder.UseNumber() + if err := decoder.Decode(&decoded); err != nil { + return body, false, fmt.Errorf("decode cross-mode failover body: %w", err) + } + if !dropOpenAIEncryptedReasoningInputItems(decoded) { + return body, false, nil + } + out, marshalErr := marshalOpenAIUpstreamJSON(decoded) + if marshalErr != nil { + return body, false, fmt.Errorf("serialize cross-mode failover body: %w", marshalErr) + } + return out, true, nil +} + +// dropOpenAIEncryptedReasoningInputItems removes reasoning input items that carry +// provider-specific encrypted_content in full — including their coupled id and +// summary — and reports whether anything changed. Contrast with +// trimOpenAIEncryptedReasoningItems, which only strips fields while keeping the +// reasoning item skeleton. +func dropOpenAIEncryptedReasoningInputItems(reqBody map[string]any) bool { + if len(reqBody) == 0 { + return false + } + inputValue, has := reqBody["input"] + if !has { + return false + } + switch input := inputValue.(type) { + case []any: + filtered := input[:0] + changed := false + for _, item := range input { + if isOpenAIEncryptedReasoningInputItem(item) { + changed = true + continue + } + filtered = append(filtered, item) + } + if !changed { + return false + } + if len(filtered) == 0 { + delete(reqBody, "input") + return true + } + reqBody["input"] = filtered + return true + case []map[string]any: + filtered := input[:0] + changed := false + for _, item := range input { + if isOpenAIEncryptedReasoningInputItem(item) { + changed = true + continue + } + filtered = append(filtered, item) + } + if !changed { + return false + } + if len(filtered) == 0 { + delete(reqBody, "input") + return true + } + reqBody["input"] = filtered + return true + case map[string]any: + if isOpenAIEncryptedReasoningInputItem(input) { + delete(reqBody, "input") + return true + } + return false + default: + return false + } +} + +func isOpenAIEncryptedReasoningInputItem(item any) bool { + inputItem, ok := item.(map[string]any) + if !ok { + return false + } + if itemType, _ := inputItem["type"].(string); strings.TrimSpace(itemType) != "reasoning" { + return false + } + _, has := inputItem["encrypted_content"] + return has +} + func IsOpenAIResponsesCompactPathForTest(c *gin.Context) bool { return isOpenAIResponsesCompactPath(c) } diff --git a/backend/internal/service/openai_gateway_request_body_reasoning_test.go b/backend/internal/service/openai_gateway_request_body_reasoning_test.go index 62a8855114..ad14f4d8d8 100644 --- a/backend/internal/service/openai_gateway_request_body_reasoning_test.go +++ b/backend/internal/service/openai_gateway_request_body_reasoning_test.go @@ -5,6 +5,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/tidwall/gjson" ) func TestTrimOpenAIEncryptedReasoningItems_ContentNull(t *testing.T) { @@ -98,6 +99,53 @@ func TestTrimOpenAIEncryptedReasoningItems_NoReasoningItems(t *testing.T) { assert.False(t, changed) } +func TestSanitizeOpenAICrossModeFailoverReasoning_DropsWholeEncryptedItem(t *testing.T) { + body := []byte(`{"model":"gpt-5.1","input":[` + + `{"type":"message","role":"user","content":"hi"},` + + `{"type":"reasoning","id":"rs_kiro_1","encrypted_content":"ENC","summary":[{"type":"summary_text","text":"t"}]},` + + `{"type":"message","role":"assistant","content":"yo"}` + + `]}`) + + sanitized, changed, err := SanitizeOpenAICrossModeFailoverReasoning(body) + require.NoError(t, err) + require.True(t, changed) + // The whole reasoning item is gone — id and summary go with encrypted_content, + // unlike trimOpenAIEncryptedReasoningItems which keeps the skeleton. + require.NotContains(t, string(sanitized), "reasoning") + require.NotContains(t, string(sanitized), "rs_kiro_1") + require.NotContains(t, string(sanitized), "summary_text") + require.Equal(t, int64(2), gjson.GetBytes(sanitized, "input.#").Int()) +} + +func TestSanitizeOpenAICrossModeFailoverReasoning_NoEncryptedIsNoop(t *testing.T) { + body := []byte(`{"model":"gpt-5.1","input":[{"type":"reasoning","summary":[{"type":"summary_text","text":"t"}]}]}`) + sanitized, changed, err := SanitizeOpenAICrossModeFailoverReasoning(body) + require.NoError(t, err) + require.False(t, changed, "reasoning without encrypted_content must be preserved") + require.Equal(t, string(body), string(sanitized)) +} + +func TestSanitizeOpenAICrossModeFailoverReasoning_NoInputIsNoop(t *testing.T) { + body := []byte(`{"model":"gpt-5.1"}`) + sanitized, changed, err := SanitizeOpenAICrossModeFailoverReasoning(body) + require.NoError(t, err) + require.False(t, changed) + require.Equal(t, string(body), string(sanitized)) +} + +func TestSanitizeOpenAICrossModeFailoverReasoning_PreservesLargeIntegers(t *testing.T) { + body := []byte(`{"model":"gpt-5.1","input":[` + + `{"type":"reasoning","id":"rs_kiro_1","encrypted_content":"ENC"},` + + `{"type":"message","role":"user","content":"hi"}` + + `],"tools":[{"type":"function","function":{"name":"lookup","parameters":{"type":"object","properties":{"id":{"const":9007199254740993}}}}}]}`) + + sanitized, changed, err := SanitizeOpenAICrossModeFailoverReasoning(body) + require.NoError(t, err) + require.True(t, changed) + require.Contains(t, string(sanitized), `"const":9007199254740993`, + "sanitization must not round JSON integers through float64") +} + func TestTrimOpenAIEncryptedReasoningItems_ContentNullDropsBareSkeleton(t *testing.T) { reqBody := map[string]any{ "input": []any{