Merge pull request #7074 from Brisbanehuang/codex/fix-codex-quota-windows

fix(scheduler): 按规范 5h/7d 窗口读取 Codex 配额与重置时间
This commit is contained in:
Wesley Liddick
2026-09-15 16:17:37 +08:00
committed by GitHub
3 changed files with 235 additions and 33 deletions
@@ -961,15 +961,15 @@ func (s *defaultOpenAIAccountScheduler) buildOpenAIAccountLoadPlan(
}
}
// Reset 因子(use-it-or-lose-it):在拥有「未来会话窗口结束时间」的账号中,
// Reset 因子(use-it-or-lose-it):优先读取 Codex 5h 重置时间,回退到会话窗口。
// 剩余时间越短 → 因子越接近 1(越早重置越优先用尽)。无活跃窗口的账号因子为 0。
// 仅在 weights.Reset > 0 时计算,默认关闭不影响原有行为。
minResetRemaining, maxResetRemaining := 0.0, 0.0
hasResetSample := false
if weights.Reset > 0 {
for _, candidate := range candidates {
end := candidate.account.SessionWindowEnd
if end == nil || !now.Before(*end) {
end, ok := openAISchedulingResetWindowEnd(candidate.account, now)
if !ok {
continue
}
remaining := end.Sub(now).Seconds()
@@ -1002,7 +1002,7 @@ func (s *defaultOpenAIAccountScheduler) buildOpenAIAccountLoadPlan(
}
resetFactor := 0.0
if weights.Reset > 0 && hasResetSample {
if end := item.account.SessionWindowEnd; end != nil && now.Before(*end) {
if end, ok := openAISchedulingResetWindowEnd(item.account, now); ok {
if maxResetRemaining > minResetRemaining {
resetFactor = 1 - clamp01((end.Sub(now).Seconds()-minResetRemaining)/(maxResetRemaining-minResetRemaining))
} else {
@@ -2732,8 +2732,8 @@ func buildOpenAIAccountSchedulerScoreSnapshot(
}
if weights.Reset > 0 {
for _, candidate := range candidates {
end := candidate.account.SessionWindowEnd
if end == nil || !now.Before(*end) {
end, ok := openAISchedulingResetWindowEnd(candidate.account, now)
if !ok {
continue
}
remaining := end.Sub(now).Seconds()
@@ -2763,7 +2763,7 @@ func buildOpenAIAccountSchedulerScoreSnapshot(
ttftFactor := 0.5
resetFactor := 0.0
if weights.Reset > 0 && hasResetSample {
if end := candidate.account.SessionWindowEnd; end != nil && now.Before(*end) {
if end, ok := openAISchedulingResetWindowEnd(candidate.account, now); ok {
if maxResetRemaining > minResetRemaining {
resetFactor = 1 - clamp01((end.Sub(now).Seconds()-minResetRemaining)/(maxResetRemaining-minResetRemaining))
} else {
@@ -2952,22 +2952,89 @@ func openAIQuotaHeadroomFactor(account *Account, now time.Time) float64 {
if account == nil || len(account.Extra) == 0 || openAIQuotaHeadroomSnapshotStale(account.Extra, now) {
return openAIQuotaHeadroomNeutralFactor
}
primaryUsedPercent, ok := resolveAccountExtraNumber(account.Extra, "codex_primary_used_percent", "codex_7d_used_percent")
if !ok || openAIQuotaWindowResetAny(account.Extra, now, "primary", "7d") {
window5h, window7d := openAICanonicalQuotaWindows(account.Extra, now)
if !window7d.hasUsed || window7d.reset {
return openAIQuotaHeadroomNeutralFactor
}
factor := 1 - clamp01(primaryUsedPercent/100)
if secondaryUsedPercent, ok := resolveAccountExtraNumber(account.Extra, "codex_secondary_used_percent", "codex_5h_used_percent"); ok &&
!openAIQuotaWindowResetAny(account.Extra, now, "secondary", "5h") {
secondaryRemaining := 1 - clamp01(secondaryUsedPercent/100)
if secondaryRemaining < openAIQuotaHeadroomSecondaryLowRemain {
factor := 1 - clamp01(window7d.usedPercent/100)
if window5h.hasUsed && !window5h.reset {
remaining := 1 - clamp01(window5h.usedPercent/100)
if remaining < openAIQuotaHeadroomSecondaryLowRemain {
factor *= openAIQuotaHeadroomNeutralFactor
}
}
return factor
}
type openAICanonicalQuotaWindow struct {
usedPercent float64
hasUsed bool
reset bool
}
// 规范字段优先;历史原始字段复用写入端 Normalize 的窗口分类,不能固定把 primary 当作 7d。
func openAICanonicalQuotaWindows(extra map[string]any, now time.Time) (window5h, window7d openAICanonicalQuotaWindow) {
if used, ok := resolveAccountExtraNumber(extra, "codex_5h_used_percent"); ok {
window5h = openAICanonicalQuotaWindow{usedPercent: used, hasUsed: true, reset: openAIQuotaWindowReset(extra, "5h", now)}
}
if used, ok := resolveAccountExtraNumber(extra, "codex_7d_used_percent"); ok {
window7d = openAICanonicalQuotaWindow{usedPercent: used, hasUsed: true, reset: openAIQuotaWindowReset(extra, "7d", now)}
}
if window5h.hasUsed && window7d.hasUsed {
return window5h, window7d
}
snapshot := &OpenAICodexUsageSnapshot{}
if used, ok := resolveAccountExtraNumber(extra, "codex_primary_used_percent"); ok {
snapshot.PrimaryUsedPercent = &used
}
if used, ok := resolveAccountExtraNumber(extra, "codex_secondary_used_percent"); ok {
snapshot.SecondaryUsedPercent = &used
}
if minutes := parseExtraInt(extra["codex_primary_window_minutes"]); minutes > 0 {
snapshot.PrimaryWindowMinutes = &minutes
}
if minutes := parseExtraInt(extra["codex_secondary_window_minutes"]); minutes > 0 {
snapshot.SecondaryWindowMinutes = &minutes
}
normalized := snapshot.Normalize()
if normalized == nil {
return window5h, window7d
}
fromRaw := func(used *float64) openAICanonicalQuotaWindow {
if used == nil {
return openAICanonicalQuotaWindow{}
}
// Normalize 保留原始字段指针,据此配对同一个窗口的用量和重置时间。
window := "secondary"
if used == snapshot.PrimaryUsedPercent {
window = "primary"
}
return openAICanonicalQuotaWindow{usedPercent: *used, hasUsed: true, reset: openAIQuotaWindowReset(extra, window, now)}
}
if !window5h.hasUsed {
window5h = fromRaw(normalized.Used5hPercent)
}
if !window7d.hasUsed {
window7d = fromRaw(normalized.Used7dPercent)
}
return window5h, window7d
}
func openAISchedulingResetWindowEnd(account *Account, now time.Time) (time.Time, bool) {
if account == nil {
return time.Time{}, false
}
if end, ok := openAICodexWindowResetAt(account.Extra, "5h"); ok && now.Before(end) {
return end, true
}
if end := account.SessionWindowEnd; end != nil && now.Before(*end) {
return *end, true
}
return time.Time{}, false
}
func openAIQuotaHeadroomSnapshotStale(extra map[string]any, now time.Time) bool {
updatedRaw, ok := extra["codex_usage_updated_at"]
if !ok {
@@ -2980,15 +3047,6 @@ func openAIQuotaHeadroomSnapshotStale(extra map[string]any, now time.Time) bool
return now.Sub(updatedAt) >= openAIQuotaHeadroomSnapshotStaleAfter
}
func openAIQuotaWindowResetAny(extra map[string]any, now time.Time, windows ...string) bool {
for _, window := range windows {
if openAIQuotaWindowReset(extra, window, now) {
return true
}
}
return false
}
func clamp01(value float64) float64 {
switch {
case value < 0:
@@ -0,0 +1,141 @@
package service
import (
"context"
"strings"
"testing"
"time"
"github.com/stretchr/testify/require"
)
func TestOpenAIQuotaHeadroomFactorUsesCanonicalWindows(t *testing.T) {
now := time.Now()
shortMins, longMins := 300, 10080
shortReset, longReset := 3600, 86400
shortUsed, longUsed := 96.0, 25.0
for _, primaryIs5h := range []bool{true, false} {
name := "primary_is_7d"
snapshot := &OpenAICodexUsageSnapshot{
PrimaryUsedPercent: &longUsed, PrimaryWindowMinutes: &longMins, PrimaryResetAfterSeconds: &longReset,
SecondaryUsedPercent: &shortUsed, SecondaryWindowMinutes: &shortMins, SecondaryResetAfterSeconds: &shortReset,
}
if primaryIs5h {
name = "primary_is_5h"
snapshot = &OpenAICodexUsageSnapshot{
PrimaryUsedPercent: &shortUsed, PrimaryWindowMinutes: &shortMins, PrimaryResetAfterSeconds: &shortReset,
SecondaryUsedPercent: &longUsed, SecondaryWindowMinutes: &longMins, SecondaryResetAfterSeconds: &longReset,
}
}
t.Run(name, func(t *testing.T) {
for _, rawOnly := range []bool{false, true} {
fields := "canonical_and_raw"
if rawOnly {
fields = "raw_only"
}
t.Run(fields, func(t *testing.T) {
extra := buildCodexUsageExtraUpdates(snapshot, now)
if rawOnly {
for key := range extra {
if strings.HasPrefix(key, "codex_5h_") || strings.HasPrefix(key, "codex_7d_") {
delete(extra, key)
}
}
}
// 周余量 75%,5h 余量低于 10% 时沿用现有的减半规则。
require.InDelta(t, 0.375, openAIQuotaHeadroomFactor(&Account{Extra: extra}, now), 1e-9)
})
}
})
}
}
func TestOpenAIQuotaHeadroomFactorCanonicalPrecedence(t *testing.T) {
now := time.Now()
extra := map[string]any{
"codex_primary_used_percent": 40.0,
"codex_secondary_used_percent": 20.0,
"codex_usage_updated_at": now.Format(time.RFC3339),
}
// 没有窗口长度的历史数据仍沿用 primary=7d、secondary=5h。
require.InDelta(t, 0.6, openAIQuotaHeadroomFactor(&Account{Extra: extra}, now), 1e-9)
extra["codex_7d_used_percent"] = 80.0
extra["codex_5h_used_percent"] = 10.0
require.InDelta(t, 0.2, openAIQuotaHeadroomFactor(&Account{Extra: extra}, now), 1e-9)
// 规范窗口有效时,不应被另一个原始窗口的过期时间误判为已重置。
extra["codex_7d_reset_at"] = now.Add(time.Hour).Format(time.RFC3339)
extra["codex_primary_reset_at"] = now.Add(-time.Minute).Format(time.RFC3339)
require.InDelta(t, 0.2, openAIQuotaHeadroomFactor(&Account{Extra: extra}, now), 1e-9)
extra["codex_7d_reset_at"] = now.Add(-time.Minute).Format(time.RFC3339)
require.Equal(t, openAIQuotaHeadroomNeutralFactor, openAIQuotaHeadroomFactor(&Account{Extra: extra}, now))
}
func TestOpenAIQuotaHeadroomFactorRawWindowReset(t *testing.T) {
now := time.Now()
extra := map[string]any{
"codex_primary_used_percent": 96.0,
"codex_primary_window_minutes": 300,
"codex_primary_reset_at": now.Add(-time.Minute).Format(time.RFC3339),
"codex_secondary_used_percent": 25.0,
"codex_secondary_reset_at": now.Add(time.Hour).Format(time.RFC3339),
"codex_usage_updated_at": now.Add(-time.Hour).Format(time.RFC3339),
}
require.InDelta(t, 0.75, openAIQuotaHeadroomFactor(&Account{Extra: extra}, now), 1e-9)
extra["codex_secondary_reset_at"] = now.Add(-time.Minute).Format(time.RFC3339)
require.Equal(t, openAIQuotaHeadroomNeutralFactor, openAIQuotaHeadroomFactor(&Account{Extra: extra}, now))
}
func TestOpenAISchedulerCanonicalResetScoresMatchSnapshot(t *testing.T) {
now := time.Now()
scheduler := openAIResetTestScheduler(1)
accounts := []*Account{
{ID: 1, Extra: map[string]any{"codex_5h_reset_at": now.Add(10 * time.Minute).Format(time.RFC3339)}},
{ID: 2, Extra: map[string]any{"codex_5h_reset_at": now.Add(4 * time.Hour).Format(time.RFC3339)}},
}
plan := scheduler.buildOpenAIAccountLoadPlan(context.Background(), OpenAIAccountScheduleRequest{}, accounts, nil)
scores := openAIPlanScores(plan)
require.InDelta(t, 1, scores[1]-scores[2], 1e-9)
snapshots := buildOpenAIAccountSchedulerScoreSnapshot(accounts, nil, scheduler.service.openAIWSSchedulerWeights(), false, defaultOpenAIOAuthSchedulingRateMultiplier)
for _, account := range accounts {
require.InDelta(t, scores[account.ID], snapshots[account.ID].BaseScore, 1e-9)
}
}
func TestOpenAISchedulingResetWindowEnd(t *testing.T) {
now := time.Date(2026, 9, 1, 12, 0, 0, 0, time.UTC)
updated := now.Add(-30 * time.Minute)
account := &Account{Extra: map[string]any{
"codex_usage_updated_at": updated.Format(time.RFC3339),
"codex_5h_reset_after_seconds": 3600,
}}
end, ok := openAISchedulingResetWindowEnd(account, now)
require.True(t, ok)
require.Equal(t, updated.Add(time.Hour), end)
account.Extra["codex_5h_reset_at"] = now.Add(10 * time.Minute).Format(time.RFC3339)
end, ok = openAISchedulingResetWindowEnd(account, now)
require.True(t, ok)
require.Equal(t, now.Add(10*time.Minute), end)
delete(account.Extra, "codex_5h_reset_at")
account.Extra["codex_5h_reset_after_seconds"] = 60
_, ok = openAISchedulingResetWindowEnd(account, now)
require.False(t, ok)
account.Extra["codex_5h_reset_after_seconds"] = 3600
delete(account.Extra, "codex_usage_updated_at")
_, ok = openAISchedulingResetWindowEnd(account, now)
require.False(t, ok, "缺少采样时间时不能把相对倒计时锚定到当前时间")
account.Extra["codex_usage_updated_at"] = "invalid"
_, ok = openAISchedulingResetWindowEnd(account, now)
require.False(t, ok)
sessionEnd := now.Add(2 * time.Hour)
account.SessionWindowEnd = &sessionEnd
end, ok = openAISchedulingResetWindowEnd(account, now)
require.True(t, ok)
require.Equal(t, sessionEnd, end)
_, ok = openAISchedulingResetWindowEnd(nil, now)
require.False(t, ok)
}
@@ -710,26 +710,29 @@ func openAICodexSnapshotStaleForPause(extra map[string]any, now time.Time) bool
// timestamp and falls back to codex_<window>_reset_after_seconds anchored at
// codex_usage_updated_at, mirroring AccountUsageService's window-progress logic.
func openAIQuotaWindowReset(extra map[string]any, window string, now time.Time) bool {
resetAt, ok := openAICodexWindowResetAt(extra, window)
return ok && !now.Before(resetAt)
}
// 绝对时间优先;相对倒计时必须锚定快照采样时间,不能随每次评分向后滑动。
func openAICodexWindowResetAt(extra map[string]any, window string) (time.Time, bool) {
if len(extra) == 0 {
return false
return time.Time{}, false
}
if resetAtRaw, ok := extra["codex_"+window+"_reset_at"]; ok {
if resetAt, err := parseTime(fmt.Sprint(resetAtRaw)); err == nil {
return !now.Before(resetAt)
return resetAt, true
}
}
resetAfter := parseExtraInt(extra["codex_"+window+"_reset_after_seconds"])
if resetAfter <= 0 {
return false
return time.Time{}, false
}
base := now
if updatedRaw, ok := extra["codex_usage_updated_at"]; ok {
if updatedAt, err := parseTime(fmt.Sprint(updatedRaw)); err == nil {
base = updatedAt
}
updatedAt, err := parseTime(fmt.Sprint(extra["codex_usage_updated_at"]))
if err != nil {
return time.Time{}, false
}
resetAt := base.Add(time.Duration(resetAfter) * time.Second)
return !now.Before(resetAt)
return updatedAt.Add(time.Duration(resetAfter) * time.Second), true
}
func readOpenAIQuotaUsedPercent(extra map[string]any, window string) float64 {