diff --git a/backend/internal/handler/openai_gateway_handler.go b/backend/internal/handler/openai_gateway_handler.go index 951c263dd4..6730529743 100644 --- a/backend/internal/handler/openai_gateway_handler.go +++ b/backend/internal/handler/openai_gateway_handler.go @@ -1493,6 +1493,18 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) { if channelMappingWS.Mapped { wsFirstMessage = h.gatewayService.ReplaceModelInBody(firstMessage, channelMappingWS.MappedModel) } + // 切组/会话失配防护:previous_response_id 未在当前分组命中粘连账号(StickyPreviousHit=false), + // 说明该会话链不属于本次调度到的账号,原样转发会触发上游会话链鉴权失败(“鉴权失败,请检查 API Key”)。 + // 故剥离首包里的 previous_response_id,改用首包内 input 重建上下文;带 function_call_output 的 + // 工具续链无法重建,保持原样。仅作用于首轮首包,后续 turn 的续链由 WS 转发层既有逻辑处理。 + if previousResponseID != "" && !scheduleDecision.StickyPreviousHit && + !service.ValidateFunctionCallOutputContextBytes(wsFirstMessage).HasFunctionCallOutput { + wsFirstMessage = service.RemovePreviousResponseIDFromBody(wsFirstMessage) + reqLog.Debug("openai.websocket_previous_response_id_stripped_cross_group", + zap.Int64("account_id", account.ID), + zap.String("schedule_layer", scheduleDecision.Layer), + ) + } // WebSocket 首包可能很大,hash 必须在 hooks 外算成字符串,避免 AfterTurn 闭包保活请求体。 requestPayloadHash = service.HashUsageRequestPayload(wsFirstMessage) diff --git a/backend/internal/service/channel_service.go b/backend/internal/service/channel_service.go index 4bf0147f38..18bd4d5324 100644 --- a/backend/internal/service/channel_service.go +++ b/backend/internal/service/channel_service.go @@ -572,6 +572,21 @@ func ReplaceModelInBody(body []byte, newModel string) []byte { return newBody } +// RemovePreviousResponseIDFromBody 删除请求体中的 previous_response_id,用于会话失配时改用完整 input 重建上下文。 +func RemovePreviousResponseIDFromBody(body []byte) []byte { + if len(body) == 0 { + return body + } + if !gjson.GetBytes(body, "previous_response_id").Exists() { + return body + } + newBody, err := sjson.DeleteBytes(body, "previous_response_id") + if err != nil { + return body + } + return newBody +} + // validateChannelConfig 校验渠道的定价和映射配置(冲突检测 + 区间校验 + 计费模式校验)。 // Create 和 Update 共用此函数,避免重复。 func validateChannelConfig(pricing []ChannelModelPricing, mapping map[string]map[string]string) error { diff --git a/backend/internal/service/channel_service_test.go b/backend/internal/service/channel_service_test.go index e737a21125..381b8c6c25 100644 --- a/backend/internal/service/channel_service_test.go +++ b/backend/internal/service/channel_service_test.go @@ -9,6 +9,7 @@ import ( "github.com/Wei-Shaw/sub2api/internal/pkg/pagination" "github.com/stretchr/testify/require" + "github.com/tidwall/gjson" ) // --------------------------------------------------------------------------- @@ -1921,6 +1922,33 @@ func TestReplaceModelInBody_InvalidJSON(t *testing.T) { require.Equal(t, arrayBody, result2) } +func TestRemovePreviousResponseIDFromBody(t *testing.T) { + t.Run("empty body returned as-is", func(t *testing.T) { + require.Equal(t, []byte{}, RemovePreviousResponseIDFromBody([]byte{})) + require.Nil(t, RemovePreviousResponseIDFromBody(nil)) + }) + + t.Run("no previous_response_id field is a no-op", func(t *testing.T) { + body := []byte(`{"model":"gpt-5","input":"hi"}`) + result := RemovePreviousResponseIDFromBody(body) + require.Equal(t, body, result) + }) + + t.Run("strips previous_response_id and preserves other fields", func(t *testing.T) { + body := []byte(`{"model":"gpt-5","previous_response_id":"resp_abc","input":"hi"}`) + result := RemovePreviousResponseIDFromBody(body) + require.False(t, gjson.GetBytes(result, "previous_response_id").Exists()) + require.Equal(t, "gpt-5", gjson.GetBytes(result, "model").String()) + require.Equal(t, "hi", gjson.GetBytes(result, "input").String()) + }) + + t.Run("empty-string previous_response_id is also stripped", func(t *testing.T) { + body := []byte(`{"model":"gpt-5","previous_response_id":""}`) + result := RemovePreviousResponseIDFromBody(body) + require.False(t, gjson.GetBytes(result, "previous_response_id").Exists()) + }) +} + // =========================================================================== // 7. isPlatformPricingMatch // =========================================================================== diff --git a/backend/internal/service/openai_ws_state_store.go b/backend/internal/service/openai_ws_state_store.go index b606baa1a3..d3b6891b0a 100644 --- a/backend/internal/service/openai_ws_state_store.go +++ b/backend/internal/service/openai_ws_state_store.go @@ -100,9 +100,10 @@ func (s *defaultOpenAIWSStateStore) BindResponseAccount(ctx context.Context, gro s.maybeCleanup() expiresAt := time.Now().Add(ttl) + mapKey := openAIWSResponseAccountMapKey(groupID, id) s.responseToAccountMu.Lock() - ensureBindingCapacity(s.responseToAccount, id, openAIWSStateStoreMaxEntriesPerMap) - s.responseToAccount[id] = openAIWSAccountBinding{accountID: accountID, expiresAt: expiresAt} + ensureBindingCapacity(s.responseToAccount, mapKey, openAIWSStateStoreMaxEntriesPerMap) + s.responseToAccount[mapKey] = openAIWSAccountBinding{accountID: accountID, expiresAt: expiresAt} s.responseToAccountMu.Unlock() if s.cache == nil { @@ -122,8 +123,9 @@ func (s *defaultOpenAIWSStateStore) GetResponseAccount(ctx context.Context, grou s.maybeCleanup() now := time.Now() + mapKey := openAIWSResponseAccountMapKey(groupID, id) s.responseToAccountMu.RLock() - if binding, ok := s.responseToAccount[id]; ok { + if binding, ok := s.responseToAccount[mapKey]; ok { if now.Before(binding.expiresAt) { accountID := binding.accountID s.responseToAccountMu.RUnlock() @@ -153,7 +155,7 @@ func (s *defaultOpenAIWSStateStore) DeleteResponseAccount(ctx context.Context, g return nil } s.responseToAccountMu.Lock() - delete(s.responseToAccount, id) + delete(s.responseToAccount, openAIWSResponseAccountMapKey(groupID, id)) s.responseToAccountMu.Unlock() if s.cache == nil { @@ -417,6 +419,11 @@ func openAIWSResponseAccountCacheKey(responseID string) string { return openAIWSResponseAccountCachePrefix + hex.EncodeToString(sum[:]) } +// openAIWSResponseAccountMapKey 本地热缓存按分组隔离的 key,与 Redis 层保持一致,避免跨组命中。 +func openAIWSResponseAccountMapKey(groupID int64, responseID string) string { + return fmt.Sprintf("%d:%s", groupID, responseID) +} + func normalizeOpenAIWSTTL(ttl time.Duration) time.Duration { if ttl <= 0 { return time.Hour