From db76cc4e4ea6f96f9e6a82d1db45c39b32c41d1e Mon Sep 17 00:00:00 2001 From: Brisbanehuang Date: Sat, 12 Sep 2026 09:06:33 -0400 Subject: [PATCH] =?UTF-8?q?fix(scheduler):=20=E6=8C=89=E8=A7=84=E8=8C=83?= =?UTF-8?q?=E7=AA=97=E5=8F=A3=E8=AF=BB=E5=8F=96=20Codex=20=E9=85=8D?= =?UTF-8?q?=E9=A2=9D=E4=B8=8E=E9=87=8D=E7=BD=AE=E6=97=B6=E9=97=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 优先读取 codex_5h/codex_7d 规范字段，旧快照复用 Normalize 的窗口分类；用量与重置时间按同源窗口配对。重置评分与管理端快照复用规范 5h 重置时间，保留会话列回退，不改变权重与暂停阈值。 --- .../service/openai_account_scheduler.go | 104 ++++++++++--- ..._account_scheduler_canonical_quota_test.go | 141 ++++++++++++++++++ .../service/openai_gateway_scheduling.go | 23 +-- 3 files changed, 235 insertions(+), 33 deletions(-) create mode 100644 backend/internal/service/openai_account_scheduler_canonical_quota_test.go diff --git a/backend/internal/service/openai_account_scheduler.go b/backend/internal/service/openai_account_scheduler.go index 5ad63d30f8..d2c51b9b3f 100644 --- a/backend/internal/service/openai_account_scheduler.go +++ b/backend/internal/service/openai_account_scheduler.go @@ -957,15 +957,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() @@ -998,7 +998,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 { @@ -2727,8 +2727,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() @@ -2758,7 +2758,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 { @@ -2947,22 +2947,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 { @@ -2975,15 +3042,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: diff --git a/backend/internal/service/openai_account_scheduler_canonical_quota_test.go b/backend/internal/service/openai_account_scheduler_canonical_quota_test.go new file mode 100644 index 0000000000..575347e628 --- /dev/null +++ b/backend/internal/service/openai_account_scheduler_canonical_quota_test.go @@ -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) +} diff --git a/backend/internal/service/openai_gateway_scheduling.go b/backend/internal/service/openai_gateway_scheduling.go index 2ee6bea077..859e782291 100644 --- a/backend/internal/service/openai_gateway_scheduling.go +++ b/backend/internal/service/openai_gateway_scheduling.go @@ -710,26 +710,29 @@ func openAICodexSnapshotStaleForPause(extra map[string]any, now time.Time) bool // timestamp and falls back to codex__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 {