Merge pull request #3698 from Wei-Shaw/feat/anthropic-7d-oi-fable-model-rate-limit

feat(anthropic): 7d_oi(Fable 专属 7d 窗口)429 仅做模型级限流 + 用量窗口新增 7d F 进度条
This commit is contained in:
Wesley Liddick
2026-07-04 08:49:08 +08:00
committed by GitHub
11 changed files with 835 additions and 48 deletions
@@ -184,6 +184,7 @@ type UsageInfo struct {
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 日配额
@@ -236,6 +237,12 @@ type UsageInfo struct {
Error string `json:"error,omitempty"`
}
// ClaudeUsageWindow Anthropic /api/oauth/usage 返回的单个用量窗口
type ClaudeUsageWindow struct {
Utilization float64 `json:"utilization"`
ResetsAt string `json:"resets_at"`
}
// ClaudeUsageResponse Anthropic API返回的usage结构
type ClaudeUsageResponse struct {
FiveHour struct {
@@ -250,6 +257,10 @@ type ClaudeUsageResponse struct {
Utilization float64 `json:"utilization"`
ResetsAt string `json:"resets_at"`
} `json:"seven_day_sonnet"`
// Fable 专属 7d 窗口(对应响应头 7d_oi,claim 名为 seven_day_overage_included,
// 见 anthropic-ratelimit-unified-representative-claim 头)。上游 usage API
// 若不下发该字段,GetUsage 会用被动采样数据回填。
SevenDayOverageIncluded ClaudeUsageWindow `json:"seven_day_overage_included"`
}
// ClaudeUsageFetchOptions 包含获取 Claude 用量数据所需的所有选项
@@ -429,6 +440,12 @@ func (s *AccountUsageService) GetUsage(ctx context.Context, accountID int64, for
// 5. 将主动查询结果同步到被动缓存,下次 passive 加载即为最新值
s.syncActiveToPassive(ctx, account.ID, usage)
// 6. 上游 usage API 目前不一定下发 Fable 7d 窗口;缺失时回填被动采样
// (7d_oi 响应头)的数据,避免主动查询后 7d F 进度条丢失。
if usage.SevenDayFable == nil {
usage.SevenDayFable = buildPassiveUsageWindow(account.Extra, "passive_usage_7d_oi_utilization", "passive_usage_7d_oi_reset")
}
s.tryClearRecoverableAccountError(ctx, account)
return usage, nil
}
@@ -471,25 +488,10 @@ func (s *AccountUsageService) GetPassiveUsage(ctx context.Context, accountID int
}
// 构建 7d 窗口(从被动采样数据)
util7d := parseExtraFloat64(account.Extra["passive_usage_7d_utilization"])
reset7dRaw := parseExtraFloat64(account.Extra["passive_usage_7d_reset"])
if util7d > 0 || reset7dRaw > 0 {
var resetAt *time.Time
var remaining int
if reset7dRaw > 0 {
t := time.Unix(int64(reset7dRaw), 0)
resetAt = &t
remaining = int(time.Until(t).Seconds())
if remaining < 0 {
remaining = 0
}
}
info.SevenDay = &UsageProgress{
Utilization: util7d * 100,
ResetsAt: resetAt,
RemainingSeconds: remaining,
}
}
info.SevenDay = buildPassiveUsageWindow(account.Extra, "passive_usage_7d_utilization", "passive_usage_7d_reset")
// 构建 7d Fable 窗口(从被动采样的 7d_oi 响应头数据)
info.SevenDayFable = buildPassiveUsageWindow(account.Extra, "passive_usage_7d_oi_utilization", "passive_usage_7d_oi_reset")
// 添加窗口统计
s.addWindowStats(ctx, account, info)
@@ -497,6 +499,31 @@ func (s *AccountUsageService) GetPassiveUsage(ctx context.Context, accountID int
return info, nil
}
// buildPassiveUsageWindow 从 Extra 中的被动采样数据(utilization 为 0-1 小数、reset 为 Unix 秒)
// 构建用量窗口,无数据时返回 nil。
func buildPassiveUsageWindow(extra map[string]any, utilKey, resetKey string) *UsageProgress {
util := parseExtraFloat64(extra[utilKey])
resetRaw := parseExtraFloat64(extra[resetKey])
if util <= 0 && resetRaw <= 0 {
return nil
}
var resetAt *time.Time
var remaining int
if resetRaw > 0 {
t := time.Unix(int64(resetRaw), 0)
resetAt = &t
remaining = int(time.Until(t).Seconds())
if remaining < 0 {
remaining = 0
}
}
return &UsageProgress{
Utilization: util * 100,
ResetsAt: resetAt,
RemainingSeconds: remaining,
}
}
// syncActiveToPassive 将主动查询的最新数据回写到 Extra 被动缓存,
// 这样下次被动加载时能看到最新值。
func (s *AccountUsageService) syncActiveToPassive(ctx context.Context, accountID int64, usage *UsageInfo) {
@@ -511,6 +538,12 @@ func (s *AccountUsageService) syncActiveToPassive(ctx context.Context, accountID
extraUpdates["passive_usage_7d_reset"] = usage.SevenDay.ResetsAt.Unix()
}
}
if usage.SevenDayFable != nil {
extraUpdates["passive_usage_7d_oi_utilization"] = usage.SevenDayFable.Utilization / 100
if usage.SevenDayFable.ResetsAt != nil {
extraUpdates["passive_usage_7d_oi_reset"] = usage.SevenDayFable.ResetsAt.Unix()
}
}
if len(extraUpdates) > 0 {
extraUpdates["passive_usage_sampled_at"] = time.Now().UTC().Format(time.RFC3339)
@@ -1010,8 +1043,8 @@ func enrichUsageWithAccountError(info *UsageInfo, account *Account) {
// 使用独立缓存(1 分钟),与 API 缓存分离
func (s *AccountUsageService) addWindowStats(ctx context.Context, account *Account, usage *UsageInfo) {
// 修复:即使 FiveHour 为 nil,也要尝试获取统计数据
// 因为 SevenDay/SevenDaySonnet 可能需要
if usage.FiveHour == nil && usage.SevenDay == nil && usage.SevenDaySonnet == nil {
// 因为 SevenDay/SevenDaySonnet/SevenDayFable 可能需要
if usage.FiveHour == nil && usage.SevenDay == nil && usage.SevenDaySonnet == nil && usage.SevenDayFable == nil {
return
}
@@ -1347,6 +1380,22 @@ func (s *AccountUsageService) buildUsageInfo(resp *ClaudeUsageResponse, updatedA
}
}
// 7天Fable窗口(响应头 7d_oi 对应的窗口)
if fable := resp.SevenDayOverageIncluded; fable.ResetsAt != "" {
if fableReset, err := parseTime(fable.ResetsAt); err == nil {
info.SevenDayFable = &UsageProgress{
Utilization: fable.Utilization,
ResetsAt: &fableReset,
RemainingSeconds: int(time.Until(fableReset).Seconds()),
}
} else {
log.Printf("Failed to parse SevenDayFable.ResetsAt: %s, error: %v", fable.ResetsAt, err)
info.SevenDayFable = &UsageProgress{
Utilization: fable.Utilization,
}
}
}
return info
}
@@ -0,0 +1,119 @@
package service
import (
"encoding/json"
"testing"
"time"
"github.com/stretchr/testify/require"
)
func TestClaudeUsageResponse_FableWindowDecoding(t *testing.T) {
t.Run("seven_day_overage_included", func(t *testing.T) {
raw := `{
"five_hour": {"utilization": 12.0, "resets_at": "2026-07-03T10:00:00Z"},
"seven_day": {"utilization": 34.0, "resets_at": "2026-07-08T00:00:00Z"},
"seven_day_overage_included": {"utilization": 56.0, "resets_at": "2026-07-08T03:00:00Z"}
}`
var resp ClaudeUsageResponse
require.NoError(t, json.Unmarshal([]byte(raw), &resp))
require.Equal(t, 56.0, resp.SevenDayOverageIncluded.Utilization)
require.Equal(t, "2026-07-08T03:00:00Z", resp.SevenDayOverageIncluded.ResetsAt)
})
t.Run("absent", func(t *testing.T) {
raw := `{"five_hour": {"utilization": 12.0, "resets_at": "2026-07-03T10:00:00Z"}}`
var resp ClaudeUsageResponse
require.NoError(t, json.Unmarshal([]byte(raw), &resp))
require.Zero(t, resp.SevenDayOverageIncluded.Utilization)
require.Empty(t, resp.SevenDayOverageIncluded.ResetsAt)
})
}
func TestBuildUsageInfo_SevenDayFable(t *testing.T) {
svc := &AccountUsageService{}
now := time.Now()
resetAt := now.Add(72 * time.Hour).UTC().Truncate(time.Second)
var resp ClaudeUsageResponse
resp.FiveHour.Utilization = 10
resp.SevenDayOverageIncluded = ClaudeUsageWindow{
Utilization: 88,
ResetsAt: resetAt.Format(time.RFC3339),
}
info := svc.buildUsageInfo(&resp, &now)
require.NotNil(t, info.SevenDayFable)
require.Equal(t, 88.0, info.SevenDayFable.Utilization)
require.NotNil(t, info.SevenDayFable.ResetsAt)
require.True(t, info.SevenDayFable.ResetsAt.Equal(resetAt))
require.Greater(t, info.SevenDayFable.RemainingSeconds, 0)
// 无 Fable 数据时不应创建窗口
var empty ClaudeUsageResponse
empty.FiveHour.Utilization = 10
info = svc.buildUsageInfo(&empty, &now)
require.Nil(t, info.SevenDayFable)
}
func TestBuildPassiveUsageWindow(t *testing.T) {
future := time.Now().Add(48 * time.Hour).Unix()
t.Run("utilization and reset", func(t *testing.T) {
window := buildPassiveUsageWindow(map[string]any{
"passive_usage_7d_oi_utilization": 0.87,
"passive_usage_7d_oi_reset": float64(future),
}, "passive_usage_7d_oi_utilization", "passive_usage_7d_oi_reset")
require.NotNil(t, window)
require.InDelta(t, 87.0, window.Utilization, 1e-9)
require.NotNil(t, window.ResetsAt)
require.Equal(t, future, window.ResetsAt.Unix())
require.Greater(t, window.RemainingSeconds, 0)
})
t.Run("no data returns nil", func(t *testing.T) {
require.Nil(t, buildPassiveUsageWindow(nil, "u", "r"))
require.Nil(t, buildPassiveUsageWindow(map[string]any{}, "u", "r"))
})
t.Run("expired reset clamps remaining to zero", func(t *testing.T) {
past := time.Now().Add(-time.Hour).Unix()
window := buildPassiveUsageWindow(map[string]any{
"u": 0.5,
"r": float64(past),
}, "u", "r")
require.NotNil(t, window)
require.Equal(t, 0, window.RemainingSeconds)
})
t.Run("utilization only", func(t *testing.T) {
window := buildPassiveUsageWindow(map[string]any{"u": 0.25}, "u", "r")
require.NotNil(t, window)
require.InDelta(t, 25.0, window.Utilization, 1e-9)
require.Nil(t, window.ResetsAt)
})
}
func TestSyncActiveToPassive_WritesFableExtras(t *testing.T) {
repo := &accountUsageCodexProbeRepo{updateExtraCh: make(chan map[string]any, 1)}
svc := &AccountUsageService{accountRepo: repo}
resetAt := time.Now().Add(72 * time.Hour).Truncate(time.Second)
usage := &UsageInfo{
SevenDayFable: &UsageProgress{
Utilization: 87,
ResetsAt: &resetAt,
},
}
svc.syncActiveToPassive(t.Context(), 1, usage)
select {
case updates := <-repo.updateExtraCh:
require.InDelta(t, 0.87, updates["passive_usage_7d_oi_utilization"], 1e-9)
require.Equal(t, resetAt.Unix(), updates["passive_usage_7d_oi_reset"])
require.Contains(t, updates, "passive_usage_sampled_at")
default:
t.Fatal("expected UpdateExtra to be called with fable extras")
}
}
@@ -12,6 +12,9 @@ const (
modelRateLimitsKey = "model_rate_limits"
antigravityGeminiModelRateLimitKey = "antigravity:gemini"
openAIImageGenerationRateLimitKey = "openai:image_generation"
// anthropicFableRateLimitKey 是 Anthropic 7d_oi(Fable 专属 7d 窗口)限流的
// 家族级 scope:命中后所有 Fable 变体(含 [1m] 等后缀)都不再调度到该账号。
anthropicFableRateLimitKey = "claude-fable-5"
)
// isRateLimitActiveForKey 检查指定 key 的限流是否生效
@@ -82,10 +85,19 @@ func (a *Account) modelRateLimitKeysForRequest(ctx context.Context, requestedMod
if openAIImageGenerationRateLimitApplies(ctx, requestedModel, modelKey) && modelKey != openAIImageGenerationRateLimitKey {
keys = append(keys, openAIImageGenerationRateLimitKey)
}
case PlatformAnthropic:
if isAnthropicFableModel(modelKey) && modelKey != anthropicFableRateLimitKey {
keys = append(keys, anthropicFableRateLimitKey)
}
}
return keys
}
// isAnthropicFableModel 判断是否为 Fable 模型家族(claude-fable-5、claude-fable-5[1m] 等变体)
func isAnthropicFableModel(model string) bool {
return strings.Contains(strings.ToLower(model), "fable")
}
func openAIImageGenerationRateLimitApplies(ctx context.Context, requestedModel, modelKey string) bool {
if isOpenAIImageGenerationModel(requestedModel) || isOpenAIImageGenerationModel(modelKey) {
return true
@@ -499,3 +499,47 @@ func TestGetRateLimitRemainingTime(t *testing.T) {
})
}
}
func TestIsModelRateLimited_AnthropicFableFamilyKey(t *testing.T) {
now := time.Now()
future := now.Add(48 * time.Hour).Format(time.RFC3339)
account := &Account{
Platform: PlatformAnthropic,
Extra: map[string]any{
modelRateLimitsKey: map[string]any{
anthropicFableRateLimitKey: map[string]any{
"rate_limit_reset_at": future,
},
},
},
}
tests := []struct {
requestedModel string
expected bool
}{
{"claude-fable-5", true},
{"claude-fable-5[1m]", true}, // 家族 key 覆盖变体
{"Claude-Fable-5-20260601", true}, // 大小写不敏感
{"claude-sonnet-4-6", false}, // 其他模型不受影响
{"claude-opus-4-8", false},
}
for _, tc := range tests {
t.Run(tc.requestedModel, func(t *testing.T) {
got := account.isModelRateLimitedWithContext(context.Background(), tc.requestedModel)
require.Equal(t, tc.expected, got)
remaining := account.GetModelRateLimitRemainingTimeWithContext(context.Background(), tc.requestedModel)
require.Equal(t, tc.expected, remaining > 0)
})
}
}
func TestIsAnthropicFableModel(t *testing.T) {
require.True(t, isAnthropicFableModel("claude-fable-5"))
require.True(t, isAnthropicFableModel("claude-fable-5[1m]"))
require.True(t, isAnthropicFableModel("Claude-Fable-5"))
require.False(t, isAnthropicFableModel("claude-sonnet-4-6"))
require.False(t, isAnthropicFableModel(""))
}
+128 -24
View File
@@ -186,9 +186,14 @@ func (s *RateLimitService) HandleUpstreamError(ctx context.Context, account *Acc
// otherwise a broad "rate limit" keyword rule can shorten a multi-hour
// cooldown to a local temporary pause.
if statusCode == http.StatusTooManyRequests && account.Platform == PlatformAnthropic {
// 7d_oi 是 Fable 模型专属的 7d 窗口:只标记模型级限流,账号对其他模型仍可调度。
fableLimited := s.persistAnthropicFableWindowLimit(ctx, account, headers)
if s.persistAnthropicExhaustedWindowLimit(ctx, account, headers) {
return false
}
if fableLimited {
return false
}
}
// 先尝试临时不可调度规则(401除外)
@@ -1140,11 +1145,25 @@ func selectAnthropicExhaustedWindow(headers http.Header, now time.Time) *anthrop
}
func isAnthropic5hRejected(headers http.Header) bool {
return strings.EqualFold(strings.TrimSpace(headers.Get("anthropic-ratelimit-unified-5h-status")), "rejected")
return isAnthropicWindowRejected(headers, "5h")
}
func isAnthropicWindowRejected(headers http.Header, window string) bool {
return strings.EqualFold(strings.TrimSpace(headers.Get("anthropic-ratelimit-unified-"+window+"-status")), "rejected")
}
func parseAnthropicWindowReset(headers http.Header, window string, now time.Time) (time.Time, bool) {
raw := strings.TrimSpace(headers.Get("anthropic-ratelimit-unified-" + window + "-reset"))
maxAge := 8 * 24 * time.Hour
if window == "5h" {
maxAge = 6 * time.Hour
}
return parseAnthropicResetTimestamp(headers.Get("anthropic-ratelimit-unified-"+window+"-reset"), now, maxAge)
}
// parseAnthropicResetTimestamp 解析 Anthropic reset 头的 Unix 时间戳(自动识别毫秒),
// 并校验落在 (now, now+maxAge] 的合理区间内。
func parseAnthropicResetTimestamp(raw string, now time.Time, maxAge time.Duration) (time.Time, bool) {
raw = strings.TrimSpace(raw)
if raw == "" {
return time.Time{}, false
}
@@ -1156,15 +1175,7 @@ func parseAnthropicWindowReset(headers http.Header, window string, now time.Time
ts = ts / 1000
}
resetAt := time.Unix(ts, 0)
if !resetAt.After(now) {
return time.Time{}, false
}
maxAge := 8 * 24 * time.Hour
if window == "5h" {
maxAge = 6 * time.Hour
}
if resetAt.After(now.Add(maxAge)) {
if !resetAt.After(now) || resetAt.After(now.Add(maxAge)) {
return time.Time{}, false
}
return resetAt, true
@@ -1218,6 +1229,76 @@ func (s *RateLimitService) persistAnthropicExhaustedWindowLimit(ctx context.Cont
return true
}
const anthropicFableWindowReason = "anthropic_7d_oi_window_exhausted"
// selectAnthropicFableWindowLimit parses the Anthropic 7d_oi per-model window
// headers (the Fable-only 7d window, e.g. anthropic-ratelimit-unified-7d_oi-*).
// Unlike 5h/7d, exhaustion of this window only limits the Fable model family —
// the account must stay schedulable for other models.
//
// The 7d_oi surpassed-threshold header carries a float ("1.0") rather than
// "true", so exhaustion is detected via status=rejected or utilization >= 1.0.
// When the 7d_oi reset header is missing, the aggregated
// anthropic-ratelimit-unified-reset is used (it mirrors the binding claim's
// reset when 7d_oi is the representative claim).
func selectAnthropicFableWindowLimit(headers http.Header, now time.Time) *anthropicWindowLimit {
if !isAnthropicWindowRejected(headers, "7d_oi") && !isAnthropicWindowExceeded(headers, "7d_oi") {
return nil
}
resetAt, ok := parseAnthropicWindowReset(headers, "7d_oi", now)
if !ok {
resetAt, ok = parseAnthropicAggregateReset(headers, now)
}
if !ok {
return nil
}
return &anthropicWindowLimit{
window: "7d_oi",
resetAt: resetAt,
reason: anthropicFableWindowReason,
}
}
// parseAnthropicAggregateReset parses the aggregated
// anthropic-ratelimit-unified-reset header with the same sanity checks as the
// per-window variant (7d scale).
func parseAnthropicAggregateReset(headers http.Header, now time.Time) (time.Time, bool) {
return parseAnthropicResetTimestamp(headers.Get("anthropic-ratelimit-unified-reset"), now, 8*24*time.Hour)
}
// persistAnthropicFableWindowLimit marks the Fable model family as rate limited
// when the 7d_oi window is exhausted. Returns true when the 7d_oi window was the
// (or a) trigger of this 429, so the caller must not fall through to logic that
// would mark the whole account as rate limited.
func (s *RateLimitService) persistAnthropicFableWindowLimit(ctx context.Context, account *Account, headers http.Header) bool {
if s == nil || s.accountRepo == nil || account == nil {
return false
}
now := time.Now()
limit := selectAnthropicFableWindowLimit(headers, now)
if limit == nil {
return false
}
// 429 响应头本身携带最新的窗口用量(7d_oi utilization=1.0)。限流期内
// Fable 请求不再调度到该账号,若不在此处采样,7d F 进度条会冻结在
// 限流前的旧值直到窗口重置。
s.samplePassiveUsageFromHeaders(ctx, account, headers)
if err := s.accountRepo.SetModelRateLimit(ctx, account.ID, anthropicFableRateLimitKey, limit.resetAt, limit.reason); err != nil {
slog.Warn("anthropic_fable_window_rate_limit_set_failed",
"account_id", account.ID,
"scope", anthropicFableRateLimitKey,
"reset_at", limit.resetAt,
"error", err)
return true
}
slog.Info("anthropic_fable_window_model_rate_limited",
"account_id", account.ID,
"scope", anthropicFableRateLimitKey,
"reset_at", limit.resetAt,
"reset_in", time.Until(limit.resetAt).Truncate(time.Second))
return true
}
// calculateAnthropic429ResetTime parses Anthropic's per-window rate-limit headers
// to determine which window (5h or 7d) actually triggered the 429.
//
@@ -1541,10 +1622,12 @@ func (s *RateLimitService) UpdateSessionWindow(ctx context.Context, account *Acc
// 窗口重置时清除旧的 utilization 和被动采样数据,避免残留上个窗口的数据
if windowEnd != nil && needInitWindow {
_ = s.accountRepo.UpdateExtra(ctx, account.ID, map[string]any{
"session_window_utilization": nil,
"passive_usage_7d_utilization": nil,
"passive_usage_7d_reset": nil,
"passive_usage_sampled_at": nil,
"session_window_utilization": nil,
"passive_usage_7d_utilization": nil,
"passive_usage_7d_reset": nil,
"passive_usage_7d_oi_utilization": nil,
"passive_usage_7d_oi_reset": nil,
"passive_usage_sampled_at": nil,
})
}
@@ -1552,8 +1635,21 @@ func (s *RateLimitService) UpdateSessionWindow(ctx context.Context, account *Acc
slog.Warn("session_window_update_failed", "account_id", account.ID, "error", err)
}
// 被动采样:从响应头收集 5h + 7d utilization,合并为一次 DB 写入
extraUpdates := make(map[string]any, 4)
// 被动采样:从响应头收集 5h + 7d + 7d_oi utilization,合并为一次 DB 写入
s.samplePassiveUsageFromHeaders(ctx, account, headers)
// 如果状态为allowed且之前有限流,说明窗口已重置,清除限流状态
if status == "allowed" && account.IsRateLimited() {
if err := s.ClearRateLimit(ctx, account.ID); err != nil {
slog.Warn("rate_limit_clear_failed", "account_id", account.ID, "error", err)
}
}
}
// samplePassiveUsageFromHeaders 从 Anthropic 响应头收集 5h/7d/7d_oi 的
// utilization 与 reset 被动采样数据,合并为一次 Extra 写入。无数据时不写。
func (s *RateLimitService) samplePassiveUsageFromHeaders(ctx context.Context, account *Account, headers http.Header) {
extraUpdates := make(map[string]any, 6)
// 5h utilization(0-1 小数),供 estimateSetupTokenUsage 使用
if utilStr := headers.Get("anthropic-ratelimit-unified-5h-utilization"); utilStr != "" {
if util, err := strconv.ParseFloat(utilStr, 64); err == nil {
@@ -1575,19 +1671,27 @@ func (s *RateLimitService) UpdateSessionWindow(ctx context.Context, account *Acc
extraUpdates["passive_usage_7d_reset"] = ts
}
}
// 7d_oi (Fable 专属 7d 窗口) utilization(0-1 小数)
if utilStr := headers.Get("anthropic-ratelimit-unified-7d_oi-utilization"); utilStr != "" {
if util, err := strconv.ParseFloat(utilStr, 64); err == nil {
extraUpdates["passive_usage_7d_oi_utilization"] = util
}
}
// 7d_oi reset timestamp
if resetStr := headers.Get("anthropic-ratelimit-unified-7d_oi-reset"); resetStr != "" {
if ts, err := strconv.ParseInt(resetStr, 10, 64); err == nil {
if ts > 1e11 {
ts = ts / 1000
}
extraUpdates["passive_usage_7d_oi_reset"] = ts
}
}
if len(extraUpdates) > 0 {
extraUpdates["passive_usage_sampled_at"] = time.Now().UTC().Format(time.RFC3339)
if err := s.accountRepo.UpdateExtra(ctx, account.ID, extraUpdates); err != nil {
slog.Warn("passive_usage_update_failed", "account_id", account.ID, "error", err)
}
}
// 如果状态为allowed且之前有限流,说明窗口已重置,清除限流状态
if status == "allowed" && account.IsRateLimited() {
if err := s.ClearRateLimit(ctx, account.ID); err != nil {
slog.Warn("rate_limit_clear_failed", "account_id", account.ID, "error", err)
}
}
}
// ClearRateLimit 清除账号的限流状态
@@ -2,6 +2,7 @@ package service
import (
"net/http"
"strconv"
"testing"
"time"
)
@@ -181,6 +182,142 @@ func TestIsAnthropicWindowExceeded(t *testing.T) {
}
}
func TestSelectAnthropicFableWindowLimit_RejectedStatus(t *testing.T) {
now := time.Now()
reset := now.Add(80 * time.Hour).Truncate(time.Second)
headers := http.Header{}
headers.Set("anthropic-ratelimit-unified-7d_oi-status", "rejected")
headers.Set("anthropic-ratelimit-unified-7d_oi-utilization", "1.0")
headers.Set("anthropic-ratelimit-unified-7d_oi-surpassed-threshold", "1.0")
headers.Set("anthropic-ratelimit-unified-7d_oi-reset", strconv.FormatInt(reset.Unix(), 10))
limit := selectAnthropicFableWindowLimit(headers, now)
if limit == nil {
t.Fatal("expected non-nil limit")
}
if !limit.resetAt.Equal(reset) {
t.Errorf("expected resetAt=%v, got %v", reset, limit.resetAt)
}
if limit.reason != anthropicFableWindowReason {
t.Errorf("expected reason=%q, got %q", anthropicFableWindowReason, limit.reason)
}
}
func TestSelectAnthropicFableWindowLimit_UtilizationOnly(t *testing.T) {
// 无 status 头时,utilization >= 1.0 也应视为超限
now := time.Now()
reset := now.Add(3 * 24 * time.Hour).Truncate(time.Second)
headers := http.Header{}
headers.Set("anthropic-ratelimit-unified-7d_oi-utilization", "1.0")
headers.Set("anthropic-ratelimit-unified-7d_oi-reset", strconv.FormatInt(reset.Unix(), 10))
limit := selectAnthropicFableWindowLimit(headers, now)
if limit == nil {
t.Fatal("expected non-nil limit")
}
if !limit.resetAt.Equal(reset) {
t.Errorf("expected resetAt=%v, got %v", reset, limit.resetAt)
}
}
func TestSelectAnthropicFableWindowLimit_AllowedReturnsNil(t *testing.T) {
now := time.Now()
headers := http.Header{}
headers.Set("anthropic-ratelimit-unified-7d_oi-status", "allowed")
headers.Set("anthropic-ratelimit-unified-7d_oi-utilization", "0.56")
headers.Set("anthropic-ratelimit-unified-7d_oi-reset", strconv.FormatInt(now.Add(80*time.Hour).Unix(), 10))
if limit := selectAnthropicFableWindowLimit(headers, now); limit != nil {
t.Errorf("expected nil limit for allowed window, got %+v", limit)
}
}
func TestSelectAnthropicFableWindowLimit_NoHeadersReturnsNil(t *testing.T) {
if limit := selectAnthropicFableWindowLimit(http.Header{}, time.Now()); limit != nil {
t.Errorf("expected nil limit for empty headers, got %+v", limit)
}
}
func TestSelectAnthropicFableWindowLimit_FallsBackToAggregateReset(t *testing.T) {
// 7d_oi-reset 缺失时回退聚合 anthropic-ratelimit-unified-reset
now := time.Now()
reset := now.Add(80 * time.Hour).Truncate(time.Second)
headers := http.Header{}
headers.Set("anthropic-ratelimit-unified-7d_oi-status", "rejected")
headers.Set("anthropic-ratelimit-unified-reset", strconv.FormatInt(reset.Unix(), 10))
limit := selectAnthropicFableWindowLimit(headers, now)
if limit == nil {
t.Fatal("expected non-nil limit via aggregate reset fallback")
}
if !limit.resetAt.Equal(reset) {
t.Errorf("expected resetAt=%v, got %v", reset, limit.resetAt)
}
}
func TestSelectAnthropicFableWindowLimit_RejectedWithoutAnyResetReturnsNil(t *testing.T) {
headers := http.Header{}
headers.Set("anthropic-ratelimit-unified-7d_oi-status", "rejected")
if limit := selectAnthropicFableWindowLimit(headers, time.Now()); limit != nil {
t.Errorf("expected nil limit when no reset time available, got %+v", limit)
}
}
func TestParseAnthropicAggregateReset(t *testing.T) {
now := time.Now()
future := now.Add(80 * time.Hour).Truncate(time.Second)
tests := []struct {
name string
value string
want time.Time
wantOK bool
}{
{"valid seconds", strconv.FormatInt(future.Unix(), 10), future, true},
{"valid milliseconds", strconv.FormatInt(future.UnixMilli(), 10), future, true},
{"empty", "", time.Time{}, false},
{"garbage", "abc", time.Time{}, false},
{"in the past", strconv.FormatInt(now.Add(-time.Hour).Unix(), 10), time.Time{}, false},
{"too far in the future", strconv.FormatInt(now.Add(30*24*time.Hour).Unix(), 10), time.Time{}, false},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
headers := http.Header{}
if tc.value != "" {
headers.Set("anthropic-ratelimit-unified-reset", tc.value)
}
got, ok := parseAnthropicAggregateReset(headers, now)
if ok != tc.wantOK {
t.Fatalf("expected ok=%v, got %v", tc.wantOK, ok)
}
if ok && !got.Equal(tc.want) {
t.Errorf("expected %v, got %v", tc.want, got)
}
})
}
}
func TestIsAnthropicWindowRejected(t *testing.T) {
headers := http.Header{}
headers.Set("anthropic-ratelimit-unified-7d_oi-status", "Rejected")
headers.Set("anthropic-ratelimit-unified-5h-status", "allowed")
if !isAnthropicWindowRejected(headers, "7d_oi") {
t.Error("expected 7d_oi to be rejected (case insensitive)")
}
if isAnthropicWindowRejected(headers, "5h") {
t.Error("expected 5h not rejected")
}
if isAnthropicWindowRejected(headers, "7d") {
t.Error("expected missing 7d status not rejected")
}
}
// assertAnthropicResult is a test helper that verifies the result is non-nil and
// has the expected resetAt unix timestamp.
func assertAnthropicResult(t *testing.T, result *anthropic429Result, wantUnix int64) {
@@ -14,9 +14,14 @@ import (
type anthropicWindowLimitRepo struct {
mockAccountRepoForGemini
rateLimitCalls int
tempUnschedCalls int
lastRateLimitReset time.Time
rateLimitCalls int
tempUnschedCalls int
lastRateLimitReset time.Time
modelRateLimitCalls int
lastModelRateLimitScope string
lastModelRateLimitReset time.Time
sessionWindowCalls int
lastExtraUpdates map[string]any
}
func (r *anthropicWindowLimitRepo) SetRateLimited(_ context.Context, _ int64, resetAt time.Time) error {
@@ -30,6 +35,23 @@ func (r *anthropicWindowLimitRepo) SetTempUnschedulable(_ context.Context, _ int
return nil
}
func (r *anthropicWindowLimitRepo) SetModelRateLimit(_ context.Context, _ int64, scope string, resetAt time.Time, _ ...string) error {
r.modelRateLimitCalls++
r.lastModelRateLimitScope = scope
r.lastModelRateLimitReset = resetAt
return nil
}
func (r *anthropicWindowLimitRepo) UpdateSessionWindow(_ context.Context, _ int64, _, _ *time.Time, _ string) error {
r.sessionWindowCalls++
return nil
}
func (r *anthropicWindowLimitRepo) UpdateExtra(_ context.Context, _ int64, updates map[string]any) error {
r.lastExtraUpdates = updates
return nil
}
func TestHandleUpstreamError_AnthropicWindowLimitPreemptsTempUnschedRule(t *testing.T) {
resetAt := time.Now().Add(3 * time.Hour).Truncate(time.Second)
headers := http.Header{}
@@ -66,3 +88,142 @@ func TestHandleUpstreamError_AnthropicWindowLimitPreemptsTempUnschedRule(t *test
require.Equal(t, 1, repo.rateLimitCalls)
require.Equal(t, resetAt, repo.lastRateLimitReset)
}
// fable429Headers 构造 7d_oi(Fable 专属 7d 窗口)触发 429 的完整响应头,
// 数值取自真实抓包(5h/7d 均 allowed,仅 7d_oi rejected)。
func fable429Headers(reset5h, resetOI time.Time) http.Header {
headers := http.Header{}
headers.Set("anthropic-ratelimit-unified-5h-reset", strconv.FormatInt(reset5h.Unix(), 10))
headers.Set("anthropic-ratelimit-unified-5h-status", "allowed")
headers.Set("anthropic-ratelimit-unified-5h-utilization", "0.41")
headers.Set("anthropic-ratelimit-unified-7d-reset", strconv.FormatInt(resetOI.Unix(), 10))
headers.Set("anthropic-ratelimit-unified-7d-status", "allowed")
headers.Set("anthropic-ratelimit-unified-7d-utilization", "0.56")
headers.Set("anthropic-ratelimit-unified-7d_oi-reset", strconv.FormatInt(resetOI.Unix(), 10))
headers.Set("anthropic-ratelimit-unified-7d_oi-status", "rejected")
headers.Set("anthropic-ratelimit-unified-7d_oi-surpassed-threshold", "1.0")
headers.Set("anthropic-ratelimit-unified-7d_oi-utilization", "1.0")
headers.Set("anthropic-ratelimit-unified-fallback-percentage", "0.5")
headers.Set("anthropic-ratelimit-unified-overage-disabled-reason", "org_level_disabled")
headers.Set("anthropic-ratelimit-unified-overage-status", "rejected")
headers.Set("anthropic-ratelimit-unified-representative-claim", "seven_day_overage_included")
headers.Set("anthropic-ratelimit-unified-reset", strconv.FormatInt(resetOI.Unix(), 10))
headers.Set("anthropic-ratelimit-unified-status", "rejected")
return headers
}
func TestHandleUpstreamError_Anthropic7dOiOnlyMarksModelRateLimit(t *testing.T) {
now := time.Now()
reset5h := now.Add(2 * time.Hour).Truncate(time.Second)
resetOI := now.Add(80 * time.Hour).Truncate(time.Second)
headers := fable429Headers(reset5h, resetOI)
repo := &anthropicWindowLimitRepo{}
svc := NewRateLimitService(repo, nil, nil, nil, nil)
account := &Account{
ID: 42,
Type: AccountTypeOAuth,
Platform: PlatformAnthropic,
Credentials: map[string]any{
"temp_unschedulable_enabled": true,
"temp_unschedulable_rules": []any{
map[string]any{
"error_code": float64(http.StatusTooManyRequests),
"keywords": []any{"rate limit"},
"duration_minutes": float64(10),
},
},
},
}
shouldDisable := svc.HandleUpstreamError(
context.Background(),
account,
http.StatusTooManyRequests,
headers,
[]byte(`{"type":"error","error":{"type":"rate_limit_error","message":"This request would exceed your account's rate limit. Please try again later."}}`),
"claude-fable-5",
)
require.False(t, shouldDisable)
require.Zero(t, repo.rateLimitCalls, "7d_oi (Fable-only) window must not mark the whole account rate limited")
require.Zero(t, repo.tempUnschedCalls, "7d_oi window must not trigger local temp-unsched rules")
require.Zero(t, repo.sessionWindowCalls, "7d_oi window must not rewrite the 5h session window as rejected")
require.Equal(t, 1, repo.modelRateLimitCalls)
require.Equal(t, anthropicFableRateLimitKey, repo.lastModelRateLimitScope)
require.Equal(t, resetOI, repo.lastModelRateLimitReset)
// 429 响应头也要被动采样,避免 7d F 进度条在限流期内冻结在旧值
require.NotNil(t, repo.lastExtraUpdates)
require.Equal(t, 1.0, repo.lastExtraUpdates["passive_usage_7d_oi_utilization"])
require.Equal(t, resetOI.Unix(), repo.lastExtraUpdates["passive_usage_7d_oi_reset"])
require.Equal(t, 0.41, repo.lastExtraUpdates["session_window_utilization"])
}
func TestHandleUpstreamError_Anthropic5hWindowStillWinsOver7dOi(t *testing.T) {
// 5h 窗口 rejected 时必须仍按账号级限流处理(用 5h reset),同时记录 Fable 模型限流。
now := time.Now()
reset5h := now.Add(2 * time.Hour).Truncate(time.Second)
resetOI := now.Add(80 * time.Hour).Truncate(time.Second)
headers := fable429Headers(reset5h, resetOI)
headers.Set("anthropic-ratelimit-unified-5h-status", "rejected")
headers.Set("anthropic-ratelimit-unified-5h-utilization", "1.0")
repo := &anthropicWindowLimitRepo{}
svc := NewRateLimitService(repo, nil, nil, nil, nil)
account := &Account{ID: 42, Type: AccountTypeOAuth, Platform: PlatformAnthropic}
svc.HandleUpstreamError(context.Background(), account, http.StatusTooManyRequests, headers, nil, "claude-fable-5")
require.Equal(t, 1, repo.rateLimitCalls, "exhausted 5h window must still rate limit the account")
require.Equal(t, reset5h, repo.lastRateLimitReset)
require.Equal(t, 1, repo.modelRateLimitCalls)
require.Equal(t, anthropicFableRateLimitKey, repo.lastModelRateLimitScope)
}
func TestHandleUpstreamError_AnthropicAccountWindowStillWinsOver7dOi(t *testing.T) {
// 7d 窗口真超限时必须仍按账号级限流处理,同时记录 Fable 模型限流。
now := time.Now()
reset5h := now.Add(2 * time.Hour).Truncate(time.Second)
resetOI := now.Add(80 * time.Hour).Truncate(time.Second)
headers := fable429Headers(reset5h, resetOI)
headers.Set("anthropic-ratelimit-unified-7d-status", "rejected")
headers.Set("anthropic-ratelimit-unified-7d-utilization", "1.02")
repo := &anthropicWindowLimitRepo{}
svc := NewRateLimitService(repo, nil, nil, nil, nil)
account := &Account{ID: 42, Type: AccountTypeOAuth, Platform: PlatformAnthropic}
svc.HandleUpstreamError(context.Background(), account, http.StatusTooManyRequests, headers, nil, "claude-fable-5")
require.Equal(t, 1, repo.rateLimitCalls, "exhausted 7d window must still rate limit the account")
require.Equal(t, resetOI, repo.lastRateLimitReset)
require.Equal(t, 1, repo.modelRateLimitCalls, "Fable model rate limit should also be recorded")
require.Equal(t, anthropicFableRateLimitKey, repo.lastModelRateLimitScope)
}
func TestHandleUpstreamError_Anthropic429Without7dOiKeepsLegacyBehavior(t *testing.T) {
// 无 7d_oi 头、5h/7d 均未超限的 429:保持旧行为(按较早 reset 标记账号限流)。
now := time.Now()
reset5h := now.Add(2 * time.Hour).Truncate(time.Second)
reset7d := now.Add(80 * time.Hour).Truncate(time.Second)
headers := http.Header{}
headers.Set("anthropic-ratelimit-unified-5h-reset", strconv.FormatInt(reset5h.Unix(), 10))
headers.Set("anthropic-ratelimit-unified-5h-status", "allowed")
headers.Set("anthropic-ratelimit-unified-5h-utilization", "0.41")
headers.Set("anthropic-ratelimit-unified-7d-reset", strconv.FormatInt(reset7d.Unix(), 10))
headers.Set("anthropic-ratelimit-unified-7d-status", "allowed")
headers.Set("anthropic-ratelimit-unified-7d-utilization", "0.56")
repo := &anthropicWindowLimitRepo{}
svc := NewRateLimitService(repo, nil, nil, nil, nil)
account := &Account{ID: 42, Type: AccountTypeOAuth, Platform: PlatformAnthropic}
svc.HandleUpstreamError(context.Background(), account, http.StatusTooManyRequests, headers, nil, "claude-fable-5")
require.Zero(t, repo.modelRateLimitCalls, "no 7d_oi signal → no model rate limit")
require.Equal(t, 1, repo.rateLimitCalls)
require.Equal(t, reset5h, repo.lastRateLimitReset, "legacy path picks the sooner reset")
require.Equal(t, 1, repo.sessionWindowCalls)
}
@@ -367,6 +367,59 @@ func TestUpdateSessionWindow_NoClearUtilizationOnCorrection(t *testing.T) {
}
}
func TestUpdateSessionWindow_SamplesFable7dOiHeaders(t *testing.T) {
// 被动采样应收集 7d_oi(Fable 专属 7d 窗口)的 utilization 和 reset。
existingEnd := time.Now().Add(3 * time.Hour)
resetOIUnix := time.Now().Add(80 * time.Hour).Unix()
repo := &sessionWindowMockRepo{}
svc := newRateLimitServiceForTest(repo)
account := &Account{ID: 90, SessionWindowEnd: &existingEnd} // needInitWindow=false
headers := http.Header{}
headers.Set("anthropic-ratelimit-unified-5h-status", "allowed")
headers.Set("anthropic-ratelimit-unified-7d_oi-utilization", "0.87")
headers.Set("anthropic-ratelimit-unified-7d_oi-reset", fmt.Sprintf("%d", resetOIUnix))
svc.UpdateSessionWindow(context.Background(), account, headers)
if len(repo.updateExtraCalls) != 1 {
t.Fatalf("expected 1 UpdateExtra call, got %d", len(repo.updateExtraCalls))
}
updates := repo.updateExtraCalls[0].Updates
if val, ok := updates["passive_usage_7d_oi_utilization"].(float64); !ok || val != 0.87 {
t.Errorf("expected passive_usage_7d_oi_utilization=0.87, got %v", updates["passive_usage_7d_oi_utilization"])
}
if val, ok := updates["passive_usage_7d_oi_reset"].(int64); !ok || val != resetOIUnix {
t.Errorf("expected passive_usage_7d_oi_reset=%d, got %v", resetOIUnix, updates["passive_usage_7d_oi_reset"])
}
}
func TestUpdateSessionWindow_ClearsFable7dOiOnWindowReset(t *testing.T) {
// 5h 窗口重置时应连同清除 7d_oi 被动采样数据,与 7d 行为一致。
resetUnix := time.Now().Add(3 * time.Hour).Unix()
repo := &sessionWindowMockRepo{}
svc := newRateLimitServiceForTest(repo)
account := &Account{ID: 91} // no existing window → needInitWindow=true
headers := http.Header{}
headers.Set("anthropic-ratelimit-unified-5h-status", "allowed")
headers.Set("anthropic-ratelimit-unified-5h-reset", fmt.Sprintf("%d", resetUnix))
svc.UpdateSessionWindow(context.Background(), account, headers)
if len(repo.updateExtraCalls) != 1 {
t.Fatalf("expected 1 UpdateExtra (clear) call, got %d", len(repo.updateExtraCalls))
}
clearUpdates := repo.updateExtraCalls[0].Updates
for _, key := range []string{"passive_usage_7d_oi_utilization", "passive_usage_7d_oi_reset"} {
if val, present := clearUpdates[key]; !present || val != nil {
t.Errorf("expected %s cleared to nil on window reset, got present=%v val=%v", key, present, val)
}
}
}
func TestUpdateSessionWindow_NoStatusHeader(t *testing.T) {
// Should return immediately if no status header.
repo := &sessionWindowMockRepo{}
@@ -68,6 +68,15 @@
color="purple"
/>
<!-- 7d Fable Window (7d_oi) -->
<UsageProgressBar
v-if="usageInfo.seven_day_fable"
label="7d F"
:utilization="usageInfo.seven_day_fable.utilization"
:resets-at="usageInfo.seven_day_fable.resets_at"
color="amber"
/>
<!-- Passive sampling label + active query button -->
<div class="flex items-center gap-1.5 mt-0.5">
<span
@@ -655,4 +655,102 @@ describe('AccountUsageCell', () => {
expect(wrapper.text()).toContain('A $0.00')
expect(wrapper.text()).toContain('U $0.00')
})
it('Anthropic OAuth 会渲染 7d F (Fable) 进度条,且 7d S 逻辑保留', async () => {
getUsage.mockResolvedValue({
source: 'passive',
five_hour: {
utilization: 41,
resets_at: '2026-07-03T10:00:00Z',
remaining_seconds: 3600
},
seven_day: {
utilization: 56,
resets_at: '2026-07-06T22:00:00Z',
remaining_seconds: 300000
},
seven_day_sonnet: {
utilization: 30,
resets_at: '2026-07-06T22:00:00Z',
remaining_seconds: 300000
},
seven_day_fable: {
utilization: 100,
resets_at: '2026-07-06T22:00:00Z',
remaining_seconds: 300000
}
})
const wrapper = mount(AccountUsageCell, {
props: {
account: makeAccount({
id: 3001,
platform: 'anthropic',
type: 'oauth',
extra: {}
})
},
global: {
stubs: {
UsageProgressBar: {
props: ['label', 'utilization', 'resetsAt', 'color'],
template: '<div class="usage-bar">{{ label }}|{{ utilization }}</div>'
},
AccountQuotaInfo: true,
GrokQuotaProbeCell: true
}
}
})
await flushPromises()
expect(wrapper.text()).toContain('5h|41')
expect(wrapper.text()).toContain('7d|56')
expect(wrapper.text()).toContain('7d S|30')
expect(wrapper.text()).toContain('7d F|100')
})
it('Anthropic OAuth 无 Fable 数据时不渲染 7d F 进度条', async () => {
getUsage.mockResolvedValue({
source: 'passive',
five_hour: {
utilization: 41,
resets_at: '2026-07-03T10:00:00Z',
remaining_seconds: 3600
},
seven_day: {
utilization: 56,
resets_at: '2026-07-06T22:00:00Z',
remaining_seconds: 300000
}
})
const wrapper = mount(AccountUsageCell, {
props: {
account: makeAccount({
id: 3002,
platform: 'anthropic',
type: 'oauth',
extra: {}
})
},
global: {
stubs: {
UsageProgressBar: {
props: ['label', 'utilization', 'resetsAt', 'color'],
template: '<div class="usage-bar">{{ label }}|{{ utilization }}</div>'
},
AccountQuotaInfo: true,
GrokQuotaProbeCell: true
}
}
})
await flushPromises()
expect(wrapper.text()).toContain('5h|41')
expect(wrapper.text()).toContain('7d|56')
expect(wrapper.text()).not.toContain('7d S')
expect(wrapper.text()).not.toContain('7d F')
})
})
+1
View File
@@ -984,6 +984,7 @@ export interface AccountUsageInfo {
five_hour: UsageProgress | null
seven_day: UsageProgress | null
seven_day_sonnet: UsageProgress | null
seven_day_fable?: UsageProgress | null
gemini_shared_daily?: UsageProgress | null
gemini_pro_daily?: UsageProgress | null
gemini_flash_daily?: UsageProgress | null