Merge pull request #4890 from SemonCat/fix/openai-cross-mode-reasoning-failover

fix(openai): strip foreign reasoning on account failover
This commit is contained in:
Wesley Liddick
2026-07-27 11:46:49 +08:00
committed by GitHub
5 changed files with 375 additions and 1 deletions
@@ -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 {
@@ -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
}
@@ -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")
}
@@ -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)
}
@@ -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{