From fad2f215e88b974cd9073f2278b852bcbe2fdd8f Mon Sep 17 00:00:00 2001 From: Brisbanehuang Date: Sat, 1 Aug 2026 04:10:28 -0400 Subject: [PATCH] fix(profit-control): decouple gate from image intent, propagate gate via selection, restore eager sticky fallback Review fixes for the profit-control feature commit: - Image intent no longer disables the profit gate. The shared /v1/responses handler previously skipped the pricing context (and therefore the gate) whenever the platform-wide image intent predicate matched, which includes Codex's passive image_gen namespace declaration: any client could disable admission control for anthropic/gemini/antigravity groups by declaring a namespace tool in the request body. Both /v1/responses paths now always install the token pricing context; image intent only drives capability routing and image billing. Mixed token+image requests stay token-gated; only dedicated media endpoints remain out of scope. - Out-of-scope paths are now explicitly suppressed instead of implicitly ungated: Grok media (billed by media multipliers; also prevents in-flight video lookups from turning into spurious 404s), OpenAI-group count_tokens (unbilled), and Live calls (duration billed) carry a suppress marker that every install point honors, including the defensive scheduler-entry install. - The gate resolved during selection now travels back to handlers on the AccountSelectionResult. The shared gateway installed the gate only on a scheduler-local context, so handler-side post-slot terminal rechecks and post-admission sticky binding were no-ops for anthropic/gemini/ antigravity/shared-grok requests (and for composite-routed member groups on the OpenAI path). Handlers re-apply the carried gate via ContextWithSelectionProfitGate before the terminal recheck and binding; the WS acquired-selection branch gained the previously missing recheck. - Sticky binding semantics restored for ungated traffic: BindStickySessionAfterProfitAdmission falls back to the official eager bind when no gate is installed (wait paths lost their only binding point otherwise), reads the pre-existing binding at bind time only when gated (removes the unconditional per-request Redis read the feature added to the shared handlers), and the legacy engine's three selection-time binding writes are skipped under a gate so a terminally vetoed account can no longer become the new sticky target. - Responses WS connections re-freeze pricingAt and re-resolve the gate at every turn (BeforeTurn) and bill each turn with its own instant, closing the connect-at-valley/bill-at-valley window; a turn that fails the recheck closes the connection so the client reselects on reconnect. - Terminal recheck no longer swaps a DB-fresh selected account for a staler snapshot object (UpdatedAt guard), and observer counters are documented as per-evaluation. - Migrations renumbered 191/192 -> 192/193 after the passkey migration landed upstream as 191. New regressions: selection-carried gate propagation (control group proves the pre-fix no-op), image intent not disabling the shared gate, eager binding fallback without a gate (both services), gated read-failure/sentinel-miss binding semantics, legacy-engine deferred binding under a gate with eager behavior preserved ungated, turn-level pricing refresh (config re-resolution, scheduled-group precedence, suppress, mid-connection disable), and suppress-marker coverage for the request pricing context. --- backend/internal/handler/gateway_handler.go | 24 +++- .../gateway_handler_chat_completions.go | 12 +- .../handler/gateway_handler_responses.go | 21 ++-- .../internal/handler/gemini_v1beta_handler.go | 13 +- backend/internal/handler/grok_media.go | 7 +- .../internal/handler/openai_alpha_search.go | 2 +- .../handler/openai_chat_completions.go | 2 +- backend/internal/handler/openai_embeddings.go | 2 +- .../handler/openai_gateway_count_tokens.go | 3 + .../handler/openai_gateway_handler.go | 75 ++++++++++-- .../openai_profit_slot_recheck_test.go | 5 +- .../service/gateway_profit_control.go | 4 +- .../service/gateway_profit_control_v2_test.go | 89 +++++++++++++- .../internal/service/gateway_scheduling.go | 4 +- backend/internal/service/gateway_service.go | 27 ++++- .../service/openai_account_scheduler.go | 24 ++-- .../service/openai_gateway_scheduling.go | 15 +-- backend/internal/service/openai_live.go | 3 + .../internal/service/openai_profit_control.go | 114 +++++++++++++++--- .../openai_profit_control_paths_test.go | 54 +++++++++ .../openai_profit_control_pricing_test.go | 84 ++++++++++++- .../service/openai_ws_forwarder_support.go | 8 +- 22 files changed, 497 insertions(+), 95 deletions(-) diff --git a/backend/internal/handler/gateway_handler.go b/backend/internal/handler/gateway_handler.go index 9b4808d76..3d9d1e4d0 100644 --- a/backend/internal/handler/gateway_handler.go +++ b/backend/internal/handler/gateway_handler.go @@ -424,7 +424,9 @@ func (h *GatewayHandler) Messages(c *gin.Context) { // Slot acquired: no longer waiting in queue. releaseWait() } - latest, vetoed, reason := h.gatewayService.GatewayProfitControlVetoLatest(c.Request.Context(), account) + // 终检与准入后绑定使用选号结果携带的门（见 responses 同名注释）。 + admissionCtx := service.ContextWithSelectionProfitGate(c.Request.Context(), selection) + latest, vetoed, reason := h.gatewayService.GatewayProfitControlVetoLatest(admissionCtx, account) if vetoed { if accountReleaseFunc != nil { accountReleaseFunc() @@ -435,8 +437,12 @@ func (h *GatewayHandler) Messages(c *gin.Context) { } account = latest selection.Account = latest - if err := h.gatewayService.BindStickySessionAfterProfitAdmission(c.Request.Context(), apiKey.GroupID, sessionKey, account.ID, sessionBoundAccountID); err != nil { - reqLog.Warn("gateway.bind_sticky_session_after_profit_admission_failed", zap.Int64("account_id", account.ID), zap.Error(err)) + // 等待路径保持既有 eager 绑定（无门时 helper 直接绑定）；调度器已 + // 抢槽的直达路径无门时由选号内部绑定，这里只在门下补准入后绑定。 + if selection.ProfitGateActive() || !selection.Acquired { + if err := h.gatewayService.BindStickySessionAfterProfitAdmission(admissionCtx, apiKey.GroupID, sessionKey, account.ID); err != nil { + reqLog.Warn("gateway.bind_sticky_session_after_profit_admission_failed", zap.Int64("account_id", account.ID), zap.Error(err)) + } } // 账号槽位/等待计数需要在超时或断开时安全回收 accountReleaseFunc = wrapReleaseOnDone(c.Request.Context(), accountReleaseFunc) @@ -736,7 +742,9 @@ func (h *GatewayHandler) Messages(c *gin.Context) { // Slot acquired: no longer waiting in queue. releaseWait() } - latest, vetoed, reason := h.gatewayService.GatewayProfitControlVetoLatest(c.Request.Context(), account) + // 终检与准入后绑定使用选号结果携带的门（见 responses 同名注释）。 + admissionCtx := service.ContextWithSelectionProfitGate(c.Request.Context(), selection) + latest, vetoed, reason := h.gatewayService.GatewayProfitControlVetoLatest(admissionCtx, account) if vetoed { if accountReleaseFunc != nil { accountReleaseFunc() @@ -747,8 +755,12 @@ func (h *GatewayHandler) Messages(c *gin.Context) { } account = latest selection.Account = latest - if err := h.gatewayService.BindStickySessionAfterProfitAdmission(c.Request.Context(), currentAPIKey.GroupID, sessionKey, account.ID, sessionBoundAccountID); err != nil { - reqLog.Warn("gateway.bind_sticky_session_after_profit_admission_failed", zap.Int64("account_id", account.ID), zap.Error(err)) + // 等待路径保持既有 eager 绑定（无门时 helper 直接绑定）；调度器已 + // 抢槽的直达路径无门时由选号内部绑定，这里只在门下补准入后绑定。 + if selection.ProfitGateActive() || !selection.Acquired { + if err := h.gatewayService.BindStickySessionAfterProfitAdmission(admissionCtx, currentAPIKey.GroupID, sessionKey, account.ID); err != nil { + reqLog.Warn("gateway.bind_sticky_session_after_profit_admission_failed", zap.Int64("account_id", account.ID), zap.Error(err)) + } } // 账号槽位/等待计数需要在超时或断开时安全回收 accountReleaseFunc = wrapReleaseOnDone(c.Request.Context(), accountReleaseFunc) diff --git a/backend/internal/handler/gateway_handler_chat_completions.go b/backend/internal/handler/gateway_handler_chat_completions.go index cc504235a..294baae82 100644 --- a/backend/internal/handler/gateway_handler_chat_completions.go +++ b/backend/internal/handler/gateway_handler_chat_completions.go @@ -159,8 +159,6 @@ func (h *GatewayHandler) ChatCompletions(c *gin.Context) { if groupPlatform == service.PlatformGemini && selectionSessionHash != "" { selectionSessionHash = "gemini:" + selectionSessionHash } - sessionBoundAccountID, _ := h.gatewayService.GetCachedSessionAccountID(c.Request.Context(), apiKey.GroupID, selectionSessionHash) - // 3. Account selection + failover loop fs := NewFailoverState(h.maxAccountSwitches, false) if groupPlatform == service.PlatformGemini { @@ -226,7 +224,9 @@ func (h *GatewayHandler) ChatCompletions(c *gin.Context) { return } } - latest, vetoed, reason := h.gatewayService.GatewayProfitControlVetoLatest(c.Request.Context(), account) + // 终检与准入后绑定使用选号结果携带的门（见 responses 同名注释）。 + admissionCtx := service.ContextWithSelectionProfitGate(c.Request.Context(), selection) + latest, vetoed, reason := h.gatewayService.GatewayProfitControlVetoLatest(admissionCtx, account) if vetoed { if accountReleaseFunc != nil { accountReleaseFunc() @@ -237,8 +237,10 @@ func (h *GatewayHandler) ChatCompletions(c *gin.Context) { } account = latest selection.Account = latest - if err := h.gatewayService.BindStickySessionAfterProfitAdmission(c.Request.Context(), apiKey.GroupID, selectionSessionHash, account.ID, sessionBoundAccountID); err != nil { - reqLog.Warn("gateway.cc.bind_sticky_session_after_profit_admission_failed", zap.Int64("account_id", account.ID), zap.Error(err)) + if selection.ProfitGateActive() { + if err := h.gatewayService.BindStickySessionAfterProfitAdmission(admissionCtx, apiKey.GroupID, selectionSessionHash, account.ID); err != nil { + reqLog.Warn("gateway.cc.bind_sticky_session_after_profit_admission_failed", zap.Int64("account_id", account.ID), zap.Error(err)) + } } accountReleaseFunc = wrapReleaseOnDone(c.Request.Context(), accountReleaseFunc) diff --git a/backend/internal/handler/gateway_handler_responses.go b/backend/internal/handler/gateway_handler_responses.go index e50f20529..0102fa9f0 100644 --- a/backend/internal/handler/gateway_handler_responses.go +++ b/backend/internal/handler/gateway_handler_responses.go @@ -90,11 +90,13 @@ func (h *GatewayHandler) Responses(c *gin.Context) { setOpsRequestContext(c, reqModel, reqStream) setOpsEndpointContext(c, "", int16(service.RequestTypeFromLegacy(reqStream, false))) requestCtx := c.Request.Context() - pricingAt := time.Time{} + // 定价上下文无条件装配：/v1/responses 是 token 计费端点，声明生图工具的 + // 混合请求同样按 token 计费（外加图片部分），其 token 利润保护不因请求体 + // 里的任何工具声明（含 Codex 被动 image_gen namespace）而关闭。生图意图 + // 仅用于能力路由与图片计费；独立图片/视频端点才在利润门范围之外。 + requestCtx, pricingAt := service.WithGatewayTokenRequestPricing(requestCtx) if service.IsImageGenerationIntentForPlatform("/v1/responses", reqModel, body, openAICompatibleRequestPlatform(c.Request.Context(), apiKey)) { requestCtx = service.WithOpenAIImageGenerationIntent(requestCtx) - } else { - requestCtx, pricingAt = service.WithGatewayTokenRequestPricing(requestCtx) } c.Request = c.Request.WithContext(requestCtx) @@ -161,7 +163,6 @@ func (h *GatewayHandler) Responses(c *gin.Context) { APIKeyID: apiKey.ID, } sessionHash := h.gatewayService.GenerateSessionHash(parsedReq) - sessionBoundAccountID, _ := h.gatewayService.GetCachedSessionAccountID(requestCtx, apiKey.GroupID, sessionHash) // 3. Account selection + failover loop fs := NewFailoverState(h.maxAccountSwitches, false) @@ -225,7 +226,11 @@ func (h *GatewayHandler) Responses(c *gin.Context) { return } } - latest, vetoed, reason := h.gatewayService.GatewayProfitControlVetoLatest(requestCtx, account) + // 终检与准入后绑定必须使用选号结果携带的门：门安装在调度栈的局部 + // ctx 上（composite/fallback 还可能解析出与入口分组不同的门），直接用 + // requestCtx 会退化为空操作。 + admissionCtx := service.ContextWithSelectionProfitGate(requestCtx, selection) + latest, vetoed, reason := h.gatewayService.GatewayProfitControlVetoLatest(admissionCtx, account) if vetoed { if accountReleaseFunc != nil { accountReleaseFunc() @@ -236,8 +241,10 @@ func (h *GatewayHandler) Responses(c *gin.Context) { } account = latest selection.Account = latest - if err := h.gatewayService.BindStickySessionAfterProfitAdmission(requestCtx, apiKey.GroupID, sessionHash, account.ID, sessionBoundAccountID); err != nil { - reqLog.Warn("gateway.responses.bind_sticky_session_after_profit_admission_failed", zap.Int64("account_id", account.ID), zap.Error(err)) + if selection.ProfitGateActive() { + if err := h.gatewayService.BindStickySessionAfterProfitAdmission(admissionCtx, apiKey.GroupID, sessionHash, account.ID); err != nil { + reqLog.Warn("gateway.responses.bind_sticky_session_after_profit_admission_failed", zap.Int64("account_id", account.ID), zap.Error(err)) + } } accountReleaseFunc = wrapReleaseOnDone(c.Request.Context(), accountReleaseFunc) diff --git a/backend/internal/handler/gemini_v1beta_handler.go b/backend/internal/handler/gemini_v1beta_handler.go index 8caf86267..45136b306 100644 --- a/backend/internal/handler/gemini_v1beta_handler.go +++ b/backend/internal/handler/gemini_v1beta_handler.go @@ -355,7 +355,6 @@ func (h *GatewayHandler) GeminiV1BetaModels(c *gin.Context) { // 判断是否真的绑定了粘性会话：有 sessionKey 且已经绑定到某个账号 hasBoundSession := sessionKey != "" && sessionBoundAccountID > 0 - profitStickyAccountID := sessionBoundAccountID cleanedForUnknownBinding := false fs := NewFailoverState(h.maxAccountSwitchesGemini, hasBoundSession) @@ -470,7 +469,9 @@ func (h *GatewayHandler) GeminiV1BetaModels(c *gin.Context) { accountWaitCounted = false } } - latest, vetoed, reason := h.gatewayService.GatewayProfitControlVetoLatest(c.Request.Context(), account) + // 终检与准入后绑定使用选号结果携带的门（见 responses 同名注释）。 + admissionCtx := service.ContextWithSelectionProfitGate(c.Request.Context(), selection) + latest, vetoed, reason := h.gatewayService.GatewayProfitControlVetoLatest(admissionCtx, account) if vetoed { if accountReleaseFunc != nil { accountReleaseFunc() @@ -481,8 +482,12 @@ func (h *GatewayHandler) GeminiV1BetaModels(c *gin.Context) { } account = latest selection.Account = latest - if err := h.gatewayService.BindStickySessionAfterProfitAdmission(c.Request.Context(), apiKey.GroupID, sessionKey, account.ID, profitStickyAccountID); err != nil { - reqLog.Warn("gemini.bind_sticky_session_after_profit_admission_failed", zap.Int64("account_id", account.ID), zap.Error(err)) + // 等待路径保持既有 eager 绑定（无门时 helper 直接绑定）；调度器已抢槽 + // 的直达路径无门时由选号内部绑定，这里只在门下补准入后绑定。 + if selection.ProfitGateActive() || !selection.Acquired { + if err := h.gatewayService.BindStickySessionAfterProfitAdmission(admissionCtx, apiKey.GroupID, sessionKey, account.ID); err != nil { + reqLog.Warn("gemini.bind_sticky_session_after_profit_admission_failed", zap.Int64("account_id", account.ID), zap.Error(err)) + } } // 账号槽位/等待计数需要在超时或断开时安全回收 accountReleaseFunc = wrapReleaseOnDone(c.Request.Context(), accountReleaseFunc) diff --git a/backend/internal/handler/grok_media.go b/backend/internal/handler/grok_media.go index fb9df0228..15a93de24 100644 --- a/backend/internal/handler/grok_media.go +++ b/backend/internal/handler/grok_media.go @@ -177,7 +177,10 @@ func (h *OpenAIGatewayHandler) handleGrokMedia(c *gin.Context, endpoint service. return } } - requestCtx := c.Request.Context() + // Grok 媒体（图片/视频生成与视频查询）按媒体倍率计费，不在 token 利润门 + // 范围内：显式豁免，防止 service 层防御性装门按文本 D 误过滤媒体请求， + // 也防止已计费的在途视频任务因绑定账号被门排除而查询返回伪 404。 + requestCtx := service.WithOpenAIProfitControlSuppressed(c.Request.Context()) failedAccountIDs := make(map[int64]struct{}) sameAccountRetryCount := make(map[int64]int) var lastFailoverErr *service.UpstreamFailoverError @@ -295,7 +298,7 @@ func (h *OpenAIGatewayHandler) handleGrokMedia(c *gin.Context, endpoint service. accountReleaseFunc, slotResult := h.acquireResponsesAccountSlot(c, apiKey.GroupID, sessionHash, selection, false, &streamStarted, reqLog) if slotResult == openAISlotAcquireProfitVetoed { - // Grok 分组非 openai 平台不装利润门，此分支实际不可达；防御性排除重选。 + // 媒体路径已显式豁免利润门（suppress 标记），此分支仅防御性兜底。 failedAccountIDs[account.ID] = struct{}{} continue } diff --git a/backend/internal/handler/openai_alpha_search.go b/backend/internal/handler/openai_alpha_search.go index 186e85e54..4c11a0155 100644 --- a/backend/internal/handler/openai_alpha_search.go +++ b/backend/internal/handler/openai_alpha_search.go @@ -115,7 +115,7 @@ func (h *OpenAIGatewayHandler) AlphaSearch(c *gin.Context) { // 分组利润控制：alpha search 文本入口请求级装门并固定 pricingAt //（记录路径经 service.OpenAIPricingAtFromContext 从请求 ctx 回读）。 - asPricingCtx, _ := h.gatewayService.WithOpenAIRequestPricingContext(c.Request.Context(), apiKey.GroupID, false) + asPricingCtx, _ := h.gatewayService.WithOpenAIRequestPricingContext(c.Request.Context(), apiKey.GroupID) c.Request = c.Request.WithContext(asPricingCtx) for { diff --git a/backend/internal/handler/openai_chat_completions.go b/backend/internal/handler/openai_chat_completions.go index 8cdc577e0..0d03e034f 100644 --- a/backend/internal/handler/openai_chat_completions.go +++ b/backend/internal/handler/openai_chat_completions.go @@ -150,7 +150,7 @@ func (h *OpenAIGatewayHandler) ChatCompletions(c *gin.Context) { var oauth429FailoverState service.OpenAIOAuth429FailoverState // 分组利润控制：chat completions 文本入口请求级装门并固定 pricingAt。 - ccPricingCtx, pricingAt := h.gatewayService.WithOpenAIRequestPricingContext(c.Request.Context(), apiKey.GroupID, false) + ccPricingCtx, pricingAt := h.gatewayService.WithOpenAIRequestPricingContext(c.Request.Context(), apiKey.GroupID) c.Request = c.Request.WithContext(ccPricingCtx) for { diff --git a/backend/internal/handler/openai_embeddings.go b/backend/internal/handler/openai_embeddings.go index 4fe168b75..531f7e137 100644 --- a/backend/internal/handler/openai_embeddings.go +++ b/backend/internal/handler/openai_embeddings.go @@ -117,7 +117,7 @@ func (h *OpenAIGatewayHandler) Embeddings(c *gin.Context) { routingStart := time.Now() // 分组利润控制：embeddings 文本入口请求级装门并固定 pricingAt。 - embPricingCtx, pricingAt := h.gatewayService.WithOpenAIRequestPricingContext(c.Request.Context(), apiKey.GroupID, false) + embPricingCtx, pricingAt := h.gatewayService.WithOpenAIRequestPricingContext(c.Request.Context(), apiKey.GroupID) c.Request = c.Request.WithContext(embPricingCtx) for { diff --git a/backend/internal/handler/openai_gateway_count_tokens.go b/backend/internal/handler/openai_gateway_count_tokens.go index 2d642e6c7..ac6deda4a 100644 --- a/backend/internal/handler/openai_gateway_count_tokens.go +++ b/backend/internal/handler/openai_gateway_count_tokens.go @@ -143,6 +143,9 @@ func (h *OpenAIGatewayHandler) CountTokens(c *gin.Context) { } requestStart := time.Now() + // count_tokens 不计费：显式豁免利润门，避免高倍率账号池被门排除后连 + // token 计数都返回 no available accounts。 + c.Request = c.Request.WithContext(service.WithOpenAIProfitControlSuppressed(c.Request.Context())) sessionHash := h.gatewayService.GenerateSessionHash(c, body) currentRoutingModel := routingModel if preferredMappedModel != "" { diff --git a/backend/internal/handler/openai_gateway_handler.go b/backend/internal/handler/openai_gateway_handler.go index d7fc3c882..17ffb1b28 100644 --- a/backend/internal/handler/openai_gateway_handler.go +++ b/backend/internal/handler/openai_gateway_handler.go @@ -9,6 +9,7 @@ import ( "runtime/debug" "strconv" "strings" + "sync" "sync/atomic" "time" @@ -440,9 +441,10 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { requiredCapability := openAIResponsesRequiredCapability(imageIntent, requestPlatform) // 分组利润控制：请求级装配定价上下文——pricingAt 固定本请求的 - // D 与计费高峰因子，选号、槽位终检与全部 failover 重入共用同一门与阈值； - // 显式生图意图跳门（图片/视频不在利润门范围）。 - pricingCtx, pricingAt := h.gatewayService.WithOpenAIRequestPricingContext(c.Request.Context(), apiKey.GroupID, imageIntent) + // D 与计费高峰因子，选号、槽位终检与全部 failover 重入共用同一门与阈值。 + // 生图意图只影响能力路由与图片计费，不关门：混合 /v1/responses 请求的 + // token 计费部分仍受利润门保护，独立图片/视频端点才在门外。 + pricingCtx, pricingAt := h.gatewayService.WithOpenAIRequestPricingContext(c.Request.Context(), apiKey.GroupID) c.Request = c.Request.WithContext(pricingCtx) for { @@ -1007,7 +1009,7 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) { effectiveMappedModel := preferredMappedModel // 分组利润控制：Messages 文本入口同样请求级装门并固定 pricingAt。 - msgPricingCtx, pricingAt := h.gatewayService.WithOpenAIRequestPricingContext(c.Request.Context(), apiKey.GroupID, false) + msgPricingCtx, pricingAt := h.gatewayService.WithOpenAIRequestPricingContext(c.Request.Context(), apiKey.GroupID) c.Request = c.Request.WithContext(msgPricingCtx) for { @@ -1398,7 +1400,9 @@ func (h *OpenAIGatewayHandler) acquireResponsesAccountSlot( return nil, openAISlotAcquireFailed } - ctx := c.Request.Context() + // 终检与准入后绑定使用选号结果携带的门：composite 等跨分组调度解析出的 + // 门只存在于调度栈的局部 ctx，必须经选号结果重放到本函数的 ctx 上。 + ctx := service.ContextWithSelectionProfitGate(c.Request.Context(), selection) account := selection.Account if selection.Acquired { latest, vetoed, reason := h.gatewayService.ProfitControlVetoLatest(ctx, account) @@ -1411,6 +1415,13 @@ func (h *OpenAIGatewayHandler) acquireResponsesAccountSlot( } account = latest selection.Account = latest + // 调度器已抢槽路径无门时由选号内部完成 eager 绑定；门下选号内部 + // 推迟绑定，这里在终检通过后补准入后绑定。 + if selection.ProfitGateActive() { + if err := h.gatewayService.BindStickySessionAfterProfitAdmission(ctx, groupID, sessionHash, account.ID); err != nil { + reqLog.Warn("openai.bind_sticky_session_after_profit_admission_failed", zap.Int64("account_id", account.ID), zap.Error(err)) + } + } return wrapReleaseOnDone(ctx, selection.ReleaseFunc), openAISlotAcquireOK } if selection.WaitPlan == nil { @@ -1786,10 +1797,11 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) { requiredCapability = service.OpenAIEndpointCapabilityResponses } - // 分组利润控制：WS 桥按连接装配定价上下文并装门（选号与抢槽 - // 共用该 ctx）；连接内不重选号，每个 turn 的计费共用同一 pricingAt——长 - // 连接跨峰谷边界属已知残余风险，由安全缓冲承担。显式生图意图跳门。 - wsPricingCtx, wsPricingAt := h.gatewayService.WithOpenAIRequestPricingContext(ctx, apiKey.GroupID, imageIntent) + // 分组利润控制：WS 桥按连接装配定价上下文并装门（选号与抢槽共用该 + // ctx）。连接内不重选号，但每个 turn 开始经 BeforeTurn 重新冻结 pricingAt + // 并按最新门复核当前账号（准入与计费同源），峰前建连保活不能让后续 turn + // 继续按建连时刻的谷价计费。生图意图只影响能力路由与图片计费，不关门。 + wsPricingCtx, wsPricingAt := h.gatewayService.WithOpenAIRequestPricingContext(ctx, apiKey.GroupID) ctx = wsPricingCtx for { @@ -1837,7 +1849,24 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) { if selection.WaitPlan != nil && selection.WaitPlan.MaxConcurrency > 0 { accountMaxConcurrency = selection.WaitPlan.MaxConcurrency } + // 终检、准入后绑定与后续 turn 级复核都使用选号结果携带的门（composite + // 等跨分组调度的门只存在于调度栈局部 ctx）；准入成功后并入连接 ctx。 + admissionCtx := service.ContextWithSelectionProfitGate(ctx, selection) accountReleaseFunc := selection.ReleaseFunc + if selection.Acquired { + // 调度器已抢槽路径同样终检：选号与抢槽之间账号倍率可能刷新。 + latest, vetoed, reason := h.gatewayService.ProfitControlVetoLatest(admissionCtx, account) + if vetoed { + if accountReleaseFunc != nil { + accountReleaseFunc() + } + reqLog.Debug("openai.websocket_account_slot_profit_vetoed", zap.Int64("account_id", account.ID), zap.String("reason", reason)) + failedAccountIDs[account.ID] = struct{}{} + continue + } + account = latest + selection.Account = latest + } if !selection.Acquired { if selection.WaitPlan == nil { closeOpenAIClientWS(wsConn, coderws.StatusTryAgainLater, "account is busy, please retry later") @@ -1859,7 +1888,7 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) { } // 分组利润控制：WS 快速抢槽成功后终检，越线则释放 // 槽位、排除该账号重新选号，全池耗尽由下一轮选号关闭连接。 - latest, vetoed, reason := h.gatewayService.ProfitControlVetoLatest(ctx, account) + latest, vetoed, reason := h.gatewayService.ProfitControlVetoLatest(admissionCtx, account) if vetoed { if fastReleaseFunc != nil { fastReleaseFunc() @@ -1872,6 +1901,8 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) { selection.Account = latest accountReleaseFunc = fastReleaseFunc } + // 准入完成：门并入连接 ctx，turn 级复核与 failover 重选共用。 + ctx = admissionCtx currentAccountRelease = wrapReleaseOnDone(ctx, accountReleaseFunc) if err := h.gatewayService.BindStickySessionAfterProfitAdmission(ctx, apiKey.GroupID, sessionHash, account.ID); err != nil { reqLog.Warn("openai.websocket_bind_sticky_session_after_profit_admission_failed", zap.Int64("account_id", account.ID), zap.Error(err)) @@ -1912,6 +1943,11 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) { // turn-tagged slot preserves the exact mapping used for the in-flight request. var turnChannelMapping atomic.Pointer[openAIWSTurnChannelMappingSnapshot] turnChannelMapping.Store(&openAIWSTurnChannelMappingSnapshot{turn: 1, mapping: channelMappingWS}) + // turn 级定价：BeforeTurn 重新冻结 pricingAt 并按最新门复核当前账号， + // AfterTurn 的计费读取所属 turn 的时刻；turn 在连接内串行推进，互斥锁 + // 只为跨用量提交 goroutine 的读取安全。 + var turnPricingMu sync.Mutex + turnPricingAt := wsPricingAt hooks := &service.OpenAIWSIngressHooks{ InitialRequestModel: reqModel, MaxReasoningEffort: maxReasoningEffort, @@ -1957,6 +1993,20 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) { if cyberBlockedThisConn { return service.NewOpenAIWSClientCloseError(coderws.StatusPolicyViolation, cyberSessionBlockedClientMsg, nil) } + // 长连接跨峰谷/倍率刷新防护：每个 turn 按当前时刻重装门并复核 + // 当前账号，越线即要求客户端重连重选（连接绑定单一上游账号， + // 无法中途换号）。本 turn 的准入与计费共用同一 pricingAt。 + turnCtx, turnAt := h.gatewayService.WithOpenAITurnPricingContext(ctx, apiKey.GroupID) + if _, vetoed, reason := h.gatewayService.ProfitControlVetoLatest(turnCtx, account); vetoed { + reqLog.Info("openai.websocket_turn_profit_vetoed", + zap.Int("turn", turn), + zap.Int64("account_id", account.ID), + zap.String("reason", reason)) + return service.NewOpenAIWSClientCloseError(coderws.StatusTryAgainLater, "account is no longer eligible for this connection, please reconnect", nil) + } + turnPricingMu.Lock() + turnPricingAt = turnAt + turnPricingMu.Unlock() if turn == 1 { return nil } @@ -2055,6 +2105,9 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) { upstreamEndpoint := resolveOpenAIUpstreamEndpoint(c, account, result) quotaPlatform := service.QuotaPlatform(c.Request.Context(), apiKey) sessionID := service.ExtractClientSessionID(c) + turnPricingMu.Lock() + turnRecordPricingAt := turnPricingAt + turnPricingMu.Unlock() cyberBlocked := service.GetOpsCyberPolicy(c) != nil h.submitOpenAIUsageRecordTask(ctx, result, func(taskCtx context.Context) { if err := h.gatewayService.RecordUsage(taskCtx, &service.OpenAIRecordUsageInput{ @@ -2072,7 +2125,7 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) { QuotaPlatform: quotaPlatform, SessionID: sessionID, ChannelUsageFields: turnUsageFields, - PricingAt: wsPricingAt, + PricingAt: turnRecordPricingAt, CyberBlocked: cyberBlocked, }); err != nil { reqLog.Error("openai.websocket_record_usage_failed", diff --git a/backend/internal/handler/openai_profit_slot_recheck_test.go b/backend/internal/handler/openai_profit_slot_recheck_test.go index 224631cf0..b40fbc4e2 100644 --- a/backend/internal/handler/openai_profit_slot_recheck_test.go +++ b/backend/internal/handler/openai_profit_slot_recheck_test.go @@ -70,7 +70,10 @@ func profitSlotTestContext(t *testing.T, gw *service.OpenAIGatewayService, group ProfitMinMargin: 0.5, } base := context.WithValue(context.Background(), ctxkey.Group, group) - ctx, pricingAt := gw.WithOpenAIRequestPricingContext(base, &groupID, suppress) + if suppress { + base = service.WithOpenAIProfitControlSuppressed(base) + } + ctx, pricingAt := gw.WithOpenAIRequestPricingContext(base, &groupID) require.False(t, pricingAt.IsZero()) return ctx } diff --git a/backend/internal/service/gateway_profit_control.go b/backend/internal/service/gateway_profit_control.go index 20e0e4d91..c2888a1b4 100644 --- a/backend/internal/service/gateway_profit_control.go +++ b/backend/internal/service/gateway_profit_control.go @@ -94,7 +94,9 @@ func profitControlVetoLatest(ctx context.Context, selected *Account, snapshot *S if err != nil || refreshed == nil { slog.Warn("profit_control_account_refresh_failed", "group_id", gate.groupID, "platform", gate.platform, "account_id", selected.ID, "error", err) openAIProfitControlObserverInstance.recordRefreshFailure(gate.groupID, gate.platform, gate.threshold) - } else { + } else if !refreshed.UpdatedAt.Before(selected.UpdatedAt) { + // 选号路径可能已做过 DB recheck，selected 比缓存快照更新鲜；只有 + // 快照不落后时才替换，避免终检把新鲜账号换回较旧的缓存对象。 latest = refreshed } } diff --git a/backend/internal/service/gateway_profit_control_v2_test.go b/backend/internal/service/gateway_profit_control_v2_test.go index 075eeb17b..312ade0fd 100644 --- a/backend/internal/service/gateway_profit_control_v2_test.go +++ b/backend/internal/service/gateway_profit_control_v2_test.go @@ -275,7 +275,6 @@ func TestGatewayProfitControlStickyVetoKeepsBindingUntilRateRecovers(t *testing. &group.ID, "sticky-profit", cheap.ID, - expensive.ID, )) require.Equal(t, expensive.ID, cache.sessionBindings["sticky-profit"], "终检通过的 fallback 账号也不得覆盖旧绑定") require.Zero(t, cache.deletedSessions["sticky-profit"]) @@ -383,3 +382,91 @@ func TestGatewayProfitControlTerminalRefreshFailureFallsBackToSelectedObject(t * require.False(t, vetoed) require.Empty(t, reason) } + +// 选号结果携带门：门安装在调度栈局部 ctx 上，handler 必须经 +// ContextWithSelectionProfitGate 重放后终检与准入后绑定才可见（评审修复回归）。 +func TestGatewayProfitControlSelectionCarriesGateToHandlerContext(t *testing.T) { + group := gatewayProfitTestGroup(1, PlatformAnthropic) + svc := &GatewayService{} + expensive := gatewayProfitTestAccount(161, PlatformAnthropic, 0.9, group.ID) + + gateCtx := svc.withGatewayProfitControlGate(gatewayProfitTestContext(group), &group.ID) + selection, err := svc.newSelectionResult(gateCtx, &expensive, true, nil, nil) + require.NoError(t, err) + require.True(t, selection.ProfitGateActive(), "选号结果必须携带调度栈内生效的门") + + // 修复前的缺陷形态：handler 原始 ctx 不含门，终检退化为空操作。 + _, vetoed, _ := svc.GatewayProfitControlVetoLatest(context.Background(), &expensive) + require.False(t, vetoed, "对照组：不重放门时终检确实看不到门") + + handlerCtx := ContextWithSelectionProfitGate(context.Background(), selection) + latest, vetoed, reason := svc.GatewayProfitControlVetoLatest(handlerCtx, &expensive) + require.True(t, vetoed, "重放门后终检必须真实生效") + require.Equal(t, openAIProfitFilterReasonThreshold, reason) + require.NotNil(t, latest) + + // 无门选号不携带门，重放为无操作。 + plain, err := svc.newSelectionResult(context.Background(), &expensive, true, nil, nil) + require.NoError(t, err) + require.False(t, plain.ProfitGateActive()) + require.Equal(t, context.Background(), ContextWithSelectionProfitGate(context.Background(), plain)) +} + +// 生图意图不关门（H1/H2 回归锚点）：/v1/responses 混合请求即使带生图声明， +// token 定价上下文照常装配，共享门照常安装并否决越线账号。 +func TestGatewayProfitControlImageIntentDoesNotDisableGate(t *testing.T) { + group := gatewayProfitTestGroup(2, PlatformAnthropic) + svc := &GatewayService{} + expensive := gatewayProfitTestAccount(162, PlatformAnthropic, 0.9, group.ID) + + ctx := gatewayProfitTestContext(group) + ctx = WithOpenAIImageGenerationIntent(ctx) + gateCtx := svc.withGatewayProfitControlGate(ctx, &group.ID) + require.False(t, svc.isGatewayAccountProfitEligible(gateCtx, &expensive), + "请求体里的生图声明（含被动 image_gen namespace）不得关闭利润门") +} + +// 无门时准入后绑定回退官方 eager 语义；门下读失败保守不写（评审 M-Bind 回归）。 +func TestGatewayProfitControlAfterAdmissionBindSemantics(t *testing.T) { + groupID := int64(3) + expensiveID := int64(171) + cheapID := int64(172) + + t.Run("eager without gate", func(t *testing.T) { + cache := &mockGatewayCacheForPlatform{sessionBindings: map[string]int64{"s": expensiveID}} + svc := &GatewayService{cache: cache} + require.NoError(t, svc.BindStickySessionAfterProfitAdmission(context.Background(), &groupID, "s", cheapID)) + require.Equal(t, cheapID, cache.sessionBindings["s"], "无门时保持既有 eager 绑定行为") + }) + + t.Run("gated read failure is conservative", func(t *testing.T) { + // mock 的 miss 返回非 sentinel 错误，等价于 Redis 读失败：门下保守不写。 + cache := &mockGatewayCacheForPlatform{sessionBindings: map[string]int64{}} + svc := &GatewayService{cache: cache} + gate := &openAIProfitControlGate{groupID: groupID, platform: PlatformAnthropic, threshold: 0.5} + gateCtx := context.WithValue(context.Background(), openAIProfitControlGateCtxKey{}, gate) + require.NoError(t, svc.BindStickySessionAfterProfitAdmission(gateCtx, &groupID, "absent", cheapID)) + require.NotContains(t, cache.sessionBindings, "absent") + }) + + t.Run("gated sentinel miss binds", func(t *testing.T) { + cache := &sentinelMissGatewayCache{mockGatewayCacheForPlatform: &mockGatewayCacheForPlatform{sessionBindings: map[string]int64{}}} + svc := &GatewayService{cache: cache} + gate := &openAIProfitControlGate{groupID: groupID, platform: PlatformAnthropic, threshold: 0.5} + gateCtx := context.WithValue(context.Background(), openAIProfitControlGateCtxKey{}, gate) + require.NoError(t, svc.BindStickySessionAfterProfitAdmission(gateCtx, &groupID, "fresh", cheapID)) + require.Equal(t, cheapID, cache.sessionBindings["fresh"], "门下无既有绑定（sentinel miss）应建立粘性") + }) +} + +// sentinelMissGatewayCache 让 miss 返回与真实仓库一致的 ErrStickySessionNotFound。 +type sentinelMissGatewayCache struct { + *mockGatewayCacheForPlatform +} + +func (c *sentinelMissGatewayCache) GetSessionAccountID(ctx context.Context, groupID int64, sessionHash string) (int64, error) { + if id, ok := c.sessionBindings[sessionHash]; ok { + return id, nil + } + return 0, ErrStickySessionNotFound +} diff --git a/backend/internal/service/gateway_scheduling.go b/backend/internal/service/gateway_scheduling.go index b90bd5a7e..9660abdd6 100644 --- a/backend/internal/service/gateway_scheduling.go +++ b/backend/internal/service/gateway_scheduling.go @@ -1453,12 +1453,12 @@ func (s *GatewayService) newSelectionResult(ctx context.Context, account *Accoun if err != nil { return nil, err } - return &AccountSelectionResult{ + return attachSelectionProfitGate(ctx, &AccountSelectionResult{ Account: hydrated, Acquired: acquired, ReleaseFunc: release, WaitPlan: waitPlan, - }, nil + }), nil } // filterByMinPriority 过滤出优先级最小的账号集合 diff --git a/backend/internal/service/gateway_service.go b/backend/internal/service/gateway_service.go index 7717f7d34..979542534 100644 --- a/backend/internal/service/gateway_service.go +++ b/backend/internal/service/gateway_service.go @@ -547,6 +547,15 @@ type AccountSelectionResult struct { Acquired bool ReleaseFunc func() WaitPlan *AccountWaitPlan // nil means no wait allowed + // profitGate 携带本次选号真实生效的利润门（无门为 nil）。门安装在调度栈的 + // 局部 ctx 上，handler 必须经 ContextWithSelectionProfitGate 重放后才能在 + // 调度栈之外做抢槽后终检与准入后粘性绑定。 + profitGate *openAIProfitControlGate +} + +// ProfitGateActive 报告本次选号是否处于利润门之下。 +func (r *AccountSelectionResult) ProfitGateActive() bool { + return r != nil && r.profitGate != nil } // ClaudeUsage 表示Claude API返回的usage信息 @@ -894,15 +903,23 @@ func (s *GatewayService) bindGatewayStickySessionDuringSelection(ctx context.Con return s.BindStickySession(ctx, groupID, sessionHash, accountID) } -// BindStickySessionAfterProfitAdmission records a successful terminal -// selection for a profit-controlled request without replacing the binding -// observed at request start. A temporarily ineligible sticky account remains -// bound and automatically becomes eligible again if its account rate recovers. -func (s *GatewayService) BindStickySessionAfterProfitAdmission(ctx context.Context, groupID *int64, sessionHash string, accountID, existingAccountID int64) error { +// BindStickySessionAfterProfitAdmission records a terminally admitted +// account. Without a profit gate it preserves the pre-existing eager binding +// behavior at the handler bind points. With a gate it never replaces a +// different binding that already exists: a temporarily ineligible sticky +// account remains bound and automatically becomes eligible again if its +// account rate recovers. +func (s *GatewayService) BindStickySessionAfterProfitAdmission(ctx context.Context, groupID *int64, sessionHash string, accountID int64) error { if sessionHash == "" || accountID <= 0 || s.cache == nil { return nil } if !gatewayProfitControlGateActive(ctx) { + return s.BindStickySession(ctx, groupID, sessionHash, accountID) + } + existingAccountID, err := s.cache.GetSessionAccountID(ctx, derefGroupID(groupID), sessionHash) + if err != nil && !errors.Is(err, ErrStickySessionNotFound) { + // 读失败时无法判断既有绑定，保守跳过而不是冒着覆盖健康绑定的风险写入。 + slog.Warn("profit_control_sticky_binding_read_failed", "group_id", derefGroupID(groupID), "account_id", accountID, "error", err) return nil } if existingAccountID > 0 && existingAccountID != accountID { diff --git a/backend/internal/service/openai_account_scheduler.go b/backend/internal/service/openai_account_scheduler.go index bf5ef751e..497c183eb 100644 --- a/backend/internal/service/openai_account_scheduler.go +++ b/backend/internal/service/openai_account_scheduler.go @@ -512,11 +512,11 @@ func (s *defaultOpenAIAccountScheduler) selectBySessionHash( result, acquireErr := s.service.tryAcquireAccountSlot(ctx, accountID, account.Concurrency) if acquireErr == nil && result != nil && result.Acquired { _ = s.service.refreshStickySessionTTL(ctx, req.GroupID, sessionHash, s.service.openAIWSSessionStickyTTL()) - return &AccountSelectionResult{ + return attachSelectionProfitGate(ctx, &AccountSelectionResult{ Account: account, Acquired: true, ReleaseFunc: result.ReleaseFunc, - }, false, nil + }), false, nil } cfg := s.service.schedulingConfig() @@ -532,7 +532,7 @@ func (s *defaultOpenAIAccountScheduler) selectBySessionHash( ) return nil, true, nil } - return &AccountSelectionResult{ + return attachSelectionProfitGate(ctx, &AccountSelectionResult{ Account: account, WaitPlan: &AccountWaitPlan{ AccountID: accountID, @@ -540,7 +540,7 @@ func (s *defaultOpenAIAccountScheduler) selectBySessionHash( Timeout: cfg.StickySessionWaitTimeout, MaxWaiting: cfg.StickySessionMaxWaiting, }, - }, false, nil + }), false, nil } return nil, false, nil } @@ -1170,11 +1170,11 @@ func (s *defaultOpenAIAccountScheduler) tryAcquireOpenAISelectionOrderWithBudget if req.SessionHash != "" && !req.PreserveStickyBinding { _ = s.service.bindOpenAIStickySessionDuringSelection(ctx, req.GroupID, req.SessionHash, fresh.ID) } - return &AccountSelectionResult{ + return attachSelectionProfitGate(ctx, &AccountSelectionResult{ Account: fresh, Acquired: true, ReleaseFunc: result.ReleaseFunc, - }, compactBlocked, nil + }), compactBlocked, nil } return nil, compactBlocked, nil } @@ -1249,15 +1249,15 @@ func (s *defaultOpenAIAccountScheduler) tryFallbackToWeightedSticky( if req.SessionHash != "" && !req.PreserveStickyBinding { _ = s.service.bindOpenAIStickySessionDuringSelection(ctx, req.GroupID, req.SessionHash, account.ID) } - return &AccountSelectionResult{ + return attachSelectionProfitGate(ctx, &AccountSelectionResult{ Account: account, Acquired: true, ReleaseFunc: result.ReleaseFunc, - }, nil + }), nil } if s.service.concurrencyService != nil { cfg := s.service.schedulingConfig() - return &AccountSelectionResult{ + return attachSelectionProfitGate(ctx, &AccountSelectionResult{ Account: account, WaitPlan: &AccountWaitPlan{ AccountID: account.ID, @@ -1265,7 +1265,7 @@ func (s *defaultOpenAIAccountScheduler) tryFallbackToWeightedSticky( Timeout: cfg.StickySessionWaitTimeout, MaxWaiting: cfg.StickySessionMaxWaiting, }, - }, nil + }), nil } } return nil, nil @@ -1622,7 +1622,7 @@ func (s *defaultOpenAIAccountScheduler) finishLoadBalanceSelectionFallback( compactBlocked = true continue } - return &AccountSelectionResult{ + return attachSelectionProfitGate(ctx, &AccountSelectionResult{ Account: fresh, WaitPlan: &AccountWaitPlan{ AccountID: fresh.ID, @@ -1630,7 +1630,7 @@ func (s *defaultOpenAIAccountScheduler) finishLoadBalanceSelectionFallback( Timeout: cfg.FallbackWaitTimeout, MaxWaiting: cfg.FallbackMaxWaiting, }, - }, candidateCount, topK, loadSkew, nil + }), candidateCount, topK, loadSkew, nil } } diff --git a/backend/internal/service/openai_gateway_scheduling.go b/backend/internal/service/openai_gateway_scheduling.go index ca9ce44b0..c70a03802 100644 --- a/backend/internal/service/openai_gateway_scheduling.go +++ b/backend/internal/service/openai_gateway_scheduling.go @@ -670,9 +670,10 @@ func (s *OpenAIGatewayService) selectAccountForModelWithExclusions(ctx context.C return nil, err } - // 4. 设置粘性会话绑定 - // Set sticky session binding - if sessionHash != "" { + // 4. 设置粘性会话绑定（利润门下推迟到 handler 终检通过后再绑定， + // 终检否决的账号不得成为新的粘性目标；无门保持既有 eager 绑定与 TTL） + // Set sticky session binding (deferred until terminal admission under a profit gate) + if sessionHash != "" && !gatewayProfitControlGateActive(ctx) { _ = s.setStickySessionAccountID(ctx, groupID, sessionHash, selected.ID, openaiStickySessionTTL) } @@ -1103,7 +1104,7 @@ func (s *OpenAIGatewayService) selectAccountWithLoadAwareness(ctx context.Contex if selectErr != nil { return nil, true, selectErr } - if sessionHash != "" { + if sessionHash != "" && !gatewayProfitControlGateActive(ctx) { _ = s.setStickySessionAccountID(ctx, groupID, sessionHash, fresh.ID, openaiStickySessionTTL) } return selection, true, nil @@ -1142,7 +1143,7 @@ func (s *OpenAIGatewayService) selectAccountWithLoadAwareness(ctx context.Contex if selectErr != nil { return nil, selectErr } - if sessionHash != "" { + if sessionHash != "" && !gatewayProfitControlGateActive(ctx) { _ = s.setStickySessionAccountID(ctx, groupID, sessionHash, fresh.ID, openaiStickySessionTTL) } return selection, nil @@ -1354,12 +1355,12 @@ func (s *OpenAIGatewayService) newSelectionResult(ctx context.Context, account * if err != nil { return nil, err } - return &AccountSelectionResult{ + return attachSelectionProfitGate(ctx, &AccountSelectionResult{ Account: hydrated, Acquired: acquired, ReleaseFunc: release, WaitPlan: waitPlan, - }, nil + }), nil } func (s *OpenAIGatewayService) newAcquiredSelectionResult(ctx context.Context, account *Account, release func()) (*AccountSelectionResult, error) { diff --git a/backend/internal/service/openai_live.go b/backend/internal/service/openai_live.go index 6cce50573..0d9f10554 100644 --- a/backend/internal/service/openai_live.go +++ b/backend/internal/service/openai_live.go @@ -144,6 +144,9 @@ func (s *OpenAIGatewayService) CreateLiveCall( } excluded := make(map[int64]struct{}) + // Live 按通话时长计费，不属于 token 利润门的语义范围：显式豁免，避免 + // 防御性装门按文本 D 过滤 Live 账号池且门与计费时刻不同源。 + ctx = WithOpenAIProfitControlSuppressed(ctx) var lastErr error for attempt := 0; attempt <= 3; attempt++ { selection, _, selectErr := s.SelectAccountWithSchedulerForCapability( diff --git a/backend/internal/service/openai_profit_control.go b/backend/internal/service/openai_profit_control.go index 2e4229546..fa76dfa82 100644 --- a/backend/internal/service/openai_profit_control.go +++ b/backend/internal/service/openai_profit_control.go @@ -22,14 +22,23 @@ package service // // 装门点（gate 随 ctx 传播，请求内复用，覆盖等待/重试/failover/抢槽后终检）： // - handler 各文本入口经 WithOpenAIRequestPricingContext 在请求开始统一装门并 -// 固定 pricingAt；显式生图意图（responses image_generation）以抑制标记跳门， -// 图片/视频保持既有调度不装门。 +// 固定 pricingAt。生图意图只用于能力路由与图片计费，不决定是否装门：混合 +// /v1/responses 请求（含声明了生图工具的请求）的 token 计费部分仍受利润门 +// 保护，请求体内的任何工具声明都不能作为关门开关。 +// - 门范围之外的路径显式携带 WithOpenAIProfitControlSuppressed 标记：独立 +// 图片/视频端点与 Grok 媒体（媒体计费另有倍率来源）、count_tokens（不计费）、 +// live（按时长计费）。所有装门点（含防御性装门）都尊重该标记。 // - selectAccountWithScheduler 顶部：唯一文本调度入口的防御性装门（ctx 已有 -// 同分组门则复用，保证 failover 阈值稳定）；requiredImageCapability != "" 或 -// requiredCapability == OpenAIEndpointCapabilityResponses（该值当前仅显式 -// 生图意图设置）不装门。 +// 同分组门则复用，保证 failover 阈值稳定）；requiredImageCapability != "" +// 的专用图片调度不装门。 // - 公开 SelectAccountWithLoadAwareness / SelectAccountByPreviousResponseID： // 防御性装门，保证不经唯一入口的调用方无法绕过。 +// - 选号结果通过 AccountSelectionResult 携带真实生效的门（composite/fallback +// 路由可能解析出与入口分组不同的门），handler 经 +// ContextWithSelectionProfitGate 重放后再做抢槽后终检与准入后粘性绑定。 +// - 长连接（Responses WS）在每个 turn 开始经 WithOpenAITurnPricingContext +// 重新冻结 pricingAt 并重装门：turn 的准入与计费同源，跨峰谷边界不再共用 +// 建连时刻的 D。 // // 否决点（消费 gate，任何 fallback 都无法把已排除账号重新放回）： // - defaultOpenAIAccountScheduler.isAccountRequestCompatibleReason：候选池 @@ -78,9 +87,9 @@ const ( type openAIProfitControlGateCtxKey struct{} -// openAIProfitControlSuppressCtxKey 标记本请求显式跳过利润门（生图意图等 -// 利润门范围外流量）。所有装门点看到该标记后一律不装门，防止 service 层防御性 -// 装门把边界外流量重新拉回利润过滤。 +// openAIProfitControlSuppressCtxKey 标记本请求显式跳过利润门（独立图片/视频 +// 端点、Grok 媒体、count_tokens、live 等利润门范围外流量）。所有装门点看到该 +// 标记后一律不装门，防止 service 层防御性装门把边界外流量重新拉回利润过滤。 type openAIProfitControlSuppressCtxKey struct{} // openAIPricingAtCtxKey 携带请求级定价时刻 pricingAt：门的 D 与 RecordUsage @@ -103,17 +112,49 @@ type openAIProfitControlGate struct { // WithOpenAIRequestPricingContext 在请求开始处装配请求级定价上下文：固定 // pricingAt（返回给调用方，供 RecordUsage 入参共用同一时刻），并按分组安装 -// 利润门；suppressProfitGate 为 true（显式生图意图）时只固定 pricingAt、写入 -// 抑制标记，不装门。handler 各文本入口应在选号循环前调用一次。 -func (s *OpenAIGatewayService) WithOpenAIRequestPricingContext(ctx context.Context, groupID *int64, suppressProfitGate bool) (context.Context, time.Time) { +// 利润门。ctx 携带 WithOpenAIProfitControlSuppressed 标记（门范围外流量）时 +// 只固定 pricingAt、不装门。handler 各文本入口应在选号循环前调用一次。 +func (s *OpenAIGatewayService) WithOpenAIRequestPricingContext(ctx context.Context, groupID *int64) (context.Context, time.Time) { pricingAt := timezone.Now() ctx = context.WithValue(ctx, openAIPricingAtCtxKey{}, pricingAt) - if suppressProfitGate { - return context.WithValue(ctx, openAIProfitControlSuppressCtxKey{}, struct{}{}), pricingAt - } return s.withOpenAIProfitControlGate(ctx, groupID), pricingAt } +// WithOpenAIProfitControlSuppressed 标记本请求在利润门范围之外（独立图片/视频 +// 端点、Grok 媒体、count_tokens、live）。所有装门点（含 service 层防御性装门） +// 都尊重该标记；它只关闭利润准入过滤，不影响定价上下文与计费。 +func WithOpenAIProfitControlSuppressed(ctx context.Context) context.Context { + return context.WithValue(ctx, openAIProfitControlSuppressCtxKey{}, struct{}{}) +} + +// WithOpenAITurnPricingContext 在长连接（Responses WS）的每个 turn 开始重新 +// 冻结 pricingAt 并按当前配置重装利润门，使 turn 的准入与计费同源：峰前建连 +// 保活不再让后续 turn 继续按建连时刻的谷价定价。连接可能被调度到与入口分组 +// 不同的分组（composite 成员分组），turn 级重装以连接上已装门的调度分组为准； +// 连接从未装门时才回退入口分组。抑制标记下只刷新 pricingAt。 +func (s *OpenAIGatewayService) WithOpenAITurnPricingContext(ctx context.Context, groupID *int64) (context.Context, time.Time) { + pricingAt := timezone.Now() + ctx = context.WithValue(ctx, openAIPricingAtCtxKey{}, pricingAt) + if _, suppressed := ctx.Value(openAIProfitControlSuppressCtxKey{}).(struct{}); suppressed { + return ctx, pricingAt + } + if existing, ok := ctx.Value(openAIProfitControlGateCtxKey{}).(*openAIProfitControlGate); ok && existing != nil { + gid := existing.groupID + groupID = &gid + } + gate := s.resolveOpenAIProfitControlGate(ctx, groupID) + if gate == nil { + // 分组已关门（或配置读取失败 fail-open）：清除旧 turn 的门，后续 turn + // 按无门放行，与 HTTP 路径的开关语义一致。 + if existing, ok := ctx.Value(openAIProfitControlGateCtxKey{}).(*openAIProfitControlGate); ok && existing != nil { + return context.WithValue(ctx, openAIProfitControlGateCtxKey{}, (*openAIProfitControlGate)(nil)), pricingAt + } + return ctx, pricingAt + } + openAIProfitControlObserverInstance.recordInstall(gate.groupID, gate.platform, gate.threshold) + return context.WithValue(ctx, openAIProfitControlGateCtxKey{}, gate), pricingAt +} + // openAIPricingAtFromContext 返回请求级定价时刻；未装配（内部调用、非文本 // 入口）时 ok=false，调用方回退 timezone.Now() 保持既有行为。 func openAIPricingAtFromContext(ctx context.Context) (time.Time, bool) { @@ -215,6 +256,34 @@ func (s *OpenAIGatewayService) resolveOpenAIProfitControlGate(ctx context.Contex } } +// attachSelectionProfitGate 把调度上下文里生效的利润门记录到选号结果上。门在 +// 选号函数内部的局部 ctx 上安装（composite/fallback 路由还可能解析出与入口 +// 分组不同的门），不随返回值离开调度栈；结果携带后 handler 才能对"真实过滤了 +// 候选的那个门"做抢槽后终检与准入后绑定。 +func attachSelectionProfitGate(ctx context.Context, sel *AccountSelectionResult) *AccountSelectionResult { + if sel == nil { + return nil + } + if gate, ok := ctx.Value(openAIProfitControlGateCtxKey{}).(*openAIProfitControlGate); ok && gate != nil { + sel.profitGate = gate + } + return sel +} + +// ContextWithSelectionProfitGate 把选号时真实生效的利润门重放到 ctx 上。 +// handler 在拿到选号结果后必须用返回的 ctx 做抢槽后终检 +// （ProfitControlVetoLatest / GatewayProfitControlVetoLatest）与准入后粘性 +// 绑定，否则这两步会因为看不到调度栈内安装的门而退化为空操作。 +func ContextWithSelectionProfitGate(ctx context.Context, sel *AccountSelectionResult) context.Context { + if sel == nil || sel.profitGate == nil { + return ctx + } + if existing, ok := ctx.Value(openAIProfitControlGateCtxKey{}).(*openAIProfitControlGate); ok && existing == sel.profitGate { + return ctx + } + return context.WithValue(ctx, openAIProfitControlGateCtxKey{}, sel.profitGate) +} + // openAIProfitControlVetoReason 报告利润门是否否决该账号。ctx 中没有门 // （分组未启用利润控制或本请求跳门）或账号为 nil 时一律放行。 func openAIProfitControlVetoReason(ctx context.Context, account *Account) (bool, string) { @@ -267,13 +336,18 @@ func (s *OpenAIGatewayService) bindOpenAIStickySessionDuringSelection(ctx contex } // BindStickySessionAfterProfitAdmission records the terminally admitted -// account without overwriting a different binding that already exists. A -// temporarily ineligible account therefore remains sticky and becomes -// eligible again automatically after its rate recovers. +// account. Without a profit gate it preserves the pre-existing eager binding +// behavior at the handler bind points. With a gate it never overwrites a +// different binding that already exists, so a temporarily ineligible account +// remains sticky and becomes eligible again automatically after its rate +// recovers. func (s *OpenAIGatewayService) BindStickySessionAfterProfitAdmission(ctx context.Context, groupID *int64, sessionHash string, accountID int64) error { - if sessionHash == "" || accountID <= 0 || !gatewayProfitControlGateActive(ctx) { + if sessionHash == "" || accountID <= 0 { return nil } + if !gatewayProfitControlGateActive(ctx) { + return s.BindStickySession(ctx, groupID, sessionHash, accountID) + } existingAccountID, err := s.getStickySessionAccountID(ctx, groupID, sessionHash) if err != nil && !errors.Is(err, ErrStickySessionNotFound) { slog.Warn("profit_control_sticky_binding_read_failed", "group_id", derefGroupID(groupID), "account_id", accountID, "error", err) @@ -286,6 +360,10 @@ func (s *OpenAIGatewayService) BindStickySessionAfterProfitAdmission(ctx context } // ---- 可观测性：按分组累计计数 + 采样日志（无逐请求输出） ---- +// +// 计数按"每次准入评估"累计，而非每请求：粘性层校验与候选池过滤可能对同一账号 +// 各评估一次，failover 重入也会再次计数。计数用于确认门在真实流量上生效及否决 +// 构成，不能当作精确的请求数或账号数。 type openAIProfitControlGroupStats struct { installs atomic.Int64 diff --git a/backend/internal/service/openai_profit_control_paths_test.go b/backend/internal/service/openai_profit_control_paths_test.go index 58bb46bbc..4c02c887b 100644 --- a/backend/internal/service/openai_profit_control_paths_test.go +++ b/backend/internal/service/openai_profit_control_paths_test.go @@ -314,3 +314,57 @@ func TestProfitControl_EligibilityFunctionVetoes(t *testing.T) { // 无门时保持既有行为。 require.True(t, isOpenAICompatibleAccountEligibleForRequest(context.Background(), expensive, PlatformOpenAI, "", false, "")) } + +// legacy 引擎粘性写回（评审 M-Legacy 回归）：门下选号阶段不得直写粘性—— +// 终检否决的账号不能成为新绑定；无门保持官方 eager 绑定与原 TTL 语义。 +func TestProfitControl_LegacyEngineDefersStickyBindingUnderGate(t *testing.T) { + now := time.Now() + cheap := upstreamCostTestAccount(45, UpstreamBillingProbeStatusOK, 0.3, now.Add(-time.Minute), 30*time.Minute) + expensive := upstreamCostTestAccount(46, UpstreamBillingProbeStatusOK, 0.8, now.Add(-time.Minute), 30*time.Minute) + profitControlTestAccountWithRate(cheap, 0.3) + profitControlTestAccountWithRate(expensive, 0.8) + for _, account := range []*Account{cheap, expensive} { + account.Status = StatusActive + account.Schedulable = true + account.Concurrency = 2 + } + groupID := int64(9) + const sessionHash = "legacy-sticky" + newSvc := func(bindings map[string]int64) (*OpenAIGatewayService, *schedulerTestGatewayCache) { + cache := &schedulerTestGatewayCache{sessionBindings: bindings} + return &OpenAIGatewayService{ + accountRepo: stubOpenAIAccountRepo{accounts: []Account{*cheap, *expensive}}, + cfg: &config.Config{}, + rateLimitService: newOpenAIAdvancedSchedulerRateLimitService("false"), + concurrencyService: NewConcurrencyService(stubConcurrencyCache{}), + cache: cache, + }, cache + } + + t.Run("gated selection defers binding to terminal admission", func(t *testing.T) { + svc, cache := newSvc(map[string]int64{}) + ctx := profitControlTestCtx(profitControlTestGroup(groupID, 0.5, 0)) + selection, _, err := svc.SelectAccountWithScheduler(ctx, &groupID, "", sessionHash, "gpt-test", nil, OpenAIUpstreamTransportAny, false) + require.NoError(t, err) + require.NotNil(t, selection) + require.Equal(t, cheap.ID, selection.Account.ID) + require.Empty(t, cache.sessionBindings, "门下 legacy 选号阶段不得直写粘性绑定") + require.True(t, selection.ProfitGateActive(), "legacy 选号结果同样携带门") + if selection.ReleaseFunc != nil { + selection.ReleaseFunc() + } + }) + + t.Run("ungated selection keeps official eager binding", func(t *testing.T) { + svc, cache := newSvc(map[string]int64{}) + group := profitControlTestGroup(groupID, 0.5, 0) + group.ProfitControlEnabled = false + selection, _, err := svc.SelectAccountWithScheduler(profitControlTestCtx(group), &groupID, "", sessionHash, "gpt-test", nil, OpenAIUpstreamTransportAny, false) + require.NoError(t, err) + require.NotNil(t, selection) + require.NotEmpty(t, cache.sessionBindings, "无门时 legacy 选号保持官方 eager 绑定") + if selection.ReleaseFunc != nil { + selection.ReleaseFunc() + } + }) +} diff --git a/backend/internal/service/openai_profit_control_pricing_test.go b/backend/internal/service/openai_profit_control_pricing_test.go index f155a1f5c..03077d2a8 100644 --- a/backend/internal/service/openai_profit_control_pricing_test.go +++ b/backend/internal/service/openai_profit_control_pricing_test.go @@ -14,8 +14,8 @@ import ( "github.com/stretchr/testify/require" ) -// WithOpenAIRequestPricingContext：装门 + 固定 pricingAt；抑制标记跳门且防御性 -// 装门无法把门加回来。 +// WithOpenAIRequestPricingContext：装门 + 固定 pricingAt；显式抑制标记 +// （媒体/count_tokens/live 等门范围外路径）跳门且防御性装门无法把门加回来。 func TestProfitControl_RequestPricingContext(t *testing.T) { svc := &OpenAIGatewayService{} groupID := int64(61) @@ -25,7 +25,7 @@ func TestProfitControl_RequestPricingContext(t *testing.T) { t.Run("installs gate and pricing instant", func(t *testing.T) { base := profitControlTestCtx(profitControlTestGroup(groupID, 0.5, 0)) - ctx, pricingAt := svc.WithOpenAIRequestPricingContext(base, &groupID, false) + ctx, pricingAt := svc.WithOpenAIRequestPricingContext(base, &groupID) require.False(t, pricingAt.IsZero()) require.Equal(t, pricingAt, OpenAIPricingAtFromContext(ctx)) vetoed, reason := OpenAIProfitControlVeto(ctx, expensive) @@ -33,9 +33,9 @@ func TestProfitControl_RequestPricingContext(t *testing.T) { require.Equal(t, openAIProfitFilterReasonThreshold, reason) }) - t.Run("image intent suppresses gate everywhere", func(t *testing.T) { - base := profitControlTestCtx(profitControlTestGroup(groupID, 0.5, 0)) - ctx, pricingAt := svc.WithOpenAIRequestPricingContext(base, &groupID, true) + t.Run("suppress marker skips gate everywhere", func(t *testing.T) { + base := WithOpenAIProfitControlSuppressed(profitControlTestCtx(profitControlTestGroup(groupID, 0.5, 0))) + ctx, pricingAt := svc.WithOpenAIRequestPricingContext(base, &groupID) require.False(t, pricingAt.IsZero(), "跳门时 pricingAt 仍需固定供计费共用") vetoed, _ := OpenAIProfitControlVeto(ctx, expensive) require.False(t, vetoed) @@ -199,3 +199,75 @@ func TestOpenAIProfitControlStickyBindingOccursOnlyAfterTerminalAdmission(t *tes require.NoError(t, svc.BindStickySessionAfterProfitAdmission(ctx, &groupID, sessionHash, cheapID)) require.Equal(t, cheapID, cache.sessionBindings[cacheKey], "无既有绑定时应在终检通过后建立粘性") } + +// WithOpenAITurnPricingContext：长连接 turn 边界重新冻结 pricingAt 并按当前 +// 配置重装门（区别于请求级同门复用）；已装门时以门所属调度分组为准。 +func TestProfitControl_TurnPricingContext(t *testing.T) { + svc := &OpenAIGatewayService{} + groupID := int64(63) + expensive := upstreamCostTestAccount(3, UpstreamBillingProbeStatusOK, 0.8, time.Now().Add(-time.Minute), 30*time.Minute) + profitControlTestAccountWithRate(expensive, 0.8) + + t.Run("refreshes instant and re-resolves gate config", func(t *testing.T) { + group := profitControlTestGroup(groupID, 0.5, 0) + base := profitControlTestCtx(group) + connCtx, connAt := svc.WithOpenAIRequestPricingContext(base, &groupID) + vetoed, _ := OpenAIProfitControlVeto(connCtx, expensive) + require.True(t, vetoed) + + // 连接中途运营者放宽 margin：turn 级重装必须生效（请求级复用不生效）。 + group.ProfitMinMargin = 0.1 + turnCtx, turnAt := svc.WithOpenAITurnPricingContext(connCtx, &groupID) + require.False(t, turnAt.Before(connAt)) + require.Equal(t, turnAt, OpenAIPricingAtFromContext(turnCtx)) + vetoed, _ = OpenAIProfitControlVeto(turnCtx, expensive) + require.False(t, vetoed, "turn 级重装应采用最新分组配置") + }) + + t.Run("keeps scheduled group of the existing gate", func(t *testing.T) { + scheduledGroupID := int64(64) + scheduled := profitControlTestGroup(scheduledGroupID, 0.5, 0) + connCtx, _ := svc.WithOpenAIRequestPricingContext(profitControlTestCtx(scheduled), &scheduledGroupID) + // 入口分组与调度分组不同（composite 成员分组场景）：turn 重装取门的分组。 + entryGroupID := int64(65) + turnCtx, _ := svc.WithOpenAITurnPricingContext(connCtx, &entryGroupID) + gate, ok := turnCtx.Value(openAIProfitControlGateCtxKey{}).(*openAIProfitControlGate) + require.True(t, ok) + require.NotNil(t, gate) + require.Equal(t, scheduledGroupID, gate.groupID) + }) + + t.Run("suppress marker only refreshes instant", func(t *testing.T) { + base := WithOpenAIProfitControlSuppressed(profitControlTestCtx(profitControlTestGroup(groupID, 0.5, 0))) + turnCtx, turnAt := svc.WithOpenAITurnPricingContext(base, &groupID) + require.False(t, turnAt.IsZero()) + vetoed, _ := OpenAIProfitControlVeto(turnCtx, expensive) + require.False(t, vetoed) + }) + + t.Run("clears gate when group disables profit control mid-connection", func(t *testing.T) { + group := profitControlTestGroup(groupID, 0.5, 0) + connCtx, _ := svc.WithOpenAIRequestPricingContext(profitControlTestCtx(group), &groupID) + group.ProfitControlEnabled = false + turnCtx, _ := svc.WithOpenAITurnPricingContext(connCtx, &groupID) + vetoed, _ := OpenAIProfitControlVeto(turnCtx, expensive) + require.False(t, vetoed, "关门后 turn 级复核应放行") + }) +} + +// 无门时准入后绑定回退官方 eager 语义：等待/抢槽路径不得因利润控制关闭而 +// 失去粘性绑定（评审 M-Bind 回归锚点）。 +func TestOpenAIProfitControlAfterAdmissionBindEagerWithoutGate(t *testing.T) { + groupID := int64(82) + expensiveID := int64(903) + cheapID := int64(904) + const sessionHash = "no-gate-sticky" + const cacheKey = "openai:" + sessionHash + cache := &schedulerTestGatewayCache{ + sessionBindings: map[string]int64{cacheKey: expensiveID}, + } + svc := &OpenAIGatewayService{cache: cache} + + require.NoError(t, svc.BindStickySessionAfterProfitAdmission(context.Background(), &groupID, sessionHash, cheapID)) + require.Equal(t, cheapID, cache.sessionBindings[cacheKey], "无门时保持既有 eager 绑定行为") +} diff --git a/backend/internal/service/openai_ws_forwarder_support.go b/backend/internal/service/openai_ws_forwarder_support.go index 681d5d82c..6c6d7c9cf 100644 --- a/backend/internal/service/openai_ws_forwarder_support.go +++ b/backend/internal/service/openai_ws_forwarder_support.go @@ -415,16 +415,16 @@ func (s *OpenAIGatewayService) selectAccountByPreviousResponseIDForCapability( responseID, store.BindResponseAccount(ctx, derefGroupID(groupID), responseID, accountID, s.openAIWSResponseStickyTTL()), ) - return &AccountSelectionResult{ + return attachSelectionProfitGate(ctx, &AccountSelectionResult{ Account: account, Acquired: true, ReleaseFunc: result.ReleaseFunc, - }, nil + }), nil } cfg := s.schedulingConfig() if s.concurrencyService != nil { - return &AccountSelectionResult{ + return attachSelectionProfitGate(ctx, &AccountSelectionResult{ Account: account, WaitPlan: &AccountWaitPlan{ AccountID: accountID, @@ -432,7 +432,7 @@ func (s *OpenAIGatewayService) selectAccountByPreviousResponseIDForCapability( Timeout: cfg.StickySessionWaitTimeout, MaxWaiting: cfg.StickySessionMaxWaiting, }, - }, nil + }), nil } return nil, nil }