diff --git a/CODEX_QUOTA_OVERDRAFT_CUSTOMIZATION.md b/CODEX_QUOTA_OVERDRAFT_CUSTOMIZATION.md index 2df3b7a41..7a6cf5a16 100644 --- a/CODEX_QUOTA_OVERDRAFT_CUSTOMIZATION.md +++ b/CODEX_QUOTA_OVERDRAFT_CUSTOMIZATION.md @@ -16,7 +16,7 @@ 源码 Docker 构建会读取根目录的 `FORK_VERSION`，并以 `BuildType=source` 写入二进制。后台更新服务通过 GitHub Contents API 读取 Fork 分支上的同名文件，使用语义化版本比较判断是否有新版本。Redis 缓存同时记录仓库和构建类型，因此旧的官方更新缓存不会继续生效。 -源码构建仅显示 `git pull` 更新提示，`PerformUpdate`、指定版本回退和在线回退列表均被禁用，防止官方二进制覆盖透支功能。维护者每次发布 Fork 更新都必须递增 `FORK_VERSION`；当前版本为 `0.1.177-overdraft.5`，同步下一上游版本后使用 `0.1.178-overdraft.1`。 +源码构建仅显示 `git pull` 更新提示，`PerformUpdate`、指定版本回退和在线回退列表均被禁用，防止官方二进制覆盖透支功能。维护者每次发布 Fork 更新都必须递增 `FORK_VERSION`；当前版本为 `0.1.177-overdraft.6`，同步下一上游版本后使用 `0.1.178-overdraft.1`。 Sub2API `v0.1.177` 将原生 remote compaction v2 保留在 `/responses` 路径。本定制同时检查旧 Compact 路径和原生 v2 请求信号，二者都不会开启额度透支调度或注入透支请求形态。 @@ -76,6 +76,18 @@ gateway: `/responses/compact`、生图请求、Embedding、Count Tokens、Live 等端点不启用额度透支。 +## 低冲突集成结构 + +从 `0.1.177-overdraft.6` 起，透支行为保持不变，但实现按“独立模块 + 固定钩子”组织，以减少同步官方更新时的冲突： + +- 四个 PostgreSQL 原子状态方法及临时暂停查询扩展位于 `backend/internal/repository/account_repo_codex_overdraft.go`，`account_repo.go` 不再承载大段透支 SQL。 +- Gateway 通过 `sync.Once` 懒加载并持有全进程唯一协调器；账号用量服务和账号测试服务从同一个 Gateway 获取实例，不再向 Wire 图添加 Fork 专用 Provider。 +- `backend/cmd/server/wire_gen.go` 由官方 Wire 图正常生成，不包含透支专用变量或参数。 +- 调度、429、业务成功、用量快照和账号测试接线统一放在 `openai_codex_quota_overdraft_hooks.go`、`openai_codex_quota_overdraft_integration.go` 和 `account_test_codex_overdraft.go`；官方热点文件仅保留短调用。 +- 前端状态徽标和透支统计分别位于 `CodexOverdraftStatus.vue`、`CodexOverdraftStats.vue`，官方账号用量组件仅负责传入原有数据。 + +这次调整没有修改 95% 注入门槛、Payload 内容、额度信号解析、429 分类、探测次数、状态转换、事务边界、暂停/恢复规则、统计口径或页面文案。升级时应优先保留独立文件，再逐一核对官方文件中的短钩子。 + ## 修改文件 后端配置、路由和调度： @@ -84,10 +96,14 @@ gateway: - `backend/internal/handler/openai_chat_completions.go` - `backend/internal/handler/openai_gateway_handler.go` - `backend/internal/repository/account_repo.go` +- `backend/internal/repository/account_repo_codex_overdraft.go` - `backend/internal/repository/account_repo_schedulable_projection_test.go` +- `backend/internal/service/account_test_codex_overdraft.go` - `backend/internal/service/openai_account_runtime_block_fastpath.go` - `backend/internal/service/openai_account_scheduler.go` - `backend/internal/service/openai_codex_quota_overdraft.go` +- `backend/internal/service/openai_codex_quota_overdraft_hooks.go` +- `backend/internal/service/openai_codex_quota_overdraft_integration.go` - `backend/internal/service/openai_codex_quota_overdraft_probe.go` - `backend/internal/service/openai_codex_quota_overdraft_test.go` - `backend/internal/service/openai_codex_quota_overdraft_probe_test.go` @@ -106,11 +122,12 @@ gateway: - `backend/internal/service/account_test_service_openai_test.go` - `backend/internal/service/account_test_service_openai_compact_test.go` - `backend/internal/service/wire.go` -- `backend/cmd/server/wire_gen.go` 前端、配置示例和测试： - `frontend/src/types/index.ts` +- `frontend/src/components/account/CodexOverdraftStats.vue` +- `frontend/src/components/account/CodexOverdraftStatus.vue` - `frontend/src/components/account/UsageProgressBar.vue` - `frontend/src/components/account/AccountUsageCell.vue` - `frontend/src/components/account/__tests__/UsageProgressBar.spec.ts` diff --git a/FORK_VERSION b/FORK_VERSION index 3c106b5d2..0dc05cc29 100644 --- a/FORK_VERSION +++ b/FORK_VERSION @@ -1 +1 @@ -0.1.177-overdraft.5 +0.1.177-overdraft.6 diff --git a/backend/cmd/server/wire_gen.go b/backend/cmd/server/wire_gen.go index 5f2d62856..ab9f4530b 100644 --- a/backend/cmd/server/wire_gen.go +++ b/backend/cmd/server/wire_gen.go @@ -199,9 +199,8 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) { grokQuotaService := service.ProvideGrokQuotaService(accountRepository, proxyRepository, grokTokenProvider, httpUpstream, configConfig, usageLogRepository, settingService) openAIQuotaService := service.ProvideOpenAIQuotaService(accountRepository, proxyRepository, openAITokenProvider, privacyClientFactory, openAIGatewayService) usageCache := service.NewUsageCache() - codexQuotaOverdraftCoordinator := service.ProvideCodexQuotaOverdraftCoordinator(accountRepository, httpUpstream, openAITokenProvider, tlsFingerprintProfileService, configConfig, tempUnschedCache, openAIGatewayService, rateLimitService) - accountUsageService := service.ProvideAccountUsageService(accountRepository, usageLogRepository, claudeUsageFetcher, geminiQuotaService, antigravityQuotaFetcher, grokQuotaFetcher, grokQuotaService, openAIQuotaService, usageCache, identityCache, tlsFingerprintProfileService, openAIGatewayService, codexQuotaOverdraftCoordinator) - accountTestService := service.ProvideAccountTestService(accountRepository, geminiTokenProvider, claudeTokenProvider, grokTokenProvider, antigravityGatewayService, httpUpstream, configConfig, tlsFingerprintProfileService, openAIGatewayService, settingService, codexQuotaOverdraftCoordinator) + accountUsageService := service.ProvideAccountUsageService(accountRepository, usageLogRepository, claudeUsageFetcher, geminiQuotaService, antigravityQuotaFetcher, grokQuotaFetcher, grokQuotaService, openAIQuotaService, usageCache, identityCache, tlsFingerprintProfileService, openAIGatewayService) + accountTestService := service.ProvideAccountTestService(accountRepository, geminiTokenProvider, claudeTokenProvider, grokTokenProvider, antigravityGatewayService, httpUpstream, configConfig, tlsFingerprintProfileService, openAIGatewayService, settingService) crsSyncService := service.NewCRSSyncService(accountRepository, proxyRepository, oAuthService, openAIOAuthService, geminiOAuthService, configConfig) accountHandler := admin.ProvideAccountHandler(adminService, oAuthService, openAIOAuthService, geminiOAuthService, antigravityOAuthService, grokOAuthService, rateLimitService, accountUsageService, accountTestService, concurrencyService, crsSyncService, sessionLimitCache, rpmCache, compositeTokenCacheInvalidator, grokQuotaService) adminAnnouncementHandler := admin.NewAnnouncementHandler(announcementService) diff --git a/backend/internal/repository/account_repo.go b/backend/internal/repository/account_repo.go index 0c2fe0706..1b9b0958c 100644 --- a/backend/internal/repository/account_repo.go +++ b/backend/internal/repository/account_repo.go @@ -2593,207 +2593,6 @@ func (r *accountRepository) UpdateExtra(ctx context.Context, id int64, updates m return nil } -// ClaimCodexQuotaOverdraftProbe atomically reserves one quota cycle. A cycle is -// claimed at most once across all sub2api replicas. -func (r *accountRepository) ClaimCodexQuotaOverdraftProbe( - ctx context.Context, - id int64, - state *service.CodexQuotaOverdraftProbeState, -) (bool, error) { - if state == nil || strings.TrimSpace(state.CycleKey) == "" { - return false, nil - } - payload, err := json.Marshal(state) - if err != nil { - return false, err - } - result, err := r.sql.ExecContext(ctx, ` - UPDATE accounts - SET extra = COALESCE(extra, '{}'::jsonb) || jsonb_build_object($1::text, $2::jsonb), - updated_at = NOW() - WHERE id = $3 - AND deleted_at IS NULL - AND COALESCE(extra #>> '{codex_quota_overdraft_probe,cycle_key}', '') <> $4 - `, service.CodexQuotaOverdraftProbeExtraKey, string(payload), id, state.CycleKey) - if err != nil { - return false, err - } - affected, err := result.RowsAffected() - if err != nil || affected == 0 { - return false, err - } - r.syncSchedulerAccountSnapshot(ctx, id) - return true, nil -} - -// PersistCodexQuotaOverdraftProbeUnlessFailed stores a non-failure result while -// preserving a terminal failure already confirmed for the same quota cycle. -func (r *accountRepository) PersistCodexQuotaOverdraftProbeUnlessFailed( - ctx context.Context, - id int64, - state *service.CodexQuotaOverdraftProbeState, -) (bool, error) { - if state == nil || state.Status == "failed" || strings.TrimSpace(state.CycleKey) == "" { - return false, nil - } - payload, err := json.Marshal(state) - if err != nil { - return false, err - } - result, err := r.sql.ExecContext(ctx, ` - UPDATE accounts - SET extra = COALESCE(extra, '{}'::jsonb) || jsonb_build_object($1::text, $2::jsonb), - updated_at = NOW() - WHERE id = $3 - AND deleted_at IS NULL - AND ( - COALESCE(extra #>> '{codex_quota_overdraft_probe,cycle_key}', '') <> $4 - OR COALESCE(extra #>> '{codex_quota_overdraft_probe,status}', '') <> 'failed' - ) - `, service.CodexQuotaOverdraftProbeExtraKey, string(payload), id, state.CycleKey) - if err != nil { - return false, err - } - affected, err := result.RowsAffected() - if err != nil || affected == 0 { - return false, err - } - r.syncSchedulerAccountSnapshot(ctx, id) - return true, nil -} - -// ClearCodexQuotaOverdraftPauseIfState clears stale scheduling state only when -// the persisted overdraft result is still the expected available state. -func (r *accountRepository) ClearCodexQuotaOverdraftPauseIfState( - ctx context.Context, - id int64, - cycleKey string, - status string, - expectedTempReason string, -) (bool, bool, error) { - if id <= 0 || strings.TrimSpace(cycleKey) == "" || (status != "passed" && status != "recovered") { - return false, false, nil - } - beginner, ok := r.sql.(codexQuotaOverdraftTxBeginner) - if !ok { - return false, false, errors.New("account repository does not support transactions") - } - tx, err := beginner.BeginTx(ctx, nil) - if err != nil { - return false, false, err - } - defer func() { _ = tx.Rollback() }() - - var clearedRateLimit, clearedTemp bool - err = tx.QueryRowContext(ctx, ` - WITH target AS ( - SELECT id, - (rate_limited_at IS NOT NULL OR rate_limit_reset_at IS NOT NULL) AS clear_rate_limit, - ($4::text <> '' AND COALESCE(temp_unschedulable_reason = $4, FALSE)) AS clear_temp - FROM accounts - WHERE id = $1 - AND deleted_at IS NULL - AND extra #>> '{codex_quota_overdraft_probe,cycle_key}' = $2 - AND extra #>> '{codex_quota_overdraft_probe,status}' = $3 - FOR UPDATE - ) - UPDATE accounts AS a - SET rate_limited_at = CASE WHEN target.clear_rate_limit THEN NULL ELSE a.rate_limited_at END, - rate_limit_reset_at = CASE WHEN target.clear_rate_limit THEN NULL ELSE a.rate_limit_reset_at END, - temp_unschedulable_until = CASE WHEN target.clear_temp THEN NULL ELSE a.temp_unschedulable_until END, - temp_unschedulable_reason = CASE WHEN target.clear_temp THEN NULL ELSE a.temp_unschedulable_reason END, - updated_at = NOW() - FROM target - WHERE a.id = target.id - AND (target.clear_rate_limit OR target.clear_temp) - RETURNING target.clear_rate_limit, target.clear_temp - `, id, cycleKey, status, expectedTempReason).Scan(&clearedRateLimit, &clearedTemp) - if errors.Is(err, sql.ErrNoRows) { - return false, false, nil - } - if err != nil { - return false, false, err - } - if err := enqueueSchedulerOutbox(ctx, tx, service.SchedulerOutboxEventAccountChanged, &id, nil, nil); err != nil { - return false, false, err - } - if err := tx.Commit(); err != nil { - return false, false, err - } - r.syncSchedulerAccountSnapshotDetached(ctx, id) - return clearedRateLimit, clearedTemp, nil -} - -type codexQuotaOverdraftTxBeginner interface { - BeginTx(context.Context, *sql.TxOptions) (*sql.Tx, error) -} - -// FinalizeCodexQuotaOverdraftProbeFailed persists the terminal probe state, -// account pause, and scheduler notification in one transaction. -func (r *accountRepository) FinalizeCodexQuotaOverdraftProbeFailed( - ctx context.Context, - id int64, - state *service.CodexQuotaOverdraftProbeState, - until time.Time, - reason string, -) (bool, error) { - if state == nil || state.Status != "failed" || strings.TrimSpace(state.CycleKey) == "" || until.IsZero() { - return false, nil - } - payload, err := json.Marshal(state) - if err != nil { - return false, err - } - beginner, ok := r.sql.(codexQuotaOverdraftTxBeginner) - if !ok { - return false, errors.New("account repository does not support transactions") - } - tx, err := beginner.BeginTx(ctx, nil) - if err != nil { - return false, err - } - defer func() { _ = tx.Rollback() }() - - result, err := tx.ExecContext(ctx, ` - UPDATE accounts - SET extra = COALESCE(extra, '{}'::jsonb) || jsonb_build_object($1::text, $2::jsonb), - temp_unschedulable_until = CASE - WHEN temp_unschedulable_until IS NULL OR temp_unschedulable_until < $3 THEN $3 - ELSE temp_unschedulable_until - END, - temp_unschedulable_reason = CASE - WHEN temp_unschedulable_until IS NULL OR temp_unschedulable_until < $3 THEN $4 - ELSE temp_unschedulable_reason - END, - updated_at = NOW() - WHERE id = $5 - AND deleted_at IS NULL - AND extra #>> '{codex_quota_overdraft_probe,cycle_key}' = $6 - AND ( - extra #>> '{codex_quota_overdraft_probe,status}' IN ('pending', 'failed') - OR ($7::boolean AND extra #>> '{codex_quota_overdraft_probe,status}' IN ('passed', 'inconclusive', 'recovered')) - ) - `, service.CodexQuotaOverdraftProbeExtraKey, string(payload), until, reason, id, state.CycleKey, state.ReasonCode == "business_quota_limited") - if err != nil { - return false, err - } - affected, err := result.RowsAffected() - if err != nil { - return false, err - } - if affected == 0 { - return false, nil - } - if err := enqueueSchedulerOutbox(ctx, tx, service.SchedulerOutboxEventAccountChanged, &id, nil, nil); err != nil { - return false, err - } - if err := tx.Commit(); err != nil { - return false, err - } - r.syncSchedulerAccountSnapshotDetached(ctx, id) - return true, nil -} - // UpdateUpstreamBillingProbeSnapshot stores a probe result only while the // network identity used by that probe is still current. func (r *accountRepository) UpdateUpstreamBillingProbeSnapshot( @@ -3343,15 +3142,7 @@ func tempUnschedulablePredicate(ctx context.Context) dbpredicate.Account { entsql.IsNull(col), entsql.LTE(col, entsql.Expr("NOW()")), } - if service.CodexQuotaOverdraftSchedulingEnabled(ctx) { - reasonCol := s.C("temp_unschedulable_reason") - predicates = append(predicates, entsql.And( - entsql.EQ(s.C("platform"), service.PlatformOpenAI), - entsql.EQ(s.C("type"), service.AccountTypeOAuth), - entsql.IsNull(s.C("parent_account_id")), - entsql.Contains(reasonCol, `"source":"`+service.AccountSchedulingThresholdReasonSource+`"`), - )) - } + predicates = extendCodexQuotaOverdraftTempUnschedulablePredicates(ctx, s, predicates) s.Where(entsql.Or(predicates...)) }) } diff --git a/backend/internal/repository/account_repo_codex_overdraft.go b/backend/internal/repository/account_repo_codex_overdraft.go new file mode 100644 index 000000000..46c63d17b --- /dev/null +++ b/backend/internal/repository/account_repo_codex_overdraft.go @@ -0,0 +1,232 @@ +package repository + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "strings" + "time" + + "github.com/Wei-Shaw/sub2api/internal/service" + + entsql "entgo.io/ent/dialect/sql" +) + +// ClaimCodexQuotaOverdraftProbe atomically reserves one quota cycle. A cycle is +// claimed at most once across all sub2api replicas. +func (r *accountRepository) ClaimCodexQuotaOverdraftProbe( + ctx context.Context, + id int64, + state *service.CodexQuotaOverdraftProbeState, +) (bool, error) { + if state == nil || strings.TrimSpace(state.CycleKey) == "" { + return false, nil + } + payload, err := json.Marshal(state) + if err != nil { + return false, err + } + result, err := r.sql.ExecContext(ctx, ` + UPDATE accounts + SET extra = COALESCE(extra, '{}'::jsonb) || jsonb_build_object($1::text, $2::jsonb), + updated_at = NOW() + WHERE id = $3 + AND deleted_at IS NULL + AND COALESCE(extra #>> '{codex_quota_overdraft_probe,cycle_key}', '') <> $4 + `, service.CodexQuotaOverdraftProbeExtraKey, string(payload), id, state.CycleKey) + if err != nil { + return false, err + } + affected, err := result.RowsAffected() + if err != nil || affected == 0 { + return false, err + } + r.syncSchedulerAccountSnapshot(ctx, id) + return true, nil +} + +// PersistCodexQuotaOverdraftProbeUnlessFailed stores a non-failure result while +// preserving a terminal failure already confirmed for the same quota cycle. +func (r *accountRepository) PersistCodexQuotaOverdraftProbeUnlessFailed( + ctx context.Context, + id int64, + state *service.CodexQuotaOverdraftProbeState, +) (bool, error) { + if state == nil || state.Status == "failed" || strings.TrimSpace(state.CycleKey) == "" { + return false, nil + } + payload, err := json.Marshal(state) + if err != nil { + return false, err + } + result, err := r.sql.ExecContext(ctx, ` + UPDATE accounts + SET extra = COALESCE(extra, '{}'::jsonb) || jsonb_build_object($1::text, $2::jsonb), + updated_at = NOW() + WHERE id = $3 + AND deleted_at IS NULL + AND ( + COALESCE(extra #>> '{codex_quota_overdraft_probe,cycle_key}', '') <> $4 + OR COALESCE(extra #>> '{codex_quota_overdraft_probe,status}', '') <> 'failed' + ) + `, service.CodexQuotaOverdraftProbeExtraKey, string(payload), id, state.CycleKey) + if err != nil { + return false, err + } + affected, err := result.RowsAffected() + if err != nil || affected == 0 { + return false, err + } + r.syncSchedulerAccountSnapshot(ctx, id) + return true, nil +} + +// ClearCodexQuotaOverdraftPauseIfState clears stale scheduling state only when +// the persisted overdraft result is still the expected available state. +func (r *accountRepository) ClearCodexQuotaOverdraftPauseIfState( + ctx context.Context, + id int64, + cycleKey string, + status string, + expectedTempReason string, +) (bool, bool, error) { + if id <= 0 || strings.TrimSpace(cycleKey) == "" || (status != "passed" && status != "recovered") { + return false, false, nil + } + beginner, ok := r.sql.(codexQuotaOverdraftTxBeginner) + if !ok { + return false, false, errors.New("account repository does not support transactions") + } + tx, err := beginner.BeginTx(ctx, nil) + if err != nil { + return false, false, err + } + defer func() { _ = tx.Rollback() }() + + var clearedRateLimit, clearedTemp bool + err = tx.QueryRowContext(ctx, ` + WITH target AS ( + SELECT id, + (rate_limited_at IS NOT NULL OR rate_limit_reset_at IS NOT NULL) AS clear_rate_limit, + ($4::text <> '' AND COALESCE(temp_unschedulable_reason = $4, FALSE)) AS clear_temp + FROM accounts + WHERE id = $1 + AND deleted_at IS NULL + AND extra #>> '{codex_quota_overdraft_probe,cycle_key}' = $2 + AND extra #>> '{codex_quota_overdraft_probe,status}' = $3 + FOR UPDATE + ) + UPDATE accounts AS a + SET rate_limited_at = CASE WHEN target.clear_rate_limit THEN NULL ELSE a.rate_limited_at END, + rate_limit_reset_at = CASE WHEN target.clear_rate_limit THEN NULL ELSE a.rate_limit_reset_at END, + temp_unschedulable_until = CASE WHEN target.clear_temp THEN NULL ELSE a.temp_unschedulable_until END, + temp_unschedulable_reason = CASE WHEN target.clear_temp THEN NULL ELSE a.temp_unschedulable_reason END, + updated_at = NOW() + FROM target + WHERE a.id = target.id + AND (target.clear_rate_limit OR target.clear_temp) + RETURNING target.clear_rate_limit, target.clear_temp + `, id, cycleKey, status, expectedTempReason).Scan(&clearedRateLimit, &clearedTemp) + if errors.Is(err, sql.ErrNoRows) { + return false, false, nil + } + if err != nil { + return false, false, err + } + if err := enqueueSchedulerOutbox(ctx, tx, service.SchedulerOutboxEventAccountChanged, &id, nil, nil); err != nil { + return false, false, err + } + if err := tx.Commit(); err != nil { + return false, false, err + } + r.syncSchedulerAccountSnapshotDetached(ctx, id) + return clearedRateLimit, clearedTemp, nil +} + +type codexQuotaOverdraftTxBeginner interface { + BeginTx(context.Context, *sql.TxOptions) (*sql.Tx, error) +} + +// FinalizeCodexQuotaOverdraftProbeFailed persists the terminal probe state, +// account pause, and scheduler notification in one transaction. +func (r *accountRepository) FinalizeCodexQuotaOverdraftProbeFailed( + ctx context.Context, + id int64, + state *service.CodexQuotaOverdraftProbeState, + until time.Time, + reason string, +) (bool, error) { + if state == nil || state.Status != "failed" || strings.TrimSpace(state.CycleKey) == "" || until.IsZero() { + return false, nil + } + payload, err := json.Marshal(state) + if err != nil { + return false, err + } + beginner, ok := r.sql.(codexQuotaOverdraftTxBeginner) + if !ok { + return false, errors.New("account repository does not support transactions") + } + tx, err := beginner.BeginTx(ctx, nil) + if err != nil { + return false, err + } + defer func() { _ = tx.Rollback() }() + + result, err := tx.ExecContext(ctx, ` + UPDATE accounts + SET extra = COALESCE(extra, '{}'::jsonb) || jsonb_build_object($1::text, $2::jsonb), + temp_unschedulable_until = CASE + WHEN temp_unschedulable_until IS NULL OR temp_unschedulable_until < $3 THEN $3 + ELSE temp_unschedulable_until + END, + temp_unschedulable_reason = CASE + WHEN temp_unschedulable_until IS NULL OR temp_unschedulable_until < $3 THEN $4 + ELSE temp_unschedulable_reason + END, + updated_at = NOW() + WHERE id = $5 + AND deleted_at IS NULL + AND extra #>> '{codex_quota_overdraft_probe,cycle_key}' = $6 + AND ( + extra #>> '{codex_quota_overdraft_probe,status}' IN ('pending', 'failed') + OR ($7::boolean AND extra #>> '{codex_quota_overdraft_probe,status}' IN ('passed', 'inconclusive', 'recovered')) + ) + `, service.CodexQuotaOverdraftProbeExtraKey, string(payload), until, reason, id, state.CycleKey, state.ReasonCode == "business_quota_limited") + if err != nil { + return false, err + } + affected, err := result.RowsAffected() + if err != nil { + return false, err + } + if affected == 0 { + return false, nil + } + if err := enqueueSchedulerOutbox(ctx, tx, service.SchedulerOutboxEventAccountChanged, &id, nil, nil); err != nil { + return false, err + } + if err := tx.Commit(); err != nil { + return false, err + } + r.syncSchedulerAccountSnapshotDetached(ctx, id) + return true, nil +} + +func extendCodexQuotaOverdraftTempUnschedulablePredicates( + ctx context.Context, + s *entsql.Selector, + predicates []*entsql.Predicate, +) []*entsql.Predicate { + if !service.CodexQuotaOverdraftSchedulingEnabled(ctx) { + return predicates + } + reasonCol := s.C("temp_unschedulable_reason") + return append(predicates, entsql.And( + entsql.EQ(s.C("platform"), service.PlatformOpenAI), + entsql.EQ(s.C("type"), service.AccountTypeOAuth), + entsql.IsNull(s.C("parent_account_id")), + entsql.Contains(reasonCol, `"source":"`+service.AccountSchedulingThresholdReasonSource+`"`), + )) +} diff --git a/backend/internal/service/account_test_codex_overdraft.go b/backend/internal/service/account_test_codex_overdraft.go new file mode 100644 index 000000000..f95531958 --- /dev/null +++ b/backend/internal/service/account_test_codex_overdraft.go @@ -0,0 +1,67 @@ +package service + +import ( + "context" + "net/http" + "time" +) + +type codexQuotaOverdraftAccountTestCoordinator interface { + ObserveAccount(account *Account, preferredModel string) + ObserveBusinessSuccess(account *Account, preferredModel string) + HandleQuota429(ctx context.Context, account *Account, headers http.Header, body []byte, preferredModel string) bool +} + +func (s *AccountTestService) SetCodexQuotaOverdraftCoordinator(coordinator *CodexQuotaOverdraftCoordinator) { + if s != nil { + s.codexQuotaOverdraft = coordinator + } +} + +func (s *AccountTestService) prepareCodexQuotaOverdraftTestRequest( + ctx context.Context, + account *Account, + payload []byte, +) (context.Context, []byte, bool) { + if !s.codexQuotaOverdraftTestEnabled(account) || !codexQuotaOverdraftInjectionEligible(account, time.Now().UTC()) { + return ctx, payload, false + } + ctx = WithCodexQuotaOverdraftScheduling(ctx) + updated, changed, err := injectCodexQuotaOverdraft(payload) + if err != nil || !changed { + return ctx, payload, false + } + markCodexQuotaOverdraftInjected(ctx, account.ID) + return ctx, updated, true +} + +func (s *AccountTestService) handleCodexQuotaOverdraftTest429( + ctx context.Context, + account *Account, + headers http.Header, + body []byte, + preferredModel string, +) bool { + return s.codexQuotaOverdraft != nil && s.codexQuotaOverdraftTestEnabled(account) && + s.codexQuotaOverdraft.HandleQuota429(ctx, account, headers, body, preferredModel) +} + +func (s *AccountTestService) observeCodexQuotaOverdraftTestResult( + account *Account, + preferredModel string, + injected bool, +) { + if s.codexQuotaOverdraft == nil || !s.codexQuotaOverdraftTestEnabled(account) { + return + } + if injected { + s.codexQuotaOverdraft.ObserveBusinessSuccess(account, preferredModel) + return + } + s.codexQuotaOverdraft.ObserveAccount(account, preferredModel) +} + +func (s *AccountTestService) codexQuotaOverdraftTestEnabled(account *Account) bool { + return s != nil && s.cfg != nil && s.cfg.Gateway.CodexQuotaOverdraftEnabled && + isCodexQuotaOverdraftAccount(account) +} diff --git a/backend/internal/service/account_test_service.go b/backend/internal/service/account_test_service.go index b4f21f20e..6d4857873 100644 --- a/backend/internal/service/account_test_service.go +++ b/backend/internal/service/account_test_service.go @@ -154,24 +154,12 @@ type AccountTestService struct { grokWSDialer openAIWSClientDialer } -type codexQuotaOverdraftAccountTestCoordinator interface { - ObserveAccount(account *Account, preferredModel string) - ObserveBusinessSuccess(account *Account, preferredModel string) - HandleQuota429(ctx context.Context, account *Account, headers http.Header, body []byte, preferredModel string) bool -} - func (s *AccountTestService) SetSettingService(settingService *SettingService) { if s != nil { s.settingService = settingService } } -func (s *AccountTestService) SetCodexQuotaOverdraftCoordinator(coordinator *CodexQuotaOverdraftCoordinator) { - if s != nil { - s.codexQuotaOverdraft = coordinator - } -} - // NewAccountTestService creates a new AccountTestService func NewAccountTestService( accountRepo AccountRepository, @@ -707,15 +695,7 @@ func (s *AccountTestService) testOpenAIAccountConnection(c *gin.Context, account } payload := createOpenAITestPayload(upstreamTestModelID, isOAuth) payloadBytes, _ := json.Marshal(payload) - overdraftInjected := false - if s.codexQuotaOverdraftTestEnabled(account) && codexQuotaOverdraftInjectionEligible(account, time.Now().UTC()) { - ctx = WithCodexQuotaOverdraftScheduling(ctx) - if updated, changed, err := injectCodexQuotaOverdraft(payloadBytes); err == nil && changed { - payloadBytes = updated - overdraftInjected = true - markCodexQuotaOverdraftInjected(ctx, account.ID) - } - } + ctx, payloadBytes, overdraftInjected := s.prepareCodexQuotaOverdraftTestRequest(ctx, account, payloadBytes) // Send test_start event once. A task-invalid Agent Identity response may // restart this probe after registering a replacement task. @@ -799,9 +779,7 @@ func (s *AccountTestService) testOpenAIAccountConnection(c *gin.Context, account return s.testOpenAIAccountConnection(c, account, modelID, prompt, mode) } if resp.StatusCode == http.StatusTooManyRequests { - handled := s.codexQuotaOverdraft != nil && s.codexQuotaOverdraftTestEnabled(account) && - s.codexQuotaOverdraft.HandleQuota429(ctx, account, resp.Header, body, upstreamTestModelID) - if !handled { + if !s.handleCodexQuotaOverdraftTest429(ctx, account, resp.Header, body, upstreamTestModelID) { s.reconcileOpenAI429State(ctx, account, resp.Header, body) } } @@ -816,21 +794,10 @@ func (s *AccountTestService) testOpenAIAccountConnection(c *gin.Context, account if err := s.processOpenAIStream(c, resp.Body); err != nil { return err } - if s.codexQuotaOverdraft != nil && s.codexQuotaOverdraftTestEnabled(account) { - if overdraftInjected { - s.codexQuotaOverdraft.ObserveBusinessSuccess(account, upstreamTestModelID) - } else { - s.codexQuotaOverdraft.ObserveAccount(account, upstreamTestModelID) - } - } + s.observeCodexQuotaOverdraftTestResult(account, upstreamTestModelID, overdraftInjected) return nil } -func (s *AccountTestService) codexQuotaOverdraftTestEnabled(account *Account) bool { - return s != nil && s.cfg != nil && s.cfg.Gateway.CodexQuotaOverdraftEnabled && - isCodexQuotaOverdraftAccount(account) -} - // testGrokAccountConnection routes Grok admin connectivity tests by explicit mode first, // then by selected model family for media. Standalone modes (search/tts/stt) never share // the text Responses path; image/video never hit Responses either. diff --git a/backend/internal/service/account_usage_service.go b/backend/internal/service/account_usage_service.go index e9a3e78fa..b489cdb84 100644 --- a/backend/internal/service/account_usage_service.go +++ b/backend/internal/service/account_usage_service.go @@ -312,12 +312,6 @@ type AccountUsageService struct { agentIdentityWS agentIdentityWSConnectionInvalidator } -func (s *AccountUsageService) SetCodexQuotaOverdraftCoordinator(coordinator *CodexQuotaOverdraftCoordinator) { - if s != nil { - s.codexQuotaOverdraft = coordinator - } -} - // NewAccountUsageService 创建AccountUsageService实例 func NewAccountUsageService( accountRepo AccountRepository, diff --git a/backend/internal/service/openai_account_runtime_block_fastpath.go b/backend/internal/service/openai_account_runtime_block_fastpath.go index 076004067..f5aa41b27 100644 --- a/backend/internal/service/openai_account_runtime_block_fastpath.go +++ b/backend/internal/service/openai_account_runtime_block_fastpath.go @@ -73,12 +73,7 @@ func (s *OpenAIGatewayService) handleOpenAIAccountUpstreamError(ctx context.Cont return false } stateCtx = withTempUnschedulableModel(stateCtx, canonicalModel) - preferredModel := "" - if len(canonicalModel) > 0 { - preferredModel = canonicalModel[0] - } - if statusCode == http.StatusTooManyRequests && s.codexQuotaOverdraft != nil && - s.codexQuotaOverdraft.HandleQuota429(stateCtx, account, headers, responseBody, preferredModel) { + if s.handleCodexQuotaOverdraftUpstream429(stateCtx, account, statusCode, headers, responseBody, canonicalModel) { return false } if s.rateLimitService != nil && len(canonicalModel) > 0 && s.rateLimitService.HandleUpstreamModelNotFound(stateCtx, account, canonicalModel[0], statusCode, responseBody) { diff --git a/backend/internal/service/openai_account_scheduler.go b/backend/internal/service/openai_account_scheduler.go index a5429668c..239e7854f 100644 --- a/backend/internal/service/openai_account_scheduler.go +++ b/backend/internal/service/openai_account_scheduler.go @@ -2302,9 +2302,7 @@ func (s *OpenAIGatewayService) isOpenAIAccountTransportCompatible(account *Accou func (s *OpenAIGatewayService) ReportOpenAIAccountScheduleResult(accountID int64, model string, success bool, firstTokenMs *int, requestCtx ...context.Context) { if success { s.clearOpenAIAccountModelTransientState(accountID, normalizeOpenAIAccountModelTransientModel(model)) - if len(requestCtx) > 0 && s.codexQuotaOverdraft != nil && codexQuotaOverdraftWasInjected(requestCtx[0], accountID) { - s.codexQuotaOverdraft.ObserveBusinessSuccessByID(accountID, model) - } + s.observeCodexQuotaOverdraftScheduleSuccess(accountID, model, requestCtx) } scheduler := s.getOpenAIAccountScheduler(context.Background()) if scheduler == nil { diff --git a/backend/internal/service/openai_codex_quota_overdraft_hooks.go b/backend/internal/service/openai_codex_quota_overdraft_hooks.go new file mode 100644 index 000000000..40fbd181c --- /dev/null +++ b/backend/internal/service/openai_codex_quota_overdraft_hooks.go @@ -0,0 +1,122 @@ +package service + +import ( + "context" + "fmt" + "net/http" + "time" + + "github.com/Wei-Shaw/sub2api/internal/config" +) + +func codexQuotaOverdraftBypassesSchedulingThreshold(ctx context.Context, account *Account) bool { + return codexQuotaOverdraftSchedulingEnabled(ctx) && isCodexQuotaOverdraftAccount(account) && + codexQuotaOverdraftSchedulingAllowed(account, time.Now().UTC()) +} + +func (s *RateLimitService) notifyCodexQuotaOverdraftAwareSchedulingBlock( + account *Account, + until time.Time, +) { + if !CodexQuotaOverdraftEnabled() || !isCodexQuotaOverdraftAccount(account) { + s.notifyAccountSchedulingBlocked(account, until, "account_scheduling_threshold") + } +} + +func (s *OpenAIGatewayService) listCodexQuotaOverdraftSchedulableAccounts( + ctx context.Context, + groupID *int64, + platform string, +) ([]Account, bool, error) { + if !CodexQuotaOverdraftSchedulingEnabled(ctx) || platform != PlatformOpenAI || s.accountRepo == nil { + return nil, false, nil + } + var accounts []Account + var err error + if s.cfg != nil && s.cfg.RunMode == config.RunModeSimple { + accounts, err = s.accountRepo.ListSchedulableByPlatform(ctx, platform) + } else if groupID != nil { + accounts, err = s.accountRepo.ListSchedulableByGroupIDAndPlatform(ctx, *groupID, platform) + } else { + accounts, err = s.accountRepo.ListSchedulableUngroupedByPlatform(ctx, platform) + } + if err != nil { + return nil, true, fmt.Errorf("query overdraft accounts failed: %w", err) + } + accounts = normalizeCodexQuotaOverdraftAccountsForScheduling(ctx, accounts) + return s.filterOpenAIAccountsBySchedulingThreshold(ctx, accounts), true, nil +} + +func (s *OpenAIGatewayService) handleCodexQuotaOverdraftUpstream429( + ctx context.Context, + account *Account, + statusCode int, + headers http.Header, + responseBody []byte, + canonicalModel []string, +) bool { + if statusCode != http.StatusTooManyRequests || s.codexQuotaOverdraft == nil { + return false + } + preferredModel := "" + if len(canonicalModel) > 0 { + preferredModel = canonicalModel[0] + } + return s.codexQuotaOverdraft.HandleQuota429(ctx, account, headers, responseBody, preferredModel) +} + +func (s *OpenAIGatewayService) processCodexQuotaOverdraftUsageSnapshot( + ctx context.Context, + accountID int64, + now time.Time, + updates map[string]any, +) { + persistSnapshot := codexQuotaOverdraftSnapshotPrearmReached(updates) || s.getCodexSnapshotThrottle().Allow(accountID, now) + businessSuccess := codexQuotaOverdraftWasInjected(ctx, accountID) + + go func() { + updateCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + var account *Account + if !persistSnapshot && s.codexQuotaOverdraft != nil { + current, err := s.accountRepo.GetByID(updateCtx, accountID) + if err == nil && current != nil { + account = current + state, hasState := codexQuotaOverdraftStateFromAccount(current) + _, wasExhausted := codexQuotaOverdraftSignalFromAccount(current, state, now) + persistSnapshot = wasExhausted || hasState && state.Status != codexQuotaOverdraftProbeRecovered + } + } + if persistSnapshot { + if err := s.accountRepo.UpdateExtra(updateCtx, accountID, updates); err != nil { + return + } + } + if s.codexQuotaOverdraft == nil { + return + } + if account == nil { + current, err := s.accountRepo.GetByID(updateCtx, accountID) + if err != nil || current == nil { + return + } + account = current + } + mergeAccountExtra(account, updates) + if businessSuccess { + s.codexQuotaOverdraft.observeBusinessSuccess(account, "") + } else { + s.codexQuotaOverdraft.observeAccount(account, "") + } + }() +} + +func (s *OpenAIGatewayService) observeCodexQuotaOverdraftScheduleSuccess( + accountID int64, + model string, + requestCtx []context.Context, +) { + if len(requestCtx) > 0 && s.codexQuotaOverdraft != nil && codexQuotaOverdraftWasInjected(requestCtx[0], accountID) { + s.codexQuotaOverdraft.ObserveBusinessSuccessByID(accountID, model) + } +} diff --git a/backend/internal/service/openai_codex_quota_overdraft_integration.go b/backend/internal/service/openai_codex_quota_overdraft_integration.go new file mode 100644 index 000000000..36388781a --- /dev/null +++ b/backend/internal/service/openai_codex_quota_overdraft_integration.go @@ -0,0 +1,44 @@ +package service + +func (s *OpenAIGatewayService) SetCodexQuotaOverdraftCoordinator(coordinator *CodexQuotaOverdraftCoordinator) { + if s != nil { + s.codexQuotaOverdraft = coordinator + } +} + +func (s *AccountUsageService) SetCodexQuotaOverdraftCoordinator(coordinator *CodexQuotaOverdraftCoordinator) { + if s != nil { + s.codexQuotaOverdraft = coordinator + } +} + +// codexQuotaOverdraftCoordinator returns the single coordinator owned by the +// OpenAI gateway. Keeping construction here avoids adding a fork-only provider +// to the generated Wire graph. +func (s *OpenAIGatewayService) codexQuotaOverdraftCoordinator( + tlsFPProfileService *TLSFingerprintProfileService, +) *CodexQuotaOverdraftCoordinator { + if s == nil { + return nil + } + s.codexQuotaOverdraftOnce.Do(func() { + if s.codexQuotaOverdraft != nil { + return + } + var tempUnschedCache TempUnschedCache + if s.rateLimitService != nil { + tempUnschedCache = s.rateLimitService.tempUnschedCache + } + s.codexQuotaOverdraft = NewCodexQuotaOverdraftCoordinator( + s.accountRepo, + s.httpUpstream, + s.openAITokenProvider, + tlsFPProfileService, + s.cfg, + tempUnschedCache, + s, + s.rateLimitService, + ) + }) + return s.codexQuotaOverdraft +} diff --git a/backend/internal/service/openai_codex_quota_overdraft_test.go b/backend/internal/service/openai_codex_quota_overdraft_test.go index 86825cdb7..0b62c1c01 100644 --- a/backend/internal/service/openai_codex_quota_overdraft_test.go +++ b/backend/internal/service/openai_codex_quota_overdraft_test.go @@ -184,3 +184,14 @@ func TestRateLimitServiceCodexQuotaOverdraftDoesNotCreateRuntimeThresholdBlock(t func int64PtrForCodexQuotaOverdraftTest(value int64) *int64 { return &value } + +func TestCodexQuotaOverdraftCoordinatorOwnedByGateway(t *testing.T) { + gateway := &OpenAIGatewayService{} + + first := gateway.codexQuotaOverdraftCoordinator(nil) + second := gateway.codexQuotaOverdraftCoordinator(&TLSFingerprintProfileService{}) + + require.NotNil(t, first) + require.Same(t, first, second) + require.Same(t, gateway, first.runtimeBlocker) +} diff --git a/backend/internal/service/openai_gateway_scheduling.go b/backend/internal/service/openai_gateway_scheduling.go index ffd0e4fc7..2a6bf9ac6 100644 --- a/backend/internal/service/openai_gateway_scheduling.go +++ b/backend/internal/service/openai_gateway_scheduling.go @@ -445,8 +445,7 @@ func shouldAutoPauseOpenAIAccountByQuota(ctx context.Context, account *Account) if account == nil || !account.IsOpenAI() { return false, openAIQuotaAutoPauseDecision{} } - if codexQuotaOverdraftSchedulingEnabled(ctx) && isCodexQuotaOverdraftAccount(account) && - codexQuotaOverdraftSchedulingAllowed(account, time.Now().UTC()) { + if codexQuotaOverdraftBypassesSchedulingThreshold(ctx, account) { return false, openAIQuotaAutoPauseDecision{} } // Per-account explicit-disable flags must take precedence over the global default. @@ -1309,21 +1308,8 @@ func (s *OpenAIGatewayService) selectAccountWithLoadAwareness(ctx context.Contex func (s *OpenAIGatewayService) listSchedulableAccounts(ctx context.Context, groupID *int64, platform string) ([]Account, error) { platform = normalizeOpenAICompatiblePlatform(platform) - if CodexQuotaOverdraftSchedulingEnabled(ctx) && platform == PlatformOpenAI && s.accountRepo != nil { - var accounts []Account - var err error - if s.cfg != nil && s.cfg.RunMode == config.RunModeSimple { - accounts, err = s.accountRepo.ListSchedulableByPlatform(ctx, platform) - } else if groupID != nil { - accounts, err = s.accountRepo.ListSchedulableByGroupIDAndPlatform(ctx, *groupID, platform) - } else { - accounts, err = s.accountRepo.ListSchedulableUngroupedByPlatform(ctx, platform) - } - if err != nil { - return nil, fmt.Errorf("query overdraft accounts failed: %w", err) - } - accounts = normalizeCodexQuotaOverdraftAccountsForScheduling(ctx, accounts) - return s.filterOpenAIAccountsBySchedulingThreshold(ctx, accounts), nil + if accounts, handled, err := s.listCodexQuotaOverdraftSchedulableAccounts(ctx, groupID, platform); handled { + return accounts, err } if s.schedulerSnapshot != nil { accounts, _, err := s.schedulerSnapshot.ListSchedulableAccounts(ctx, groupID, platform, false) @@ -1542,8 +1528,7 @@ func (s *OpenAIGatewayService) isOpenAIAccountBlockedBySchedulingThreshold(ctx c if s == nil || s.rateLimitService == nil || account == nil { return false } - if codexQuotaOverdraftSchedulingEnabled(ctx) && isCodexQuotaOverdraftAccount(account) && - codexQuotaOverdraftSchedulingAllowed(account, time.Now().UTC()) { + if codexQuotaOverdraftBypassesSchedulingThreshold(ctx, account) { return false } return s.rateLimitService.ApplyAccountSchedulingThreshold(ctx, account) diff --git a/backend/internal/service/openai_gateway_service.go b/backend/internal/service/openai_gateway_service.go index 0c4e18a12..a134c76e1 100644 --- a/backend/internal/service/openai_gateway_service.go +++ b/backend/internal/service/openai_gateway_service.go @@ -436,6 +436,7 @@ type OpenAIGatewayService struct { openaiWSPoolOnce sync.Once openaiWSStateStoreOnce sync.Once openaiSchedulerOnce sync.Once + codexQuotaOverdraftOnce sync.Once openaiProxyStreamCircuitOnce sync.Once openaiWSPassthroughDialerOnce sync.Once openaiModelTransientOnce sync.Once @@ -470,12 +471,6 @@ type OpenAIGatewayService struct { openaiCodexTurnStateWrites atomic.Uint64 } -func (s *OpenAIGatewayService) SetCodexQuotaOverdraftCoordinator(coordinator *CodexQuotaOverdraftCoordinator) { - if s != nil { - s.codexQuotaOverdraft = coordinator - } -} - // NewOpenAIGatewayService creates a new OpenAIGatewayService func NewOpenAIGatewayService( accountRepo AccountRepository, diff --git a/backend/internal/service/openai_gateway_usage.go b/backend/internal/service/openai_gateway_usage.go index 43973f90c..9dfc1724c 100644 --- a/backend/internal/service/openai_gateway_usage.go +++ b/backend/internal/service/openai_gateway_usage.go @@ -1028,44 +1028,7 @@ func (s *OpenAIGatewayService) updateCodexUsageSnapshot(ctx context.Context, acc if len(updates) == 0 { return } - persistSnapshot := codexQuotaOverdraftSnapshotPrearmReached(updates) || s.getCodexSnapshotThrottle().Allow(accountID, now) - businessSuccess := codexQuotaOverdraftWasInjected(ctx, accountID) - - go func() { - updateCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - var account *Account - if !persistSnapshot && s.codexQuotaOverdraft != nil { - current, err := s.accountRepo.GetByID(updateCtx, accountID) - if err == nil && current != nil { - account = current - state, hasState := codexQuotaOverdraftStateFromAccount(current) - _, wasExhausted := codexQuotaOverdraftSignalFromAccount(current, state, now) - persistSnapshot = wasExhausted || hasState && state.Status != codexQuotaOverdraftProbeRecovered - } - } - if persistSnapshot { - if err := s.accountRepo.UpdateExtra(updateCtx, accountID, updates); err != nil { - return - } - } - if s.codexQuotaOverdraft == nil { - return - } - if account == nil { - current, err := s.accountRepo.GetByID(updateCtx, accountID) - if err != nil || current == nil { - return - } - account = current - } - mergeAccountExtra(account, updates) - if businessSuccess { - s.codexQuotaOverdraft.observeBusinessSuccess(account, "") - } else { - s.codexQuotaOverdraft.observeAccount(account, "") - } - }() + s.processCodexQuotaOverdraftUsageSnapshot(ctx, accountID, now, updates) } func (s *OpenAIGatewayService) UpdateCodexUsageSnapshotFromHeaders(ctx context.Context, accountID int64, headers http.Header) { diff --git a/backend/internal/service/ratelimit_service.go b/backend/internal/service/ratelimit_service.go index b56b2b015..e51ce491a 100644 --- a/backend/internal/service/ratelimit_service.go +++ b/backend/internal/service/ratelimit_service.go @@ -149,8 +149,7 @@ func (s *RateLimitService) ApplyAccountSchedulingThreshold(ctx context.Context, if !account.IsActive() || !account.Schedulable { return false } - if codexQuotaOverdraftSchedulingEnabled(ctx) && isCodexQuotaOverdraftAccount(account) && - codexQuotaOverdraftSchedulingAllowed(account, time.Now().UTC()) { + if codexQuotaOverdraftBypassesSchedulingThreshold(ctx, account) { return false } @@ -180,12 +179,7 @@ func (s *RateLimitService) ApplyAccountSchedulingThreshold(ctx context.Context, account.TempUnschedulableUntil = cloneTimePtr(decision.Until) account.TempUnschedulableReason = reason - // When overdraft is globally enabled, the persisted threshold pause remains - // visible to ordinary endpoints, but it must not create an untyped runtime - // blocker that would also hide the account from an overdraft-marked request. - if !CodexQuotaOverdraftEnabled() || !isCodexQuotaOverdraftAccount(account) { - s.notifyAccountSchedulingBlocked(account, *decision.Until, "account_scheduling_threshold") - } + s.notifyCodexQuotaOverdraftAwareSchedulingBlock(account, *decision.Until) if err := s.accountRepo.SetTempUnschedulable(ctx, account.ID, *decision.Until, reason); err != nil { slog.Warn("account_scheduling_threshold_set_temp_unsched_failed", diff --git a/backend/internal/service/wire.go b/backend/internal/service/wire.go index 1e853b64c..007f3ba6c 100644 --- a/backend/internal/service/wire.go +++ b/backend/internal/service/wire.go @@ -202,7 +202,6 @@ func ProvideAccountUsageService( identityCache IdentityCache, tlsFPProfileService *TLSFingerprintProfileService, openAIGatewayService *OpenAIGatewayService, - codexQuotaOverdraft *CodexQuotaOverdraftCoordinator, ) *AccountUsageService { service := NewAccountUsageService( accountRepo, @@ -218,36 +217,10 @@ func ProvideAccountUsageService( tlsFPProfileService, ) service.agentIdentityWS = openAIGatewayService - service.SetCodexQuotaOverdraftCoordinator(codexQuotaOverdraft) + service.SetCodexQuotaOverdraftCoordinator(openAIGatewayService.codexQuotaOverdraftCoordinator(tlsFPProfileService)) return service } -func ProvideCodexQuotaOverdraftCoordinator( - accountRepo AccountRepository, - httpUpstream HTTPUpstream, - openAITokenProvider *OpenAITokenProvider, - tlsFPProfileService *TLSFingerprintProfileService, - cfg *config.Config, - tempUnschedCache TempUnschedCache, - runtimeBlocker AccountRuntimeBlocker, - rateLimitService *RateLimitService, -) *CodexQuotaOverdraftCoordinator { - coordinator := NewCodexQuotaOverdraftCoordinator( - accountRepo, - httpUpstream, - openAITokenProvider, - tlsFPProfileService, - cfg, - tempUnschedCache, - runtimeBlocker, - rateLimitService, - ) - if gateway, ok := runtimeBlocker.(*OpenAIGatewayService); ok { - gateway.SetCodexQuotaOverdraftCoordinator(coordinator) - } - return coordinator -} - func ProvideAccountTestService( accountRepo AccountRepository, geminiTokenProvider *GeminiTokenProvider, @@ -259,7 +232,6 @@ func ProvideAccountTestService( tlsFPProfileService *TLSFingerprintProfileService, openAIGatewayService *OpenAIGatewayService, settingService *SettingService, - codexQuotaOverdraft *CodexQuotaOverdraftCoordinator, ) *AccountTestService { service := NewAccountTestService( accountRepo, @@ -273,7 +245,7 @@ func ProvideAccountTestService( ) service.agentIdentityWS = openAIGatewayService service.SetSettingService(settingService) - service.SetCodexQuotaOverdraftCoordinator(codexQuotaOverdraft) + service.SetCodexQuotaOverdraftCoordinator(openAIGatewayService.codexQuotaOverdraftCoordinator(tlsFPProfileService)) return service } @@ -829,7 +801,6 @@ var ProviderSet = wire.NewSet( ProvideClaudeTokenProvider, NewAntigravityGatewayService, ProvideRateLimitService, - ProvideCodexQuotaOverdraftCoordinator, ProvideAccountUsageService, ProvideAccountTestService, ProvideUpstreamBillingProbeService, diff --git a/frontend/src/components/account/AccountUsageCell.vue b/frontend/src/components/account/AccountUsageCell.vue index 96deb961c..cb8e082ba 100644 --- a/frontend/src/components/account/AccountUsageCell.vue +++ b/frontend/src/components/account/AccountUsageCell.vue @@ -119,22 +119,7 @@