From e91ba0a8dca3efb65d55e7f4b31597fbe432f658 Mon Sep 17 00:00:00 2001 From: "bmiddle1.backend.zhangfucheng" <5286482+zhang_fucheng@user.noreply.gitee.com> Date: Fri, 14 Aug 2026 10:20:00 +0800 Subject: [PATCH] feat: add Codex quota overdraft support --- CODEX_QUOTA_OVERDRAFT_CUSTOMIZATION.md | 143 +++ backend/cmd/server/wire_gen.go | 3 +- backend/internal/config/config.go | 4 + .../handler/openai_chat_completions.go | 3 +- .../handler/openai_gateway_handler.go | 13 +- backend/internal/repository/account_repo.go | 78 +- ...ccount_repo_schedulable_projection_test.go | 31 + .../internal/service/account_usage_service.go | 43 +- .../openai_account_runtime_block_fastpath.go | 8 + .../service/openai_account_scheduler.go | 2 +- .../service/openai_codex_quota_overdraft.go | 210 ++++ .../openai_codex_quota_overdraft_probe.go | 942 ++++++++++++++++++ ...openai_codex_quota_overdraft_probe_test.go | 319 ++++++ .../openai_codex_quota_overdraft_test.go | 142 +++ .../service/openai_gateway_forward.go | 1 + .../service/openai_gateway_passthrough.go | 1 + .../service/openai_gateway_scheduling.go | 29 + .../service/openai_gateway_service.go | 8 + .../internal/service/openai_gateway_usage.go | 32 +- .../service/openai_ws_forwarder_ingress.go | 3 +- .../service/openai_ws_forwarder_support.go | 1 + .../service/openai_ws_forwarder_v2.go | 1 + .../openai_ws_v2_passthrough_adapter.go | 2 + backend/internal/service/ratelimit_service.go | 11 +- backend/internal/service/wire.go | 29 + deploy/config.example.yaml | 5 + .../components/account/AccountUsageCell.vue | 89 +- .../components/account/UsageProgressBar.vue | 37 + .../__tests__/AccountUsageCell.spec.ts | 45 + .../__tests__/UsageProgressBar.spec.ts | 22 + frontend/src/i18n/locales/en/dashboard.ts | 8 + frontend/src/i18n/locales/zh/dashboard.ts | 8 + frontend/src/types/index.ts | 25 + 33 files changed, 2262 insertions(+), 36 deletions(-) create mode 100644 CODEX_QUOTA_OVERDRAFT_CUSTOMIZATION.md create mode 100644 backend/internal/service/openai_codex_quota_overdraft.go create mode 100644 backend/internal/service/openai_codex_quota_overdraft_probe.go create mode 100644 backend/internal/service/openai_codex_quota_overdraft_probe_test.go create mode 100644 backend/internal/service/openai_codex_quota_overdraft_test.go diff --git a/CODEX_QUOTA_OVERDRAFT_CUSTOMIZATION.md b/CODEX_QUOTA_OVERDRAFT_CUSTOMIZATION.md new file mode 100644 index 000000000..3821cead8 --- /dev/null +++ b/CODEX_QUOTA_OVERDRAFT_CUSTOMIZATION.md @@ -0,0 +1,143 @@ +# Codex 5h / 7d 额度透支定制记录 + +## 基本信息 + +- 实施基线：`fbfdcef81` +- 参考实现： +- 功能开关：`gateway.codex_quota_overdraft_enabled` +- 代码默认值：关闭；`deploy/config.example.yaml` 部署示例默认开启 + +## PostgreSQL 兼容性修复 + +探测计划使用原子 claim 写入 `accounts.extra`。`jsonb_build_object` 的键参数必须显式 +转换为 `text`，否则 PostgreSQL 会报 `could not determine data type of parameter $1`，导致 +所有探测在启动前失败。对应 SQL 使用： + +```sql +jsonb_build_object($1::text, $2::jsonb) +``` + +## 启用方式 + +在实际使用的 `config.yaml` 中加入： + +```yaml +gateway: + codex_quota_overdraft_enabled: true +``` + +修改配置后重启 sub2api。出现上游兼容问题时可改为 `false` 并重启，立即恢复官方调度和请求行为。 + +## 完整行为 + +开启后，对 OpenAI OAuth 普通账号和 Agent Identity 账号生效，Shadow 账号除外。支持路径： + +- `/v1/responses` +- `/v1/chat/completions` 转换到 Responses 的请求 +- `/v1/messages` 转换到 Responses 的请求 +- Responses WebSocket v2 首轮和后续轮 + +每次最终发往上游前，如果 `input` 最后一项是用户消息，则追加一对与参考项目相同的无操作 `custom_tool_call` 和 `custom_tool_call_output`。请求体超过 32 MiB、JSON 无效、形状不匹配或已经注入时保持原样。 + +5h 或 7d 额度首次达到 100%，或者收到包含明确额度证据的真实 429 时，按参考项目执行复核： + +1. 同一额度周期只启动一个探测计划，状态持久化在 `accounts.extra.codex_quota_overdraft_probe`，多实例使用数据库原子 claim 去重。 +2. 最多执行 5 次真实 Responses 请求，每次独立 20 秒超时。 +3. 模型按“首选模型、`gpt-5.5`、`gpt-5.4-mini`”循环。例如首选为 `gpt-5.4` 时，顺序是 `gpt-5.4`、`gpt-5.5`、`gpt-5.4-mini`、`gpt-5.4`、`gpt-5.5`。 +4. 任意一次返回有效的 `response.completed` 或 `response.output_item.done` 即判定 `passed`，账号继续调度并开始透支统计。 +5. 只有 5 次均返回明确 `quota_limited` 才判定 `failed`，账号暂停到对应额度恢复时间；5h 和 7d 同时耗尽时取最晚恢复时间。 +6. 网络错误、超时、5xx、普通瞬时 429、无效响应、认证依赖不可用均判定 `inconclusive`，不会误停账号，一分钟后允许重试。 +7. 401/403 仍交给原有认证失效策略处理；账号禁用、过载、代理/传输故障、模型级冷却及其他临时不可调度原因不被绕过。 +8. 额度恢复后状态改为 `recovered`，清理本功能或额度阈值产生的暂停；5h 与 7d 分别维护透支起点，不会因另一窗口后来耗尽而重置已有统计。 + +状态值包括 `pending`、`passed`、`failed`、`inconclusive`、`recovered`。账号用量页面显示探测状态、尝试次数、额度周期、透支期成功请求数、Token、账号金额及预计恢复时间。 + +不需要新增数据库表或执行 schema migration；周期状态保存在现有 `accounts.extra` JSONB 字段，透支统计读取现有 `usage_logs`。 + +`/responses/compact`、生图请求、Embedding、Count Tokens、Live 等端点不启用额度透支。 + +## 修改文件 + +后端配置、路由和调度： + +- `backend/internal/config/config.go` +- `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_schedulable_projection_test.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_probe.go` +- `backend/internal/service/openai_codex_quota_overdraft_test.go` +- `backend/internal/service/openai_codex_quota_overdraft_probe_test.go` +- `backend/internal/service/openai_gateway_forward.go` +- `backend/internal/service/openai_gateway_passthrough.go` +- `backend/internal/service/openai_gateway_scheduling.go` +- `backend/internal/service/openai_gateway_service.go` +- `backend/internal/service/openai_gateway_usage.go` +- `backend/internal/service/openai_ws_forwarder_ingress.go` +- `backend/internal/service/openai_ws_forwarder_support.go` +- `backend/internal/service/openai_ws_forwarder_v2.go` +- `backend/internal/service/openai_ws_v2_passthrough_adapter.go` +- `backend/internal/service/ratelimit_service.go` +- `backend/internal/service/account_usage_service.go` +- `backend/internal/service/wire.go` +- `backend/cmd/server/wire_gen.go` + +前端、配置示例和测试： + +- `frontend/src/types/index.ts` +- `frontend/src/components/account/UsageProgressBar.vue` +- `frontend/src/components/account/AccountUsageCell.vue` +- `frontend/src/components/account/__tests__/UsageProgressBar.spec.ts` +- `frontend/src/components/account/__tests__/AccountUsageCell.spec.ts` +- `frontend/src/i18n/locales/zh/dashboard.ts` +- `frontend/src/i18n/locales/en/dashboard.ts` +- `deploy/config.example.yaml` + +## 官方更新后的重新实施 + +普通 `git pull` 不会静默丢弃未提交修改：能自动合并时会保留，冲突时会停止。安装脚本若强制重置、删除目录后重拉或重新解压，则可能覆盖本定制。 + +1. 更新前保存补丁：`git diff > codex-quota-overdraft.patch` +2. 拉取官方代码并处理冲突。 +3. 尝试恢复补丁：`git apply --3way codex-quota-overdraft.patch` +4. 无法自动应用时，按本文件“完整行为”和“修改文件”逐项重新接入。 +5. 在 `backend` 执行 `go generate ./cmd/server`，重新生成 Wire。 +6. 确认实际 `config.yaml` 仍有 `gateway.codex_quota_overdraft_enabled: true`。 +7. 运行验证： + +```bash +cd backend +gofmt -w internal/service/openai_codex_quota_overdraft*.go +go test -tags=unit ./internal/repository ./internal/handler ./internal/service \ + -run 'CodexQuotaOverdraft|AccountSchedulingThreshold|SchedulerSnapshot|SchedulableAccountQueryScopesCodex' -count=1 +go test ./internal/repository ./internal/handler ./internal/service ./cmd/server -run '^$' -count=1 + +cd ../frontend +pnpm install --frozen-lockfile +pnpm run typecheck +pnpm exec vitest run \ + src/components/account/__tests__/UsageProgressBar.spec.ts \ + src/components/account/__tests__/AccountUsageCell.spec.ts + +cd .. +git diff --check +``` + +人工核对所有 Responses 出站路径仍在最终发送前注入，并检查日志关键字： + +```text +codex_quota_overdraft_probe_passed +codex_quota_overdraft_probe_failed +codex_quota_overdraft_probe_inconclusive +``` + +数据库状态可这样检查： + +```sql +SELECT id, name, extra->'codex_quota_overdraft_probe' +FROM accounts +WHERE platform = 'openai' AND type = 'oauth'; +``` diff --git a/backend/cmd/server/wire_gen.go b/backend/cmd/server/wire_gen.go index ab9f4530b..83b97e68f 100644 --- a/backend/cmd/server/wire_gen.go +++ b/backend/cmd/server/wire_gen.go @@ -199,7 +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() - accountUsageService := service.ProvideAccountUsageService(accountRepository, usageLogRepository, claudeUsageFetcher, geminiQuotaService, antigravityQuotaFetcher, grokQuotaFetcher, grokQuotaService, openAIQuotaService, usageCache, identityCache, tlsFingerprintProfileService, openAIGatewayService) + 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) 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) diff --git a/backend/internal/config/config.go b/backend/internal/config/config.go index 0ed2b1721..81b257b8a 100644 --- a/backend/internal/config/config.go +++ b/backend/internal/config/config.go @@ -922,6 +922,9 @@ type GatewayConfig struct { // CodexImageGenerationBridgeEnabled: 是否为 Codex `/v1/responses` 自动注入 image_generation 工具和桥接指令。 // 默认关闭，避免纯文本 Codex 请求被意外改写；显式携带 image_generation 工具的请求仍按分组能力转发。 CodexImageGenerationBridgeEnabled bool `mapstructure:"codex_image_generation_bridge_enabled"` + // CodexQuotaOverdraftEnabled: 是否启用 OpenAI OAuth Codex 5h/7d 额度透支。 + // 包括请求注入、五次真实复核、调度门控和透支期用量统计；默认关闭。 + CodexQuotaOverdraftEnabled bool `mapstructure:"codex_quota_overdraft_enabled"` // ForcedCodexInstructionsTemplateFile: 服务端强制附加到 Codex 顶层 instructions 的模板文件路径。 // 模板渲染后会直接覆盖最终 instructions；若需要保留客户端 system 转换结果，请在模板中显式引用 {{ .ExistingInstructions }}。 ForcedCodexInstructionsTemplateFile string `mapstructure:"forced_codex_instructions_template_file"` @@ -2272,6 +2275,7 @@ func setDefaults() { viper.SetDefault("gateway.disable_codex_identity_enforcement", false) viper.SetDefault("gateway.disable_codex_originator_normalization", false) viper.SetDefault("gateway.codex_image_generation_bridge_enabled", false) + viper.SetDefault("gateway.codex_quota_overdraft_enabled", false) viper.SetDefault("gateway.openai_passthrough_allow_timeout_headers", false) viper.SetDefault("gateway.openai_compact_model", "gpt-5.4") viper.SetDefault("gateway.live.max_session_duration_seconds", 3600) diff --git a/backend/internal/handler/openai_chat_completions.go b/backend/internal/handler/openai_chat_completions.go index 42dc13112..17b479058 100644 --- a/backend/internal/handler/openai_chat_completions.go +++ b/backend/internal/handler/openai_chat_completions.go @@ -149,7 +149,8 @@ func (h *OpenAIGatewayHandler) ChatCompletions(c *gin.Context) { var oauth429FailoverState service.OpenAIOAuth429FailoverState // 分组利润控制：chat completions 文本入口请求级装门并固定 pricingAt。 - ccPricingCtx, pricingAt := h.gatewayService.WithOpenAIRequestPricingContext(c.Request.Context(), apiKey.GroupID) + ccPricingCtx := service.WithCodexQuotaOverdraftScheduling(c.Request.Context()) + ccPricingCtx, pricingAt := h.gatewayService.WithOpenAIRequestPricingContext(ccPricingCtx, apiKey.GroupID) c.Request = c.Request.WithContext(ccPricingCtx) for { diff --git a/backend/internal/handler/openai_gateway_handler.go b/backend/internal/handler/openai_gateway_handler.go index 26b5dd03c..b71abfe10 100644 --- a/backend/internal/handler/openai_gateway_handler.go +++ b/backend/internal/handler/openai_gateway_handler.go @@ -443,7 +443,11 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { // D 与计费高峰因子，选号、槽位终检与全部 failover 重入共用同一门与阈值。 // 生图意图只影响能力路由与图片计费，不关门：混合 /v1/responses 请求的 // token 计费部分仍受利润门保护，独立图片/视频端点才在门外。 - pricingCtx, pricingAt := h.gatewayService.WithOpenAIRequestPricingContext(c.Request.Context(), apiKey.GroupID) + requestCtx := c.Request.Context() + if !requireCompact && !imageIntent { + requestCtx = service.WithCodexQuotaOverdraftScheduling(requestCtx) + } + pricingCtx, pricingAt := h.gatewayService.WithOpenAIRequestPricingContext(requestCtx, apiKey.GroupID) c.Request = c.Request.WithContext(pricingCtx) for { @@ -1016,7 +1020,8 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) { effectiveMappedModel := preferredMappedModel // 分组利润控制：Messages 文本入口同样请求级装门并固定 pricingAt。 - msgPricingCtx, pricingAt := h.gatewayService.WithOpenAIRequestPricingContext(c.Request.Context(), apiKey.GroupID) + msgPricingCtx := service.WithCodexQuotaOverdraftScheduling(c.Request.Context()) + msgPricingCtx, pricingAt := h.gatewayService.WithOpenAIRequestPricingContext(msgPricingCtx, apiKey.GroupID) c.Request = c.Request.WithContext(msgPricingCtx) for { @@ -1735,6 +1740,10 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) { closeOpenAIClientWS(wsConn, coderws.StatusPolicyViolation, service.ImageGenerationPermissionMessage()) return } + if !imageIntent { + ctx = service.WithCodexQuotaOverdraftScheduling(ctx) + c.Request = c.Request.WithContext(ctx) + } // F5a: 握手层会话屏蔽检查。WS 握手无 body，显式标识仅来自握手 header // （session_id / conversation_id）；无标识则放行，连接内仍有本地 flag 兜底。 diff --git a/backend/internal/repository/account_repo.go b/backend/internal/repository/account_repo.go index 010442eb6..65a592fcb 100644 --- a/backend/internal/repository/account_repo.go +++ b/backend/internal/repository/account_repo.go @@ -56,6 +56,7 @@ var schedulerNeutralExtraKeyPrefixes = []string{ "codex_secondary_", "codex_5h_", "codex_7d_", + "codex_quota_overdraft_", "codex_reset_credit_", "passive_usage_", "upstream_billing_probe", @@ -1832,7 +1833,7 @@ func (r *accountRepository) BindGroups(ctx context.Context, accountID int64, gro } func (r *accountRepository) ListSchedulable(ctx context.Context) ([]service.Account, error) { - accounts, err := r.schedulableAccountsQuery(time.Now()).All(ctx) + accounts, err := r.schedulableAccountsQuery(ctx, time.Now()).All(ctx) if err != nil { return nil, err } @@ -1840,7 +1841,7 @@ func (r *accountRepository) ListSchedulable(ctx context.Context) ([]service.Acco } func (r *accountRepository) ListSchedulableAccountLoads(ctx context.Context) ([]service.AccountWithConcurrency, error) { - accounts, err := r.schedulableAccountsQuery(time.Now()). + accounts, err := r.schedulableAccountsQuery(ctx, time.Now()). Select( dbaccount.FieldID, dbaccount.FieldConcurrency, @@ -1866,12 +1867,12 @@ func (r *accountRepository) ListSchedulableAccountLoads(ctx context.Context) ([] return loads, nil } -func (r *accountRepository) schedulableAccountsQuery(now time.Time) *dbent.AccountQuery { +func (r *accountRepository) schedulableAccountsQuery(ctx context.Context, now time.Time) *dbent.AccountQuery { return r.client.Account.Query(). Where( dbaccount.StatusEQ(service.StatusActive), dbaccount.SchedulableEQ(true), - tempUnschedulablePredicate(), + tempUnschedulablePredicate(ctx), notExpiredPredicate(now), dbaccount.Or(dbaccount.OverloadUntilIsNil(), dbaccount.OverloadUntilLTE(now)), dbaccount.Or(dbaccount.RateLimitResetAtIsNil(), dbaccount.RateLimitResetAtLTE(now)), @@ -1977,7 +1978,7 @@ func (r *accountRepository) ListSchedulableByPlatform(ctx context.Context, platf dbaccount.PlatformEQ(platform), dbaccount.StatusEQ(service.StatusActive), dbaccount.SchedulableEQ(true), - tempUnschedulablePredicate(), + tempUnschedulablePredicate(ctx), notExpiredPredicate(now), dbaccount.Or(dbaccount.OverloadUntilIsNil(), dbaccount.OverloadUntilLTE(now)), dbaccount.Or(dbaccount.RateLimitResetAtIsNil(), dbaccount.RateLimitResetAtLTE(now)), @@ -2011,7 +2012,7 @@ func (r *accountRepository) ListSchedulableByPlatforms(ctx context.Context, plat dbaccount.PlatformIn(platforms...), dbaccount.StatusEQ(service.StatusActive), dbaccount.SchedulableEQ(true), - tempUnschedulablePredicate(), + tempUnschedulablePredicate(ctx), notExpiredPredicate(now), dbaccount.Or(dbaccount.OverloadUntilIsNil(), dbaccount.OverloadUntilLTE(now)), dbaccount.Or(dbaccount.RateLimitResetAtIsNil(), dbaccount.RateLimitResetAtLTE(now)), @@ -2032,7 +2033,7 @@ func (r *accountRepository) ListSchedulableUngroupedByPlatform(ctx context.Conte dbaccount.StatusEQ(service.StatusActive), dbaccount.SchedulableEQ(true), dbaccount.Not(dbaccount.HasAccountGroups()), - tempUnschedulablePredicate(), + tempUnschedulablePredicate(ctx), notExpiredPredicate(now), dbaccount.Or(dbaccount.OverloadUntilIsNil(), dbaccount.OverloadUntilLTE(now)), dbaccount.Or(dbaccount.RateLimitResetAtIsNil(), dbaccount.RateLimitResetAtLTE(now)), @@ -2056,7 +2057,7 @@ func (r *accountRepository) ListSchedulableUngroupedByPlatforms(ctx context.Cont dbaccount.StatusEQ(service.StatusActive), dbaccount.SchedulableEQ(true), dbaccount.Not(dbaccount.HasAccountGroups()), - tempUnschedulablePredicate(), + tempUnschedulablePredicate(ctx), notExpiredPredicate(now), dbaccount.Or(dbaccount.OverloadUntilIsNil(), dbaccount.OverloadUntilLTE(now)), dbaccount.Or(dbaccount.RateLimitResetAtIsNil(), dbaccount.RateLimitResetAtLTE(now)), @@ -2592,6 +2593,49 @@ func (r *accountRepository) UpdateExtra(ctx context.Context, id int64, updates m return nil } +// ClaimCodexQuotaOverdraftProbe atomically reserves one quota cycle. This +// prevents duplicate five-request probe plans across multiple 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 + OR ( + extra #>> '{codex_quota_overdraft_probe,status}' = 'inconclusive' + AND COALESCE(NULLIF(extra #>> '{codex_quota_overdraft_probe,retry_at}', '')::timestamptz, '1970-01-01'::timestamptz) <= NOW() + ) + OR ( + extra #>> '{codex_quota_overdraft_probe,status}' = 'pending' + AND COALESCE(NULLIF(extra #>> '{codex_quota_overdraft_probe,started_at}', '')::timestamptz, '1970-01-01'::timestamptz) <= NOW() - INTERVAL '2 minutes' + ) + ) + `, 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 +} + // UpdateUpstreamBillingProbeSnapshot stores a probe result only while the // network identity used by that probe is still current. func (r *accountRepository) UpdateUpstreamBillingProbeSnapshot( @@ -3029,7 +3073,7 @@ func (r *accountRepository) queryAccountsByGroup(ctx context.Context, groupID in if !opts.ignoreTransientState { now := time.Now() preds = append(preds, - tempUnschedulablePredicate(), + tempUnschedulablePredicate(ctx), notExpiredPredicate(now), dbaccount.Or(dbaccount.OverloadUntilIsNil(), dbaccount.OverloadUntilLTE(now)), dbaccount.Or(dbaccount.RateLimitResetAtIsNil(), dbaccount.RateLimitResetAtLTE(now)), @@ -3134,13 +3178,23 @@ func (r *accountRepository) accountsToService(ctx context.Context, accounts []*d return outAccounts, nil } -func tempUnschedulablePredicate() dbpredicate.Account { +func tempUnschedulablePredicate(ctx context.Context) dbpredicate.Account { return dbpredicate.Account(func(s *entsql.Selector) { col := s.C("temp_unschedulable_until") - s.Where(entsql.Or( + predicates := []*entsql.Predicate{ 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+`"`), + )) + } + s.Where(entsql.Or(predicates...)) }) } diff --git a/backend/internal/repository/account_repo_schedulable_projection_test.go b/backend/internal/repository/account_repo_schedulable_projection_test.go index 76d7e0bdc..328e9f285 100644 --- a/backend/internal/repository/account_repo_schedulable_projection_test.go +++ b/backend/internal/repository/account_repo_schedulable_projection_test.go @@ -9,6 +9,7 @@ import ( "github.com/DATA-DOG/go-sqlmock" dbent "github.com/Wei-Shaw/sub2api/ent" _ "github.com/Wei-Shaw/sub2api/ent/runtime" + "github.com/Wei-Shaw/sub2api/internal/service" "github.com/stretchr/testify/require" "entgo.io/ent/dialect" @@ -80,3 +81,33 @@ func TestListSchedulableAccountLoadsUsesSingleProjectionQuery(t *testing.T) { require.True(t, hasOrder, "projection query must preserve schedulable account order: %s", normalized) require.Contains(t, orderClause, `"priority" ASC`) } + +func TestSchedulableAccountQueryScopesCodexQuotaOverdraftToMarkedContext(t *testing.T) { + service.SetCodexQuotaOverdraftEnabled(true) + t.Cleanup(func() { service.SetCodexQuotaOverdraftEnabled(false) }) + + buildQuery := func(ctx context.Context) string { + var capturedSQL string + db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(captureEntQueryMatcher{actual: &capturedSQL})) + require.NoError(t, err) + t.Cleanup(func() { _ = db.Close() }) + driver := entsql.OpenDB(dialect.Postgres, db) + client := dbent.NewClient(dbent.Driver(driver)) + t.Cleanup(func() { _ = client.Close() }) + repo := newAccountRepositoryWithSQL(client, db, nil) + mock.ExpectQuery("schedulable query").WillReturnRows(sqlmock.NewRows([]string{"id", "concurrency", "load_factor"})) + _, err = repo.ListSchedulableAccountLoads(ctx) + require.NoError(t, err) + require.NoError(t, mock.ExpectationsWereMet()) + return normalizeSQLWhitespace(capturedSQL) + } + + ordinarySQL := buildQuery(context.Background()) + require.NotContains(t, ordinarySQL, `"temp_unschedulable_reason" LIKE`) + + overdraftSQL := buildQuery(service.WithCodexQuotaOverdraftScheduling(context.Background())) + require.Contains(t, overdraftSQL, `"temp_unschedulable_reason" LIKE`) + require.Contains(t, overdraftSQL, `"platform" =`) + require.Contains(t, overdraftSQL, `"type" =`) + require.Contains(t, overdraftSQL, `"parent_account_id" IS NULL`) +} diff --git a/backend/internal/service/account_usage_service.go b/backend/internal/service/account_usage_service.go index 084d153ff..e9a3e78fa 100644 --- a/backend/internal/service/account_usage_service.go +++ b/backend/internal/service/account_usage_service.go @@ -153,6 +153,10 @@ type UsageProgress struct { WindowStats *WindowStats `json:"window_stats,omitempty"` // 窗口期统计（从窗口开始到当前的使用量） UsedRequests int64 `json:"used_requests,omitempty"` LimitRequests int64 `json:"limit_requests,omitempty"` + OverdraftActive bool `json:"overdraft_active,omitempty"` + OverdraftStats *WindowStats `json:"overdraft_stats,omitempty"` + OverdraftStarted *time.Time `json:"overdraft_started_at,omitempty"` + OverdraftRecover *time.Time `json:"overdraft_recover_at,omitempty"` } // AntigravityModelQuota Antigravity 单个模型的配额信息 @@ -182,18 +186,19 @@ type AICredit struct { // UsageInfo 账号使用量信息 type UsageInfo struct { - Source string `json:"source,omitempty"` // "passive" or "active" - UpdatedAt *time.Time `json:"updated_at,omitempty"` // 更新时间 - FiveHour *UsageProgress `json:"five_hour"` // 5小时窗口 - SevenDay *UsageProgress `json:"seven_day,omitempty"` // 7天窗口 - SevenDaySonnet *UsageProgress `json:"seven_day_sonnet,omitempty"` // 7天Sonnet窗口 - SevenDayFable *UsageProgress `json:"seven_day_fable,omitempty"` // 7天Fable窗口（响应头 7d_oi） - GeminiSharedDaily *UsageProgress `json:"gemini_shared_daily,omitempty"` // Gemini shared pool RPD (Google One / Code Assist) - GeminiProDaily *UsageProgress `json:"gemini_pro_daily,omitempty"` // Gemini Pro 日配额 - GeminiFlashDaily *UsageProgress `json:"gemini_flash_daily,omitempty"` // Gemini Flash 日配额 - GeminiSharedMinute *UsageProgress `json:"gemini_shared_minute,omitempty"` // Gemini shared pool RPM (Google One / Code Assist) - GeminiProMinute *UsageProgress `json:"gemini_pro_minute,omitempty"` // Gemini Pro RPM - GeminiFlashMinute *UsageProgress `json:"gemini_flash_minute,omitempty"` // Gemini Flash RPM + Source string `json:"source,omitempty"` // "passive" or "active" + UpdatedAt *time.Time `json:"updated_at,omitempty"` // 更新时间 + FiveHour *UsageProgress `json:"five_hour"` // 5小时窗口 + SevenDay *UsageProgress `json:"seven_day,omitempty"` // 7天窗口 + SevenDaySonnet *UsageProgress `json:"seven_day_sonnet,omitempty"` // 7天Sonnet窗口 + SevenDayFable *UsageProgress `json:"seven_day_fable,omitempty"` // 7天Fable窗口（响应头 7d_oi） + CodexQuotaOverdraft *CodexQuotaOverdraftProbeState `json:"codex_quota_overdraft,omitempty"` + GeminiSharedDaily *UsageProgress `json:"gemini_shared_daily,omitempty"` // Gemini shared pool RPD (Google One / Code Assist) + GeminiProDaily *UsageProgress `json:"gemini_pro_daily,omitempty"` // Gemini Pro 日配额 + GeminiFlashDaily *UsageProgress `json:"gemini_flash_daily,omitempty"` // Gemini Flash 日配额 + GeminiSharedMinute *UsageProgress `json:"gemini_shared_minute,omitempty"` // Gemini shared pool RPM (Google One / Code Assist) + GeminiProMinute *UsageProgress `json:"gemini_pro_minute,omitempty"` // Gemini Pro RPM + GeminiFlashMinute *UsageProgress `json:"gemini_flash_minute,omitempty"` // Gemini Flash RPM // Antigravity 多模型配额 AntigravityQuota map[string]*AntigravityModelQuota `json:"antigravity_quota,omitempty"` @@ -302,10 +307,17 @@ type AccountUsageService struct { cache *UsageCache identityCache IdentityCache tlsFPProfileService *TLSFingerprintProfileService + codexQuotaOverdraft *CodexQuotaOverdraftCoordinator agentIdentityTaskMu sync.Mutex agentIdentityWS agentIdentityWSConnectionInvalidator } +func (s *AccountUsageService) SetCodexQuotaOverdraftCoordinator(coordinator *CodexQuotaOverdraftCoordinator) { + if s != nil { + s.codexQuotaOverdraft = coordinator + } +} + // NewAccountUsageService 创建AccountUsageService实例 func NewAccountUsageService( accountRepo AccountRepository, @@ -711,6 +723,7 @@ func (s *AccountUsageService) getOpenAIUsage(ctx context.Context, account *Accou } applyExtraToUsage(usage, account.Extra, now) + usage.CodexQuotaOverdraft, _ = codexQuotaOverdraftStateFromAccount(account) if (force || shouldRefreshOpenAICodexSnapshot(account, usage, now)) && s.shouldProbeOpenAICodexSnapshot(account.ID, now, force) { if account.IsShadow() { @@ -741,6 +754,10 @@ func (s *AccountUsageService) getOpenAIUsage(ctx context.Context, account *Accou } } + if s.codexQuotaOverdraft != nil { + s.codexQuotaOverdraft.ObserveAccount(account, "") + } + usage.CodexQuotaOverdraft, _ = codexQuotaOverdraftStateFromAccount(account) if s.usageLogRepo == nil { return usage, nil } @@ -759,6 +776,8 @@ func (s *AccountUsageService) getOpenAIUsage(ctx context.Context, account *Accou usage.SevenDay.WindowStats = windowStatsFromAccountStats(stats) } + applyCodexQuotaOverdraftUsage(ctx, s.usageLogRepo, account, usage, now) + return usage, nil } diff --git a/backend/internal/service/openai_account_runtime_block_fastpath.go b/backend/internal/service/openai_account_runtime_block_fastpath.go index ace8366a0..076004067 100644 --- a/backend/internal/service/openai_account_runtime_block_fastpath.go +++ b/backend/internal/service/openai_account_runtime_block_fastpath.go @@ -73,6 +73,14 @@ 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) { + return false + } if s.rateLimitService != nil && len(canonicalModel) > 0 && s.rateLimitService.HandleUpstreamModelNotFound(stateCtx, account, canonicalModel[0], statusCode, responseBody) { return true } diff --git a/backend/internal/service/openai_account_scheduler.go b/backend/internal/service/openai_account_scheduler.go index 6457ef8fc..154dd1dc6 100644 --- a/backend/internal/service/openai_account_scheduler.go +++ b/backend/internal/service/openai_account_scheduler.go @@ -1393,7 +1393,7 @@ func (s *defaultOpenAIAccountScheduler) selectByLoadBalance( filtered := make([]*Account, 0, len(accounts)) loadReq := make([]AccountWithConcurrency, 0, len(accounts)) for i := range accounts { - account := &accounts[i] + account := normalizeCodexQuotaOverdraftAccountForScheduling(ctx, &accounts[i]) if req.ExcludedIDs != nil { if _, excluded := req.ExcludedIDs[account.ID]; excluded { filterStats.exclude("excluded") diff --git a/backend/internal/service/openai_codex_quota_overdraft.go b/backend/internal/service/openai_codex_quota_overdraft.go new file mode 100644 index 000000000..2aba1fd71 --- /dev/null +++ b/backend/internal/service/openai_codex_quota_overdraft.go @@ -0,0 +1,210 @@ +package service + +import ( + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "strings" + "sync/atomic" + "time" + + "github.com/tidwall/sjson" +) + +const ( + codexQuotaOverdraftCallIDPrefix = "call_sub2api_overdraft_" + codexQuotaOverdraftExecInput = `const r = await tools.exec_command({"cmd":"true","yield_time_ms":1000,"max_output_tokens":1000}); text(r.output);` + codexQuotaOverdraftMaxBodyBytes = 32 << 20 +) + +var codexQuotaOverdraftEnabled atomic.Bool + +// SetCodexQuotaOverdraftEnabled publishes the process-wide scheduling switch. +// Request mutation still reads the gateway instance config directly. +func SetCodexQuotaOverdraftEnabled(enabled bool) { + codexQuotaOverdraftEnabled.Store(enabled) +} + +// CodexQuotaOverdraftEnabled is exported for repository scheduling predicates. +func CodexQuotaOverdraftEnabled() bool { + return codexQuotaOverdraftEnabled.Load() +} + +func isCodexQuotaOverdraftAccount(account *Account) bool { + return account != nil && + account.Platform == PlatformOpenAI && + account.Type == AccountTypeOAuth && + !account.IsShadow() +} + +type codexQuotaOverdraftSchedulingCtxKey struct{} + +// WithCodexQuotaOverdraftScheduling marks normal text-generation requests as +// eligible for the experimental quota-overdraft behavior. The process-wide +// configuration switch is still checked at every scheduling and mutation gate. +func WithCodexQuotaOverdraftScheduling(ctx context.Context) context.Context { + if ctx == nil { + ctx = context.Background() + } + return context.WithValue(ctx, codexQuotaOverdraftSchedulingCtxKey{}, true) +} + +// CodexQuotaOverdraftSchedulingEnabled reports whether the global switch and +// the request-scoped endpoint marker are both enabled. +func CodexQuotaOverdraftSchedulingEnabled(ctx context.Context) bool { + if !CodexQuotaOverdraftEnabled() || ctx == nil { + return false + } + enabled, _ := ctx.Value(codexQuotaOverdraftSchedulingCtxKey{}).(bool) + return enabled +} + +func codexQuotaOverdraftSchedulingEnabled(ctx context.Context) bool { + return CodexQuotaOverdraftSchedulingEnabled(ctx) +} + +func (s *OpenAIGatewayService) shouldInjectCodexQuotaOverdraft(ctx context.Context, account *Account, compact bool) bool { + return codexQuotaOverdraftSchedulingEnabled(ctx) && !compact && + s != nil && s.cfg != nil && s.cfg.Gateway.CodexQuotaOverdraftEnabled && + isCodexQuotaOverdraftAccount(account) +} + +func (s *OpenAIGatewayService) prepareCodexQuotaOverdraftBody(ctx context.Context, account *Account, compact bool, body []byte) []byte { + if !s.shouldInjectCodexQuotaOverdraft(ctx, account, compact) { + return body + } + updated, changed, _ := injectCodexQuotaOverdraft(body) + if !changed { + return body + } + return updated +} + +func (s *OpenAIGatewayService) prepareCodexQuotaOverdraftPayload(ctx context.Context, account *Account, payload map[string]any) map[string]any { + if !s.shouldInjectCodexQuotaOverdraft(ctx, account, false) || payload == nil { + return payload + } + raw, err := json.Marshal(payload) + if err != nil { + return payload + } + updated, changed, _ := injectCodexQuotaOverdraft(raw) + if !changed { + return payload + } + var out map[string]any + if err := json.Unmarshal(updated, &out); err != nil { + return payload + } + return out +} + +type codexQuotaOverdraftDocument struct { + Input []json.RawMessage `json:"input"` +} + +type codexQuotaOverdraftInputItem struct { + Type string `json:"type"` + Role string `json:"role"` + CallID string `json:"call_id"` +} + +// injectCodexQuotaOverdraft appends the same no-op custom tool call pair used by +// cpa-account-config-manager. Unsupported request shapes fail open unchanged. +func injectCodexQuotaOverdraft(body []byte) ([]byte, bool, error) { + if len(body) == 0 || len(body) > codexQuotaOverdraftMaxBodyBytes { + return body, false, nil + } + + var document codexQuotaOverdraftDocument + if err := json.Unmarshal(body, &document); err != nil { + return body, false, nil + } + if len(document.Input) == 0 { + return body, false, nil + } + + for _, raw := range document.Input { + var item codexQuotaOverdraftInputItem + if err := json.Unmarshal(raw, &item); err == nil && + item.Type == "custom_tool_call" && + strings.HasPrefix(item.CallID, codexQuotaOverdraftCallIDPrefix) { + return body, false, nil + } + } + + var last codexQuotaOverdraftInputItem + if err := json.Unmarshal(document.Input[len(document.Input)-1], &last); err != nil || last.Type != "message" || last.Role != "user" { + return body, false, nil + } + + callID, ok := newCodexQuotaOverdraftCallID() + if !ok { + return body, false, nil + } + call, err := json.Marshal(map[string]any{ + "type": "custom_tool_call", + "name": "exec", + "call_id": callID, + "input": codexQuotaOverdraftExecInput, + }) + if err != nil { + return body, false, nil + } + output, err := json.Marshal(map[string]any{ + "type": "custom_tool_call_output", + "call_id": callID, + "output": []map[string]string{{ + "type": "input_text", + "text": "Script completed\nWall time 0.0 seconds\nOutput:\n", + }}, + }) + if err != nil { + return body, false, nil + } + + document.Input = append(document.Input, call, output) + updatedInput, err := json.Marshal(document.Input) + if err != nil { + return body, false, nil + } + updated, err := sjson.SetRawBytes(body, "input", updatedInput) + if err != nil { + return body, false, nil + } + if len(updated) > codexQuotaOverdraftMaxBodyBytes { + return body, false, nil + } + return updated, true, nil +} + +func normalizeCodexQuotaOverdraftAccountForScheduling(ctx context.Context, account *Account) *Account { + if !codexQuotaOverdraftSchedulingEnabled(ctx) || !isCodexQuotaOverdraftAccount(account) || + !codexQuotaOverdraftSchedulingAllowed(account, time.Now().UTC()) || + account.TempUnschedulableUntil == nil || !time.Now().Before(*account.TempUnschedulableUntil) || + !IsAccountSchedulingThresholdReason(account.TempUnschedulableReason) { + return account + } + clone := *account + clone.TempUnschedulableUntil = nil + clone.TempUnschedulableReason = "" + return &clone +} + +func normalizeCodexQuotaOverdraftAccountsForScheduling(ctx context.Context, accounts []Account) []Account { + for i := range accounts { + if normalized := normalizeCodexQuotaOverdraftAccountForScheduling(ctx, &accounts[i]); normalized != &accounts[i] { + accounts[i] = *normalized + } + } + return accounts +} + +func newCodexQuotaOverdraftCallID() (string, bool) { + var random [12]byte + if _, err := rand.Read(random[:]); err != nil { + return "", false + } + return codexQuotaOverdraftCallIDPrefix + hex.EncodeToString(random[:]), true +} diff --git a/backend/internal/service/openai_codex_quota_overdraft_probe.go b/backend/internal/service/openai_codex_quota_overdraft_probe.go new file mode 100644 index 000000000..cde930e0a --- /dev/null +++ b/backend/internal/service/openai_codex_quota_overdraft_probe.go @@ -0,0 +1,942 @@ +package service + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "net/http" + "strings" + "sync" + "time" + + "github.com/Wei-Shaw/sub2api/internal/config" + "github.com/Wei-Shaw/sub2api/internal/pkg/openai" + "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint" +) + +const ( + CodexQuotaOverdraftProbeExtraKey = "codex_quota_overdraft_probe" + + codexQuotaOverdraftProbePending = "pending" + codexQuotaOverdraftProbePassed = "passed" + codexQuotaOverdraftProbeFailed = "failed" + codexQuotaOverdraftProbeInconclusive = "inconclusive" + codexQuotaOverdraftProbeRecovered = "recovered" + + codexQuotaOverdraftProbeAttemptLimit = 5 + codexQuotaOverdraftProbeAttemptTimeout = 20 * time.Second + codexQuotaOverdraftProbePlanTimeout = codexQuotaOverdraftProbeAttemptLimit * codexQuotaOverdraftProbeAttemptTimeout + codexQuotaOverdraftProbeRetryDelay = time.Minute + codexQuotaOverdraftProbeBodyLimit = 256 << 10 + codexQuotaOverdraftPauseSource = "codex_quota_overdraft" + + codexQuotaOverdraftFallbackModel = "gpt-5.5" + codexQuotaOverdraftCompatibilityModel = "gpt-5.4-mini" +) + +// CodexQuotaOverdraftProbeState is persisted in accounts.extra so quota-cycle +// decisions survive restarts and are visible through the account usage API. +type CodexQuotaOverdraftProbeState struct { + Status string `json:"status"` + QuotaWindow string `json:"quota_window"` + CycleKey string `json:"cycle_key"` + Attempts int `json:"attempts"` + Limit int `json:"limit"` + Model string `json:"model,omitempty"` + ReasonCode string `json:"reason_code,omitempty"` + StartedAt time.Time `json:"started_at"` + TestedAt *time.Time `json:"tested_at,omitempty"` + RetryAt *time.Time `json:"retry_at,omitempty"` + RecoverAt *time.Time `json:"recover_at,omitempty"` + FiveHourRecoverAt *time.Time `json:"five_hour_recover_at,omitempty"` + SevenDayRecoverAt *time.Time `json:"seven_day_recover_at,omitempty"` + OverdraftStartedAt *time.Time `json:"overdraft_started_at,omitempty"` + FiveHourStartedAt *time.Time `json:"five_hour_overdraft_started_at,omitempty"` + SevenDayStartedAt *time.Time `json:"seven_day_overdraft_started_at,omitempty"` + ObservedRateLimit *time.Time `json:"observed_rate_limit_reset_at,omitempty"` +} + +type codexQuotaOverdraftSignal struct { + Window string + CycleKey string + RecoverAt time.Time + FiveHourRecoverAt *time.Time + SevenDayRecoverAt *time.Time +} + +type codexQuotaOverdraftProbeResult struct { + Status string + ReasonCode string + StatusCode int + Model string + Headers http.Header + Body []byte +} + +type codexQuotaOverdraftProbeClaimer interface { + ClaimCodexQuotaOverdraftProbe(context.Context, int64, *CodexQuotaOverdraftProbeState) (bool, error) +} + +// CodexQuotaOverdraftCoordinator implements the bounded real-request gate used +// by cpa-account-config-manager before it disables a quota-exhausted account. +type CodexQuotaOverdraftCoordinator struct { + accountRepo AccountRepository + httpUpstream HTTPUpstream + openAITokenProvider *OpenAITokenProvider + tlsFPProfileService *TLSFingerprintProfileService + cfg *config.Config + tempUnschedCache TempUnschedCache + runtimeBlocker AccountRuntimeBlocker + rateLimitService *RateLimitService + agentIdentityWS agentIdentityWSConnectionInvalidator + agentIdentityTaskMu sync.Mutex + running sync.Map + now func() time.Time + probeAttemptForTest func(context.Context, *Account, string) codexQuotaOverdraftProbeResult +} + +func NewCodexQuotaOverdraftCoordinator( + accountRepo AccountRepository, + httpUpstream HTTPUpstream, + openAITokenProvider *OpenAITokenProvider, + tlsFPProfileService *TLSFingerprintProfileService, + cfg *config.Config, + tempUnschedCache TempUnschedCache, + runtimeBlocker AccountRuntimeBlocker, + rateLimitService *RateLimitService, +) *CodexQuotaOverdraftCoordinator { + coordinator := &CodexQuotaOverdraftCoordinator{ + accountRepo: accountRepo, + httpUpstream: httpUpstream, + openAITokenProvider: openAITokenProvider, + tlsFPProfileService: tlsFPProfileService, + cfg: cfg, + tempUnschedCache: tempUnschedCache, + runtimeBlocker: runtimeBlocker, + rateLimitService: rateLimitService, + now: time.Now, + } + coordinator.agentIdentityWS, _ = runtimeBlocker.(agentIdentityWSConnectionInvalidator) + return coordinator +} + +func (c *CodexQuotaOverdraftCoordinator) enabled() bool { + return c != nil && c.cfg != nil && c.cfg.Gateway.CodexQuotaOverdraftEnabled && + c.accountRepo != nil && c.httpUpstream != nil +} + +// ObserveAccount starts a probe when a persisted 5h/7d snapshot first reaches +// 100%, and closes an old overdraft cycle after the quota window recovers. +func (c *CodexQuotaOverdraftCoordinator) ObserveAccount(account *Account, preferredModel string) { + if !c.enabled() || !isCodexQuotaOverdraftAccount(account) || account.ID <= 0 { + return + } + accountCopy := cloneCodexQuotaOverdraftAccount(account) + go c.observeAccount(accountCopy, preferredModel) +} + +func (c *CodexQuotaOverdraftCoordinator) observeAccount(account *Account, preferredModel string) { + if account == nil { + return + } + now := c.currentTime() + state, _ := codexQuotaOverdraftStateFromAccount(account) + if state != nil && clearRecoveredCodexQuotaOverdraftWindows(state, account, now) { + c.persistState(account.ID, state) + mergeAccountExtra(account, map[string]any{CodexQuotaOverdraftProbeExtraKey: state}) + } + signal, exhausted := codexQuotaOverdraftSignalFromAccount(account, state, now) + if !exhausted { + c.recoverCycle(account, state, now) + return + } + c.startProbe(account, signal, preferredModel) +} + +// HandleQuota429 intercepts only definite subscription-quota 429s. Ordinary +// transient 429s continue through the existing rate-limit policy unchanged. +func (c *CodexQuotaOverdraftCoordinator) HandleQuota429( + ctx context.Context, + account *Account, + headers http.Header, + body []byte, + preferredModel string, +) bool { + if !c.enabled() || !isCodexQuotaOverdraftAccount(account) || account.ID <= 0 || + !codexQuotaOverdraftResponseIsQuotaLimited(headers, body) { + return false + } + + accountCopy := cloneCodexQuotaOverdraftAccount(account) + if snapshot := ParseCodexRateLimitHeaders(headers); snapshot != nil { + updates := buildCodexUsageExtraUpdates(snapshot, c.currentTime()) + mergeAccountExtra(accountCopy, updates) + persistCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 3*time.Second) + _ = c.accountRepo.UpdateExtra(persistCtx, account.ID, updates) + cancel() + } + state, _ := codexQuotaOverdraftStateFromAccount(accountCopy) + signal, exhausted := codexQuotaOverdraftSignalFromAccount(accountCopy, state, c.currentTime()) + if !exhausted { + signal = codexQuotaOverdraftFallbackSignal(headers, body, state, c.currentTime()) + } + if signal.CycleKey == "" { + return false + } + c.startProbe(accountCopy, signal, preferredModel) + return true +} + +func (c *CodexQuotaOverdraftCoordinator) startProbe(account *Account, signal codexQuotaOverdraftSignal, preferredModel string) { + if account == nil || signal.CycleKey == "" { + return + } + current, hasCurrent := codexQuotaOverdraftStateFromAccount(account) + if hasCurrent && codexQuotaOverdraftStateCoversSignal(current, signal) { + switch current.Status { + case codexQuotaOverdraftProbePassed: + return + case codexQuotaOverdraftProbeFailed: + c.ensureFailedPause(account, current) + return + case codexQuotaOverdraftProbePending: + if c.currentTime().Sub(current.StartedAt) < 2*time.Minute { + return + } + case codexQuotaOverdraftProbeInconclusive: + if current.RetryAt != nil && c.currentTime().Before(*current.RetryAt) { + return + } + } + } + + startedAt := c.currentTime().UTC() + state := &CodexQuotaOverdraftProbeState{ + Status: codexQuotaOverdraftProbePending, + QuotaWindow: signal.Window, + CycleKey: signal.CycleKey, + Limit: codexQuotaOverdraftProbeAttemptLimit, + StartedAt: startedAt, + RecoverAt: codexQuotaOverdraftTimePtr(signal.RecoverAt), + FiveHourRecoverAt: cloneTimePtr(signal.FiveHourRecoverAt), + SevenDayRecoverAt: cloneTimePtr(signal.SevenDayRecoverAt), + ObservedRateLimit: cloneTimePtr(account.RateLimitResetAt), + } + carryCodexQuotaOverdraftWindowStarts(state, current, signal, startedAt) + claimed, err := c.claimProbe(account.ID, state) + if err != nil { + slog.Warn("codex_quota_overdraft_probe_claim_failed", "account_id", account.ID, "error", err) + return + } + if !claimed { + return + } + mergeAccountExtra(account, map[string]any{CodexQuotaOverdraftProbeExtraKey: state}) + + taskKey := fmt.Sprintf("%d:%s", account.ID, signal.CycleKey) + if _, loaded := c.running.LoadOrStore(taskKey, struct{}{}); loaded { + return + } + go func() { + defer c.running.Delete(taskKey) + c.runProbePlan(account.ID, signal, preferredModel, state) + }() +} + +func (c *CodexQuotaOverdraftCoordinator) claimProbe(accountID int64, state *CodexQuotaOverdraftProbeState) (bool, error) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if claimer, ok := c.accountRepo.(codexQuotaOverdraftProbeClaimer); ok { + return claimer.ClaimCodexQuotaOverdraftProbe(ctx, accountID, state) + } + if err := c.accountRepo.UpdateExtra(ctx, accountID, map[string]any{CodexQuotaOverdraftProbeExtraKey: state}); err != nil { + return false, err + } + return true, nil +} + +func (c *CodexQuotaOverdraftCoordinator) runProbePlan( + accountID int64, + signal codexQuotaOverdraftSignal, + preferredModel string, + state *CodexQuotaOverdraftProbeState, +) { + planCtx, cancel := context.WithTimeout(context.Background(), codexQuotaOverdraftProbePlanTimeout) + defer cancel() + + account, err := c.accountRepo.GetByID(planCtx, accountID) + if err != nil || !isCodexQuotaOverdraftAccount(account) { + c.finishInconclusive(accountID, state, "upstream_unavailable") + return + } + models := codexQuotaOverdraftProbeModels(preferredModel) + lastReason := "invalid_response" + quotaLimitedAttempts := 0 + for attempt := 0; attempt < codexQuotaOverdraftProbeAttemptLimit; attempt++ { + if planCtx.Err() != nil { + c.finishInconclusive(accountID, state, "request_timeout") + return + } + model := models[attempt%len(models)] + attemptCtx, attemptCancel := context.WithTimeout(planCtx, codexQuotaOverdraftProbeAttemptTimeout) + result := c.runProbeAttempt(attemptCtx, account, model) + attemptCancel() + state.Attempts = attempt + 1 + state.Model = result.Model + state.ReasonCode = result.ReasonCode + testedAt := c.currentTime().UTC() + state.TestedAt = &testedAt + lastReason = result.ReasonCode + if result.ReasonCode == "quota_limited" { + quotaLimitedAttempts++ + } + c.persistState(accountID, state) + + if result.Status == "available" { + state.Status = codexQuotaOverdraftProbePassed + state.ReasonCode = "model_response_ok" + state.RetryAt = nil + startCodexQuotaOverdraftWindows(state, signal, testedAt) + c.persistState(accountID, state) + c.clearQuotaPause(accountID, state) + slog.Info("codex_quota_overdraft_probe_passed", "account_id", accountID, "attempts", state.Attempts, "model", state.Model, "quota_window", signal.Window) + return + } + if result.Status == "inconclusive" { + c.finishInconclusive(accountID, state, result.ReasonCode) + return + } + if result.Status == "authentication_failed" { + if c.rateLimitService != nil { + c.rateLimitService.HandleUpstreamError(context.Background(), account, result.StatusCode, result.Headers, result.Body) + } + c.finishInconclusive(accountID, state, result.ReasonCode) + return + } + } + if quotaLimitedAttempts < codexQuotaOverdraftProbeAttemptLimit { + c.finishInconclusive(accountID, state, lastReason) + return + } + + state.Status = codexQuotaOverdraftProbeFailed + state.ReasonCode = lastReason + state.RetryAt = nil + c.persistState(accountID, state) + c.ensureFailedPause(account, state) + slog.Warn("codex_quota_overdraft_probe_failed", "account_id", accountID, "attempts", state.Attempts, "quota_window", signal.Window, "recover_at", state.RecoverAt) +} + +func (c *CodexQuotaOverdraftCoordinator) finishInconclusive(accountID int64, state *CodexQuotaOverdraftProbeState, reason string) { + now := c.currentTime().UTC() + retryAt := now.Add(codexQuotaOverdraftProbeRetryDelay) + state.Status = codexQuotaOverdraftProbeInconclusive + state.ReasonCode = reason + state.TestedAt = &now + state.RetryAt = &retryAt + c.persistState(accountID, state) + slog.Warn("codex_quota_overdraft_probe_inconclusive", "account_id", accountID, "attempts", state.Attempts, "reason", reason) +} + +func (c *CodexQuotaOverdraftCoordinator) runProbeAttempt(ctx context.Context, account *Account, model string) codexQuotaOverdraftProbeResult { + if c.probeAttemptForTest != nil { + return c.probeAttemptForTest(ctx, account, model) + } + result := codexQuotaOverdraftProbeResult{Model: model} + upstreamModel := normalizeOpenAIModelForUpstream(account, account.GetMappedModel(model)) + payload := map[string]any{ + "model": upstreamModel, + "input": []map[string]any{{ + "type": "message", + "role": "user", + "content": []map[string]string{{ + "type": "input_text", + "text": "hi", + }}, + }}, + "instructions": "Reply with OK only.", + "stream": true, + "store": false, + } + payloadBytes, err := json.Marshal(payload) + if err != nil { + result.Status, result.ReasonCode = "inconclusive", "experimental_probe_unavailable" + return result + } + payloadBytes, changed, err := injectCodexQuotaOverdraft(payloadBytes) + if err != nil || !changed { + result.Status, result.ReasonCode = "inconclusive", "experimental_probe_unavailable" + return result + } + + req, err := http.NewRequestWithContext(WithHTTPUpstreamProfile(ctx, HTTPUpstreamProfileOpenAI), http.MethodPost, chatgptCodexAPIURL, bytes.NewReader(payloadBytes)) + if err != nil { + result.Status, result.ReasonCode = "inconclusive", "upstream_unavailable" + return result + } + req.Host = "chatgpt.com" + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "text/event-stream") + req.Header.Set("OpenAI-Beta", "responses=experimental") + req.Header.Set("Originator", openai.CodexDefaultOriginator) + req.Header.Set("Version", openAICodexProbeVersion) + req.Header.Set("User-Agent", codexCLIUserAgent) + + if account.IsOpenAIAgentIdentity() { + authHeaders, authErr := buildAgentIdentityAuthenticationHeaders(ctx, c.accountRepo, c.agentIdentityWS, &c.agentIdentityTaskMu, account) + if authErr != nil { + result.Status, result.ReasonCode = "inconclusive", "authentication_unavailable" + return result + } + for key, values := range authHeaders { + for _, value := range values { + req.Header.Add(key, value) + } + } + } else { + token := account.GetOpenAIAccessToken() + if c.openAITokenProvider != nil { + if refreshed, tokenErr := c.openAITokenProvider.GetAccessToken(ctx, account); tokenErr == nil { + token = refreshed + } else if strings.TrimSpace(token) == "" { + result.Status, result.ReasonCode = "inconclusive", "authentication_unavailable" + return result + } + } + if strings.TrimSpace(token) == "" { + result.Status, result.ReasonCode = "inconclusive", "authentication_unavailable" + return result + } + req.Header.Set("Authorization", "Bearer "+token) + } + setOpenAIChatGPTAccountHeaders(req.Header, account) + enforceCodexIdentityHeadersWithUA(req.Header, account.GetOpenAIUserAgent()) + account.ApplyHeaderOverrides(req.Header) + + proxyURL := "" + if account.ProxyID != nil && account.Proxy != nil { + proxyURL = account.Proxy.URL() + } + var tlsProfile = (*tlsfingerprint.Profile)(nil) + if c.tlsFPProfileService != nil { + tlsProfile = c.tlsFPProfileService.ResolveTLSProfile(account) + } + resp, err := c.httpUpstream.DoWithTLS(req, proxyURL, account.ID, account.Concurrency, tlsProfile) + if err != nil { + result.Status = "inconclusive" + if errors.Is(err, context.DeadlineExceeded) || errors.Is(ctx.Err(), context.DeadlineExceeded) { + result.ReasonCode = "request_timeout" + } else { + result.ReasonCode = "upstream_unavailable" + } + return result + } + defer func() { _ = resp.Body.Close() }() + result.StatusCode = resp.StatusCode + result.Headers = resp.Header.Clone() + body, readErr := io.ReadAll(io.LimitReader(resp.Body, codexQuotaOverdraftProbeBodyLimit+1)) + if readErr != nil || len(body) > codexQuotaOverdraftProbeBodyLimit { + result.Status, result.ReasonCode = "inconclusive", "invalid_response" + return result + } + result.Body = body + result.Status, result.ReasonCode = classifyCodexQuotaOverdraftProbe(resp.StatusCode, resp.Header, body) + return result +} + +func classifyCodexQuotaOverdraftProbe(statusCode int, headers http.Header, body []byte) (string, string) { + if statusCode >= http.StatusOK && statusCode < http.StatusMultipleChoices { + lower := bytes.ToLower(body) + if bytes.Contains(lower, []byte(`"type":"error"`)) || bytes.Contains(lower, []byte(`"type": "error"`)) || + bytes.Contains(lower, []byte(`"response.failed"`)) { + return "retry", "invalid_response" + } + if bytes.Contains(lower, []byte(`"response.completed"`)) || bytes.Contains(lower, []byte(`"response.output_item.done"`)) { + return "available", "model_response_ok" + } + return "retry", "invalid_response" + } + switch statusCode { + case http.StatusUnauthorized, http.StatusForbidden: + return "authentication_failed", "authentication_failed" + case http.StatusPaymentRequired: + lower := strings.ToLower(string(body)) + if strings.Contains(lower, "deactivated_workspace") || strings.Contains(lower, "account_deactivated") { + return "authentication_failed", "account_deactivated" + } + if codexQuotaOverdraftResponseIsQuotaLimited(headers, body) { + return "retry", "quota_limited" + } + return "retry", "invalid_response" + case http.StatusTooManyRequests: + if codexQuotaOverdraftResponseIsQuotaLimited(headers, body) { + return "retry", "quota_limited" + } + return "inconclusive", "transient_failure" + case http.StatusRequestTimeout, http.StatusGatewayTimeout: + return "inconclusive", "request_timeout" + default: + if statusCode >= http.StatusInternalServerError { + return "inconclusive", "upstream_unavailable" + } + if statusCode == http.StatusBadRequest || statusCode == http.StatusNotFound { + return "inconclusive", "model_not_found" + } + return "inconclusive", "invalid_response" + } +} + +func (c *CodexQuotaOverdraftCoordinator) persistState(accountID int64, state *CodexQuotaOverdraftProbeState) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := c.accountRepo.UpdateExtra(ctx, accountID, map[string]any{CodexQuotaOverdraftProbeExtraKey: state}); err != nil { + slog.Warn("codex_quota_overdraft_state_persist_failed", "account_id", accountID, "error", err) + } +} + +func (c *CodexQuotaOverdraftCoordinator) ensureFailedPause(account *Account, state *CodexQuotaOverdraftProbeState) { + if account == nil || state == nil || state.RecoverAt == nil || !state.RecoverAt.After(c.currentTime()) { + return + } + reason := BuildTempUnschedReasonPayload(codexQuotaOverdraftPauseSource, "five real overdraft probes confirmed quota exhaustion") + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := c.accountRepo.SetTempUnschedulable(ctx, account.ID, *state.RecoverAt, reason); err != nil { + slog.Warn("codex_quota_overdraft_pause_failed", "account_id", account.ID, "error", err) + return + } + if c.tempUnschedCache != nil { + _ = c.tempUnschedCache.SetTempUnsched(ctx, account.ID, &TempUnschedState{ + UntilUnix: state.RecoverAt.Unix(), + TriggeredAtUnix: c.currentTime().Unix(), + StatusCode: http.StatusTooManyRequests, + ErrorMessage: "five real overdraft probes confirmed quota exhaustion", + }) + } + if c.runtimeBlocker != nil { + c.runtimeBlocker.BlockAccountScheduling(account, *state.RecoverAt, codexQuotaOverdraftPauseSource) + } +} + +func (c *CodexQuotaOverdraftCoordinator) clearQuotaPause(accountID int64, state *CodexQuotaOverdraftProbeState) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + account, err := c.accountRepo.GetByID(ctx, accountID) + if err != nil || account == nil { + return + } + clearedSchedulingState := false + if state != nil && sameCodexQuotaOverdraftTime(account.RateLimitResetAt, state.ObservedRateLimit) { + _ = c.accountRepo.ClearRateLimit(ctx, accountID) + clearedSchedulingState = true + } + if IsAccountSchedulingThresholdReason(account.TempUnschedulableReason) || codexQuotaOverdraftPauseReason(account.TempUnschedulableReason) { + _ = c.accountRepo.ClearTempUnschedulable(ctx, accountID) + clearedSchedulingState = true + if c.tempUnschedCache != nil { + _ = c.tempUnschedCache.DeleteTempUnsched(ctx, accountID) + } + } + if clearedSchedulingState && c.runtimeBlocker != nil { + c.runtimeBlocker.ClearAccountSchedulingBlock(accountID) + } +} + +func (c *CodexQuotaOverdraftCoordinator) recoverCycle(account *Account, state *CodexQuotaOverdraftProbeState, now time.Time) { + if account == nil || state == nil || state.Status == "" || state.Status == codexQuotaOverdraftProbeRecovered { + return + } + state.Status = codexQuotaOverdraftProbeRecovered + state.ReasonCode = "quota_recovered" + state.RetryAt = nil + state.OverdraftStartedAt = nil + state.FiveHourStartedAt = nil + state.SevenDayStartedAt = nil + testedAt := now.UTC() + state.TestedAt = &testedAt + c.persistState(account.ID, state) + c.clearQuotaPause(account.ID, state) +} + +func (c *CodexQuotaOverdraftCoordinator) currentTime() time.Time { + if c != nil && c.now != nil { + return c.now().UTC() + } + return time.Now().UTC() +} + +func codexQuotaOverdraftProbeModels(preferred string) []string { + models := []string{strings.TrimSpace(preferred), codexQuotaOverdraftFallbackModel, codexQuotaOverdraftCompatibilityModel} + if models[0] == "" { + models[0] = openai.DefaultTestModel + } + out := make([]string, 0, len(models)) + seen := make(map[string]struct{}, len(models)) + for _, model := range models { + key := strings.ToLower(strings.TrimSpace(model)) + if key == "" { + continue + } + if _, ok := seen[key]; ok { + continue + } + seen[key] = struct{}{} + out = append(out, model) + } + return out +} + +func codexQuotaOverdraftResponseIsQuotaLimited(headers http.Header, body []byte) bool { + if snapshot := ParseCodexRateLimitHeaders(headers); snapshot != nil { + if normalized := snapshot.Normalize(); normalized != nil && + (normalized.Used5hPercent != nil && *normalized.Used5hPercent >= 100 || + normalized.Used7dPercent != nil && *normalized.Used7dPercent >= 100) { + return true + } + } + text := strings.ToLower(strings.Join(strings.Fields(string(body)), " ")) + for _, marker := range []string{"usage_limit_reached", "usage limit has been reached", "quota exhausted", "weekly limit reached"} { + if strings.Contains(text, marker) { + return true + } + } + return false +} + +func codexQuotaOverdraftSignalFromAccount(account *Account, state *CodexQuotaOverdraftProbeState, now time.Time) (codexQuotaOverdraftSignal, bool) { + if account == nil || len(account.Extra) == 0 { + return codexQuotaOverdraftSignal{}, false + } + fiveUsed := parseExtraFloat64(account.Extra["codex_5h_used_percent"]) + sevenUsed := parseExtraFloat64(account.Extra["codex_7d_used_percent"]) + fiveReset := codexQuotaOverdraftResetAt(account.Extra["codex_5h_reset_at"], now) + sevenReset := codexQuotaOverdraftResetAt(account.Extra["codex_7d_reset_at"], now) + if state != nil { + fiveReset = stabilizeCodexQuotaOverdraftReset(fiveReset, state.FiveHourRecoverAt, now) + sevenReset = stabilizeCodexQuotaOverdraftReset(sevenReset, state.SevenDayRecoverAt, now) + } + fiveExhausted := fiveUsed >= 100 && (fiveReset == nil || fiveReset.After(now)) + sevenExhausted := sevenUsed >= 100 && (sevenReset == nil || sevenReset.After(now)) + if !fiveExhausted && !sevenExhausted { + return codexQuotaOverdraftSignal{}, false + } + if fiveExhausted && fiveReset == nil { + fallback := now.Add(5 * time.Hour) + if state != nil && state.FiveHourRecoverAt != nil && state.FiveHourRecoverAt.After(now) { + fallback = *state.FiveHourRecoverAt + } + fiveReset = &fallback + } + if sevenExhausted && sevenReset == nil { + fallback := now.Add(7 * 24 * time.Hour) + if state != nil && state.SevenDayRecoverAt != nil && state.SevenDayRecoverAt.After(now) { + fallback = *state.SevenDayRecoverAt + } + sevenReset = &fallback + } + + signal := codexQuotaOverdraftSignal{FiveHourRecoverAt: fiveReset, SevenDayRecoverAt: sevenReset} + switch { + case fiveExhausted && sevenExhausted: + signal.Window = "multiple" + signal.CycleKey = fmt.Sprintf("5h:%d|7d:%d", fiveReset.Unix(), sevenReset.Unix()) + signal.RecoverAt = laterTime(*fiveReset, *sevenReset) + case fiveExhausted: + signal.Window = "5h" + signal.CycleKey = fmt.Sprintf("5h:%d", fiveReset.Unix()) + signal.RecoverAt = *fiveReset + signal.SevenDayRecoverAt = nil + default: + signal.Window = "7d" + signal.CycleKey = fmt.Sprintf("7d:%d", sevenReset.Unix()) + signal.RecoverAt = *sevenReset + signal.FiveHourRecoverAt = nil + } + return signal, true +} + +func clearRecoveredCodexQuotaOverdraftWindows(state *CodexQuotaOverdraftProbeState, account *Account, now time.Time) bool { + if state == nil || account == nil { + return false + } + changed := false + fiveUsed := parseExtraFloat64(account.Extra["codex_5h_used_percent"]) + sevenUsed := parseExtraFloat64(account.Extra["codex_7d_used_percent"]) + if state.FiveHourStartedAt != nil && (fiveUsed < 100 || state.FiveHourRecoverAt == nil || !state.FiveHourRecoverAt.After(now)) { + state.FiveHourStartedAt = nil + state.FiveHourRecoverAt = nil + changed = true + } + if state.SevenDayStartedAt != nil && (sevenUsed < 100 || state.SevenDayRecoverAt == nil || !state.SevenDayRecoverAt.After(now)) { + state.SevenDayStartedAt = nil + state.SevenDayRecoverAt = nil + changed = true + } + state.OverdraftStartedAt = earlierCodexQuotaOverdraftStart(state.FiveHourStartedAt, state.SevenDayStartedAt) + return changed +} + +func codexQuotaOverdraftFallbackSignal(headers http.Header, body []byte, state *CodexQuotaOverdraftProbeState, now time.Time) codexQuotaOverdraftSignal { + recoverAt := now.Add(5 * time.Hour) + if resetAt := calculateOpenAI429ResetTime(headers); resetAt != nil && resetAt.After(now) { + recoverAt = *resetAt + } else if resetUnix := parseOpenAIRateLimitResetTime(body); resetUnix != nil { + if parsed := time.Unix(*resetUnix, 0).UTC(); parsed.After(now) { + recoverAt = parsed + } + } else if state != nil && state.RecoverAt != nil && state.RecoverAt.After(now) { + recoverAt = *state.RecoverAt + } + return codexQuotaOverdraftSignal{ + Window: "multiple", + CycleKey: fmt.Sprintf("multiple:%d", recoverAt.Unix()), + RecoverAt: recoverAt, + FiveHourRecoverAt: codexQuotaOverdraftTimePtr(recoverAt), + SevenDayRecoverAt: codexQuotaOverdraftTimePtr(recoverAt), + } +} + +func codexQuotaOverdraftStateFromAccount(account *Account) (*CodexQuotaOverdraftProbeState, bool) { + if account == nil || account.Extra == nil { + return nil, false + } + raw, ok := account.Extra[CodexQuotaOverdraftProbeExtraKey] + if !ok || raw == nil { + return nil, false + } + encoded, err := json.Marshal(raw) + if err != nil { + return nil, false + } + var state CodexQuotaOverdraftProbeState + if err := json.Unmarshal(encoded, &state); err != nil || state.CycleKey == "" { + return nil, false + } + return &state, true +} + +func codexQuotaOverdraftSchedulingAllowed(account *Account, now time.Time) bool { + if !isCodexQuotaOverdraftAccount(account) { + return false + } + state, _ := codexQuotaOverdraftStateFromAccount(account) + signal, exhausted := codexQuotaOverdraftSignalFromAccount(account, state, now) + if !exhausted || state == nil || !codexQuotaOverdraftStateCoversSignal(state, signal) { + return true + } + return state.Status != codexQuotaOverdraftProbeFailed +} + +func codexQuotaOverdraftSnapshotExhausted(updates map[string]any) bool { + if len(updates) == 0 { + return false + } + return parseExtraFloat64(updates["codex_5h_used_percent"]) >= 100 || + parseExtraFloat64(updates["codex_7d_used_percent"]) >= 100 +} + +func applyCodexQuotaOverdraftUsage( + ctx context.Context, + repo UsageLogRepository, + account *Account, + usage *UsageInfo, + now time.Time, +) { + if repo == nil || account == nil || usage == nil { + return + } + state, ok := codexQuotaOverdraftStateFromAccount(account) + if !ok || state.Status == codexQuotaOverdraftProbeRecovered { + return + } + apply := func(progress *UsageProgress, startedAt, recoverAt *time.Time) { + if progress == nil || progress.Utilization < 100 || startedAt == nil || recoverAt == nil || !recoverAt.After(now) { + return + } + stats, err := repo.GetAccountWindowStats(ctx, account.ID, *startedAt) + if err != nil { + return + } + progress.OverdraftActive = true + progress.OverdraftStats = windowStatsFromAccountStats(stats) + progress.OverdraftStarted = cloneTimePtr(startedAt) + progress.OverdraftRecover = cloneTimePtr(recoverAt) + } + fiveStarted, sevenStarted := codexQuotaOverdraftWindowStarts(state) + apply(usage.FiveHour, fiveStarted, state.FiveHourRecoverAt) + apply(usage.SevenDay, sevenStarted, state.SevenDayRecoverAt) +} + +func stabilizeCodexQuotaOverdraftReset(current, persisted *time.Time, now time.Time) *time.Time { + if current == nil || persisted == nil || !persisted.After(now) { + return current + } + delta := current.Sub(*persisted) + if delta < 0 { + delta = -delta + } + if delta <= 2*time.Minute { + return cloneTimePtr(persisted) + } + return current +} + +func codexQuotaOverdraftStateCoversSignal(state *CodexQuotaOverdraftProbeState, signal codexQuotaOverdraftSignal) bool { + if state == nil || signal.CycleKey == "" { + return false + } + if state.CycleKey == signal.CycleKey { + return true + } + resetMatches := func(current, persisted *time.Time) bool { + if current == nil || persisted == nil { + return false + } + delta := current.Sub(*persisted) + if delta < 0 { + delta = -delta + } + return delta <= 2*time.Minute + } + switch signal.Window { + case "5h": + return resetMatches(signal.FiveHourRecoverAt, state.FiveHourRecoverAt) + case "7d": + return resetMatches(signal.SevenDayRecoverAt, state.SevenDayRecoverAt) + case "multiple": + return resetMatches(signal.FiveHourRecoverAt, state.FiveHourRecoverAt) && + resetMatches(signal.SevenDayRecoverAt, state.SevenDayRecoverAt) + default: + return false + } +} + +func carryCodexQuotaOverdraftWindowStarts(target, current *CodexQuotaOverdraftProbeState, signal codexQuotaOverdraftSignal, now time.Time) { + if target == nil || current == nil { + return + } + if current.FiveHourRecoverAt != nil && current.FiveHourRecoverAt.After(now) && signal.FiveHourRecoverAt != nil { + delta := signal.FiveHourRecoverAt.Sub(*current.FiveHourRecoverAt) + if delta >= -2*time.Minute && delta <= 2*time.Minute { + target.FiveHourStartedAt = cloneTimePtr(current.FiveHourStartedAt) + } + } + if current.SevenDayRecoverAt != nil && current.SevenDayRecoverAt.After(now) && signal.SevenDayRecoverAt != nil { + delta := signal.SevenDayRecoverAt.Sub(*current.SevenDayRecoverAt) + if delta >= -2*time.Minute && delta <= 2*time.Minute { + target.SevenDayStartedAt = cloneTimePtr(current.SevenDayStartedAt) + } + } + target.OverdraftStartedAt = earlierCodexQuotaOverdraftStart(target.FiveHourStartedAt, target.SevenDayStartedAt) +} + +func startCodexQuotaOverdraftWindows(state *CodexQuotaOverdraftProbeState, signal codexQuotaOverdraftSignal, testedAt time.Time) { + if state == nil { + return + } + switch signal.Window { + case "5h": + if state.FiveHourStartedAt == nil { + state.FiveHourStartedAt = codexQuotaOverdraftTimePtr(testedAt) + } + case "7d": + if state.SevenDayStartedAt == nil { + state.SevenDayStartedAt = codexQuotaOverdraftTimePtr(testedAt) + } + default: + if state.FiveHourStartedAt == nil { + state.FiveHourStartedAt = codexQuotaOverdraftTimePtr(testedAt) + } + if state.SevenDayStartedAt == nil { + state.SevenDayStartedAt = codexQuotaOverdraftTimePtr(testedAt) + } + } + state.OverdraftStartedAt = earlierCodexQuotaOverdraftStart(state.FiveHourStartedAt, state.SevenDayStartedAt) +} + +func codexQuotaOverdraftWindowStarts(state *CodexQuotaOverdraftProbeState) (*time.Time, *time.Time) { + if state == nil { + return nil, nil + } + fiveStarted := cloneTimePtr(state.FiveHourStartedAt) + sevenStarted := cloneTimePtr(state.SevenDayStartedAt) + if fiveStarted == nil && sevenStarted == nil && state.OverdraftStartedAt != nil { + switch state.QuotaWindow { + case "5h": + fiveStarted = cloneTimePtr(state.OverdraftStartedAt) + case "7d": + sevenStarted = cloneTimePtr(state.OverdraftStartedAt) + default: + fiveStarted = cloneTimePtr(state.OverdraftStartedAt) + sevenStarted = cloneTimePtr(state.OverdraftStartedAt) + } + } + return fiveStarted, sevenStarted +} + +func earlierCodexQuotaOverdraftStart(left, right *time.Time) *time.Time { + switch { + case left == nil: + return cloneTimePtr(right) + case right == nil: + return cloneTimePtr(left) + case left.Before(*right): + return cloneTimePtr(left) + default: + return cloneTimePtr(right) + } +} + +func codexQuotaOverdraftPauseReason(reason string) bool { + payload, ok := parseTempUnschedReasonPayload(reason) + return ok && payload.Source == codexQuotaOverdraftPauseSource +} + +func sameCodexQuotaOverdraftTime(left, right *time.Time) bool { + if left == nil || right == nil { + return false + } + delta := left.Sub(*right) + if delta < 0 { + delta = -delta + } + return delta <= time.Second +} + +func codexQuotaOverdraftResetAt(raw any, now time.Time) *time.Time { + if raw == nil { + return nil + } + parsed, err := parseTime(strings.TrimSpace(fmt.Sprint(raw))) + if err != nil { + return nil + } + parsed = parsed.UTC() + return &parsed +} + +func cloneCodexQuotaOverdraftAccount(account *Account) *Account { + if account == nil { + return nil + } + clone := *account + clone.Extra = shallowCopyMap(account.Extra) + clone.Credentials = shallowCopyMap(account.Credentials) + return &clone +} + +func laterTime(left, right time.Time) time.Time { + if right.After(left) { + return right + } + return left +} + +func codexQuotaOverdraftTimePtr(value time.Time) *time.Time { + value = value.UTC() + return &value +} diff --git a/backend/internal/service/openai_codex_quota_overdraft_probe_test.go b/backend/internal/service/openai_codex_quota_overdraft_probe_test.go new file mode 100644 index 000000000..fa1d08354 --- /dev/null +++ b/backend/internal/service/openai_codex_quota_overdraft_probe_test.go @@ -0,0 +1,319 @@ +//go:build unit + +package service + +import ( + "context" + "encoding/json" + "net/http" + "strconv" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +type codexOverdraftProbeRepoStub struct { + AccountRepository + account *Account + states []*CodexQuotaOverdraftProbeState + tempPauseCalls int + clearTempCalls int + clearLimitCalls int +} + +func (r *codexOverdraftProbeRepoStub) GetByID(context.Context, int64) (*Account, error) { + return r.account, nil +} + +func (r *codexOverdraftProbeRepoStub) UpdateExtra(_ context.Context, _ int64, updates map[string]any) error { + if r.account.Extra == nil { + r.account.Extra = make(map[string]any) + } + for key, value := range updates { + r.account.Extra[key] = value + } + if state, ok := updates[CodexQuotaOverdraftProbeExtraKey].(*CodexQuotaOverdraftProbeState); ok { + clone := *state + r.states = append(r.states, &clone) + } + return nil +} + +func (r *codexOverdraftProbeRepoStub) SetTempUnschedulable(_ context.Context, _ int64, until time.Time, reason string) error { + r.tempPauseCalls++ + r.account.TempUnschedulableUntil = codexQuotaOverdraftTimePtr(until) + r.account.TempUnschedulableReason = reason + return nil +} + +func (r *codexOverdraftProbeRepoStub) ClearTempUnschedulable(context.Context, int64) error { + r.clearTempCalls++ + r.account.TempUnschedulableUntil = nil + r.account.TempUnschedulableReason = "" + return nil +} + +func (r *codexOverdraftProbeRepoStub) ClearRateLimit(context.Context, int64) error { + r.clearLimitCalls++ + r.account.RateLimitResetAt = nil + return nil +} + +func newCodexOverdraftProbeTestAccount(now time.Time) *Account { + return &Account{ + ID: 77, + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + Status: StatusActive, + Schedulable: true, + Extra: map[string]any{ + "codex_5h_used_percent": 100, + "codex_5h_reset_at": now.Add(5 * time.Hour).Format(time.RFC3339), + }, + } +} + +func TestCodexQuotaOverdraftProbeModelsMatchPluginRotation(t *testing.T) { + models := codexQuotaOverdraftProbeModels("gpt-5.4") + require.Equal(t, []string{"gpt-5.4", "gpt-5.5", "gpt-5.4-mini"}, models) + + got := make([]string, 0, codexQuotaOverdraftProbeAttemptLimit) + for attempt := 0; attempt < codexQuotaOverdraftProbeAttemptLimit; attempt++ { + got = append(got, models[attempt%len(models)]) + } + require.Equal(t, []string{"gpt-5.4", "gpt-5.5", "gpt-5.4-mini", "gpt-5.4", "gpt-5.5"}, got) +} + +func TestCodexQuotaOverdraftSignalsKeepFiveHourAndSevenDayCyclesSeparate(t *testing.T) { + now := time.Date(2026, time.August, 13, 14, 0, 0, 0, time.UTC) + fiveReset := now.Add(5 * time.Hour) + sevenReset := now.Add(6 * 24 * time.Hour) + account := newCodexOverdraftProbeTestAccount(now) + + five, exhausted := codexQuotaOverdraftSignalFromAccount(account, nil, now) + require.True(t, exhausted) + require.Equal(t, "5h", five.Window) + require.Equal(t, "5h:"+formatCodexOverdraftUnix(fiveReset), five.CycleKey) + require.WithinDuration(t, fiveReset, five.RecoverAt, time.Second) + + account.Extra["codex_7d_used_percent"] = 100 + account.Extra["codex_7d_reset_at"] = sevenReset.Format(time.RFC3339) + multiple, exhausted := codexQuotaOverdraftSignalFromAccount(account, nil, now) + require.True(t, exhausted) + require.Equal(t, "multiple", multiple.Window) + require.WithinDuration(t, sevenReset, multiple.RecoverAt, time.Second, "多个额度周期必须等待最晚恢复时间") + require.Contains(t, multiple.CycleKey, "5h:") + require.Contains(t, multiple.CycleKey, "|7d:") +} + +func TestCodexQuotaOverdraftProbePassesOnAnySuccessfulAttempt(t *testing.T) { + now := time.Date(2026, time.August, 13, 14, 0, 0, 0, time.UTC) + account := newCodexOverdraftProbeTestAccount(now) + repo := &codexOverdraftProbeRepoStub{account: account} + coordinator := &CodexQuotaOverdraftCoordinator{accountRepo: repo, now: func() time.Time { return now }} + models := make([]string, 0, 3) + coordinator.probeAttemptForTest = func(_ context.Context, _ *Account, model string) codexQuotaOverdraftProbeResult { + models = append(models, model) + if len(models) == 3 { + return codexQuotaOverdraftProbeResult{Status: "available", ReasonCode: "model_response_ok", StatusCode: http.StatusOK, Model: model} + } + return codexQuotaOverdraftProbeResult{Status: "retry", ReasonCode: "quota_limited", StatusCode: http.StatusTooManyRequests, Model: model} + } + signal, exhausted := codexQuotaOverdraftSignalFromAccount(account, nil, now) + require.True(t, exhausted) + state := newCodexOverdraftPendingState(signal, now) + + coordinator.runProbePlan(account.ID, signal, "gpt-5.4", state) + + require.Equal(t, codexQuotaOverdraftProbePassed, state.Status) + require.Equal(t, 3, state.Attempts) + require.Equal(t, []string{"gpt-5.4", "gpt-5.5", "gpt-5.4-mini"}, models) + require.NotNil(t, state.FiveHourStartedAt) + require.Zero(t, repo.tempPauseCalls) +} + +func TestCodexQuotaOverdraftProbeRequiresFiveExplicitQuotaFailures(t *testing.T) { + now := time.Date(2026, time.August, 13, 14, 0, 0, 0, time.UTC) + account := newCodexOverdraftProbeTestAccount(now) + repo := &codexOverdraftProbeRepoStub{account: account} + coordinator := &CodexQuotaOverdraftCoordinator{accountRepo: repo, now: func() time.Time { return now }} + models := make([]string, 0, codexQuotaOverdraftProbeAttemptLimit) + coordinator.probeAttemptForTest = func(_ context.Context, _ *Account, model string) codexQuotaOverdraftProbeResult { + models = append(models, model) + return codexQuotaOverdraftProbeResult{Status: "retry", ReasonCode: "quota_limited", StatusCode: http.StatusTooManyRequests, Model: model} + } + signal, _ := codexQuotaOverdraftSignalFromAccount(account, nil, now) + state := newCodexOverdraftPendingState(signal, now) + + coordinator.runProbePlan(account.ID, signal, "gpt-5.4", state) + + require.Equal(t, codexQuotaOverdraftProbeFailed, state.Status) + require.Equal(t, codexQuotaOverdraftProbeAttemptLimit, state.Attempts) + require.Len(t, models, codexQuotaOverdraftProbeAttemptLimit) + require.Equal(t, 1, repo.tempPauseCalls) + require.True(t, codexQuotaOverdraftPauseReason(account.TempUnschedulableReason)) + require.False(t, codexQuotaOverdraftSchedulingAllowed(account, now), "同周期确认失败后不得继续绕过限额") +} + +func TestCodexQuotaOverdraftProbeInconclusiveNeverPauses(t *testing.T) { + now := time.Date(2026, time.August, 13, 14, 0, 0, 0, time.UTC) + for _, result := range []codexQuotaOverdraftProbeResult{ + {Status: "inconclusive", ReasonCode: "request_timeout", StatusCode: http.StatusGatewayTimeout, Model: "gpt-5.4"}, + {Status: "inconclusive", ReasonCode: "upstream_unavailable", StatusCode: http.StatusServiceUnavailable, Model: "gpt-5.4"}, + } { + t.Run(result.ReasonCode, func(t *testing.T) { + account := newCodexOverdraftProbeTestAccount(now) + repo := &codexOverdraftProbeRepoStub{account: account} + coordinator := &CodexQuotaOverdraftCoordinator{accountRepo: repo, now: func() time.Time { return now }} + coordinator.probeAttemptForTest = func(context.Context, *Account, string) codexQuotaOverdraftProbeResult { return result } + signal, _ := codexQuotaOverdraftSignalFromAccount(account, nil, now) + state := newCodexOverdraftPendingState(signal, now) + + coordinator.runProbePlan(account.ID, signal, "gpt-5.4", state) + + require.Equal(t, codexQuotaOverdraftProbeInconclusive, state.Status) + require.Equal(t, 1, state.Attempts) + require.NotNil(t, state.RetryAt) + require.Zero(t, repo.tempPauseCalls) + }) + } +} + +func TestCodexQuotaOverdraftNewWindowPreservesExistingWindowBaseline(t *testing.T) { + now := time.Date(2026, time.August, 13, 14, 0, 0, 0, time.UTC) + fiveStarted := now.Add(-time.Hour) + fiveRecover := now.Add(4 * time.Hour) + sevenRecover := now.Add(6 * 24 * time.Hour) + current := &CodexQuotaOverdraftProbeState{ + Status: codexQuotaOverdraftProbePassed, + QuotaWindow: "5h", + CycleKey: "5h:" + formatCodexOverdraftUnix(fiveRecover), + FiveHourRecoverAt: codexQuotaOverdraftTimePtr(fiveRecover), + FiveHourStartedAt: codexQuotaOverdraftTimePtr(fiveStarted), + } + signal := codexQuotaOverdraftSignal{ + Window: "multiple", + CycleKey: "5h:" + formatCodexOverdraftUnix(fiveRecover) + "|7d:" + formatCodexOverdraftUnix(sevenRecover), + RecoverAt: sevenRecover, + FiveHourRecoverAt: codexQuotaOverdraftTimePtr(fiveRecover), + SevenDayRecoverAt: codexQuotaOverdraftTimePtr(sevenRecover), + } + target := newCodexOverdraftPendingState(signal, now) + carryCodexQuotaOverdraftWindowStarts(target, current, signal, now) + startCodexQuotaOverdraftWindows(target, signal, now) + + require.NotNil(t, target.FiveHourStartedAt) + require.True(t, target.FiveHourStartedAt.Equal(fiveStarted), "已有 5h 透支统计起点不能因 7d 后续耗尽而重置") + require.NotNil(t, target.SevenDayStartedAt) + require.True(t, target.SevenDayStartedAt.Equal(now)) +} + +func TestCodexQuotaOverdraftFailedMultipleCycleStillCoversRemainingWindow(t *testing.T) { + now := time.Date(2026, time.August, 13, 14, 0, 0, 0, time.UTC) + fiveRecover := now.Add(5 * time.Hour) + sevenRecover := now.Add(6 * 24 * time.Hour) + state := &CodexQuotaOverdraftProbeState{ + Status: codexQuotaOverdraftProbeFailed, + QuotaWindow: "multiple", + CycleKey: "5h:" + formatCodexOverdraftUnix(fiveRecover) + "|7d:" + formatCodexOverdraftUnix(sevenRecover), + FiveHourRecoverAt: codexQuotaOverdraftTimePtr(fiveRecover), + SevenDayRecoverAt: codexQuotaOverdraftTimePtr(sevenRecover), + } + remaining := codexQuotaOverdraftSignal{ + Window: "7d", + CycleKey: "7d:" + formatCodexOverdraftUnix(sevenRecover), + RecoverAt: sevenRecover, + SevenDayRecoverAt: codexQuotaOverdraftTimePtr(sevenRecover), + } + + require.True(t, codexQuotaOverdraftStateCoversSignal(state, remaining)) +} + +func TestCodexQuotaOverdraftRecoveryClearsWindowState(t *testing.T) { + now := time.Date(2026, time.August, 13, 14, 0, 0, 0, time.UTC) + account := newCodexOverdraftProbeTestAccount(now) + account.Extra["codex_5h_used_percent"] = 0 + started := now.Add(-time.Hour) + recoverAt := now.Add(4 * time.Hour) + state := &CodexQuotaOverdraftProbeState{ + Status: codexQuotaOverdraftProbePassed, + QuotaWindow: "5h", + CycleKey: "5h:" + formatCodexOverdraftUnix(recoverAt), + RecoverAt: codexQuotaOverdraftTimePtr(recoverAt), + FiveHourRecoverAt: codexQuotaOverdraftTimePtr(recoverAt), + OverdraftStartedAt: codexQuotaOverdraftTimePtr(started), + FiveHourStartedAt: codexQuotaOverdraftTimePtr(started), + } + account.Extra[CodexQuotaOverdraftProbeExtraKey] = state + repo := &codexOverdraftProbeRepoStub{account: account} + coordinator := &CodexQuotaOverdraftCoordinator{accountRepo: repo, now: func() time.Time { return now }} + + coordinator.observeAccount(account, "gpt-5.4") + + recovered, ok := codexQuotaOverdraftStateFromAccount(account) + require.True(t, ok) + require.Equal(t, codexQuotaOverdraftProbeRecovered, recovered.Status) + require.Equal(t, "quota_recovered", recovered.ReasonCode) + require.Nil(t, recovered.OverdraftStartedAt) + require.Nil(t, recovered.FiveHourStartedAt) + require.Nil(t, recovered.FiveHourRecoverAt) + require.Zero(t, repo.tempPauseCalls) +} + +func TestClassifyCodexQuotaOverdraftProbeResponses(t *testing.T) { + status, reason := classifyCodexQuotaOverdraftProbe(http.StatusOK, nil, []byte(`data: {"type":"response.completed"}`)) + require.Equal(t, "available", status) + require.Equal(t, "model_response_ok", reason) + + status, reason = classifyCodexQuotaOverdraftProbe(http.StatusTooManyRequests, nil, []byte(`{"error":{"type":"usage_limit_reached"}}`)) + require.Equal(t, "retry", status) + require.Equal(t, "quota_limited", reason) + + status, reason = classifyCodexQuotaOverdraftProbe(http.StatusServiceUnavailable, nil, nil) + require.Equal(t, "inconclusive", status) + require.Equal(t, "upstream_unavailable", reason) + + status, reason = classifyCodexQuotaOverdraftProbe(http.StatusUnauthorized, nil, nil) + require.Equal(t, "authentication_failed", status) + require.Equal(t, "authentication_failed", reason) + + status, reason = classifyCodexQuotaOverdraftProbe(http.StatusOK, nil, []byte(`data: {"type":"response.failed"}\ndata: {"type":"response.completed"}`)) + require.Equal(t, "retry", status) + require.Equal(t, "invalid_response", reason) +} + +func TestCodexQuotaOverdraftProbePayloadAcceptsInjection(t *testing.T) { + payload := map[string]any{ + "model": "gpt-5.4", + "input": []map[string]any{{ + "type": "message", + "role": "user", + "content": []map[string]string{{"type": "input_text", "text": "hi"}}, + }}, + } + raw, err := json.Marshal(payload) + require.NoError(t, err) + updated, changed, err := injectCodexQuotaOverdraft(raw) + require.NoError(t, err) + require.True(t, changed) + require.Contains(t, string(updated), `"custom_tool_call"`) +} + +func newCodexOverdraftPendingState(signal codexQuotaOverdraftSignal, now time.Time) *CodexQuotaOverdraftProbeState { + return &CodexQuotaOverdraftProbeState{ + Status: codexQuotaOverdraftProbePending, + QuotaWindow: signal.Window, + CycleKey: signal.CycleKey, + Limit: codexQuotaOverdraftProbeAttemptLimit, + StartedAt: now, + RecoverAt: codexQuotaOverdraftTimePtr(signal.RecoverAt), + FiveHourRecoverAt: cloneTimePtr(signal.FiveHourRecoverAt), + SevenDayRecoverAt: cloneTimePtr(signal.SevenDayRecoverAt), + OverdraftStartedAt: nil, + } +} + +func formatCodexOverdraftUnix(value time.Time) string { + return strconv.FormatInt(value.Unix(), 10) +} diff --git a/backend/internal/service/openai_codex_quota_overdraft_test.go b/backend/internal/service/openai_codex_quota_overdraft_test.go new file mode 100644 index 000000000..4d20e2bc4 --- /dev/null +++ b/backend/internal/service/openai_codex_quota_overdraft_test.go @@ -0,0 +1,142 @@ +//go:build unit + +package service + +import ( + "context" + "encoding/json" + "strings" + "testing" + "time" + + "github.com/Wei-Shaw/sub2api/internal/config" + "github.com/stretchr/testify/require" +) + +func TestCodexQuotaOverdraftInjection(t *testing.T) { + t.Cleanup(func() { SetCodexQuotaOverdraftEnabled(false) }) + SetCodexQuotaOverdraftEnabled(true) + ctx := WithCodexQuotaOverdraftScheduling(context.Background()) + svc := &OpenAIGatewayService{cfg: &config.Config{Gateway: config.GatewayConfig{CodexQuotaOverdraftEnabled: true}}} + oauth := &Account{Platform: PlatformOpenAI, Type: AccountTypeOAuth} + body := []byte(`{"model":"gpt-5.4","input":[{"type":"message","role":"user","content":[{"type":"input_text","text":"hello"}]}]}`) + + updated := svc.prepareCodexQuotaOverdraftBody(ctx, oauth, false, body) + require.NotEqual(t, string(body), string(updated)) + + var document codexQuotaOverdraftDocument + require.NoError(t, json.Unmarshal(updated, &document)) + require.Len(t, document.Input, 3) + var call, output codexQuotaOverdraftInputItem + require.NoError(t, json.Unmarshal(document.Input[1], &call)) + require.NoError(t, json.Unmarshal(document.Input[2], &output)) + require.Equal(t, "custom_tool_call", call.Type) + require.Equal(t, "custom_tool_call_output", output.Type) + require.True(t, strings.HasPrefix(call.CallID, codexQuotaOverdraftCallIDPrefix)) + require.Equal(t, call.CallID, output.CallID) + + again := svc.prepareCodexQuotaOverdraftBody(ctx, oauth, false, updated) + require.Equal(t, string(updated), string(again), "重复处理不能再次注入") +} + +func TestCodexQuotaOverdraftInjectionGuards(t *testing.T) { + t.Cleanup(func() { SetCodexQuotaOverdraftEnabled(false) }) + body := []byte(`{"input":[{"type":"message","role":"user"}]}`) + oauth := &Account{Platform: PlatformOpenAI, Type: AccountTypeOAuth} + enabledCfg := &config.Config{Gateway: config.GatewayConfig{CodexQuotaOverdraftEnabled: true}} + svc := &OpenAIGatewayService{cfg: enabledCfg} + + SetCodexQuotaOverdraftEnabled(false) + require.Equal(t, string(body), string(svc.prepareCodexQuotaOverdraftBody(WithCodexQuotaOverdraftScheduling(context.Background()), oauth, false, body))) + + SetCodexQuotaOverdraftEnabled(true) + require.Equal(t, string(body), string(svc.prepareCodexQuotaOverdraftBody(context.Background(), oauth, false, body)), "未标记的端点不能注入") + require.Equal(t, string(body), string(svc.prepareCodexQuotaOverdraftBody(WithCodexQuotaOverdraftScheduling(context.Background()), oauth, true, body)), "compact 不能注入") + + ctx := WithCodexQuotaOverdraftScheduling(context.Background()) + for _, account := range []*Account{ + {Platform: PlatformOpenAI, Type: AccountTypeAPIKey}, + {Platform: PlatformOpenAI, Type: AccountTypeOAuth, ParentAccountID: int64PtrForCodexQuotaOverdraftTest(1)}, + } { + require.Equal(t, string(body), string(svc.prepareCodexQuotaOverdraftBody(ctx, account, false, body))) + } + agentIdentity := &Account{Platform: PlatformOpenAI, Type: AccountTypeOAuth, Credentials: map[string]any{openAIAuthModeCredentialKey: OpenAIAuthModeAgentIdentity}} + require.NotEqual(t, string(body), string(svc.prepareCodexQuotaOverdraftBody(ctx, agentIdentity, false, body)), "Agent Identity 必须支持透支请求注入") + + notUserLast := []byte(`{"input":[{"type":"message","role":"assistant"}]}`) + require.Equal(t, string(notUserLast), string(svc.prepareCodexQuotaOverdraftBody(ctx, oauth, false, notUserLast))) + invalid := []byte(`{"input":`) + require.Equal(t, string(invalid), string(svc.prepareCodexQuotaOverdraftBody(ctx, oauth, false, invalid))) + oversized := make([]byte, codexQuotaOverdraftMaxBodyBytes+1) + require.Equal(t, oversized, svc.prepareCodexQuotaOverdraftBody(ctx, oauth, false, oversized)) +} + +func TestCodexQuotaOverdraftSchedulingOnlyBypassesQuotaThresholds(t *testing.T) { + t.Cleanup(func() { SetCodexQuotaOverdraftEnabled(false) }) + SetCodexQuotaOverdraftEnabled(true) + ctx := WithCodexQuotaOverdraftScheduling(context.Background()) + now := time.Now().UTC() + reset := now.Add(time.Hour) + account := &Account{ + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + Status: StatusActive, + Schedulable: true, + Extra: map[string]any{ + "codex_5h_used_percent": 100, + "codex_5h_reset_at": reset.Format(time.RFC3339), + }, + } + quotaCtx := withOpenAIQuotaAutoPauseSettings(ctx, OpsOpenAIAccountQuotaAutoPauseSettings{DefaultThreshold5h: 0.8}) + paused, _ := shouldAutoPauseOpenAIAccountByQuota(quotaCtx, account) + require.False(t, paused) + + account.RateLimitResetAt = &reset + require.False(t, account.IsSchedulableForModelWithContext(quotaCtx, "gpt-5.4"), "真实 429 限流仍必须生效") + account.RateLimitResetAt = nil + account.TempUnschedulableUntil = &reset + account.TempUnschedulableReason = BuildTempUnschedReasonPayload("oauth_401", "unauthorized") + require.Same(t, account, normalizeCodexQuotaOverdraftAccountForScheduling(quotaCtx, account), "其他临时暂停不能绕过") + + account.TempUnschedulableReason = BuildAccountSchedulingThresholdReason("") + normalized := normalizeCodexQuotaOverdraftAccountForScheduling(quotaCtx, account) + require.NotSame(t, account, normalized) + require.Nil(t, normalized.TempUnschedulableUntil) + require.Empty(t, normalized.TempUnschedulableReason) + require.NotNil(t, account.TempUnschedulableUntil, "不能修改缓存或数据库账号原对象") +} + +func TestRateLimitServiceCodexQuotaOverdraftDoesNotCreateRuntimeThresholdBlock(t *testing.T) { + t.Cleanup(func() { SetCodexQuotaOverdraftEnabled(false) }) + SetCodexQuotaOverdraftEnabled(true) + accountSchedulingThresholdsSF.Forget(SettingKeyAccountSchedulingThresholds) + accountSchedulingThresholdsCache.Store(&cachedAccountSchedulingThresholds{}) + + settingsRepo := newMockSettingRepo() + settingsRepo.data[SettingKeyAccountSchedulingThresholds] = `{"openai":80}` + accountRepo := &rateLimitAccountRepoStub{} + runtimeBlocker := &runtimeBlockRecorder{} + rl := NewRateLimitService(accountRepo, nil, &config.Config{}, nil, nil) + rl.SetSettingService(NewSettingService(settingsRepo, &config.Config{})) + rl.SetAccountRuntimeBlocker(runtimeBlocker) + reset := time.Now().UTC().Add(time.Hour) + account := &Account{ + ID: 9001, + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + Status: StatusActive, + Schedulable: true, + Extra: map[string]any{ + "codex_7d_used_percent": 100, + "codex_7d_reset_at": reset.Format(time.RFC3339), + }, + } + + require.True(t, rl.ApplyAccountSchedulingThreshold(context.Background(), account)) + require.Equal(t, 1, accountRepo.tempCalls, "普通端点仍应持久化阈值暂停") + require.Empty(t, runtimeBlocker.reasons, "阈值暂停不能误写为所有请求共享的 runtime blocker") +} + +func int64PtrForCodexQuotaOverdraftTest(value int64) *int64 { + return &value +} diff --git a/backend/internal/service/openai_gateway_forward.go b/backend/internal/service/openai_gateway_forward.go index f7eb75934..dfb29d1aa 100644 --- a/backend/internal/service/openai_gateway_forward.go +++ b/backend/internal/service/openai_gateway_forward.go @@ -1029,6 +1029,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco } func (s *OpenAIGatewayService) buildUpstreamRequest(ctx context.Context, c *gin.Context, account *Account, body []byte, token string, isStream bool, promptCacheKey string, isCodexCLI bool) (*http.Request, error) { + body = s.prepareCodexQuotaOverdraftBody(ctx, account, isOpenAIResponsesCompactPath(c), body) // Determine target URL based on account type var targetURL string switch account.Type { diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index 8c029f0aa..d712ba25a 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -340,6 +340,7 @@ func (s *OpenAIGatewayService) buildUpstreamRequestOpenAIPassthrough( body []byte, token string, ) (*http.Request, error) { + body = s.prepareCodexQuotaOverdraftBody(ctx, account, isOpenAIResponsesCompactPath(c), body) targetURL := openaiPlatformAPIURL switch account.Type { case AccountTypeOAuth: diff --git a/backend/internal/service/openai_gateway_scheduling.go b/backend/internal/service/openai_gateway_scheduling.go index be25b72b0..301ba997d 100644 --- a/backend/internal/service/openai_gateway_scheduling.go +++ b/backend/internal/service/openai_gateway_scheduling.go @@ -445,6 +445,10 @@ 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()) { + return false, openAIQuotaAutoPauseDecision{} + } // Per-account explicit-disable flags must take precedence over the global default. // Without these, leaving the account threshold blank means "use global default", // so an admin has no way to exempt a single account from auto-pause once a global @@ -1277,11 +1281,28 @@ 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 s.schedulerSnapshot != nil { accounts, _, err := s.schedulerSnapshot.ListSchedulableAccounts(ctx, groupID, platform, false) if err != nil { return accounts, err } + accounts = normalizeCodexQuotaOverdraftAccountsForScheduling(ctx, accounts) accounts = s.filterOpenAIAccountsBySchedulingThreshold(ctx, accounts) if platform == PlatformGrok { accounts = s.filterGrokFreeQuotaAccountsForOpenAI(ctx, accounts) @@ -1300,6 +1321,7 @@ func (s *OpenAIGatewayService) listSchedulableAccounts(ctx context.Context, grou if err != nil { return nil, fmt.Errorf("query accounts failed: %w", err) } + accounts = normalizeCodexQuotaOverdraftAccountsForScheduling(ctx, accounts) accounts = s.filterOpenAIAccountsBySchedulingThreshold(ctx, accounts) if platform == PlatformGrok { accounts = s.filterGrokFreeQuotaAccountsForOpenAI(ctx, accounts) @@ -1339,6 +1361,7 @@ func (s *OpenAIGatewayService) resolveFreshSchedulableOpenAIAccountBeforeProfit( } fresh = current } + fresh = normalizeCodexQuotaOverdraftAccountForScheduling(ctx, fresh) if !isOpenAICompatibleAccountEligibleForRequestBeforeProfit(ctx, fresh, platform, requestedModel, requireCompact, requiredCapability) { return nil @@ -1408,6 +1431,7 @@ func (s *OpenAIGatewayService) recheckSelectedOpenAIAccountFromDBBeforeProfit(ct if err != nil || latest == nil { return nil } + latest = normalizeCodexQuotaOverdraftAccountForScheduling(ctx, latest) if !s.openAIAccountMatchesSchedulingGroup(latest, groupID) { return nil } @@ -1449,6 +1473,7 @@ func (s *OpenAIGatewayService) getSchedulableAccount(ctx context.Context, accoun if err != nil || account == nil { return account, err } + account = normalizeCodexQuotaOverdraftAccountForScheduling(ctx, account) if s.isOpenAIAccountBlockedBySchedulingThreshold(ctx, account) { return nil, nil } @@ -1489,6 +1514,10 @@ 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()) { + 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 55353e1d0..1f0893afa 100644 --- a/backend/internal/service/openai_gateway_service.go +++ b/backend/internal/service/openai_gateway_service.go @@ -416,6 +416,7 @@ type OpenAIGatewayService struct { concurrencyService *ConcurrencyService billingService *BillingService rateLimitService *RateLimitService + codexQuotaOverdraft *CodexQuotaOverdraftCoordinator billingCacheService *BillingCacheService userGroupRateResolver *userGroupRateResolver httpUpstream HTTPUpstream @@ -464,6 +465,12 @@ type OpenAIGatewayService struct { openaiCompatAnthropicDigestSessions sync.Map } +func (s *OpenAIGatewayService) SetCodexQuotaOverdraftCoordinator(coordinator *CodexQuotaOverdraftCoordinator) { + if s != nil { + s.codexQuotaOverdraft = coordinator + } +} + // NewOpenAIGatewayService creates a new OpenAIGatewayService func NewOpenAIGatewayService( accountRepo AccountRepository, @@ -493,6 +500,7 @@ func NewOpenAIGatewayService( // 拿不到配置，故在此发布进程级开关快照。配置取反义，零值即「强制统一出口开启」。 if cfg != nil { SetCodexIdentityEnforcementEnabled(!cfg.Gateway.DisableCodexIdentityEnforcement) + SetCodexQuotaOverdraftEnabled(cfg.Gateway.CodexQuotaOverdraftEnabled) } svc := &OpenAIGatewayService{ accountRepo: accountRepo, diff --git a/backend/internal/service/openai_gateway_usage.go b/backend/internal/service/openai_gateway_usage.go index c55bcdbea..baa644ff5 100644 --- a/backend/internal/service/openai_gateway_usage.go +++ b/backend/internal/service/openai_gateway_usage.go @@ -1028,14 +1028,38 @@ func (s *OpenAIGatewayService) updateCodexUsageSnapshot(ctx context.Context, acc if len(updates) == 0 { return } - if !s.getCodexSnapshotThrottle().Allow(accountID, now) { - return - } + persistSnapshot := codexQuotaOverdraftSnapshotExhausted(updates) || s.getCodexSnapshotThrottle().Allow(accountID, now) go func() { updateCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - _ = s.accountRepo.UpdateExtra(updateCtx, accountID, updates) + 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) + s.codexQuotaOverdraft.ObserveAccount(account, "") }() } diff --git a/backend/internal/service/openai_ws_forwarder_ingress.go b/backend/internal/service/openai_ws_forwarder_ingress.go index ce7e8f038..3c5c5f03a 100644 --- a/backend/internal/service/openai_ws_forwarder_ingress.go +++ b/backend/internal/service/openai_ws_forwarder_ingress.go @@ -804,7 +804,8 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient( } turnStart := time.Now() wroteDownstream := false - if err := lease.WriteJSONWithContextTimeout(ctx, json.RawMessage(payload), s.openAIWSWriteTimeout()); err != nil { + outboundPayload := s.prepareCodexQuotaOverdraftBody(ctx, account, false, payload) + if err := lease.WriteJSONWithContextTimeout(ctx, json.RawMessage(outboundPayload), s.openAIWSWriteTimeout()); err != nil { return nil, wrapOpenAIWSIngressTurnError( "write_upstream", fmt.Errorf("write upstream websocket request: %w", err), diff --git a/backend/internal/service/openai_ws_forwarder_support.go b/backend/internal/service/openai_ws_forwarder_support.go index 6c6d7c9cf..d5a0ee34b 100644 --- a/backend/internal/service/openai_ws_forwarder_support.go +++ b/backend/internal/service/openai_ws_forwarder_support.go @@ -524,6 +524,7 @@ func (s *OpenAIGatewayService) resolveAccountByPreviousResponseIDForCapability( _ = store.DeleteResponseAccount(ctx, derefGroupID(groupID), responseID) return 0, nil, "", nil } + latest = normalizeCodexQuotaOverdraftAccountForScheduling(ctx, latest) if shouldClearStickySession(latest, requestedModel) || !latest.IsOpenAI() || !latest.IsSchedulable() { _ = store.DeleteResponseAccount(ctx, derefGroupID(groupID), responseID) return 0, nil, "", nil diff --git a/backend/internal/service/openai_ws_forwarder_v2.go b/backend/internal/service/openai_ws_forwarder_v2.go index e0dbd58bd..5ea286779 100644 --- a/backend/internal/service/openai_ws_forwarder_v2.go +++ b/backend/internal/service/openai_ws_forwarder_v2.go @@ -324,6 +324,7 @@ func (s *OpenAIGatewayService) forwardOpenAIWSV2( ); err != nil { return nil, err } + payload = s.prepareCodexQuotaOverdraftPayload(ctx, account, payload) if err := lease.WriteJSONWithContextTimeout(ctx, payload, s.openAIWSWriteTimeout()); err != nil { lease.MarkBroken() diff --git a/backend/internal/service/openai_ws_v2_passthrough_adapter.go b/backend/internal/service/openai_ws_v2_passthrough_adapter.go index 775e26980..3bf082a92 100644 --- a/backend/internal/service/openai_ws_v2_passthrough_adapter.go +++ b/backend/internal/service/openai_ws_v2_passthrough_adapter.go @@ -752,6 +752,7 @@ func (s *OpenAIGatewayService) proxyResponsesWebSocketV2Passthrough( return NewOpenAIWSClientCloseError(coderws.StatusPolicyViolation, blocked.Message, blocked) } firstClientMessage = updatedFirst + firstClientMessage = s.prepareCodexQuotaOverdraftBody(ctx, account, false, firstClientMessage) // 在 policy filter 之后再提取 service_tier / reasoning_effort 用于 // usage 上报：filter @@ -1034,6 +1035,7 @@ func (s *OpenAIGatewayService) proxyResponsesWebSocketV2Passthrough( // 覆盖（Store(nil)），因为 OpenAI 上游对该帧实际不传 // service_tier 时按 default 处理，billing 应如实反映。 if policyErr == nil && blocked == nil && isResponseCreate { + out = s.prepareCodexQuotaOverdraftBody(ctx, account, false, out) usageMeta.updateFromResponseCreate(out, model, requestModelForThisFrame) acceptedTurn = true } diff --git a/backend/internal/service/ratelimit_service.go b/backend/internal/service/ratelimit_service.go index ee5b09bbb..b56b2b015 100644 --- a/backend/internal/service/ratelimit_service.go +++ b/backend/internal/service/ratelimit_service.go @@ -149,6 +149,10 @@ func (s *RateLimitService) ApplyAccountSchedulingThreshold(ctx context.Context, if !account.IsActive() || !account.Schedulable { return false } + if codexQuotaOverdraftSchedulingEnabled(ctx) && isCodexQuotaOverdraftAccount(account) && + codexQuotaOverdraftSchedulingAllowed(account, time.Now().UTC()) { + return false + } now := time.Now().UTC() thresholds := s.settingService.GetAccountSchedulingThresholds(ctx) @@ -176,7 +180,12 @@ func (s *RateLimitService) ApplyAccountSchedulingThreshold(ctx context.Context, account.TempUnschedulableUntil = cloneTimePtr(decision.Until) account.TempUnschedulableReason = reason - s.notifyAccountSchedulingBlocked(account, *decision.Until, "account_scheduling_threshold") + // 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") + } 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 2e26c0035..2b58b8797 100644 --- a/backend/internal/service/wire.go +++ b/backend/internal/service/wire.go @@ -202,6 +202,7 @@ func ProvideAccountUsageService( identityCache IdentityCache, tlsFPProfileService *TLSFingerprintProfileService, openAIGatewayService *OpenAIGatewayService, + codexQuotaOverdraft *CodexQuotaOverdraftCoordinator, ) *AccountUsageService { service := NewAccountUsageService( accountRepo, @@ -217,9 +218,36 @@ func ProvideAccountUsageService( tlsFPProfileService, ) service.agentIdentityWS = openAIGatewayService + service.SetCodexQuotaOverdraftCoordinator(codexQuotaOverdraft) 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, @@ -799,6 +827,7 @@ var ProviderSet = wire.NewSet( ProvideClaudeTokenProvider, NewAntigravityGatewayService, ProvideRateLimitService, + ProvideCodexQuotaOverdraftCoordinator, ProvideAccountUsageService, ProvideAccountTestService, ProvideUpstreamBillingProbeService, diff --git a/deploy/config.example.yaml b/deploy/config.example.yaml index 095c55099..8d61d38ad 100644 --- a/deploy/config.example.yaml +++ b/deploy/config.example.yaml @@ -287,6 +287,11 @@ gateway: # 默认 false：保持纯文本 Codex 请求不被改写；客户端显式提供 image_generation tool 时， # 仍会在分组允许图片生成的情况下正常转发。 codex_image_generation_bridge_enabled: false + # Experimental: allow eligible OpenAI OAuth Codex accounts to keep serving + # after the reported 5h/7d quota reaches the scheduling threshold. + # 实验性 Codex 额度透支：为符合条件的 OAuth 请求注入空操作工具调用，并忽略 + # 5h/7d 用量百分比预暂停。真实上游 429 仍会触发账号冷却。此部署示例默认开启。 + codex_quota_overdraft_enabled: true # Optional: template file used to build the final top-level Codex `instructions`. # 可选：用于构建最终 Codex 顶层 `instructions` 的模板文件路径。 # diff --git a/frontend/src/components/account/AccountUsageCell.vue b/frontend/src/components/account/AccountUsageCell.vue index 485b4b870..4e7aefb51 100644 --- a/frontend/src/components/account/AccountUsageCell.vue +++ b/frontend/src/components/account/AccountUsageCell.vue @@ -119,12 +119,32 @@