From 8d88b6666803fdf225f26086196a2a3c71a38665 Mon Sep 17 00:00:00 2001 From: xlplbo Date: Wed, 16 Sep 2026 14:42:51 +0800 Subject: [PATCH 1/2] =?UTF-8?q?fix(openai):=20=E8=A1=A5=E9=BD=90=E9=9D=9E?= =?UTF-8?q?=E9=AB=98=E7=BA=A7=E8=B0=83=E5=BA=A6=E9=80=89=E5=8F=B7=E5=86=B3?= =?UTF-8?q?=E7=AD=96=E6=A0=87=E7=AD=BE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/internal/service/gateway_service.go | 2 + .../service/openai_account_scheduler.go | 16 ++++ .../service/openai_gateway_scheduling.go | 43 ++++++--- .../openai_legacy_scheduler_decision_test.go | 87 +++++++++++++++++++ 4 files changed, 136 insertions(+), 12 deletions(-) create mode 100644 backend/internal/service/openai_legacy_scheduler_decision_test.go diff --git a/backend/internal/service/gateway_service.go b/backend/internal/service/gateway_service.go index 97e35fdb1b..0b6094fcf8 100644 --- a/backend/internal/service/gateway_service.go +++ b/backend/internal/service/gateway_service.go @@ -580,6 +580,8 @@ type AccountSelectionResult struct { Acquired bool ReleaseFunc func() WaitPlan *AccountWaitPlan // nil means no wait allowed + // stickySessionHit 标记账号来自会话粘性绑定命中，供非高级调度路径回填决策标签。 + stickySessionHit bool // profitGate 携带本次选号真实生效的利润门（无门为 nil）。门安装在调度栈的 // 局部 ctx 上，handler 必须经 ContextWithSelectionProfitGate 重放后才能在 // 调度栈之外做抢槽后终检与准入后粘性绑定。 diff --git a/backend/internal/service/openai_account_scheduler.go b/backend/internal/service/openai_account_scheduler.go index 28e30dfadf..0085182333 100644 --- a/backend/internal/service/openai_account_scheduler.go +++ b/backend/internal/service/openai_account_scheduler.go @@ -2221,6 +2221,20 @@ func (s *OpenAIGatewayService) loadOpenAIGroupRequiresPrivacySet(ctx context.Con return group != nil && group.RequirePrivacySet } +// applyLegacySelectionDecision 把非高级调度路径的选号结果回填到决策：记录选中账号， +// 命中会话粘性时标记 session_hash 层，否则保持 load_balance。 +func applyLegacySelectionDecision(decision *OpenAIAccountScheduleDecision, selection *AccountSelectionResult) { + if decision == nil || selection == nil || selection.Account == nil { + return + } + decision.SelectedAccountID = selection.Account.ID + decision.SelectedAccountType = selection.Account.Type + if selection.stickySessionHit { + decision.Layer = openAIAccountScheduleLayerSessionSticky + decision.StickySessionHit = true + } +} + func (s *OpenAIGatewayService) selectAccountWithSchedulerOnce( ctx context.Context, groupID *int64, @@ -2302,6 +2316,7 @@ func (s *OpenAIGatewayService) selectAccountWithSchedulerOnce( return selection, decision, nil } if accountSupportsOpenAICapabilities(selection.Account, requiredCapability, requiredImageCapability) { + applyLegacySelectionDecision(&decision, selection) return selection, decision, nil } if selection.ReleaseFunc != nil { @@ -2328,6 +2343,7 @@ func (s *OpenAIGatewayService) selectAccountWithSchedulerOnce( } if s.isOpenAIAccountTransportCompatible(selection.Account, requiredTransport) && accountSupportsOpenAICapabilities(selection.Account, requiredCapability, requiredImageCapability) { + applyLegacySelectionDecision(&decision, selection) return selection, decision, nil } if selection.ReleaseFunc != nil { diff --git a/backend/internal/service/openai_gateway_scheduling.go b/backend/internal/service/openai_gateway_scheduling.go index 859e782291..2f9a750caa 100644 --- a/backend/internal/service/openai_gateway_scheduling.go +++ b/backend/internal/service/openai_gateway_scheduling.go @@ -882,25 +882,32 @@ func resolveOpenAIErrorSchedulingModel(billingModel, upstreamModel string) strin } func (s *OpenAIGatewayService) selectAccountForModelWithExclusions(ctx context.Context, groupID *int64, platform string, sessionHash string, requestedModel string, excludedIDs map[int64]struct{}, requireCompact bool, stickyAccountID int64, requiredCapability OpenAIEndpointCapability, preferLowUpstreamRate bool) (*Account, error) { + account, _, err := s.selectAccountForModelWithExclusionsStickyHit(ctx, groupID, platform, sessionHash, requestedModel, excludedIDs, requireCompact, stickyAccountID, requiredCapability, preferLowUpstreamRate) + return account, err +} + +// selectAccountForModelWithExclusionsStickyHit 与 selectAccountForModelWithExclusions 相同， +// 另返回账号是否来自粘性会话命中。 +func (s *OpenAIGatewayService) selectAccountForModelWithExclusionsStickyHit(ctx context.Context, groupID *int64, platform string, sessionHash string, requestedModel string, excludedIDs map[int64]struct{}, requireCompact bool, stickyAccountID int64, requiredCapability OpenAIEndpointCapability, preferLowUpstreamRate bool) (*Account, bool, error) { platform = NormalizeOpenAICompatiblePlatform(platform) if s.checkChannelPricingRestriction(ctx, groupID, requestedModel) { slog.Warn("channel pricing restriction blocked request", "group_id", derefGroupID(groupID), "model", requestedModel) - return nil, fmt.Errorf("%w supporting model: %s (channel pricing restriction)", ErrNoAvailableAccounts, requestedModel) + return nil, false, fmt.Errorf("%w supporting model: %s (channel pricing restriction)", ErrNoAvailableAccounts, requestedModel) } // 1. 尝试粘性会话命中 // Try sticky session hit if account := s.tryStickySessionHit(ctx, groupID, platform, sessionHash, requestedModel, excludedIDs, requireCompact, stickyAccountID, requiredCapability); account != nil { - return account, nil + return account, true, nil } // 2. 获取可调度的 OpenAI 账号 // Get schedulable OpenAI accounts accounts, err := s.listSchedulableAccounts(ctx, groupID, platform) if err != nil { - return nil, fmt.Errorf("query accounts failed: %w", err) + return nil, false, fmt.Errorf("query accounts failed: %w", err) } // 3. 按优先级 + LRU 选择最佳账号 @@ -908,12 +915,12 @@ func (s *OpenAIGatewayService) selectAccountForModelWithExclusions(ctx context.C selected, compactBlocked, filterStats := s.selectBestAccount(ctx, groupID, platform, accounts, requestedModel, excludedIDs, requireCompact, requiredCapability, preferLowUpstreamRate) if selected == nil { - return nil, noAvailableOpenAISelectionError(requestedModel, compactBlocked, filterStats.summary("")) + return nil, false, noAvailableOpenAISelectionError(requestedModel, compactBlocked, filterStats.summary("")) } hydrated, err := s.hydrateSelectedAccount(ctx, selected) if err != nil { - return nil, err + return nil, false, err } // 4. 设置粘性会话绑定（利润门下推迟到 handler 终检通过后再绑定， @@ -923,7 +930,7 @@ func (s *OpenAIGatewayService) selectAccountForModelWithExclusions(ctx context.C _ = s.setStickySessionAccountID(ctx, groupID, sessionHash, selected.ID, openaiStickySessionTTL) } - return hydrated, nil + return hydrated, false, nil } // tryStickySessionHit 尝试从粘性会话获取账号。 @@ -1133,31 +1140,34 @@ func (s *OpenAIGatewayService) selectAccountWithLoadAwareness(ctx context.Contex } } if s.concurrencyService == nil || !cfg.LoadBatchEnabled { - account, err := s.selectAccountForModelWithExclusions(ctx, groupID, platform, sessionHash, requestedModel, excludedIDs, requireCompact, stickyAccountID, requiredCapability, preferLowUpstreamRate) + account, stickyHit, err := s.selectAccountForModelWithExclusionsStickyHit(ctx, groupID, platform, sessionHash, requestedModel, excludedIDs, requireCompact, stickyAccountID, requiredCapability, preferLowUpstreamRate) if err != nil { return nil, err } result, err := s.tryAcquireAccountSlot(ctx, account.ID, account.Concurrency) if err == nil && result != nil && result.Acquired { - return s.newAcquiredSelectionResult(ctx, account, result.ReleaseFunc) + selection, selectErr := s.newAcquiredSelectionResult(ctx, account, result.ReleaseFunc) + return markStickySessionHit(selection, stickyHit), selectErr } if stickyAccountID > 0 && stickyAccountID == account.ID && s.concurrencyService != nil { waitingCount, _ := s.concurrencyService.GetAccountWaitingCount(ctx, account.ID) if waitingCount < cfg.StickySessionMaxWaiting { - return s.newSelectionResult(ctx, account, false, nil, &AccountWaitPlan{ + selection, selectErr := s.newSelectionResult(ctx, account, false, nil, &AccountWaitPlan{ AccountID: account.ID, MaxConcurrency: account.Concurrency, Timeout: cfg.StickySessionWaitTimeout, MaxWaiting: cfg.StickySessionMaxWaiting, }) + return markStickySessionHit(selection, stickyHit), selectErr } } - return s.newSelectionResult(ctx, account, false, nil, &AccountWaitPlan{ + selection, selectErr := s.newSelectionResult(ctx, account, false, nil, &AccountWaitPlan{ AccountID: account.ID, MaxConcurrency: account.Concurrency, Timeout: cfg.FallbackWaitTimeout, MaxWaiting: cfg.FallbackMaxWaiting, }) + return markStickySessionHit(selection, stickyHit), selectErr } accounts, err := s.listSchedulableAccounts(ctx, groupID, platform) @@ -1211,17 +1221,18 @@ func (s *OpenAIGatewayService) selectAccountWithLoadAwareness(ctx context.Contex return nil, selectErr } _ = s.refreshStickySessionTTL(ctx, groupID, sessionHash, openaiStickySessionTTL) - return selection, nil + return markStickySessionHit(selection, true), nil } waitingCount, _ := s.concurrencyService.GetAccountWaitingCount(ctx, accountID) if waitingCount < cfg.StickySessionMaxWaiting { - return s.newSelectionResult(ctx, account, false, nil, &AccountWaitPlan{ + selection, selectErr := s.newSelectionResult(ctx, account, false, nil, &AccountWaitPlan{ AccountID: accountID, MaxConcurrency: account.Concurrency, Timeout: cfg.StickySessionWaitTimeout, MaxWaiting: cfg.StickySessionMaxWaiting, }) + return markStickySessionHit(selection, true), selectErr } stickySpillover = true } @@ -1730,6 +1741,14 @@ func (s *OpenAIGatewayService) newAcquiredSelectionResult(ctx context.Context, a return selection, err } +// markStickySessionHit 在选号结果上记录账号是否来自会话粘性命中。 +func markStickySessionHit(selection *AccountSelectionResult, hit bool) *AccountSelectionResult { + if selection != nil && hit { + selection.stickySessionHit = true + } + return selection +} + func (s *OpenAIGatewayService) schedulingConfig() config.GatewaySchedulingConfig { if s.cfg != nil { return s.cfg.Gateway.Scheduling diff --git a/backend/internal/service/openai_legacy_scheduler_decision_test.go b/backend/internal/service/openai_legacy_scheduler_decision_test.go new file mode 100644 index 0000000000..9e30512a61 --- /dev/null +++ b/backend/internal/service/openai_legacy_scheduler_decision_test.go @@ -0,0 +1,87 @@ +package service + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func newLegacySchedulerDecisionTestAccounts(groupID int64, secondWSEnabled bool) []Account { + return []Account{ + { + ID: 38101, Platform: PlatformOpenAI, Type: AccountTypeAPIKey, + Status: StatusActive, Schedulable: true, Concurrency: 1, Priority: 0, + GroupIDs: []int64{groupID}, + Extra: map[string]any{"openai_apikey_responses_websockets_v2_enabled": true}, + }, + { + ID: 38102, Platform: PlatformOpenAI, Type: AccountTypeAPIKey, + Status: StatusActive, Schedulable: true, Concurrency: 1, Priority: 5, + GroupIDs: []int64{groupID}, + Extra: map[string]any{"openai_apikey_responses_websockets_v2_enabled": secondWSEnabled}, + }, + } +} + +func newLegacySchedulerDecisionTestService(accounts []Account, loadBatch bool, concurrency schedulerTestConcurrencyCache) *OpenAIGatewayService { + cfg := newSchedulerTestOpenAIWSV2Config() + cfg.Gateway.Scheduling.LoadBatchEnabled = loadBatch + return &OpenAIGatewayService{ + accountRepo: schedulerTestOpenAIAccountRepo{accounts: accounts}, + cache: &schedulerTestGatewayCache{}, + cfg: cfg, + concurrencyService: NewConcurrencyService(concurrency), + } +} + +func releaseLegacySchedulerDecisionSelection(selection *AccountSelectionResult) { + if selection != nil && selection.ReleaseFunc != nil { + selection.ReleaseFunc() + } +} + +func TestLegacySchedulerDecision_StickySessionLayer(t *testing.T) { + ctx := context.Background() + groupID := int64(38100) + for _, loadBatch := range []bool{true, false} { + name := "load_batch" + if !loadBatch { + name = "priority_lru" + } + t.Run(name, func(t *testing.T) { + svc := newLegacySchedulerDecisionTestService(newLegacySchedulerDecisionTestAccounts(groupID, true), loadBatch, schedulerTestConcurrencyCache{}) + sessionHash := "legacy-sticky-session" + require.NoError(t, svc.setStickySessionAccountID(ctx, &groupID, sessionHash, 38102, time.Hour)) + + selection, decision, err := svc.SelectAccountWithSchedulerForCapability( + ctx, &groupID, "", sessionHash, "gpt-5.1", nil, + OpenAIUpstreamTransportAny, OpenAIEndpointCapabilityChatCompletions, false, false, true, + ) + require.NoError(t, err) + require.NotNil(t, selection) + require.NotNil(t, selection.Account) + releaseLegacySchedulerDecisionSelection(selection) + require.Equal(t, int64(38102), selection.Account.ID, "sticky binding must win over priority order") + require.Equal(t, openAIAccountScheduleLayerSessionSticky, decision.Layer) + require.True(t, decision.StickySessionHit) + require.False(t, decision.StickyPreviousHit) + require.Equal(t, int64(38102), decision.SelectedAccountID) + require.Equal(t, AccountTypeAPIKey, decision.SelectedAccountType) + + selection, decision, err = svc.SelectAccountWithSchedulerForCapability( + ctx, &groupID, "", "", "gpt-5.1", nil, + OpenAIUpstreamTransportAny, OpenAIEndpointCapabilityChatCompletions, false, false, true, + ) + require.NoError(t, err) + require.NotNil(t, selection) + require.NotNil(t, selection.Account) + releaseLegacySchedulerDecisionSelection(selection) + require.Equal(t, int64(38101), selection.Account.ID) + require.Equal(t, openAIAccountScheduleLayerLoadBalance, decision.Layer) + require.False(t, decision.StickySessionHit) + require.Equal(t, int64(38101), decision.SelectedAccountID) + }) + } +} From 0d0200059223d66c746d32e0e4980ca822cf8762 Mon Sep 17 00:00:00 2001 From: xlplbo Date: Sun, 20 Sep 2026 19:28:13 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix(openai):=20=E9=9D=9E=E9=AB=98=E7=BA=A7?= =?UTF-8?q?=E8=B0=83=E5=BA=A6=E6=8C=89=20previous=5Fresponse=20=E8=B7=AF?= =?UTF-8?q?=E7=94=B1=E5=88=B0=E6=8C=81=E6=9C=89=E5=93=8D=E5=BA=94=E7=9A=84?= =?UTF-8?q?=E8=B4=A6=E5=8F=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/openai_account_scheduler.go | 63 +++++++ .../openai_legacy_scheduler_decision_test.go | 178 ++++++++++++++++++ 2 files changed, 241 insertions(+) diff --git a/backend/internal/service/openai_account_scheduler.go b/backend/internal/service/openai_account_scheduler.go index 0085182333..c136f8c3fb 100644 --- a/backend/internal/service/openai_account_scheduler.go +++ b/backend/internal/service/openai_account_scheduler.go @@ -2235,6 +2235,60 @@ func applyLegacySelectionDecision(decision *OpenAIAccountScheduleDecision, selec } } +// selectLegacyAccountByPreviousResponse 在非高级调度路径按 previous_response_id 命中持有该响应的账号， +// 先按请求模型做渠道限制检查，再复用高级调度器 previous_response 层的账号兼容校验 +// （分组、隐私、运行期封禁、代理隔离、上游模型渠道限制、传输与能力）；未命中返回 (nil, false, nil)。 +func (s *OpenAIGatewayService) selectLegacyAccountByPreviousResponse( + ctx context.Context, + groupID *int64, + previousResponseID string, + sessionHash string, + requestedModel string, + excludedIDs map[int64]struct{}, + requiredTransport OpenAIUpstreamTransport, + requiredCapability OpenAIEndpointCapability, + requiredImageCapability OpenAIImagesCapability, + requireCompact bool, + platform string, +) (*AccountSelectionResult, bool, error) { + if strings.TrimSpace(previousResponseID) == "" || platform != PlatformOpenAI { + return nil, false, nil + } + if s.checkChannelPricingRestriction(ctx, groupID, requestedModel) { + return nil, false, fmt.Errorf("%w supporting model: %s (channel pricing restriction)", ErrNoAvailableAccounts, requestedModel) + } + selection, err := s.selectAccountByPreviousResponseIDForCapability(ctx, groupID, previousResponseID, requestedModel, excludedIDs, requiredCapability, requireCompact) + if err != nil { + return nil, false, err + } + if selection == nil || selection.Account == nil { + return nil, false, nil + } + account := selection.Account + scheduler := &defaultOpenAIAccountScheduler{service: s, stats: newOpenAIAccountRuntimeStats()} + compatible, _ := scheduler.isAccountRequestCompatibleReason(ctx, account, OpenAIAccountScheduleRequest{ + GroupID: groupID, + Platform: platform, + RequestedModel: requestedModel, + RequiredTransport: requiredTransport, + RequiredCapability: requiredCapability, + RequiredImageCapability: requiredImageCapability, + RequireCompact: requireCompact, + ExcludedIDs: excludedIDs, + RequirePrivacySet: s.openAIGroupRequiresPrivacySet(ctx, groupID), + }) + if !s.openAIAccountMatchesSchedulingGroup(account, groupID) || !compatible || !scheduler.isAccountTransportCompatible(account, requiredTransport) { + if selection.ReleaseFunc != nil { + selection.ReleaseFunc() + } + return nil, false, nil + } + if sessionHash != "" { + _ = s.bindOpenAIStickySessionDuringSelection(ctx, groupID, sessionHash, account.ID) + } + return selection, true, nil +} + func (s *OpenAIGatewayService) selectAccountWithSchedulerOnce( ctx context.Context, groupID *int64, @@ -2271,6 +2325,15 @@ func (s *OpenAIGatewayService) selectAccountWithSchedulerOnce( scheduler := s.getOpenAIAccountScheduler(ctx) if scheduler == nil { decision.Layer = openAIAccountScheduleLayerLoadBalance + if selection, hit, err := s.selectLegacyAccountByPreviousResponse(ctx, groupID, previousResponseID, sessionHash, requestedModel, excludedIDs, requiredTransport, requiredCapability, requiredImageCapability, requireCompact, platform); err != nil { + return nil, decision, err + } else if hit { + decision.Layer = openAIAccountScheduleLayerPreviousResponse + decision.StickyPreviousHit = true + decision.SelectedAccountID = selection.Account.ID + decision.SelectedAccountType = selection.Account.Type + return selection, decision, nil + } if guardianParentAccountID > 0 { if s.checkChannelPricingRestriction(ctx, groupID, requestedModel) { return nil, decision, fmt.Errorf("%w supporting model: %s (channel pricing restriction)", ErrNoAvailableAccounts, requestedModel) diff --git a/backend/internal/service/openai_legacy_scheduler_decision_test.go b/backend/internal/service/openai_legacy_scheduler_decision_test.go index 9e30512a61..2e1ac52176 100644 --- a/backend/internal/service/openai_legacy_scheduler_decision_test.go +++ b/backend/internal/service/openai_legacy_scheduler_decision_test.go @@ -1,3 +1,5 @@ +//go:build unit + package service import ( @@ -43,6 +45,7 @@ func releaseLegacySchedulerDecisionSelection(selection *AccountSelectionResult) } func TestLegacySchedulerDecision_StickySessionLayer(t *testing.T) { + resetOpenAIAdvancedSchedulerSettingCacheForTest() ctx := context.Background() groupID := int64(38100) for _, loadBatch := range []bool{true, false} { @@ -85,3 +88,178 @@ func TestLegacySchedulerDecision_StickySessionLayer(t *testing.T) { }) } } + +func TestLegacySchedulerDecision_PreviousResponseRouting(t *testing.T) { + resetOpenAIAdvancedSchedulerSettingCacheForTest() + ctx := context.Background() + groupID := int64(38110) + responseID := "resp_legacy_route" + + t.Run("routes to the account owning the response", func(t *testing.T) { + svc := newLegacySchedulerDecisionTestService(newLegacySchedulerDecisionTestAccounts(groupID, true), true, schedulerTestConcurrencyCache{}) + require.NoError(t, svc.getOpenAIWSStateStore().BindResponseAccount(ctx, groupID, responseID, 38102, time.Hour)) + + selection, decision, err := svc.SelectAccountWithSchedulerForCapability( + ctx, &groupID, responseID, "legacy-prev-session", "gpt-5.1", nil, + OpenAIUpstreamTransportResponsesWebsocketV2Ingress, OpenAIEndpointCapabilityChatCompletions, false, true, true, + ) + require.NoError(t, err) + require.NotNil(t, selection) + require.NotNil(t, selection.Account) + releaseLegacySchedulerDecisionSelection(selection) + require.Equal(t, int64(38102), selection.Account.ID, "previous_response_id must route to the owning account") + require.Equal(t, openAIAccountScheduleLayerPreviousResponse, decision.Layer) + require.True(t, decision.StickyPreviousHit) + require.False(t, decision.StickySessionHit) + require.Equal(t, int64(38102), decision.SelectedAccountID) + }) + + t.Run("excluded owner falls back to load balance", func(t *testing.T) { + svc := newLegacySchedulerDecisionTestService(newLegacySchedulerDecisionTestAccounts(groupID, true), true, schedulerTestConcurrencyCache{}) + require.NoError(t, svc.getOpenAIWSStateStore().BindResponseAccount(ctx, groupID, responseID, 38102, time.Hour)) + + selection, decision, err := svc.SelectAccountWithSchedulerForCapability( + ctx, &groupID, responseID, "", "gpt-5.1", map[int64]struct{}{38102: {}}, + OpenAIUpstreamTransportAny, OpenAIEndpointCapabilityChatCompletions, false, true, true, + ) + require.NoError(t, err) + require.NotNil(t, selection) + require.NotNil(t, selection.Account) + releaseLegacySchedulerDecisionSelection(selection) + require.Equal(t, int64(38101), selection.Account.ID) + require.Equal(t, openAIAccountScheduleLayerLoadBalance, decision.Layer) + require.False(t, decision.StickyPreviousHit) + }) + + t.Run("transport-incompatible owner releases its slot and falls back", func(t *testing.T) { + released := make([]int64, 0) + svc := newLegacySchedulerDecisionTestService(newLegacySchedulerDecisionTestAccounts(groupID, false), true, schedulerTestConcurrencyCache{releasedIDs: &released}) + require.NoError(t, svc.getOpenAIWSStateStore().BindResponseAccount(ctx, groupID, responseID, 38102, time.Hour)) + + selection, decision, err := svc.SelectAccountWithSchedulerForCapability( + ctx, &groupID, responseID, "", "gpt-5.1", nil, + OpenAIUpstreamTransportResponsesWebsocketV2Ingress, OpenAIEndpointCapabilityChatCompletions, false, true, true, + ) + require.NoError(t, err) + require.NotNil(t, selection) + require.NotNil(t, selection.Account) + releaseLegacySchedulerDecisionSelection(selection) + require.Equal(t, int64(38101), selection.Account.ID) + require.Equal(t, openAIAccountScheduleLayerLoadBalance, decision.Layer) + require.False(t, decision.StickyPreviousHit) + require.Contains(t, released, int64(38102), "the owning account's slot must be released after the transport check fails") + }) + + t.Run("channel-restricted request model fails before routing", func(t *testing.T) { + released := make([]int64, 0) + svc := newLegacySchedulerDecisionTestService(newLegacySchedulerDecisionTestAccounts(groupID, true), true, schedulerTestConcurrencyCache{releasedIDs: &released}) + svc.channelService = newTestChannelService(makeStandardRepo(Channel{ + ID: 38111, + Status: StatusActive, + GroupIDs: []int64{groupID}, + RestrictModels: true, + BillingModelSource: BillingModelSourceChannelMapped, + ModelPricing: []ChannelModelPricing{{Platform: PlatformOpenAI, Models: []string{"gpt-4o"}}}, + ModelMapping: map[string]map[string]string{PlatformOpenAI: {"gpt-5.1": "o3-mini"}}, + }, map[int64]string{groupID: PlatformOpenAI})) + require.NoError(t, svc.getOpenAIWSStateStore().BindResponseAccount(ctx, groupID, responseID, 38102, time.Hour)) + + selection, _, err := svc.SelectAccountWithSchedulerForCapability( + ctx, &groupID, responseID, "", "gpt-5.1", nil, + OpenAIUpstreamTransportAny, OpenAIEndpointCapabilityChatCompletions, false, true, true, + ) + require.ErrorIs(t, err, ErrNoAvailableAccounts) + require.Contains(t, err.Error(), "channel pricing restriction") + require.Nil(t, selection) + require.Empty(t, released, "the owning account's slot must not be touched for a channel-restricted model") + }) + + t.Run("channel-restricted upstream model of the owner releases its slot and falls back", func(t *testing.T) { + released := make([]int64, 0) + accounts := newLegacySchedulerDecisionTestAccounts(groupID, true) + accounts[1].Credentials = map[string]any{"model_mapping": map[string]any{"gpt-5.1": "o3-mini"}} + svc := newLegacySchedulerDecisionTestService(accounts, true, schedulerTestConcurrencyCache{releasedIDs: &released}) + svc.channelService = newTestChannelService(makeStandardRepo(Channel{ + ID: 38112, + Status: StatusActive, + GroupIDs: []int64{groupID}, + RestrictModels: true, + BillingModelSource: BillingModelSourceUpstream, + ModelPricing: []ChannelModelPricing{{Platform: PlatformOpenAI, Models: []string{"gpt-5.1"}}}, + }, map[int64]string{groupID: PlatformOpenAI})) + require.NoError(t, svc.getOpenAIWSStateStore().BindResponseAccount(ctx, groupID, responseID, 38102, time.Hour)) + + selection, decision, err := svc.SelectAccountWithSchedulerForCapability( + ctx, &groupID, responseID, "", "gpt-5.1", nil, + OpenAIUpstreamTransportAny, OpenAIEndpointCapabilityChatCompletions, false, true, true, + ) + require.NoError(t, err) + require.NotNil(t, selection) + require.NotNil(t, selection.Account) + releaseLegacySchedulerDecisionSelection(selection) + require.Equal(t, int64(38101), selection.Account.ID) + require.Equal(t, openAIAccountScheduleLayerLoadBalance, decision.Layer) + require.False(t, decision.StickyPreviousHit) + require.Contains(t, released, int64(38102), "the owning account's slot must be released after the upstream model restriction check fails") + }) + + t.Run("quarantined owner proxy releases its slot and falls back", func(t *testing.T) { + released := make([]int64, 0) + healthyProxy, quarantinedProxy := int64(38113), int64(38114) + accounts := newLegacySchedulerDecisionTestAccounts(groupID, true) + accounts[0].ProxyID = &healthyProxy + accounts[1].ProxyID = &quarantinedProxy + svc := newLegacySchedulerDecisionTestService(accounts, false, schedulerTestConcurrencyCache{releasedIDs: &released}) + svc.openaiProxyStreamCircuit = newOpenAIProxyStreamCircuit(openAIProxyStreamCircuitSettings{ + failureThreshold: 1, + failureWindow: time.Minute, + quarantineTTL: 10 * time.Minute, + maxEntries: 16, + }) + tripped, _ := svc.openaiProxyStreamCircuit.recordFailure(quarantinedProxy, time.Now()) + require.True(t, tripped) + require.NoError(t, svc.getOpenAIWSStateStore().BindResponseAccount(ctx, groupID, responseID, 38102, time.Hour)) + + selection, decision, err := svc.SelectAccountWithSchedulerForCapability( + ctx, &groupID, responseID, "", "gpt-5.1", nil, + OpenAIUpstreamTransportAny, OpenAIEndpointCapabilityChatCompletions, false, true, true, + ) + require.NoError(t, err) + require.NotNil(t, selection) + require.NotNil(t, selection.Account) + releaseLegacySchedulerDecisionSelection(selection) + require.Equal(t, int64(38101), selection.Account.ID) + require.Equal(t, openAIAccountScheduleLayerLoadBalance, decision.Layer) + require.False(t, decision.StickyPreviousHit) + require.Contains(t, released, int64(38102), "the owning account's slot must be released after the proxy quarantine check fails") + }) + + t.Run("all proxies quarantined fails open to the owner", func(t *testing.T) { + proxy := int64(38115) + accounts := newLegacySchedulerDecisionTestAccounts(groupID, true) + accounts[0].ProxyID = &proxy + accounts[1].ProxyID = &proxy + svc := newLegacySchedulerDecisionTestService(accounts, false, schedulerTestConcurrencyCache{}) + svc.openaiProxyStreamCircuit = newOpenAIProxyStreamCircuit(openAIProxyStreamCircuitSettings{ + failureThreshold: 1, + failureWindow: time.Minute, + quarantineTTL: 10 * time.Minute, + maxEntries: 16, + }) + tripped, _ := svc.openaiProxyStreamCircuit.recordFailure(proxy, time.Now()) + require.True(t, tripped) + require.NoError(t, svc.getOpenAIWSStateStore().BindResponseAccount(ctx, groupID, responseID, 38102, time.Hour)) + + selection, decision, err := svc.SelectAccountWithSchedulerForCapability( + ctx, &groupID, responseID, "", "gpt-5.1", nil, + OpenAIUpstreamTransportAny, OpenAIEndpointCapabilityChatCompletions, false, true, true, + ) + require.NoError(t, err, "quarantine must fail open instead of returning no available accounts") + require.NotNil(t, selection) + require.NotNil(t, selection.Account) + releaseLegacySchedulerDecisionSelection(selection) + require.Equal(t, int64(38102), selection.Account.ID) + require.Equal(t, openAIAccountScheduleLayerPreviousResponse, decision.Layer) + require.True(t, decision.StickyPreviousHit) + }) +}