diff --git a/CODEX_OVERDRAFT_DEPLOYMENT_CN.md b/CODEX_OVERDRAFT_DEPLOYMENT_CN.md index 7b22e9592..b56121bef 100644 --- a/CODEX_OVERDRAFT_DEPLOYMENT_CN.md +++ b/CODEX_OVERDRAFT_DEPLOYMENT_CN.md @@ -270,9 +270,13 @@ docker logs --since 30m sub2api 2>&1 | \ | --- | --- | | `codex_quota_overdraft_probe_passed` | 至少一次真实探测成功，透支成立，账号继续参与调度 | | `codex_quota_overdraft_probe_failed` | 5 次都明确返回额度限制，账号暂停至额度恢复时间 | -| `codex_quota_overdraft_probe_inconclusive` | 网络、超时、5xx 等导致无法确认，约 1 分钟后可重试 | +| `codex_quota_overdraft_probe_attempt` | 单次探测结果，包含模型、HTTP 状态码、结果和原因，不记录凭据及完整响应正文 | +| `codex_quota_overdraft_probe_inconclusive` | 网络、超时、5xx 等导致无法确认，按 1、3、10 分钟退避重试 | +| `codex_quota_overdraft_retry_scheduled` | 已安排主动重试 | +| `codex_quota_overdraft_retry_started` | 主动重试开始；服务重启后也会由数据库到期扫描恢复 | | `codex_quota_overdraft_probe_claim_failed` | 无法在 PostgreSQL 中原子领取探测任务，需要排查数据库或版本 | | `codex_quota_overdraft_state_persist_failed` | 状态无法写入 `accounts.extra`，页面可能不显示最新结果 | +| `codex_quota_overdraft_pause_applied` | `failed` 状态、账号暂停和调度通知已经原子提交 | | `codex_quota_overdraft_stale_rate_limit_cleared` | 探测已证明账号可用，并清除了并发 429 遗留的账号级限流状态 | 真正“透支成功”的最直接证据是： @@ -312,7 +316,7 @@ docker compose \ ## 探测逻辑 -每个额度周期最多探测 5 次，每次最长 20 秒。模型按以下顺序循环： +每个额度周期的单轮计划最多探测 5 次，每次最长 20 秒。模型按以下顺序循环： ```text 首选模型 -> gpt-5.5 -> gpt-5.4-mini -> 首选模型 -> gpt-5.5 @@ -324,6 +328,8 @@ docker compose \ - 401/403 继续由原有认证失效逻辑处理。 - 账号禁用、过载、代理故障、模型冷却等非额度限制不会被本功能绕过。 +无法确认时，`retry_count` 记录当前退避轮次。后台会依次等待 1 分钟、3 分钟和 10 分钟执行最多 3 轮自动重试；每分钟的数据库扫描会恢复因容器重启而丢失的内存定时器，并接续超过 2 分钟的遗留 `pending` 状态。达到自动重试上限后不会持续消耗上游请求，但后续额度刷新、账号测试或新的明确额度响应仍可继续观察。 + 多实例部署通过 PostgreSQL 原子领取（atomic claim）保证同一账号、同一额度周期只有一个实例发起探测。 ## 反向代理 diff --git a/CODEX_QUOTA_OVERDRAFT_CUSTOMIZATION.md b/CODEX_QUOTA_OVERDRAFT_CUSTOMIZATION.md index 6ff5d812d..cd63fe977 100644 --- a/CODEX_QUOTA_OVERDRAFT_CUSTOMIZATION.md +++ b/CODEX_QUOTA_OVERDRAFT_CUSTOMIZATION.md @@ -65,10 +65,11 @@ gateway: 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 分别维护透支起点，不会因另一窗口后来耗尽而重置已有统计。 +5. 只有 5 次均返回明确 `quota_limited` 才判定 `failed`，账号暂停到对应额度恢复时间；5h 和 7d 同时耗尽时取最晚恢复时间。`failed` 状态、临时暂停和调度 outbox 在同一数据库事务中提交，避免出现状态已失败但账号没有暂停。 +6. 网络错误、超时、5xx、普通瞬时 429 和无效响应均判定 `inconclusive`，不会误停账号；后台按 1 分钟、3 分钟、10 分钟主动重试，并每分钟扫描数据库恢复服务重启前已到期的任务。重启时遗留超过 2 分钟的 `pending` 状态也会自动接续。每个额度周期最多自动重试 3 轮，之后仍可由额度刷新或新请求继续观察。 +7. 401/403、账号停用等认证问题交给原有认证异常逻辑处理，不加入额度重试；400/404 会轮换全部探测模型后再判定为 `inconclusive`，不会暂停整个账号。 +8. 账号禁用、过载、代理/传输故障、模型级冷却及其他临时不可调度原因不被绕过。 +9. 额度恢复后状态改为 `recovered`，清理本功能或额度阈值产生的暂停；5h 与 7d 分别维护透支起点，不会因另一窗口后来耗尽而重置已有统计。 状态值包括 `pending`、`passed`、`failed`、`inconclusive`、`recovered`。账号用量页面显示探测状态、尝试次数、额度周期、透支期成功请求数、Token、账号金额及预计恢复时间。 @@ -155,6 +156,10 @@ git diff --check codex_quota_overdraft_probe_passed codex_quota_overdraft_probe_failed codex_quota_overdraft_probe_inconclusive +codex_quota_overdraft_probe_attempt +codex_quota_overdraft_retry_scheduled +codex_quota_overdraft_retry_started +codex_quota_overdraft_pause_applied codex_quota_overdraft_stale_rate_limit_cleared ``` diff --git a/FORK_VERSION b/FORK_VERSION index fb7f732ab..115bae055 100644 --- a/FORK_VERSION +++ b/FORK_VERSION @@ -1 +1 @@ -0.1.177-overdraft.1 +0.1.177-overdraft.2 diff --git a/README.md b/README.md index 4e0f4be6e..6be65ab56 100644 --- a/README.md +++ b/README.md @@ -24,7 +24,8 @@ English | [中文](README_CN.md) | [日本語](README_JA.md) - Runs up to five real probes after the reported Codex 5h or 7d quota reaches 100%. - Keeps an account schedulable after a successful probe and tracks overdraft requests, tokens, cost, and recovery for both windows independently. - Exposes `pending`, `passed`, `failed`, `inconclusive`, and `recovered` states in the admin UI and PostgreSQL. -- Uses an atomic PostgreSQL claim for multi-instance deployments and requires no schema migration. +- Actively retries inconclusive probes after 1, 3, and 10 minutes, including recovery from persisted retry state after a restart. +- Uses an atomic PostgreSQL claim for multi-instance deployments and atomically commits terminal failure, account pause, and scheduler notification without a schema migration. - Supports an immediate configuration rollback to the upstream scheduling behavior. See the **[Chinese deployment and operations guide](CODEX_OVERDRAFT_DEPLOYMENT_CN.md)** for source builds, migration, verification, upgrades, rollback, Nginx, and troubleshooting. Maintainers should also read [CODEX_QUOTA_OVERDRAFT_CUSTOMIZATION.md](CODEX_QUOTA_OVERDRAFT_CUSTOMIZATION.md). diff --git a/README_CN.md b/README_CN.md index 780544e69..a26e67a95 100644 --- a/README_CN.md +++ b/README_CN.md @@ -70,7 +70,7 @@ docker logs --since 30m sub2api 2>&1 | \ grep -E 'codex_quota_overdraft_(probe|state|pause|stale_rate_limit)' ``` -出现 `codex_quota_overdraft_probe_passed`，页面显示“透支中”，并且后续真实业务请求成功，即可确认透支功能完整生效。OpenAI OAuth 常规文本“测试账号连接”也会使用透支请求形态并接入同一探测状态机；API Key、Shadow、图片和 Compact 测试除外。额度未达到 100% 时没有探测日志是正常现象。 +出现 `codex_quota_overdraft_probe_passed`，页面显示“透支中”，并且后续真实业务请求成功，即可确认透支功能完整生效。无法确认的探测会按 1、3、10 分钟主动重试，容器重启后由持久化状态继续；5 次明确额度失败时，探测状态、账号暂停和调度通知会原子提交。OpenAI OAuth 常规文本“测试账号连接”也会使用透支请求形态并接入同一探测状态机；API Key、Shadow、图片和 Compact 测试除外。额度未达到 100% 时没有探测日志是正常现象。 ## 来源、许可证与风险 diff --git a/README_JA.md b/README_JA.md index 03c1c30a5..d17d4af9f 100644 --- a/README_JA.md +++ b/README_JA.md @@ -24,7 +24,8 @@ - Codex の 5h または 7d クォータが 100% に達した後、最大 5 回の実リクエストで利用可否を検証します。 - 検証成功後もアカウントをスケジュール対象に保ち、リクエスト数、Token、コスト、回復時刻を表示します。 - `pending`、`passed`、`failed`、`inconclusive`、`recovered` の状態を管理画面と PostgreSQL に保存します。 -- PostgreSQL の atomic claim により複数インスタンスで重複検証を防ぎ、追加の schema migration は不要です。 +- 判定不能な検証は 1 分、3 分、10 分後に再試行し、再起動後も PostgreSQL の保存状態から再開します。 +- PostgreSQL の atomic claim により複数インスタンスで重複検証を防ぎ、最終失敗・アカウント停止・スケジューラ通知を同一トランザクションで保存します。追加の schema migration は不要です。 ソースビルド、既存環境からの移行、検証、更新、ロールバック、Nginx、トラブルシューティングについては、**[中国語のデプロイ・運用ガイド](CODEX_OVERDRAFT_DEPLOYMENT_CN.md)** を参照してください。 diff --git a/backend/cmd/server/wire_gen.go b/backend/cmd/server/wire_gen.go index 5f2d62856..e8d4ad8cf 100644 --- a/backend/cmd/server/wire_gen.go +++ b/backend/cmd/server/wire_gen.go @@ -199,7 +199,7 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) { grokQuotaService := service.ProvideGrokQuotaService(accountRepository, proxyRepository, grokTokenProvider, httpUpstream, configConfig, usageLogRepository, settingService) openAIQuotaService := service.ProvideOpenAIQuotaService(accountRepository, proxyRepository, openAITokenProvider, privacyClientFactory, openAIGatewayService) usageCache := service.NewUsageCache() - codexQuotaOverdraftCoordinator := service.ProvideCodexQuotaOverdraftCoordinator(accountRepository, httpUpstream, openAITokenProvider, tlsFingerprintProfileService, configConfig, tempUnschedCache, openAIGatewayService, rateLimitService) + codexQuotaOverdraftCoordinator := service.ProvideCodexQuotaOverdraftCoordinator(accountRepository, httpUpstream, openAITokenProvider, tlsFingerprintProfileService, configConfig, tempUnschedCache, openAIGatewayService, rateLimitService, timingWheelService) accountUsageService := service.ProvideAccountUsageService(accountRepository, usageLogRepository, claudeUsageFetcher, geminiQuotaService, antigravityQuotaFetcher, grokQuotaFetcher, grokQuotaService, openAIQuotaService, usageCache, identityCache, tlsFingerprintProfileService, openAIGatewayService, codexQuotaOverdraftCoordinator) accountTestService := service.ProvideAccountTestService(accountRepository, geminiTokenProvider, claudeTokenProvider, grokTokenProvider, antigravityGatewayService, httpUpstream, configConfig, tlsFingerprintProfileService, openAIGatewayService, settingService, codexQuotaOverdraftCoordinator) crsSyncService := service.NewCRSSyncService(accountRepository, proxyRepository, oAuthService, openAIOAuthService, geminiOAuthService, configConfig) diff --git a/backend/internal/repository/account_repo.go b/backend/internal/repository/account_repo.go index 65a592fcb..cc767ff48 100644 --- a/backend/internal/repository/account_repo.go +++ b/backend/internal/repository/account_repo.go @@ -2636,6 +2636,145 @@ func (r *accountRepository) ClaimCodexQuotaOverdraftProbe( return true, nil } +type codexQuotaOverdraftTxBeginner interface { + BeginTx(context.Context, *sql.TxOptions) (*sql.Tx, error) +} + +// FinalizeCodexQuotaOverdraftProbeFailed persists the terminal probe state, +// account pause, and scheduler notification in one transaction. +func (r *accountRepository) FinalizeCodexQuotaOverdraftProbeFailed( + ctx context.Context, + id int64, + state *service.CodexQuotaOverdraftProbeState, + until time.Time, + reason string, +) (bool, error) { + if state == nil || state.Status != "failed" || strings.TrimSpace(state.CycleKey) == "" || until.IsZero() { + return false, nil + } + payload, err := json.Marshal(state) + if err != nil { + return false, err + } + beginner, ok := r.sql.(codexQuotaOverdraftTxBeginner) + if !ok { + return false, errors.New("account repository does not support transactions") + } + tx, err := beginner.BeginTx(ctx, nil) + if err != nil { + return false, err + } + defer func() { _ = tx.Rollback() }() + + result, err := tx.ExecContext(ctx, ` + UPDATE accounts + SET extra = COALESCE(extra, '{}'::jsonb) || jsonb_build_object($1::text, $2::jsonb), + temp_unschedulable_until = CASE + WHEN temp_unschedulable_until IS NULL OR temp_unschedulable_until < $3 THEN $3 + ELSE temp_unschedulable_until + END, + temp_unschedulable_reason = CASE + WHEN temp_unschedulable_until IS NULL OR temp_unschedulable_until < $3 THEN $4 + ELSE temp_unschedulable_reason + END, + updated_at = NOW() + WHERE id = $5 + AND deleted_at IS NULL + AND extra #>> '{codex_quota_overdraft_probe,cycle_key}' = $6 + AND extra #>> '{codex_quota_overdraft_probe,status}' IN ('pending', 'failed') + `, service.CodexQuotaOverdraftProbeExtraKey, string(payload), until, reason, id, state.CycleKey) + if err != nil { + return false, err + } + affected, err := result.RowsAffected() + if err != nil { + return false, err + } + if affected == 0 { + return false, nil + } + if err := enqueueSchedulerOutbox(ctx, tx, service.SchedulerOutboxEventAccountChanged, &id, nil, nil); err != nil { + return false, err + } + if err := tx.Commit(); err != nil { + return false, err + } + r.syncSchedulerAccountSnapshotDetached(ctx, id) + return true, nil +} + +// ListDueCodexQuotaOverdraftProbes returns a bounded set of persisted +// inconclusive or stale pending probes so the retry sweep can resume work. +func (r *accountRepository) ListDueCodexQuotaOverdraftProbes(ctx context.Context, now time.Time, limit int) ([]*service.Account, error) { + if limit <= 0 { + limit = 100 + } + if limit > 1000 { + limit = 1000 + } + rows, err := r.sql.QueryContext(ctx, ` + SELECT id + FROM accounts + WHERE deleted_at IS NULL + AND platform = $1 + AND type = $2 + AND status = $3 + AND schedulable IS TRUE + AND parent_account_id IS NULL + AND ( + ( + extra #>> '{codex_quota_overdraft_probe,status}' = 'inconclusive' + AND COALESCE(NULLIF(extra #>> '{codex_quota_overdraft_probe,retry_at}', '')::timestamptz, 'infinity'::timestamptz) <= $4 + AND COALESCE(NULLIF(extra #>> '{codex_quota_overdraft_probe,retry_count}', '')::integer, 0) <= $5 + ) + OR ( + extra #>> '{codex_quota_overdraft_probe,status}' = 'pending' + AND COALESCE(NULLIF(extra #>> '{codex_quota_overdraft_probe,started_at}', '')::timestamptz, 'infinity'::timestamptz) <= $4 - INTERVAL '2 minutes' + ) + ) + AND COALESCE(NULLIF(extra #>> '{codex_quota_overdraft_probe,recover_at}', '')::timestamptz, 'infinity'::timestamptz) > $4 + ORDER BY COALESCE( + NULLIF(extra #>> '{codex_quota_overdraft_probe,retry_at}', '')::timestamptz, + NULLIF(extra #>> '{codex_quota_overdraft_probe,started_at}', '')::timestamptz + ) ASC, id ASC + LIMIT $6 + `, service.PlatformOpenAI, service.AccountTypeOAuth, service.StatusActive, now, 3, limit) + if err != nil { + return nil, err + } + defer rows.Close() + ids := make([]int64, 0, limit) + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + return nil, err + } + ids = append(ids, id) + } + if err := rows.Err(); err != nil { + return nil, err + } + if len(ids) == 0 { + return nil, nil + } + accounts, err := r.GetByIDs(ctx, ids) + if err != nil { + return nil, err + } + byID := make(map[int64]*service.Account, len(accounts)) + for i := range accounts { + account := accounts[i] + byID[account.ID] = account + } + ordered := make([]*service.Account, 0, len(accounts)) + for _, id := range ids { + if account := byID[id]; account != nil { + ordered = append(ordered, account) + } + } + return ordered, nil +} + // UpdateUpstreamBillingProbeSnapshot stores a probe result only while the // network identity used by that probe is still current. func (r *accountRepository) UpdateUpstreamBillingProbeSnapshot( diff --git a/backend/internal/repository/account_repo_codex_overdraft_test.go b/backend/internal/repository/account_repo_codex_overdraft_test.go new file mode 100644 index 000000000..7233698db --- /dev/null +++ b/backend/internal/repository/account_repo_codex_overdraft_test.go @@ -0,0 +1,98 @@ +package repository + +import ( + "context" + "errors" + "regexp" + "testing" + "time" + + "github.com/DATA-DOG/go-sqlmock" + "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/stretchr/testify/require" +) + +func TestFinalizeCodexQuotaOverdraftProbeFailedIsAtomic(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + t.Cleanup(func() { _ = db.Close() }) + + now := time.Now().UTC() + until := now.Add(5 * time.Hour) + state := &service.CodexQuotaOverdraftProbeState{ + Status: "failed", + CycleKey: "5h:quota-cycle", + QuotaWindow: "5h", + Attempts: 5, + Limit: 5, + StartedAt: now, + RecoverAt: &until, + } + reason := service.BuildTempUnschedReasonPayload("codex_quota_overdraft", "quota exhausted") + + mock.ExpectBegin() + mock.ExpectExec(`(?s)UPDATE accounts.*temp_unschedulable_until.*codex_quota_overdraft_probe`). + WithArgs(service.CodexQuotaOverdraftProbeExtraKey, sqlmock.AnyArg(), until, reason, int64(77), state.CycleKey). + WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectExec(regexp.QuoteMeta("INSERT INTO scheduler_outbox (event_type, account_id, group_id, payload, dedup_key)")). + WithArgs(service.SchedulerOutboxEventAccountChanged, int64(77), nil, nil, sqlmock.AnyArg()). + WillReturnResult(sqlmock.NewResult(1, 1)) + mock.ExpectCommit() + + repo := newAccountRepositoryWithSQL(nil, db, nil) + finalized, err := repo.FinalizeCodexQuotaOverdraftProbeFailed(context.Background(), 77, state, until, reason) + + require.NoError(t, err) + require.True(t, finalized) + require.NoError(t, mock.ExpectationsWereMet()) +} + +func TestFinalizeCodexQuotaOverdraftProbeFailedRollsBackWhenOutboxFails(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + t.Cleanup(func() { _ = db.Close() }) + + now := time.Now().UTC() + until := now.Add(5 * time.Hour) + state := &service.CodexQuotaOverdraftProbeState{ + Status: "failed", + CycleKey: "5h:quota-cycle", + QuotaWindow: "5h", + Attempts: 5, + Limit: 5, + StartedAt: now, + RecoverAt: &until, + } + + mock.ExpectBegin() + mock.ExpectExec(`(?s)UPDATE accounts.*temp_unschedulable_until.*codex_quota_overdraft_probe`). + WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectExec(regexp.QuoteMeta("INSERT INTO scheduler_outbox (event_type, account_id, group_id, payload, dedup_key)")). + WillReturnError(errors.New("outbox unavailable")) + mock.ExpectRollback() + + repo := newAccountRepositoryWithSQL(nil, db, nil) + finalized, err := repo.FinalizeCodexQuotaOverdraftProbeFailed(context.Background(), 77, state, until, "quota exhausted") + + require.ErrorContains(t, err, "outbox unavailable") + require.False(t, finalized) + require.NoError(t, mock.ExpectationsWereMet()) +} + +func TestListDueCodexQuotaOverdraftProbesIncludesInconclusiveAndStalePending(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + t.Cleanup(func() { _ = db.Close() }) + + now := time.Now().UTC() + mock.ExpectQuery(`(?s)SELECT id.*status}' = 'inconclusive'.*retry_count.*status}' = 'pending'.*INTERVAL '2 minutes'`). + WithArgs(service.PlatformOpenAI, service.AccountTypeOAuth, service.StatusActive, now, 3, 100). + WillReturnRows(sqlmock.NewRows([]string{"id"})) + + repo := newAccountRepositoryWithSQL(nil, db, nil) + accounts, err := repo.ListDueCodexQuotaOverdraftProbes(context.Background(), now, 100) + + require.NoError(t, err) + require.Empty(t, accounts) + require.NoError(t, mock.ExpectationsWereMet()) +} diff --git a/backend/internal/service/openai_codex_quota_overdraft_probe.go b/backend/internal/service/openai_codex_quota_overdraft_probe.go index 8ae318ca1..a1ace2fe7 100644 --- a/backend/internal/service/openai_codex_quota_overdraft_probe.go +++ b/backend/internal/service/openai_codex_quota_overdraft_probe.go @@ -30,9 +30,12 @@ const ( codexQuotaOverdraftProbeAttemptLimit = 5 codexQuotaOverdraftProbeAttemptTimeout = 20 * time.Second codexQuotaOverdraftProbePlanTimeout = codexQuotaOverdraftProbeAttemptLimit * codexQuotaOverdraftProbeAttemptTimeout - codexQuotaOverdraftProbeRetryDelay = time.Minute codexQuotaOverdraftProbeBodyLimit = 256 << 10 codexQuotaOverdraftPauseSource = "codex_quota_overdraft" + codexQuotaOverdraftPersistAttempts = 3 + codexQuotaOverdraftAutomaticRetries = 3 + codexQuotaOverdraftRetrySweepInterval = time.Minute + codexQuotaOverdraftRetrySweepLimit = 100 codexQuotaOverdraftFallbackModel = "gpt-5.5" codexQuotaOverdraftCompatibilityModel = "gpt-5.4-mini" @@ -51,6 +54,7 @@ type CodexQuotaOverdraftProbeState struct { StartedAt time.Time `json:"started_at"` TestedAt *time.Time `json:"tested_at,omitempty"` RetryAt *time.Time `json:"retry_at,omitempty"` + RetryCount int `json:"retry_count,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"` @@ -81,6 +85,20 @@ type codexQuotaOverdraftProbeClaimer interface { ClaimCodexQuotaOverdraftProbe(context.Context, int64, *CodexQuotaOverdraftProbeState) (bool, error) } +type codexQuotaOverdraftFailedFinalizer interface { + FinalizeCodexQuotaOverdraftProbeFailed(context.Context, int64, *CodexQuotaOverdraftProbeState, time.Time, string) (bool, error) +} + +type codexQuotaOverdraftDueProbeLister interface { + ListDueCodexQuotaOverdraftProbes(context.Context, time.Time, int) ([]*Account, error) +} + +type codexQuotaOverdraftRetryScheduler interface { + Schedule(string, time.Duration, func()) + ScheduleRecurring(string, time.Duration, func()) + Cancel(string) +} + // CodexQuotaOverdraftCoordinator implements the bounded real-request gate used // by cpa-account-config-manager before it disables a quota-exhausted account. type CodexQuotaOverdraftCoordinator struct { @@ -95,6 +113,7 @@ type CodexQuotaOverdraftCoordinator struct { agentIdentityWS agentIdentityWSConnectionInvalidator agentIdentityTaskMu sync.Mutex running sync.Map + retryScheduler codexQuotaOverdraftRetryScheduler now func() time.Time probeAttemptForTest func(context.Context, *Account, string) codexQuotaOverdraftProbeResult } @@ -124,6 +143,16 @@ func NewCodexQuotaOverdraftCoordinator( return coordinator } +// SetRetryScheduler enables active retry timers and a small persisted-state +// sweep so inconclusive or stale pending probes resume after a restart. +func (c *CodexQuotaOverdraftCoordinator) SetRetryScheduler(scheduler codexQuotaOverdraftRetryScheduler) { + if c == nil || scheduler == nil { + return + } + c.retryScheduler = scheduler + scheduler.ScheduleRecurring("codex-overdraft:retry-sweep", codexQuotaOverdraftRetrySweepInterval, c.resumeDueProbes) +} + func (c *CodexQuotaOverdraftCoordinator) enabled() bool { return c != nil && c.cfg != nil && c.cfg.Gateway.CodexQuotaOverdraftEnabled && c.accountRepo != nil && c.httpUpstream != nil @@ -226,6 +255,9 @@ func (c *CodexQuotaOverdraftCoordinator) startProbe(account *Account, signal cod SevenDayRecoverAt: cloneTimePtr(signal.SevenDayRecoverAt), ObservedRateLimit: cloneTimePtr(account.RateLimitResetAt), } + if current != nil && codexQuotaOverdraftStateCoversSignal(current, signal) { + state.RetryCount = current.RetryCount + } carryCodexQuotaOverdraftWindowStarts(state, current, signal, startedAt) claimed, err := c.claimProbe(account.ID, state) if err != nil { @@ -270,7 +302,7 @@ func (c *CodexQuotaOverdraftCoordinator) runProbePlan( account, err := c.accountRepo.GetByID(planCtx, accountID) if err != nil || !isCodexQuotaOverdraftAccount(account) { - c.finishInconclusive(accountID, state, "upstream_unavailable") + c.finishInconclusive(accountID, signal, preferredModel, state, "upstream_unavailable") return } models := codexQuotaOverdraftProbeModels(preferredModel) @@ -278,7 +310,7 @@ func (c *CodexQuotaOverdraftCoordinator) runProbePlan( quotaLimitedAttempts := 0 for attempt := 0; attempt < codexQuotaOverdraftProbeAttemptLimit; attempt++ { if planCtx.Err() != nil { - c.finishInconclusive(accountID, state, "request_timeout") + c.finishInconclusive(accountID, signal, preferredModel, state, "request_timeout") return } model := models[attempt%len(models)] @@ -294,6 +326,16 @@ func (c *CodexQuotaOverdraftCoordinator) runProbePlan( if result.ReasonCode == "quota_limited" { quotaLimitedAttempts++ } + slog.Info("codex_quota_overdraft_probe_attempt", + "account_id", accountID, + "attempt", state.Attempts, + "limit", codexQuotaOverdraftProbeAttemptLimit, + "model", state.Model, + "status_code", result.StatusCode, + "result_status", result.Status, + "reason", result.ReasonCode, + "quota_window", signal.Window, + ) c.persistState(accountID, state) if result.Status == "available" { @@ -302,44 +344,182 @@ func (c *CodexQuotaOverdraftCoordinator) runProbePlan( state.RetryAt = nil startCodexQuotaOverdraftWindows(state, signal, testedAt) c.persistState(accountID, state) + c.cancelRetry(accountID) 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) + c.finishInconclusive(accountID, signal, preferredModel, 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) + c.finishInconclusive(accountID, signal, preferredModel, state, result.ReasonCode) return } } if quotaLimitedAttempts < codexQuotaOverdraftProbeAttemptLimit { - c.finishInconclusive(accountID, state, lastReason) + c.finishInconclusive(accountID, signal, preferredModel, state, lastReason) return } state.Status = codexQuotaOverdraftProbeFailed state.ReasonCode = lastReason state.RetryAt = nil - c.persistState(accountID, state) - c.ensureFailedPause(account, state) + c.cancelRetry(accountID) + if !c.ensureFailedPause(account, state) { + return + } 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) { +func (c *CodexQuotaOverdraftCoordinator) finishInconclusive( + accountID int64, + signal codexQuotaOverdraftSignal, + preferredModel string, + state *CodexQuotaOverdraftProbeState, + reason string, +) { now := c.currentTime().UTC() - retryAt := now.Add(codexQuotaOverdraftProbeRetryDelay) + retryable := reason != "authentication_failed" && reason != "account_deactivated" state.Status = codexQuotaOverdraftProbeInconclusive state.ReasonCode = reason state.TestedAt = &now - state.RetryAt = &retryAt + state.RetryAt = nil + if retryable && state.RetryCount < codexQuotaOverdraftAutomaticRetries { + state.RetryCount++ + retryAt := now.Add(codexQuotaOverdraftRetryDelay(state.RetryCount)) + state.RetryAt = &retryAt + } c.persistState(accountID, state) - slog.Warn("codex_quota_overdraft_probe_inconclusive", "account_id", accountID, "attempts", state.Attempts, "reason", reason) + if state.RetryAt != nil { + c.scheduleRetry(accountID, signal.CycleKey, preferredModel, *state.RetryAt) + } + slog.Warn("codex_quota_overdraft_probe_inconclusive", + "account_id", accountID, + "attempts", state.Attempts, + "reason", reason, + "retry_count", state.RetryCount, + "retry_at", state.RetryAt, + ) +} + +func codexQuotaOverdraftRetryDelay(retryCount int) time.Duration { + switch retryCount { + case 1: + return time.Minute + case 2: + return 3 * time.Minute + default: + return 10 * time.Minute + } +} + +func codexQuotaOverdraftRetryTaskName(accountID int64) string { + return fmt.Sprintf("codex-overdraft:retry:%d", accountID) +} + +func (c *CodexQuotaOverdraftCoordinator) scheduleRetry(accountID int64, cycleKey, preferredModel string, retryAt time.Time) { + if c == nil || c.retryScheduler == nil || accountID <= 0 { + return + } + delay := retryAt.Sub(c.currentTime()) + if delay < time.Second { + delay = time.Second + } + name := codexQuotaOverdraftRetryTaskName(accountID) + c.retryScheduler.Schedule(name, delay, func() { + c.resumeProbe(accountID, cycleKey, preferredModel) + }) + slog.Info("codex_quota_overdraft_retry_scheduled", + "account_id", accountID, + "cycle_key", cycleKey, + "retry_at", retryAt, + ) +} + +func (c *CodexQuotaOverdraftCoordinator) cancelRetry(accountID int64) { + if c == nil || c.retryScheduler == nil || accountID <= 0 { + return + } + c.retryScheduler.Cancel(codexQuotaOverdraftRetryTaskName(accountID)) +} + +func (c *CodexQuotaOverdraftCoordinator) resumeProbe(accountID int64, cycleKey, preferredModel string) { + if !c.enabled() || accountID <= 0 { + return + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + account, err := c.accountRepo.GetByID(ctx, accountID) + cancel() + if err != nil || !isCodexQuotaOverdraftAccount(account) { + slog.Warn("codex_quota_overdraft_retry_load_failed", "account_id", accountID, "error", err) + return + } + state, ok := codexQuotaOverdraftStateFromAccount(account) + if !ok || (cycleKey != "" && state.CycleKey != cycleKey) { + return + } + switch state.Status { + case codexQuotaOverdraftProbeInconclusive: + if state.RetryAt == nil { + return + } + if c.currentTime().Before(*state.RetryAt) { + c.scheduleRetry(accountID, state.CycleKey, preferredModel, *state.RetryAt) + return + } + case codexQuotaOverdraftProbePending: + staleAt := state.StartedAt.Add(2 * time.Minute) + if c.currentTime().Before(staleAt) { + c.scheduleRetry(accountID, state.CycleKey, preferredModel, staleAt) + return + } + default: + return + } + if strings.TrimSpace(preferredModel) == "" { + preferredModel = state.Model + } + slog.Info("codex_quota_overdraft_retry_started", + "account_id", accountID, + "cycle_key", state.CycleKey, + "retry_count", state.RetryCount, + ) + c.observeAccount(account, preferredModel) +} + +func (c *CodexQuotaOverdraftCoordinator) resumeDueProbes() { + if !c.enabled() { + return + } + lister, ok := c.accountRepo.(codexQuotaOverdraftDueProbeLister) + if !ok { + return + } + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + accounts, err := lister.ListDueCodexQuotaOverdraftProbes(ctx, c.currentTime(), codexQuotaOverdraftRetrySweepLimit) + cancel() + if err != nil { + slog.Warn("codex_quota_overdraft_retry_sweep_failed", "error", err) + return + } + for _, account := range accounts { + if account == nil { + continue + } + state, _ := codexQuotaOverdraftStateFromAccount(account) + cycleKey := "" + preferredModel := "" + if state != nil { + cycleKey = state.CycleKey + preferredModel = state.Model + } + go c.resumeProbe(account.ID, cycleKey, preferredModel) + } } func (c *CodexQuotaOverdraftCoordinator) runProbeAttempt(ctx context.Context, account *Account, model string) codexQuotaOverdraftProbeResult { @@ -453,6 +633,9 @@ func classifyCodexQuotaOverdraftProbe(statusCode int, headers http.Header, body lower := bytes.ToLower(body) if bytes.Contains(lower, []byte(`"type":"error"`)) || bytes.Contains(lower, []byte(`"type": "error"`)) || bytes.Contains(lower, []byte(`"response.failed"`)) { + if codexQuotaOverdraftResponseIsQuotaLimited(headers, body) { + return "retry", "quota_limited" + } return "retry", "invalid_response" } if bytes.Contains(lower, []byte(`"response.completed"`)) || bytes.Contains(lower, []byte(`"response.output_item.done"`)) { @@ -480,35 +663,104 @@ func classifyCodexQuotaOverdraftProbe(statusCode int, headers http.Header, body case http.StatusRequestTimeout, http.StatusGatewayTimeout: return "inconclusive", "request_timeout" default: + if codexQuotaOverdraftResponseIsQuotaLimited(headers, body) { + return "retry", "quota_limited" + } if statusCode >= http.StatusInternalServerError { return "inconclusive", "upstream_unavailable" } if statusCode == http.StatusBadRequest || statusCode == http.StatusNotFound { - return "inconclusive", "model_not_found" + return "retry", "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) persistState(accountID int64, state *CodexQuotaOverdraftProbeState) bool { + var lastErr error + for attempt := 1; attempt <= codexQuotaOverdraftPersistAttempts; attempt++ { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + lastErr = c.accountRepo.UpdateExtra(ctx, accountID, map[string]any{CodexQuotaOverdraftProbeExtraKey: state}) + cancel() + if lastErr == nil { + return true + } + if attempt < codexQuotaOverdraftPersistAttempts { + time.Sleep(time.Duration(attempt) * 100 * time.Millisecond) + } } + slog.Warn("codex_quota_overdraft_state_persist_failed", + "account_id", accountID, + "attempts", codexQuotaOverdraftPersistAttempts, + "error", lastErr, + ) + return false } -func (c *CodexQuotaOverdraftCoordinator) ensureFailedPause(account *Account, state *CodexQuotaOverdraftProbeState) { +func (c *CodexQuotaOverdraftCoordinator) ensureFailedPause(account *Account, state *CodexQuotaOverdraftProbeState) bool { if account == nil || state == nil || state.RecoverAt == nil || !state.RecoverAt.After(c.currentTime()) { - return + return false + } + if persisted, ok := codexQuotaOverdraftStateFromAccount(account); ok && + persisted.Status == codexQuotaOverdraftProbeFailed && persisted.CycleKey == state.CycleKey && + account.TempUnschedulableUntil != nil && !account.TempUnschedulableUntil.Before(*state.RecoverAt) { + return true } reason := BuildTempUnschedReasonPayload(codexQuotaOverdraftPauseSource, "five real overdraft probes confirmed quota exhaustion") + if finalizer, ok := c.accountRepo.(codexQuotaOverdraftFailedFinalizer); ok { + finalized := false + var lastErr error + for attempt := 1; attempt <= codexQuotaOverdraftPersistAttempts; attempt++ { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + finalized, lastErr = finalizer.FinalizeCodexQuotaOverdraftProbeFailed(ctx, account.ID, state, *state.RecoverAt, reason) + cancel() + if lastErr == nil { + break + } + if attempt < codexQuotaOverdraftPersistAttempts { + time.Sleep(time.Duration(attempt) * 100 * time.Millisecond) + } + } + if lastErr != nil { + slog.Warn("codex_quota_overdraft_pause_failed", + "account_id", account.ID, + "attempts", codexQuotaOverdraftPersistAttempts, + "error", lastErr, + ) + return false + } + if !finalized { + slog.Warn("codex_quota_overdraft_pause_stale", "account_id", account.ID, "cycle_key", state.CycleKey) + return false + } + } else { + if !c.persistState(account.ID, state) { + return false + } + var lastErr error + for attempt := 1; attempt <= codexQuotaOverdraftPersistAttempts; attempt++ { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + lastErr = c.accountRepo.SetTempUnschedulable(ctx, account.ID, *state.RecoverAt, reason) + cancel() + if lastErr == nil { + break + } + if attempt < codexQuotaOverdraftPersistAttempts { + time.Sleep(time.Duration(attempt) * 100 * time.Millisecond) + } + } + if lastErr != nil { + slog.Warn("codex_quota_overdraft_pause_failed", + "account_id", account.ID, + "attempts", codexQuotaOverdraftPersistAttempts, + "error", lastErr, + ) + return false + } + } + c.cancelRetry(account.ID) 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(), @@ -520,9 +772,12 @@ func (c *CodexQuotaOverdraftCoordinator) ensureFailedPause(account *Account, sta if c.runtimeBlocker != nil { c.runtimeBlocker.BlockAccountScheduling(account, *state.RecoverAt, codexQuotaOverdraftPauseSource) } + slog.Info("codex_quota_overdraft_pause_applied", "account_id", account.ID, "until", state.RecoverAt, "cycle_key", state.CycleKey) + return true } func (c *CodexQuotaOverdraftCoordinator) clearQuotaPause(accountID int64, state *CodexQuotaOverdraftProbeState) { + c.cancelRetry(accountID) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() account, err := c.accountRepo.GetByID(ctx, accountID) @@ -605,8 +860,20 @@ func codexQuotaOverdraftResponseIsQuotaLimited(headers http.Header, body []byte) return true } } + var payload any + if len(bytes.TrimSpace(body)) > 0 && json.Unmarshal(body, &payload) == nil && codexQuotaOverdraftJSONHasQuotaEvidence(payload, 0) { + 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"} { + for _, marker := range []string{ + "usage_limit_reached", + "usage limit has been reached", + "you have reached your usage limit", + "quota exhausted", + "insufficient quota", + "weekly limit reached", + "weekly_limit_reached", + } { if strings.Contains(text, marker) { return true } @@ -614,6 +881,69 @@ func codexQuotaOverdraftResponseIsQuotaLimited(headers http.Header, body []byte) return false } +func codexQuotaOverdraftJSONHasQuotaEvidence(value any, depth int) bool { + if depth > 6 { + return false + } + switch typed := value.(type) { + case map[string]any: + for key, raw := range typed { + normalizedKey := strings.ToLower(strings.ReplaceAll(strings.TrimSpace(key), "-", "_")) + switch normalizedKey { + case "type", "code", "reason", "error_code": + if marker, ok := raw.(string); ok && codexQuotaOverdraftQuotaCode(marker) { + return true + } + case "limit_reached", "limitreached": + if reached, ok := raw.(bool); ok && reached { + return true + } + case "used_percent", "usedpercent": + if used, ok := raw.(float64); ok && used >= 100 { + return true + } + } + if codexQuotaOverdraftJSONHasQuotaEvidence(raw, depth+1) { + return true + } + } + case []any: + for _, item := range typed { + if codexQuotaOverdraftJSONHasQuotaEvidence(item, depth+1) { + return true + } + } + case string: + if codexQuotaOverdraftQuotaCode(typed) { + return true + } + text := strings.ToLower(strings.Join(strings.Fields(typed), " ")) + for _, marker := range []string{ + "usage limit has been reached", + "you have reached your usage limit", + "quota exhausted", + "insufficient quota", + "weekly limit reached", + } { + if strings.Contains(text, marker) { + return true + } + } + } + return false +} + +func codexQuotaOverdraftQuotaCode(value string) bool { + value = strings.ToLower(strings.TrimSpace(value)) + value = strings.NewReplacer("-", "_", " ", "_").Replace(value) + switch value { + case "usage_limit_reached", "weekly_limit_reached", "monthly_limit_reached", "quota_exhausted", "insufficient_quota", "billing_hard_limit_reached": + return true + default: + return false + } +} + func codexQuotaOverdraftSignalFromAccount(account *Account, state *CodexQuotaOverdraftProbeState, now time.Time) (codexQuotaOverdraftSignal, bool) { if account == nil || len(account.Extra) == 0 { return codexQuotaOverdraftSignal{}, false diff --git a/backend/internal/service/openai_codex_quota_overdraft_probe_test.go b/backend/internal/service/openai_codex_quota_overdraft_probe_test.go index e1f2f68fc..b24380560 100644 --- a/backend/internal/service/openai_codex_quota_overdraft_probe_test.go +++ b/backend/internal/service/openai_codex_quota_overdraft_probe_test.go @@ -26,6 +26,30 @@ type codexOverdraftRuntimeBlockerStub struct { clearCalls int } +type codexOverdraftRetrySchedulerStub struct { + scheduledName string + scheduledDelay time.Duration + scheduledFn func() + recurringName string + recurringDelay time.Duration + canceledName string +} + +func (s *codexOverdraftRetrySchedulerStub) Schedule(name string, delay time.Duration, fn func()) { + s.scheduledName = name + s.scheduledDelay = delay + s.scheduledFn = fn +} + +func (s *codexOverdraftRetrySchedulerStub) ScheduleRecurring(name string, delay time.Duration, _ func()) { + s.recurringName = name + s.recurringDelay = delay +} + +func (s *codexOverdraftRetrySchedulerStub) Cancel(name string) { + s.canceledName = name +} + func (b *codexOverdraftRuntimeBlockerStub) BlockAccountScheduling(*Account, time.Time, string) {} func (b *codexOverdraftRuntimeBlockerStub) ClearAccountSchedulingBlock(int64) { @@ -166,6 +190,30 @@ func TestCodexQuotaOverdraftProbeRequiresFiveExplicitQuotaFailures(t *testing.T) require.False(t, codexQuotaOverdraftSchedulingAllowed(account, now), "同周期确认失败后不得继续绕过限额") } +func TestCodexQuotaOverdraftFailedPauseIsIdempotentForPersistedCycle(t *testing.T) { + now := time.Date(2026, time.August, 13, 14, 0, 0, 0, time.UTC) + account := newCodexOverdraftProbeTestAccount(now) + recoverAt := now.Add(5 * time.Hour) + state := &CodexQuotaOverdraftProbeState{ + Status: codexQuotaOverdraftProbeFailed, + QuotaWindow: "5h", + CycleKey: "5h:" + formatCodexOverdraftUnix(recoverAt), + Attempts: codexQuotaOverdraftProbeAttemptLimit, + Limit: codexQuotaOverdraftProbeAttemptLimit, + StartedAt: now, + RecoverAt: codexQuotaOverdraftTimePtr(recoverAt), + } + account.Extra[CodexQuotaOverdraftProbeExtraKey] = state + account.TempUnschedulableUntil = codexQuotaOverdraftTimePtr(recoverAt) + account.TempUnschedulableReason = BuildTempUnschedReasonPayload(codexQuotaOverdraftPauseSource, "quota exhausted") + repo := &codexOverdraftProbeRepoStub{account: account} + coordinator := &CodexQuotaOverdraftCoordinator{accountRepo: repo, now: func() time.Time { return now }} + + require.True(t, coordinator.ensureFailedPause(account, state)) + require.Zero(t, repo.tempPauseCalls) + require.Empty(t, repo.states) +} + func TestCodexQuotaOverdraftProbeInconclusiveNeverPauses(t *testing.T) { now := time.Date(2026, time.August, 13, 14, 0, 0, 0, time.UTC) for _, result := range []codexQuotaOverdraftProbeResult{ @@ -190,6 +238,93 @@ func TestCodexQuotaOverdraftProbeInconclusiveNeverPauses(t *testing.T) { } } +func TestCodexQuotaOverdraftProbeRotatesModelsBeforeInconclusive(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: "model_not_found", StatusCode: http.StatusNotFound, Model: model} + } + 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, codexQuotaOverdraftProbeAttemptLimit, state.Attempts) + require.Equal(t, 1, state.RetryCount) + require.Equal(t, []string{"gpt-5.4", "gpt-5.5", "gpt-5.4-mini", "gpt-5.4", "gpt-5.5"}, models) + require.Zero(t, repo.tempPauseCalls) +} + +func TestCodexQuotaOverdraftInconclusiveSchedulesBackoffRetry(t *testing.T) { + now := time.Date(2026, time.August, 13, 14, 0, 0, 0, time.UTC) + account := newCodexOverdraftProbeTestAccount(now) + repo := &codexOverdraftProbeRepoStub{account: account} + scheduler := &codexOverdraftRetrySchedulerStub{} + coordinator := &CodexQuotaOverdraftCoordinator{ + accountRepo: repo, + retryScheduler: scheduler, + now: func() time.Time { return now }, + } + signal, _ := codexQuotaOverdraftSignalFromAccount(account, nil, now) + state := newCodexOverdraftPendingState(signal, now) + + coordinator.finishInconclusive(account.ID, signal, "gpt-5.4", state, "upstream_unavailable") + + require.Equal(t, codexQuotaOverdraftProbeInconclusive, state.Status) + require.Equal(t, 1, state.RetryCount) + require.WithinDuration(t, now.Add(time.Minute), *state.RetryAt, time.Second) + require.Equal(t, codexQuotaOverdraftRetryTaskName(account.ID), scheduler.scheduledName) + require.Equal(t, time.Minute, scheduler.scheduledDelay) + require.NotNil(t, scheduler.scheduledFn) + + state.RetryCount = 1 + coordinator.finishInconclusive(account.ID, signal, "gpt-5.4", state, "upstream_unavailable") + require.Equal(t, 2, state.RetryCount) + require.Equal(t, 3*time.Minute, scheduler.scheduledDelay) +} + +func TestCodexQuotaOverdraftAuthenticationFailureDoesNotScheduleQuotaRetry(t *testing.T) { + now := time.Date(2026, time.August, 13, 14, 0, 0, 0, time.UTC) + account := newCodexOverdraftProbeTestAccount(now) + repo := &codexOverdraftProbeRepoStub{account: account} + scheduler := &codexOverdraftRetrySchedulerStub{} + coordinator := &CodexQuotaOverdraftCoordinator{accountRepo: repo, retryScheduler: scheduler, now: func() time.Time { return now }} + signal, _ := codexQuotaOverdraftSignalFromAccount(account, nil, now) + state := newCodexOverdraftPendingState(signal, now) + + coordinator.finishInconclusive(account.ID, signal, "gpt-5.4", state, "authentication_failed") + + require.Empty(t, scheduler.scheduledName) + require.Nil(t, state.RetryAt) + require.Zero(t, state.RetryCount) +} + +func TestCodexQuotaOverdraftInconclusiveStopsAtAutomaticRetryLimit(t *testing.T) { + now := time.Date(2026, time.August, 13, 14, 0, 0, 0, time.UTC) + account := newCodexOverdraftProbeTestAccount(now) + repo := &codexOverdraftProbeRepoStub{account: account} + scheduler := &codexOverdraftRetrySchedulerStub{} + coordinator := &CodexQuotaOverdraftCoordinator{ + accountRepo: repo, + retryScheduler: scheduler, + now: func() time.Time { return now }, + } + signal, _ := codexQuotaOverdraftSignalFromAccount(account, nil, now) + state := newCodexOverdraftPendingState(signal, now) + state.RetryCount = codexQuotaOverdraftAutomaticRetries + + coordinator.finishInconclusive(account.ID, signal, "gpt-5.4", state, "upstream_unavailable") + + require.Equal(t, codexQuotaOverdraftAutomaticRetries, state.RetryCount) + require.Nil(t, state.RetryAt) + require.Empty(t, scheduler.scheduledName) +} + func TestCodexQuotaOverdraftNewWindowPreservesExistingWindowBaseline(t *testing.T) { now := time.Date(2026, time.August, 13, 14, 0, 0, 0, time.UTC) fiveStarted := now.Add(-time.Hour) @@ -323,6 +458,18 @@ func TestClassifyCodexQuotaOverdraftProbeResponses(t *testing.T) { require.Equal(t, "retry", status) require.Equal(t, "quota_limited", reason) + status, reason = classifyCodexQuotaOverdraftProbe(http.StatusTooManyRequests, nil, []byte(`{"error":{"code":"weekly_limit_reached"}}`)) + require.Equal(t, "retry", status) + require.Equal(t, "quota_limited", reason) + + status, reason = classifyCodexQuotaOverdraftProbe(http.StatusTooManyRequests, nil, []byte(`{"rate_limit":{"allowed":false,"limit_reached":true}}`)) + require.Equal(t, "retry", status) + require.Equal(t, "quota_limited", reason) + + status, reason = classifyCodexQuotaOverdraftProbe(http.StatusTooManyRequests, nil, []byte(`{"error":{"type":"rate_limit_exceeded","message":"too many requests"}}`)) + require.Equal(t, "inconclusive", status) + require.Equal(t, "transient_failure", reason) + status, reason = classifyCodexQuotaOverdraftProbe(http.StatusServiceUnavailable, nil, nil) require.Equal(t, "inconclusive", status) require.Equal(t, "upstream_unavailable", reason) @@ -331,6 +478,10 @@ func TestClassifyCodexQuotaOverdraftProbeResponses(t *testing.T) { require.Equal(t, "authentication_failed", status) require.Equal(t, "authentication_failed", reason) + status, reason = classifyCodexQuotaOverdraftProbe(http.StatusNotFound, nil, []byte(`{"detail":"model not found"}`)) + require.Equal(t, "retry", status) + require.Equal(t, "model_not_found", 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) diff --git a/backend/internal/service/wire.go b/backend/internal/service/wire.go index 1e853b64c..75e7406f6 100644 --- a/backend/internal/service/wire.go +++ b/backend/internal/service/wire.go @@ -231,6 +231,7 @@ func ProvideCodexQuotaOverdraftCoordinator( tempUnschedCache TempUnschedCache, runtimeBlocker AccountRuntimeBlocker, rateLimitService *RateLimitService, + timingWheel *TimingWheelService, ) *CodexQuotaOverdraftCoordinator { coordinator := NewCodexQuotaOverdraftCoordinator( accountRepo, @@ -242,6 +243,7 @@ func ProvideCodexQuotaOverdraftCoordinator( runtimeBlocker, rateLimitService, ) + coordinator.SetRetryScheduler(timingWheel) if gateway, ok := runtimeBlocker.(*OpenAIGatewayService); ok { gateway.SetCodexQuotaOverdraftCoordinator(coordinator) } diff --git a/frontend/src/components/account/AccountStatusIndicator.vue b/frontend/src/components/account/AccountStatusIndicator.vue index 4130ecd8e..ee81ad9ed 100644 --- a/frontend/src/components/account/AccountStatusIndicator.vue +++ b/frontend/src/components/account/AccountStatusIndicator.vue @@ -1,7 +1,7 @@