diff --git a/backend/cmd/server/VERSION b/backend/cmd/server/VERSION index 2619c8dd8f..06306ef965 100644 --- a/backend/cmd/server/VERSION +++ b/backend/cmd/server/VERSION @@ -1 +1 @@ -0.1.91.1 +0.1.92.10 diff --git a/backend/internal/handler/admin/account_handler.go b/backend/internal/handler/admin/account_handler.go index d66a009cad..843d51c516 100644 --- a/backend/internal/handler/admin/account_handler.go +++ b/backend/internal/handler/admin/account_handler.go @@ -288,32 +288,48 @@ func (h *AccountHandler) List(c *gin.Context) { } } - // 仅非 lite 模式获取窗口费用(PostgreSQL 聚合查询,高开销) - if !lite && len(windowCostAccountIDs) > 0 { - windowCosts = make(map[int64]float64) - var mu sync.Mutex - g, gctx := errgroup.WithContext(c.Request.Context()) - g.SetLimit(10) // 限制并发数 - - for i := range accounts { - acc := &accounts[i] - if !acc.IsAnthropicOAuthOrSetupToken() || acc.GetWindowCostLimit() <= 0 { - continue - } - accCopy := acc // 闭包捕获 - g.Go(func() error { - // 使用统一的窗口开始时间计算逻辑(考虑窗口过期情况) - startTime := accCopy.GetCurrentWindowStartTime() - stats, err := h.accountUsageService.GetAccountWindowStats(gctx, accCopy.ID, startTime) - if err == nil && stats != nil { - mu.Lock() - windowCosts[accCopy.ID] = stats.StandardCost // 使用标准费用 - mu.Unlock() + // 窗口费用获取:lite 模式从快照缓存读取,非 lite 模式执行 PostgreSQL 查询后写入缓存 + if len(windowCostAccountIDs) > 0 { + if lite { + // lite 模式:尝试从快照缓存读取 + cacheKey := buildWindowCostCacheKey(windowCostAccountIDs) + if cached, ok := accountWindowCostCache.Get(cacheKey); ok { + if costs, ok := cached.Payload.(map[int64]float64); ok { + windowCosts = costs } - return nil // 不返回错误,允许部分失败 - }) + } + // 缓存未命中则 windowCosts 保持 nil(仅发生在服务刚启动时) + } else { + // 非 lite 模式:执行 PostgreSQL 聚合查询(高开销) + windowCosts = make(map[int64]float64) + var mu sync.Mutex + g, gctx := errgroup.WithContext(c.Request.Context()) + g.SetLimit(10) // 限制并发数 + + for i := range accounts { + acc := &accounts[i] + if !acc.IsAnthropicOAuthOrSetupToken() || acc.GetWindowCostLimit() <= 0 { + continue + } + accCopy := acc // 闭包捕获 + g.Go(func() error { + // 使用统一的窗口开始时间计算逻辑(考虑窗口过期情况) + startTime := accCopy.GetCurrentWindowStartTime() + stats, err := h.accountUsageService.GetAccountWindowStats(gctx, accCopy.ID, startTime) + if err == nil && stats != nil { + mu.Lock() + windowCosts[accCopy.ID] = stats.StandardCost // 使用标准费用 + mu.Unlock() + } + return nil // 不返回错误,允许部分失败 + }) + } + _ = g.Wait() + + // 查询完毕后写入快照缓存,供 lite 模式使用 + cacheKey := buildWindowCostCacheKey(windowCostAccountIDs) + accountWindowCostCache.Set(cacheKey, windowCosts) } - _ = g.Wait() } // Build response with concurrency info diff --git a/backend/internal/handler/admin/account_window_cost_cache.go b/backend/internal/handler/admin/account_window_cost_cache.go new file mode 100644 index 0000000000..3271b63036 --- /dev/null +++ b/backend/internal/handler/admin/account_window_cost_cache.go @@ -0,0 +1,25 @@ +package admin + +import ( + "strconv" + "strings" + "time" +) + +var accountWindowCostCache = newSnapshotCache(30 * time.Second) + +func buildWindowCostCacheKey(accountIDs []int64) string { + if len(accountIDs) == 0 { + return "accounts_window_cost_empty" + } + var b strings.Builder + b.Grow(len(accountIDs) * 6) + _, _ = b.WriteString("accounts_window_cost:") + for i, id := range accountIDs { + if i > 0 { + _ = b.WriteByte(',') + } + _, _ = b.WriteString(strconv.FormatInt(id, 10)) + } + return b.String() +} diff --git a/backend/internal/handler/dto/mappers.go b/backend/internal/handler/dto/mappers.go index d3fc51ed66..c21ff57d98 100644 --- a/backend/internal/handler/dto/mappers.go +++ b/backend/internal/handler/dto/mappers.go @@ -250,6 +250,12 @@ func AccountFromServiceShallow(a *service.Account) *Account { } } + // 客户端亲和调度(Anthropic 和 Antigravity 账号) + if a.IsClientAffinityEnabled() { + enabled := true + out.ClientAffinityEnabled = &enabled + } + // 提取 API Key 账号配额限制(仅 apikey 类型有效) if a.Type == service.AccountTypeAPIKey { if limit := a.GetQuotaLimit(); limit > 0 { diff --git a/backend/internal/handler/dto/types.go b/backend/internal/handler/dto/types.go index f0d13d3f02..3626bda982 100644 --- a/backend/internal/handler/dto/types.go +++ b/backend/internal/handler/dto/types.go @@ -188,6 +188,10 @@ type Account struct { CacheTTLOverrideEnabled *bool `json:"cache_ttl_override_enabled,omitempty"` CacheTTLOverrideTarget *string `json:"cache_ttl_override_target,omitempty"` + // 客户端亲和调度(Anthropic 和 Antigravity 账号有效) + // 启用后新会话会优先调度到客户端之前使用过的账号 + ClientAffinityEnabled *bool `json:"client_affinity_enabled,omitempty"` + // API Key 账号配额限制 QuotaLimit *float64 `json:"quota_limit,omitempty"` QuotaUsed *float64 `json:"quota_used,omitempty"` diff --git a/backend/internal/handler/sora_client_handler_test.go b/backend/internal/handler/sora_client_handler_test.go index c8b7a5ff6e..78931c88ca 100644 --- a/backend/internal/handler/sora_client_handler_test.go +++ b/backend/internal/handler/sora_client_handler_test.go @@ -2132,7 +2132,7 @@ func (r *stubAccountRepoForHandler) BulkUpdate(context.Context, []int64, service return 0, nil } -func (r *stubAccountRepoForHandler) IncrementQuotaUsed(context.Context, int64, float64) error { +func (r *stubAccountRepoForHandler) IncrementQuotaUsed(context.Context, int64, float64, string) error { return nil } diff --git a/backend/internal/handler/sora_gateway_handler_test.go b/backend/internal/handler/sora_gateway_handler_test.go index 637462ad50..2d1fb0a97f 100644 --- a/backend/internal/handler/sora_gateway_handler_test.go +++ b/backend/internal/handler/sora_gateway_handler_test.go @@ -216,7 +216,7 @@ func (r *stubAccountRepo) BulkUpdate(ctx context.Context, ids []int64, updates s return 0, nil } -func (r *stubAccountRepo) IncrementQuotaUsed(ctx context.Context, id int64, amount float64) error { +func (r *stubAccountRepo) IncrementQuotaUsed(ctx context.Context, id int64, amount float64, _ string) error { return nil } diff --git a/backend/internal/pkg/openai/constants.go b/backend/internal/pkg/openai/constants.go index 4bbc68e773..b0a31a5f38 100644 --- a/backend/internal/pkg/openai/constants.go +++ b/backend/internal/pkg/openai/constants.go @@ -15,6 +15,7 @@ type Model struct { // DefaultModels OpenAI models list var DefaultModels = []Model{ + {ID: "gpt-5.4", Object: "model", Created: 1738368000, OwnedBy: "openai", Type: "model", DisplayName: "GPT-5.4"}, {ID: "gpt-5.3-codex", Object: "model", Created: 1735689600, OwnedBy: "openai", Type: "model", DisplayName: "GPT-5.3 Codex"}, {ID: "gpt-5.3-codex-spark", Object: "model", Created: 1735689600, OwnedBy: "openai", Type: "model", DisplayName: "GPT-5.3 Codex Spark"}, {ID: "gpt-5.2", Object: "model", Created: 1733875200, OwnedBy: "openai", Type: "model", DisplayName: "GPT-5.2"}, diff --git a/backend/internal/repository/account_repo.go b/backend/internal/repository/account_repo.go index ffbfd46655..6fda0c182a 100644 --- a/backend/internal/repository/account_repo.go +++ b/backend/internal/repository/account_repo.go @@ -1677,18 +1677,53 @@ func (r *accountRepository) FindByExtraField(ctx context.Context, key string, va } // IncrementQuotaUsed 原子递增账号的 extra.quota_used 字段 -func (r *accountRepository) IncrementQuotaUsed(ctx context.Context, id int64, amount float64) error { - rows, err := r.sql.QueryContext(ctx, - `UPDATE accounts SET extra = jsonb_set( - COALESCE(extra, '{}'::jsonb), - '{quota_used}', - to_jsonb(COALESCE((extra->>'quota_used')::numeric, 0) + $1) - ), updated_at = NOW() - WHERE id = $2 AND deleted_at IS NULL - RETURNING - COALESCE((extra->>'quota_used')::numeric, 0), - COALESCE((extra->>'quota_limit')::numeric, 0)`, - amount, id) +// period 为 "daily"/"weekly" 时,使用 CTE 原子重置+递增;为空时保持累计制。 +func (r *accountRepository) IncrementQuotaUsed(ctx context.Context, id int64, amount float64, period string) error { + var rows *sql.Rows + var err error + + interval := periodToInterval(period) + if interval == "" { + // 累计制:保持原有逻辑 + rows, err = r.sql.QueryContext(ctx, + `UPDATE accounts SET extra = jsonb_set( + COALESCE(extra, '{}'::jsonb), + '{quota_used}', + to_jsonb(COALESCE((extra->>'quota_used')::numeric, 0) + $1) + ), updated_at = NOW() + WHERE id = $2 AND deleted_at IS NULL + RETURNING + COALESCE((extra->>'quota_used')::numeric, 0), + COALESCE((extra->>'quota_limit')::numeric, 0)`, + amount, id) + } else { + // 周期制:单条 UPDATE 内联过期判断,避免 CTE snapshot 在 READ COMMITTED 下的 lost update + rows, err = r.sql.QueryContext(ctx, + `UPDATE accounts SET extra = jsonb_set( + jsonb_set( + COALESCE(extra, '{}'::jsonb), + '{quota_used}', + to_jsonb( + CASE WHEN COALESCE((extra->>'quota_period_start')::timestamptz, '1970-01-01'::timestamptz) + + $3::interval <= NOW() + THEN $1 + ELSE COALESCE((extra->>'quota_used')::numeric, 0) + $1 END + ) + ), + '{quota_period_start}', + to_jsonb( + CASE WHEN COALESCE((extra->>'quota_period_start')::timestamptz, '1970-01-01'::timestamptz) + + $3::interval <= NOW() + THEN to_char(NOW() AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"') + ELSE COALESCE(extra->>'quota_period_start', to_char(NOW() AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"')) END + ) + ), updated_at = NOW() + WHERE id = $2 AND deleted_at IS NULL + RETURNING + COALESCE((extra->>'quota_used')::numeric, 0), + COALESCE((extra->>'quota_limit')::numeric, 0)`, + amount, id, interval) + } if err != nil { return err } @@ -1713,14 +1748,25 @@ func (r *accountRepository) IncrementQuotaUsed(ctx context.Context, id int64, am return nil } -// ResetQuotaUsed 重置账号的 extra.quota_used 为 0 +// periodToInterval 将周期类型转为 PostgreSQL interval 字符串 +func periodToInterval(period string) string { + switch period { + case "daily": + return "24 hours" + case "weekly": + return "168 hours" + default: + return "" + } +} + +// ResetQuotaUsed 重置账号的 extra.quota_used 为 0,同时清除 quota_period_start 以开始新周期 func (r *accountRepository) ResetQuotaUsed(ctx context.Context, id int64) error { _, err := r.sql.ExecContext(ctx, - `UPDATE accounts SET extra = jsonb_set( - COALESCE(extra, '{}'::jsonb), - '{quota_used}', - '0'::jsonb - ), updated_at = NOW() + `UPDATE accounts SET extra = ( + COALESCE(extra, '{}'::jsonb) + || '{"quota_used": 0}'::jsonb + ) - 'quota_period_start', updated_at = NOW() WHERE id = $1 AND deleted_at IS NULL`, id) if err != nil { diff --git a/backend/internal/repository/gateway_cache.go b/backend/internal/repository/gateway_cache.go index 58291b6652..8fa6d0c401 100644 --- a/backend/internal/repository/gateway_cache.go +++ b/backend/internal/repository/gateway_cache.go @@ -3,13 +3,17 @@ package repository import ( "context" "fmt" + "strconv" "time" "github.com/Wei-Shaw/sub2api/internal/service" "github.com/redis/go-redis/v9" ) -const stickySessionPrefix = "sticky_session:" +const ( + stickySessionPrefix = "sticky_session:" + clientAffinityPrefix = "client_affinity:" +) type gatewayCache struct { rdb *redis.Client @@ -51,3 +55,70 @@ func (c *gatewayCache) DeleteSessionAccountID(ctx context.Context, groupID int64 key := buildSessionKey(groupID, sessionHash) return c.rdb.Del(ctx, key).Err() } + +// buildAffinityKey 构建客户端亲和 key +// 格式: client_affinity:{groupID}:{clientID} +func buildAffinityKey(groupID int64, clientID string) string { + return fmt.Sprintf("%s%d:%s", clientAffinityPrefix, groupID, clientID) +} + +// getAffinityScript: 清理过期成员后返回亲和账号列表(按最近使用降序) +// KEYS[1] = client_affinity:{groupID}:{clientID} +// ARGV[1] = 过期阈值时间戳 (now - ttl) +var getAffinityScript = redis.NewScript(` +redis.call('ZREMRANGEBYSCORE', KEYS[1], '-inf', ARGV[1]) +return redis.call('ZREVRANGE', KEYS[1], 0, -1) +`) + +// updateAffinityScript: 清理过期成员、添加/更新亲和关系、刷新 TTL +// KEYS[1] = client_affinity:{groupID}:{clientID} +// ARGV[1] = 当前时间戳 +// ARGV[2] = TTL 秒数 +// ARGV[3] = accountID +// ARGV[4] = 过期阈值时间戳 (now - ttl) +var updateAffinityScript = redis.NewScript(` +redis.call('ZREMRANGEBYSCORE', KEYS[1], '-inf', ARGV[4]) +redis.call('ZADD', KEYS[1], ARGV[1], ARGV[3]) +redis.call('EXPIRE', KEYS[1], ARGV[2]) +return 1 +`) + +func (c *gatewayCache) GetClientAffinityAccounts(ctx context.Context, groupID int64, clientID string, ttl time.Duration) ([]int64, error) { + key := buildAffinityKey(groupID, clientID) + now := time.Now().Unix() + expireThreshold := now - int64(ttl.Seconds()) + + result, err := getAffinityScript.Run(ctx, c.rdb, []string{key}, expireThreshold).StringSlice() + if err != nil { + if err == redis.Nil { + return nil, nil + } + return nil, err + } + + accountIDs := make([]int64, 0, len(result)) + for _, s := range result { + id, err := strconv.ParseInt(s, 10, 64) + if err != nil { + continue + } + accountIDs = append(accountIDs, id) + } + return accountIDs, nil +} + +func (c *gatewayCache) UpdateClientAffinity(ctx context.Context, groupID int64, clientID string, accountID int64, ttl time.Duration) error { + key := buildAffinityKey(groupID, clientID) + now := time.Now().Unix() + ttlSeconds := int64(ttl.Seconds()) + expireThreshold := now - ttlSeconds + + return updateAffinityScript.Run(ctx, c.rdb, []string{key}, + now, ttlSeconds, accountID, expireThreshold, + ).Err() +} + +func (c *gatewayCache) RemoveClientAffinity(ctx context.Context, groupID int64, clientID string, accountID int64) error { + key := buildAffinityKey(groupID, clientID) + return c.rdb.ZRem(ctx, key, accountID).Err() +} diff --git a/backend/internal/server/api_contract_test.go b/backend/internal/server/api_contract_test.go index aafbbe21cb..b6915943fc 100644 --- a/backend/internal/server/api_contract_test.go +++ b/backend/internal/server/api_contract_test.go @@ -1096,7 +1096,7 @@ func (s *stubAccountRepo) UpdateExtra(ctx context.Context, id int64, updates map return errors.New("not implemented") } -func (s *stubAccountRepo) IncrementQuotaUsed(ctx context.Context, id int64, amount float64) error { +func (s *stubAccountRepo) IncrementQuotaUsed(ctx context.Context, id int64, amount float64, _ string) error { return errors.New("not implemented") } diff --git a/backend/internal/service/account.go b/backend/internal/service/account.go index 8eb3748c69..ce1b5c309d 100644 --- a/backend/internal/service/account.go +++ b/backend/internal/service/account.go @@ -1099,6 +1099,24 @@ func (a *Account) IsSessionIDMaskingEnabled() bool { return false } +// IsClientAffinityEnabled 检查是否启用客户端亲和调度 +// 仅适用于 Anthropic 账号(OAuth/SetupToken/APIKey) +// 启用后,新会话会优先调度到之前使用过的账号 +func (a *Account) IsClientAffinityEnabled() bool { + if a.Platform != PlatformAnthropic { + return false + } + if a.Extra == nil { + return false + } + if v, ok := a.Extra["client_affinity_enabled"]; ok { + if enabled, ok := v.(bool); ok { + return enabled + } + } + return false +} + // IsCacheTTLOverrideEnabled 检查是否启用缓存 TTL 强制替换 // 仅适用于 Anthropic OAuth/SetupToken 类型账号 // 启用后将所有 cache creation tokens 归入指定的 TTL 类型(5m 或 1h) @@ -1154,12 +1172,68 @@ func (a *Account) GetQuotaUsed() float64 { return 0 } +// GetQuotaPeriod 获取配额周期类型("daily"/"weekly",空字符串表示累计制) +func (a *Account) GetQuotaPeriod() string { + if a.Extra == nil { + return "" + } + if v, ok := a.Extra["quota_period"]; ok { + if s, ok := v.(string); ok { + return s + } + } + return "" +} + +// GetQuotaPeriodStart 获取当前周期的起始时间 +func (a *Account) GetQuotaPeriodStart() time.Time { + if a.Extra == nil { + return time.Time{} + } + if v, ok := a.Extra["quota_period_start"]; ok { + if s, ok := v.(string); ok { + if t, err := time.Parse(time.RFC3339Nano, s); err == nil { + return t + } + if t, err := time.Parse(time.RFC3339, s); err == nil { + return t + } + } + } + return time.Time{} +} + +// quotaPeriodDuration 返回周期长度 +func quotaPeriodDuration(period string) time.Duration { + switch period { + case "daily": + return 24 * time.Hour + case "weekly": + return 7 * 24 * time.Hour + default: + return 0 + } +} + +// isQuotaPeriodExpired 检查配额周期是否已过期 +func isQuotaPeriodExpired(period string, periodStart time.Time) bool { + dur := quotaPeriodDuration(period) + if dur == 0 || periodStart.IsZero() { + return false + } + return time.Since(periodStart) >= dur +} + // IsQuotaExceeded 检查 API Key 账号配额是否已超限 func (a *Account) IsQuotaExceeded() bool { limit := a.GetQuotaLimit() if limit <= 0 { return false } + // 周期性配额:如果当前周期已过期,视为未超限(下次 increment 时会自动重置) + if isQuotaPeriodExpired(a.GetQuotaPeriod(), a.GetQuotaPeriodStart()) { + return false + } return a.GetQuotaUsed() >= limit } diff --git a/backend/internal/service/account_load_factor_test.go b/backend/internal/service/account_load_factor_test.go index 1cd1b17c9c..a4d78a4bba 100644 --- a/backend/internal/service/account_load_factor_test.go +++ b/backend/internal/service/account_load_factor_test.go @@ -1,3 +1,5 @@ +//go:build unit + package service import ( @@ -6,6 +8,8 @@ import ( "github.com/stretchr/testify/require" ) +func intPtrHelper(v int) *int { return &v } + func TestEffectiveLoadFactor_NilAccount(t *testing.T) { var a *Account require.Equal(t, 1, a.EffectiveLoadFactor()) @@ -22,21 +26,21 @@ func TestEffectiveLoadFactor_NilLoadFactor_ZeroConcurrency(t *testing.T) { } func TestEffectiveLoadFactor_PositiveLoadFactor(t *testing.T) { - a := &Account{Concurrency: 5, LoadFactor: intPtr(20)} + a := &Account{Concurrency: 5, LoadFactor: intPtrHelper(20)} require.Equal(t, 20, a.EffectiveLoadFactor()) } func TestEffectiveLoadFactor_ZeroLoadFactor_FallbackToConcurrency(t *testing.T) { - a := &Account{Concurrency: 5, LoadFactor: intPtr(0)} + a := &Account{Concurrency: 5, LoadFactor: intPtrHelper(0)} require.Equal(t, 5, a.EffectiveLoadFactor()) } func TestEffectiveLoadFactor_NegativeLoadFactor_FallbackToConcurrency(t *testing.T) { - a := &Account{Concurrency: 3, LoadFactor: intPtr(-1)} + a := &Account{Concurrency: 3, LoadFactor: intPtrHelper(-1)} require.Equal(t, 3, a.EffectiveLoadFactor()) } func TestEffectiveLoadFactor_ZeroLoadFactor_ZeroConcurrency(t *testing.T) { - a := &Account{Concurrency: 0, LoadFactor: intPtr(0)} + a := &Account{Concurrency: 0, LoadFactor: intPtrHelper(0)} require.Equal(t, 1, a.EffectiveLoadFactor()) } diff --git a/backend/internal/service/account_service.go b/backend/internal/service/account_service.go index 26c0b1c29a..85c1398cf3 100644 --- a/backend/internal/service/account_service.go +++ b/backend/internal/service/account_service.go @@ -69,7 +69,7 @@ type AccountRepository interface { UpdateExtra(ctx context.Context, id int64, updates map[string]any) error BulkUpdate(ctx context.Context, ids []int64, updates AccountBulkUpdate) (int64, error) // IncrementQuotaUsed 原子递增 API Key 账号的配额用量 - IncrementQuotaUsed(ctx context.Context, id int64, amount float64) error + IncrementQuotaUsed(ctx context.Context, id int64, amount float64, period string) error // ResetQuotaUsed 重置 API Key 账号的配额用量为 0 ResetQuotaUsed(ctx context.Context, id int64) error } diff --git a/backend/internal/service/account_service_delete_test.go b/backend/internal/service/account_service_delete_test.go index c96b436f02..bfb57bdb44 100644 --- a/backend/internal/service/account_service_delete_test.go +++ b/backend/internal/service/account_service_delete_test.go @@ -199,7 +199,7 @@ func (s *accountRepoStub) BulkUpdate(ctx context.Context, ids []int64, updates A panic("unexpected BulkUpdate call") } -func (s *accountRepoStub) IncrementQuotaUsed(ctx context.Context, id int64, amount float64) error { +func (s *accountRepoStub) IncrementQuotaUsed(ctx context.Context, id int64, amount float64, _ string) error { return nil } diff --git a/backend/internal/service/admin_service.go b/backend/internal/service/admin_service.go index 11f850ee5f..4a60882afa 100644 --- a/backend/internal/service/admin_service.go +++ b/backend/internal/service/admin_service.go @@ -1437,6 +1437,9 @@ func (s *adminServiceImpl) CreateAccount(ctx context.Context, input *CreateAccou account.RateMultiplier = input.RateMultiplier } if input.LoadFactor != nil && *input.LoadFactor > 0 { + if *input.LoadFactor > 10000 { + return nil, errors.New("load_factor must be <= 10000") + } account.LoadFactor = input.LoadFactor } if err := s.accountRepo.Create(ctx, account); err != nil { @@ -1516,6 +1519,8 @@ func (s *adminServiceImpl) UpdateAccount(ctx context.Context, id int64, input *U if input.LoadFactor != nil { if *input.LoadFactor <= 0 { account.LoadFactor = nil // 0 或负数表示清除 + } else if *input.LoadFactor > 10000 { + return nil, errors.New("load_factor must be <= 10000") } else { account.LoadFactor = input.LoadFactor } @@ -1654,7 +1659,13 @@ func (s *adminServiceImpl) BulkUpdateAccounts(ctx context.Context, input *BulkUp repoUpdates.RateMultiplier = input.RateMultiplier } if input.LoadFactor != nil { - repoUpdates.LoadFactor = input.LoadFactor + if *input.LoadFactor <= 0 { + repoUpdates.LoadFactor = nil // 0 或负数表示清除 + } else if *input.LoadFactor > 10000 { + return nil, errors.New("load_factor must be <= 10000") + } else { + repoUpdates.LoadFactor = input.LoadFactor + } } if input.Status != "" { repoUpdates.Status = &input.Status diff --git a/backend/internal/service/antigravity_smart_retry_test.go b/backend/internal/service/antigravity_smart_retry_test.go index 432c80e59f..7f836bf47c 100644 --- a/backend/internal/service/antigravity_smart_retry_test.go +++ b/backend/internal/service/antigravity_smart_retry_test.go @@ -9,6 +9,7 @@ import ( "net/http" "strings" "testing" + "time" "github.com/stretchr/testify/require" ) @@ -29,6 +30,15 @@ func (c *stubSmartRetryCache) DeleteSessionAccountID(_ context.Context, groupID c.deleteCalls = append(c.deleteCalls, deleteSessionCall{groupID: groupID, sessionHash: sessionHash}) return nil } +func (c *stubSmartRetryCache) GetClientAffinityAccounts(_ context.Context, _ int64, _ string, _ time.Duration) ([]int64, error) { + return nil, nil +} +func (c *stubSmartRetryCache) UpdateClientAffinity(_ context.Context, _ int64, _ string, _ int64, _ time.Duration) error { + return nil +} +func (c *stubSmartRetryCache) RemoveClientAffinity(_ context.Context, _ int64, _ string, _ int64) error { + return nil +} // mockSmartRetryUpstream 用于 handleSmartRetry 测试的 mock upstream type mockSmartRetryUpstream struct { diff --git a/backend/internal/service/billing_service.go b/backend/internal/service/billing_service.go index 5d67c80862..d058c25af7 100644 --- a/backend/internal/service/billing_service.go +++ b/backend/internal/service/billing_service.go @@ -43,15 +43,24 @@ type BillingCache interface { // ModelPricing 模型价格配置(per-token价格,与LiteLLM格式一致) type ModelPricing struct { - InputPricePerToken float64 // 每token输入价格 (USD) - OutputPricePerToken float64 // 每token输出价格 (USD) - CacheCreationPricePerToken float64 // 缓存创建每token价格 (USD) - CacheReadPricePerToken float64 // 缓存读取每token价格 (USD) - CacheCreation5mPrice float64 // 5分钟缓存创建每token价格 (USD) - CacheCreation1hPrice float64 // 1小时缓存创建每token价格 (USD) - SupportsCacheBreakdown bool // 是否支持详细的缓存分类 + InputPricePerToken float64 // 每token输入价格 (USD) + OutputPricePerToken float64 // 每token输出价格 (USD) + CacheCreationPricePerToken float64 // 缓存创建每token价格 (USD) + CacheReadPricePerToken float64 // 缓存读取每token价格 (USD) + CacheCreation5mPrice float64 // 5分钟缓存创建每token价格 (USD) + CacheCreation1hPrice float64 // 1小时缓存创建每token价格 (USD) + SupportsCacheBreakdown bool // 是否支持详细的缓存分类 + LongContextInputThreshold int // 超过阈值后按整次会话提升输入价格 + LongContextInputMultiplier float64 // 长上下文整次会话输入倍率 + LongContextOutputMultiplier float64 // 长上下文整次会话输出倍率 } +const ( + openAIGPT54LongContextInputThreshold = 272000 + openAIGPT54LongContextInputMultiplier = 2.0 + openAIGPT54LongContextOutputMultiplier = 1.5 +) + // UsageTokens 使用的token数量 type UsageTokens struct { InputTokens int @@ -161,6 +170,35 @@ func (s *BillingService) initFallbackPricing() { CacheReadPricePerToken: 0.2e-6, // $0.20 per MTok SupportsCacheBreakdown: false, } + + // OpenAI GPT-5.1(本地兜底,防止动态定价不可用时拒绝计费) + s.fallbackPrices["gpt-5.1"] = &ModelPricing{ + InputPricePerToken: 1.25e-6, // $1.25 per MTok + OutputPricePerToken: 10e-6, // $10 per MTok + CacheCreationPricePerToken: 1.25e-6, // $1.25 per MTok + CacheReadPricePerToken: 0.125e-6, + SupportsCacheBreakdown: false, + } + // OpenAI GPT-5.4(业务指定价格) + s.fallbackPrices["gpt-5.4"] = &ModelPricing{ + InputPricePerToken: 2.5e-6, // $2.5 per MTok + OutputPricePerToken: 15e-6, // $15 per MTok + CacheCreationPricePerToken: 2.5e-6, // $2.5 per MTok + CacheReadPricePerToken: 0.25e-6, // $0.25 per MTok + SupportsCacheBreakdown: false, + LongContextInputThreshold: openAIGPT54LongContextInputThreshold, + LongContextInputMultiplier: openAIGPT54LongContextInputMultiplier, + LongContextOutputMultiplier: openAIGPT54LongContextOutputMultiplier, + } + // Codex 族兜底统一按 GPT-5.1 Codex 价格计费 + s.fallbackPrices["gpt-5.1-codex"] = &ModelPricing{ + InputPricePerToken: 1.5e-6, // $1.5 per MTok + OutputPricePerToken: 12e-6, // $12 per MTok + CacheCreationPricePerToken: 1.5e-6, // $1.5 per MTok + CacheReadPricePerToken: 0.15e-6, + SupportsCacheBreakdown: false, + } + s.fallbackPrices["gpt-5.3-codex"] = s.fallbackPrices["gpt-5.1-codex"] } // getFallbackPricing 根据模型系列获取回退价格 @@ -189,12 +227,30 @@ func (s *BillingService) getFallbackPricing(model string) *ModelPricing { } return s.fallbackPrices["claude-3-haiku"] } + // Claude 未知型号统一回退到 Sonnet,避免计费中断。 + if strings.Contains(modelLower, "claude") { + return s.fallbackPrices["claude-sonnet-4"] + } if strings.Contains(modelLower, "gemini-3.1-pro") || strings.Contains(modelLower, "gemini-3-1-pro") { return s.fallbackPrices["gemini-3.1-pro"] } - // 默认使用Sonnet价格 - return s.fallbackPrices["claude-sonnet-4"] + // OpenAI 仅匹配已知 GPT-5/Codex 族,避免未知 OpenAI 型号误计价。 + if strings.Contains(modelLower, "gpt-5") || strings.Contains(modelLower, "codex") { + normalized := normalizeCodexModel(modelLower) + switch normalized { + case "gpt-5.4": + return s.fallbackPrices["gpt-5.4"] + case "gpt-5.3-codex": + return s.fallbackPrices["gpt-5.3-codex"] + case "gpt-5.1-codex", "gpt-5.1-codex-max", "gpt-5.1-codex-mini", "codex-mini-latest": + return s.fallbackPrices["gpt-5.1-codex"] + case "gpt-5.1": + return s.fallbackPrices["gpt-5.1"] + } + } + + return nil } // GetModelPricing 获取模型价格配置 @@ -212,15 +268,18 @@ func (s *BillingService) GetModelPricing(model string) (*ModelPricing, error) { price5m := litellmPricing.CacheCreationInputTokenCost price1h := litellmPricing.CacheCreationInputTokenCostAbove1hr enableBreakdown := price1h > 0 && price1h > price5m - return &ModelPricing{ - InputPricePerToken: litellmPricing.InputCostPerToken, - OutputPricePerToken: litellmPricing.OutputCostPerToken, - CacheCreationPricePerToken: litellmPricing.CacheCreationInputTokenCost, - CacheReadPricePerToken: litellmPricing.CacheReadInputTokenCost, - CacheCreation5mPrice: price5m, - CacheCreation1hPrice: price1h, - SupportsCacheBreakdown: enableBreakdown, - }, nil + return s.applyModelSpecificPricingPolicy(model, &ModelPricing{ + InputPricePerToken: litellmPricing.InputCostPerToken, + OutputPricePerToken: litellmPricing.OutputCostPerToken, + CacheCreationPricePerToken: litellmPricing.CacheCreationInputTokenCost, + CacheReadPricePerToken: litellmPricing.CacheReadInputTokenCost, + CacheCreation5mPrice: price5m, + CacheCreation1hPrice: price1h, + SupportsCacheBreakdown: enableBreakdown, + LongContextInputThreshold: litellmPricing.LongContextInputTokenThreshold, + LongContextInputMultiplier: litellmPricing.LongContextInputCostMultiplier, + LongContextOutputMultiplier: litellmPricing.LongContextOutputCostMultiplier, + }), nil } } @@ -228,7 +287,7 @@ func (s *BillingService) GetModelPricing(model string) (*ModelPricing, error) { fallback := s.getFallbackPricing(model) if fallback != nil { log.Printf("[Billing] Using fallback pricing for model: %s", model) - return fallback, nil + return s.applyModelSpecificPricingPolicy(model, fallback), nil } return nil, fmt.Errorf("pricing not found for model: %s", model) @@ -242,12 +301,18 @@ func (s *BillingService) CalculateCost(model string, tokens UsageTokens, rateMul } breakdown := &CostBreakdown{} + inputPricePerToken := pricing.InputPricePerToken + outputPricePerToken := pricing.OutputPricePerToken + if s.shouldApplySessionLongContextPricing(tokens, pricing) { + inputPricePerToken *= pricing.LongContextInputMultiplier + outputPricePerToken *= pricing.LongContextOutputMultiplier + } // 计算输入token费用(使用per-token价格) - breakdown.InputCost = float64(tokens.InputTokens) * pricing.InputPricePerToken + breakdown.InputCost = float64(tokens.InputTokens) * inputPricePerToken // 计算输出token费用 - breakdown.OutputCost = float64(tokens.OutputTokens) * pricing.OutputPricePerToken + breakdown.OutputCost = float64(tokens.OutputTokens) * outputPricePerToken // 计算缓存费用 if pricing.SupportsCacheBreakdown && (pricing.CacheCreation5mPrice > 0 || pricing.CacheCreation1hPrice > 0) { @@ -279,6 +344,45 @@ func (s *BillingService) CalculateCost(model string, tokens UsageTokens, rateMul return breakdown, nil } +func (s *BillingService) applyModelSpecificPricingPolicy(model string, pricing *ModelPricing) *ModelPricing { + if pricing == nil { + return nil + } + if !isOpenAIGPT54Model(model) { + return pricing + } + if pricing.LongContextInputThreshold > 0 && pricing.LongContextInputMultiplier > 0 && pricing.LongContextOutputMultiplier > 0 { + return pricing + } + cloned := *pricing + if cloned.LongContextInputThreshold <= 0 { + cloned.LongContextInputThreshold = openAIGPT54LongContextInputThreshold + } + if cloned.LongContextInputMultiplier <= 0 { + cloned.LongContextInputMultiplier = openAIGPT54LongContextInputMultiplier + } + if cloned.LongContextOutputMultiplier <= 0 { + cloned.LongContextOutputMultiplier = openAIGPT54LongContextOutputMultiplier + } + return &cloned +} + +func (s *BillingService) shouldApplySessionLongContextPricing(tokens UsageTokens, pricing *ModelPricing) bool { + if pricing == nil || pricing.LongContextInputThreshold <= 0 { + return false + } + if pricing.LongContextInputMultiplier <= 1 && pricing.LongContextOutputMultiplier <= 1 { + return false + } + totalInputTokens := tokens.InputTokens + tokens.CacheReadTokens + return totalInputTokens > pricing.LongContextInputThreshold +} + +func isOpenAIGPT54Model(model string) bool { + normalized := normalizeCodexModel(strings.TrimSpace(strings.ToLower(model))) + return normalized == "gpt-5.4" +} + // CalculateCostWithConfig 使用配置中的默认倍率计算费用 func (s *BillingService) CalculateCostWithConfig(model string, tokens UsageTokens) (*CostBreakdown, error) { multiplier := s.cfg.Default.RateMultiplier diff --git a/backend/internal/service/billing_service_test.go b/backend/internal/service/billing_service_test.go index 5eb278f647..0ba52e5640 100644 --- a/backend/internal/service/billing_service_test.go +++ b/backend/internal/service/billing_service_test.go @@ -133,7 +133,7 @@ func TestGetModelPricing_CaseInsensitive(t *testing.T) { require.Equal(t, p1.InputPricePerToken, p2.InputPricePerToken) } -func TestGetModelPricing_UnknownModelFallsBackToSonnet(t *testing.T) { +func TestGetModelPricing_UnknownClaudeModelFallsBackToSonnet(t *testing.T) { svc := newTestBillingService() // 不包含 opus/sonnet/haiku 关键词的 Claude 模型会走默认 Sonnet 价格 @@ -142,6 +142,93 @@ func TestGetModelPricing_UnknownModelFallsBackToSonnet(t *testing.T) { require.InDelta(t, 3e-6, pricing.InputPricePerToken, 1e-12) } +func TestGetModelPricing_UnknownOpenAIModelReturnsError(t *testing.T) { + svc := newTestBillingService() + + pricing, err := svc.GetModelPricing("gpt-unknown-model") + require.Error(t, err) + require.Nil(t, pricing) + require.Contains(t, err.Error(), "pricing not found") +} + +func TestGetModelPricing_OpenAIGPT51Fallback(t *testing.T) { + svc := newTestBillingService() + + pricing, err := svc.GetModelPricing("gpt-5.1") + require.NoError(t, err) + require.NotNil(t, pricing) + require.InDelta(t, 1.25e-6, pricing.InputPricePerToken, 1e-12) +} + +func TestGetModelPricing_OpenAIGPT54Fallback(t *testing.T) { + svc := newTestBillingService() + + pricing, err := svc.GetModelPricing("gpt-5.4") + require.NoError(t, err) + require.NotNil(t, pricing) + require.InDelta(t, 2.5e-6, pricing.InputPricePerToken, 1e-12) + require.InDelta(t, 15e-6, pricing.OutputPricePerToken, 1e-12) + require.InDelta(t, 0.25e-6, pricing.CacheReadPricePerToken, 1e-12) + require.Equal(t, 272000, pricing.LongContextInputThreshold) + require.InDelta(t, 2.0, pricing.LongContextInputMultiplier, 1e-12) + require.InDelta(t, 1.5, pricing.LongContextOutputMultiplier, 1e-12) +} + +func TestCalculateCost_OpenAIGPT54LongContextAppliesWholeSessionMultipliers(t *testing.T) { + svc := newTestBillingService() + + tokens := UsageTokens{ + InputTokens: 300000, + OutputTokens: 4000, + } + + cost, err := svc.CalculateCost("gpt-5.4-2026-03-05", tokens, 1.0) + require.NoError(t, err) + + expectedInput := float64(tokens.InputTokens) * 2.5e-6 * 2.0 + expectedOutput := float64(tokens.OutputTokens) * 15e-6 * 1.5 + require.InDelta(t, expectedInput, cost.InputCost, 1e-10) + require.InDelta(t, expectedOutput, cost.OutputCost, 1e-10) + require.InDelta(t, expectedInput+expectedOutput, cost.TotalCost, 1e-10) + require.InDelta(t, expectedInput+expectedOutput, cost.ActualCost, 1e-10) +} + +func TestGetFallbackPricing_FamilyMatching(t *testing.T) { + svc := newTestBillingService() + + tests := []struct { + name string + model string + expectedInput float64 + expectNilPricing bool + }{ + {name: "empty model", model: " ", expectNilPricing: true}, + {name: "claude opus 4.6", model: "claude-opus-4.6-20260201", expectedInput: 5e-6}, + {name: "claude opus 4.5 alt separator", model: "claude-opus-4-5-20260101", expectedInput: 5e-6}, + {name: "claude generic model fallback sonnet", model: "claude-foo-bar", expectedInput: 3e-6}, + {name: "gemini explicit fallback", model: "gemini-3-1-pro", expectedInput: 2e-6}, + {name: "gemini unknown no fallback", model: "gemini-2.0-pro", expectNilPricing: true}, + {name: "openai gpt5.1", model: "gpt-5.1", expectedInput: 1.25e-6}, + {name: "openai gpt5.4", model: "gpt-5.4", expectedInput: 2.5e-6}, + {name: "openai gpt5.3 codex", model: "gpt-5.3-codex", expectedInput: 1.5e-6}, + {name: "openai gpt5.1 codex max alias", model: "gpt-5.1-codex-max", expectedInput: 1.5e-6}, + {name: "openai codex mini latest alias", model: "codex-mini-latest", expectedInput: 1.5e-6}, + {name: "openai unknown no fallback", model: "gpt-unknown-model", expectNilPricing: true}, + {name: "non supported family", model: "qwen-max", expectNilPricing: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + pricing := svc.getFallbackPricing(tt.model) + if tt.expectNilPricing { + require.Nil(t, pricing) + return + } + require.NotNil(t, pricing) + require.InDelta(t, tt.expectedInput, pricing.InputPricePerToken, 1e-12) + }) + } +} func TestCalculateCostWithLongContext_BelowThreshold(t *testing.T) { svc := newTestBillingService() diff --git a/backend/internal/service/error_policy_test.go b/backend/internal/service/error_policy_test.go index 59375cf50a..62aba88fdd 100644 --- a/backend/internal/service/error_policy_test.go +++ b/backend/internal/service/error_policy_test.go @@ -114,7 +114,7 @@ func TestCheckErrorPolicy(t *testing.T) { account: &Account{ ID: 15, Type: AccountTypeOAuth, - Platform: PlatformAntigravity, + Platform: PlatformGemini, // 非 Antigravity 平台才有 401 升级逻辑 TempUnschedulableReason: `{"status_code":401,"until_unix":1735689600}`, Credentials: map[string]any{ "temp_unschedulable_enabled": true, @@ -131,6 +131,28 @@ func TestCheckErrorPolicy(t *testing.T) { body: []byte(`unauthorized`), expected: ErrorPolicyNone, }, + { + name: "temp_unschedulable_401_antigravity_no_escalation", + account: &Account{ + ID: 16, + Type: AccountTypeOAuth, + Platform: PlatformAntigravity, // Antigravity 跳过 401 升级,由 rules 正常处理 + TempUnschedulableReason: `{"status_code":401,"until_unix":1735689600}`, + Credentials: map[string]any{ + "temp_unschedulable_enabled": true, + "temp_unschedulable_rules": []any{ + map[string]any{ + "error_code": float64(401), + "keywords": []any{"unauthorized"}, + "duration_minutes": float64(10), + }, + }, + }, + }, + statusCode: 401, + body: []byte(`unauthorized`), + expected: ErrorPolicyTempUnscheduled, // Antigravity 不升级,继续走规则匹配 + }, { name: "temp_unschedulable_body_miss_returns_none", account: &Account{ diff --git a/backend/internal/service/gateway_hotpath_optimization_test.go b/backend/internal/service/gateway_hotpath_optimization_test.go index 161c4ba4b1..137490e8db 100644 --- a/backend/internal/service/gateway_hotpath_optimization_test.go +++ b/backend/internal/service/gateway_hotpath_optimization_test.go @@ -143,6 +143,15 @@ func (s *stickyGatewayCacheHotpathStub) RefreshSessionTTL(ctx context.Context, g func (s *stickyGatewayCacheHotpathStub) DeleteSessionAccountID(ctx context.Context, groupID int64, sessionHash string) error { return nil } +func (s *stickyGatewayCacheHotpathStub) GetClientAffinityAccounts(_ context.Context, _ int64, _ string, _ time.Duration) ([]int64, error) { + return nil, nil +} +func (s *stickyGatewayCacheHotpathStub) UpdateClientAffinity(_ context.Context, _ int64, _ string, _ int64, _ time.Duration) error { + return nil +} +func (s *stickyGatewayCacheHotpathStub) RemoveClientAffinity(_ context.Context, _ int64, _ string, _ int64) error { + return nil +} func (s *modelsListAccountRepoStub) ListSchedulableByGroupID(ctx context.Context, groupID int64) ([]Account, error) { s.listByGroupCalls.Add(1) diff --git a/backend/internal/service/gateway_multiplatform_test.go b/backend/internal/service/gateway_multiplatform_test.go index 320ceaa708..6f800f3387 100644 --- a/backend/internal/service/gateway_multiplatform_test.go +++ b/backend/internal/service/gateway_multiplatform_test.go @@ -187,7 +187,7 @@ func (m *mockAccountRepoForPlatform) BulkUpdate(ctx context.Context, ids []int64 return 0, nil } -func (m *mockAccountRepoForPlatform) IncrementQuotaUsed(ctx context.Context, id int64, amount float64) error { +func (m *mockAccountRepoForPlatform) IncrementQuotaUsed(ctx context.Context, id int64, amount float64, _ string) error { return nil } @@ -235,6 +235,16 @@ func (m *mockGatewayCacheForPlatform) DeleteSessionAccountID(ctx context.Context return nil } +func (m *mockGatewayCacheForPlatform) GetClientAffinityAccounts(_ context.Context, _ int64, _ string, _ time.Duration) ([]int64, error) { + return nil, nil +} +func (m *mockGatewayCacheForPlatform) UpdateClientAffinity(_ context.Context, _ int64, _ string, _ int64, _ time.Duration) error { + return nil +} +func (m *mockGatewayCacheForPlatform) RemoveClientAffinity(_ context.Context, _ int64, _ string, _ int64) error { + return nil +} + type mockGroupRepoForGateway struct { groups map[int64]*Group getByIDCalls int diff --git a/backend/internal/service/gateway_service.go b/backend/internal/service/gateway_service.go index ccccdf4d10..73b726c843 100644 --- a/backend/internal/service/gateway_service.go +++ b/backend/internal/service/gateway_service.go @@ -304,6 +304,10 @@ var ( sessionIDRegex = regexp.MustCompile(`session_([a-f0-9-]{36})`) claudeCliUserAgentRe = regexp.MustCompile(`^claude-cli/\d+\.\d+\.\d+`) + // clientIDFromMetadataRegex 从 metadata.user_id 中提取客户端 ID(64位 hex) + // 格式: user_{64位hex}_account_... + clientIDFromMetadataRegex = regexp.MustCompile(`^user_([a-f0-9]{64})_account_`) + // claudeCodePromptPrefixes 用于检测 Claude Code 系统提示词的前缀列表 // 支持多种变体:标准版、Agent SDK 版、Explore Agent 版、Compact 版等 // 注意:前缀之间不应存在包含关系,否则会导致冗余匹配 @@ -364,6 +368,13 @@ type GatewayCache interface { // DeleteSessionAccountID 删除粘性会话绑定,用于账号不可用时主动清理 // Delete sticky session binding, used to proactively clean up when account becomes unavailable DeleteSessionAccountID(ctx context.Context, groupID int64, sessionHash string) error + + // GetClientAffinityAccounts 获取客户端亲和账号列表(按最近使用降序),同时清理过期成员 + GetClientAffinityAccounts(ctx context.Context, groupID int64, clientID string, ttl time.Duration) ([]int64, error) + // UpdateClientAffinity 添加/更新客户端亲和关系(更新 score 为当前时间戳,刷新 key TTL) + UpdateClientAffinity(ctx context.Context, groupID int64, clientID string, accountID int64, ttl time.Duration) error + // RemoveClientAffinity 删除单个客户端亲和关系 + RemoveClientAffinity(ctx context.Context, groupID int64, clientID string, accountID int64) error } // derefGroupID safely dereferences *int64 to int64, returning 0 if nil @@ -434,6 +445,20 @@ func shouldClearStickySession(account *Account, requestedModel string) bool { return false } +// extractClientIDFromMetadata 从 metadata.user_id 中提取客户端 ID(64位 hex)。 +// 格式: user_{64位hex}_account_..._session_... +// 返回空字符串表示无法提取(非 Claude Code/Console 客户端)。 +func extractClientIDFromMetadata(metadataUserID string) string { + if metadataUserID == "" { + return "" + } + matches := clientIDFromMetadataRegex.FindStringSubmatch(metadataUserID) + if matches == nil { + return "" + } + return matches[1] +} + type AccountWaitPlan struct { AccountID int64 MaxConcurrency int @@ -1046,7 +1071,7 @@ func (s *GatewayService) SelectAccountForModelWithExclusions(ctx context.Context } // SelectAccountWithLoadAwareness selects account with load-awareness and wait plan. -// metadataUserID: 已废弃参数,会话限制现在统一使用 sessionHash +// metadataUserID: 用于客户端亲和调度,从中提取客户端 ID func (s *GatewayService) SelectAccountWithLoadAwareness(ctx context.Context, groupID *int64, sessionHash string, requestedModel string, excludedIDs map[int64]struct{}, metadataUserID string) (*AccountSelectionResult, error) { // 调试日志:记录调度入口参数 excludedIDsList := make([]int64, 0, len(excludedIDs)) @@ -1077,6 +1102,9 @@ func (s *GatewayService) SelectAccountWithLoadAwareness(ctx context.Context, gro } } + // 提取客户端 ID(用于客户端亲和调度) + affinityClientID := extractClientIDFromMetadata(metadataUserID) + if s.debugModelRoutingEnabled() && requestedModel != "" { groupPlatform := "" if group != nil { @@ -1374,6 +1402,9 @@ func (s *GatewayService) SelectAccountWithLoadAwareness(ctx context.Context, gro if sessionHash != "" && s.cache != nil { _ = s.cache.SetSessionAccountID(ctx, derefGroupID(groupID), sessionHash, item.account.ID, stickySessionTTL) } + if affinityClientID != "" && s.cache != nil && item.account.IsClientAffinityEnabled() { + _ = s.cache.UpdateClientAffinity(ctx, derefGroupID(groupID), affinityClientID, item.account.ID, stickySessionTTL) + } if s.debugModelRoutingEnabled() { logger.LegacyPrintf("service.gateway", "[ModelRoutingDebug] routed select: group_id=%v model=%s session=%s account=%d", derefGroupID(groupID), requestedModel, shortSessionHash(sessionHash), item.account.ID) } @@ -1470,6 +1501,66 @@ func (s *GatewayService) SelectAccountWithLoadAwareness(ctx context.Context, gro } } + // ============ Layer 1.6: 客户端亲和(仅在粘性会话未命中时生效) ============ + if affinityClientID != "" && s.cache != nil && stickyAccountID <= 0 { + affinityAccountIDs, err := s.cache.GetClientAffinityAccounts(ctx, derefGroupID(groupID), affinityClientID, stickySessionTTL) + if err == nil && len(affinityAccountIDs) > 0 { + for _, affinityAccID := range affinityAccountIDs { + if isExcluded(affinityAccID) { + continue + } + account, ok := accountByID[affinityAccID] + if !ok || !s.isAccountSchedulableForSelection(account) { + continue + } + if !account.IsClientAffinityEnabled() { + continue + } + if !s.isAccountAllowedForPlatform(account, platform, useMixed) { + continue + } + if requestedModel != "" && !s.isModelSupportedByAccountWithContext(ctx, account, requestedModel) { + continue + } + if !s.isAccountSchedulableForModelSelection(ctx, account, requestedModel) { + continue + } + if !s.isAccountSchedulableForQuota(account) { + continue + } + if !s.isAccountSchedulableForWindowCost(ctx, account, false) { + continue + } + if !s.isAccountSchedulableForRPM(ctx, account, false) { + continue + } + + result, err := s.tryAcquireAccountSlot(ctx, affinityAccID, account.Concurrency) + if err == nil && result.Acquired { + if !s.checkAndRegisterSession(ctx, account, sessionHash) { + result.ReleaseFunc() + continue + } + // 亲和命中:更新亲和 score + 绑定粘性会话 + _ = s.cache.UpdateClientAffinity(ctx, derefGroupID(groupID), affinityClientID, affinityAccID, stickySessionTTL) + if sessionHash != "" { + _ = s.cache.SetSessionAccountID(ctx, derefGroupID(groupID), sessionHash, affinityAccID, stickySessionTTL) + } + slog.Debug("client_affinity_hit", + "group_id", derefGroupID(groupID), + "client_id", affinityClientID[:8]+"...", + "account_id", affinityAccID) + return &AccountSelectionResult{ + Account: account, + Acquired: true, + ReleaseFunc: result.ReleaseFunc, + }, nil + } + } + // 所有亲和账号不可用,继续到 Layer 2 + } + } + // ============ Layer 2: 负载感知选择 ============ candidates := make([]*Account, 0, len(accounts)) for i := range accounts { @@ -1522,6 +1613,9 @@ func (s *GatewayService) SelectAccountWithLoadAwareness(ctx context.Context, gro loadMap, err := s.concurrencyService.GetAccountsLoadBatch(ctx, accountLoads) if err != nil { if result, ok := s.tryAcquireByLegacyOrder(ctx, candidates, groupID, sessionHash, preferOAuth); ok { + if affinityClientID != "" && s.cache != nil && result.Account != nil && result.Account.IsClientAffinityEnabled() { + _ = s.cache.UpdateClientAffinity(ctx, derefGroupID(groupID), affinityClientID, result.Account.ID, stickySessionTTL) + } return result, nil } } else { @@ -1560,6 +1654,10 @@ func (s *GatewayService) SelectAccountWithLoadAwareness(ctx context.Context, gro if sessionHash != "" && s.cache != nil { _ = s.cache.SetSessionAccountID(ctx, derefGroupID(groupID), sessionHash, selected.account.ID, stickySessionTTL) } + // 更新客户端亲和关系 + if affinityClientID != "" && s.cache != nil && selected.account.IsClientAffinityEnabled() { + _ = s.cache.UpdateClientAffinity(ctx, derefGroupID(groupID), affinityClientID, selected.account.ID, stickySessionTTL) + } return &AccountSelectionResult{ Account: selected.account, Acquired: true, @@ -6487,7 +6585,8 @@ func postUsageBilling(ctx context.Context, p *postUsageBillingParams, deps *bill // 4. 账号配额用量(账号口径:TotalCost × 账号计费倍率) if cost.TotalCost > 0 && p.Account.Type == AccountTypeAPIKey && p.Account.GetQuotaLimit() > 0 { accountCost := cost.TotalCost * p.AccountRateMultiplier - if err := deps.accountRepo.IncrementQuotaUsed(ctx, p.Account.ID, accountCost); err != nil { + period := p.Account.GetQuotaPeriod() + if err := deps.accountRepo.IncrementQuotaUsed(ctx, p.Account.ID, accountCost, period); err != nil { slog.Error("increment account quota used failed", "account_id", p.Account.ID, "cost", accountCost, "error", err) } } diff --git a/backend/internal/service/gemini_multiplatform_test.go b/backend/internal/service/gemini_multiplatform_test.go index b0b804eb00..5a9dc69a14 100644 --- a/backend/internal/service/gemini_multiplatform_test.go +++ b/backend/internal/service/gemini_multiplatform_test.go @@ -176,7 +176,7 @@ func (m *mockAccountRepoForGemini) BulkUpdate(ctx context.Context, ids []int64, return 0, nil } -func (m *mockAccountRepoForGemini) IncrementQuotaUsed(ctx context.Context, id int64, amount float64) error { +func (m *mockAccountRepoForGemini) IncrementQuotaUsed(ctx context.Context, id int64, amount float64, _ string) error { return nil } @@ -288,6 +288,16 @@ func (m *mockGatewayCacheForGemini) DeleteSessionAccountID(ctx context.Context, return nil } +func (m *mockGatewayCacheForGemini) GetClientAffinityAccounts(_ context.Context, _ int64, _ string, _ time.Duration) ([]int64, error) { + return nil, nil +} +func (m *mockGatewayCacheForGemini) UpdateClientAffinity(_ context.Context, _ int64, _ string, _ int64, _ time.Duration) error { + return nil +} +func (m *mockGatewayCacheForGemini) RemoveClientAffinity(_ context.Context, _ int64, _ string, _ int64) error { + return nil +} + // TestGeminiMessagesCompatService_SelectAccountForModelWithExclusions_GeminiPlatform 测试 Gemini 单平台选择 func TestGeminiMessagesCompatService_SelectAccountForModelWithExclusions_GeminiPlatform(t *testing.T) { ctx := context.Background() diff --git a/backend/internal/service/openai_account_scheduler.go b/backend/internal/service/openai_account_scheduler.go index 9cc6aae163..9e2b81035a 100644 --- a/backend/internal/service/openai_account_scheduler.go +++ b/backend/internal/service/openai_account_scheduler.go @@ -323,7 +323,7 @@ func (s *defaultOpenAIAccountScheduler) selectBySessionHash( _ = s.service.deleteStickySessionAccountID(ctx, req.GroupID, sessionHash) return nil, nil } - if req.RequestedModel != "" && !account.IsModelSupported(req.RequestedModel) { + if req.RequestedModel != "" && !account.IsOpenAIPassthroughEnabled() && !account.IsModelSupported(req.RequestedModel) { return nil, nil } if !s.isAccountTransportCompatible(account, req.RequiredTransport) { @@ -342,6 +342,7 @@ func (s *defaultOpenAIAccountScheduler) selectBySessionHash( } cfg := s.service.schedulingConfig() + // WaitPlan.MaxConcurrency 使用 Concurrency(非 EffectiveLoadFactor),因为 WaitPlan 控制的是 Redis 实际并发槽位等待。 if s.service.concurrencyService != nil { return &AccountSelectionResult{ Account: account, @@ -581,7 +582,7 @@ func (s *defaultOpenAIAccountScheduler) selectByLoadBalance( if !account.IsSchedulable() || !account.IsOpenAI() { continue } - if req.RequestedModel != "" && !account.IsModelSupported(req.RequestedModel) { + if req.RequestedModel != "" && !account.IsOpenAIPassthroughEnabled() && !account.IsModelSupported(req.RequestedModel) { continue } if !s.isAccountTransportCompatible(account, req.RequiredTransport) { @@ -703,6 +704,7 @@ func (s *defaultOpenAIAccountScheduler) selectByLoadBalance( } cfg := s.service.schedulingConfig() + // WaitPlan.MaxConcurrency 使用 Concurrency(非 EffectiveLoadFactor),因为 WaitPlan 控制的是 Redis 实际并发槽位等待。 candidate := selectionOrder[0] return &AccountSelectionResult{ Account: candidate.account, diff --git a/backend/internal/service/openai_codex_transform.go b/backend/internal/service/openai_codex_transform.go index 16befb8247..9bc48cf6ba 100644 --- a/backend/internal/service/openai_codex_transform.go +++ b/backend/internal/service/openai_codex_transform.go @@ -9,6 +9,13 @@ import ( var codexCLIInstructions string var codexModelMap = map[string]string{ + "gpt-5.4": "gpt-5.4", + "gpt-5.4-none": "gpt-5.4", + "gpt-5.4-low": "gpt-5.4", + "gpt-5.4-medium": "gpt-5.4", + "gpt-5.4-high": "gpt-5.4", + "gpt-5.4-xhigh": "gpt-5.4", + "gpt-5.4-chat-latest": "gpt-5.4", "gpt-5.3": "gpt-5.3-codex", "gpt-5.3-none": "gpt-5.3-codex", "gpt-5.3-low": "gpt-5.3-codex", @@ -154,6 +161,9 @@ func normalizeCodexModel(model string) string { normalized := strings.ToLower(modelID) + if strings.Contains(normalized, "gpt-5.4") || strings.Contains(normalized, "gpt 5.4") { + return "gpt-5.4" + } if strings.Contains(normalized, "gpt-5.2-codex") || strings.Contains(normalized, "gpt 5.2 codex") { return "gpt-5.2-codex" } diff --git a/backend/internal/service/openai_codex_transform_test.go b/backend/internal/service/openai_codex_transform_test.go index 27093f6c37..7ee4bbc8d5 100644 --- a/backend/internal/service/openai_codex_transform_test.go +++ b/backend/internal/service/openai_codex_transform_test.go @@ -167,6 +167,10 @@ func TestApplyCodexOAuthTransform_EmptyInput(t *testing.T) { func TestNormalizeCodexModel_Gpt53(t *testing.T) { cases := map[string]string{ + "gpt-5.4": "gpt-5.4", + "gpt-5.4-high": "gpt-5.4", + "gpt-5.4-chat-latest": "gpt-5.4", + "gpt 5.4": "gpt-5.4", "gpt-5.3": "gpt-5.3-codex", "gpt-5.3-codex": "gpt-5.3-codex", "gpt-5.3-codex-xhigh": "gpt-5.3-codex", diff --git a/backend/internal/service/openai_gateway_service_test.go b/backend/internal/service/openai_gateway_service_test.go index 4f5f7f3c12..7e53c8996e 100644 --- a/backend/internal/service/openai_gateway_service_test.go +++ b/backend/internal/service/openai_gateway_service_test.go @@ -265,6 +265,16 @@ func (c *stubGatewayCache) DeleteSessionAccountID(ctx context.Context, groupID i return nil } +func (c *stubGatewayCache) GetClientAffinityAccounts(_ context.Context, _ int64, _ string, _ time.Duration) ([]int64, error) { + return nil, nil +} +func (c *stubGatewayCache) UpdateClientAffinity(_ context.Context, _ int64, _ string, _ int64, _ time.Duration) error { + return nil +} +func (c *stubGatewayCache) RemoveClientAffinity(_ context.Context, _ int64, _ string, _ int64) error { + return nil +} + func TestOpenAISelectAccountWithLoadAwareness_FiltersUnschedulable(t *testing.T) { now := time.Now() resetAt := now.Add(10 * time.Minute) diff --git a/backend/internal/service/openai_ws_state_store_test.go b/backend/internal/service/openai_ws_state_store_test.go index 235d42331d..6f77f64356 100644 --- a/backend/internal/service/openai_ws_state_store_test.go +++ b/backend/internal/service/openai_ws_state_store_test.go @@ -193,6 +193,16 @@ func (c *openAIWSStateStoreTimeoutProbeCache) DeleteSessionAccountID(ctx context return nil } +func (c *openAIWSStateStoreTimeoutProbeCache) GetClientAffinityAccounts(_ context.Context, _ int64, _ string, _ time.Duration) ([]int64, error) { + return nil, nil +} +func (c *openAIWSStateStoreTimeoutProbeCache) UpdateClientAffinity(_ context.Context, _ int64, _ string, _ int64, _ time.Duration) error { + return nil +} +func (c *openAIWSStateStoreTimeoutProbeCache) RemoveClientAffinity(_ context.Context, _ int64, _ string, _ int64) error { + return nil +} + func TestOpenAIWSStateStore_RedisOpsUseShortTimeout(t *testing.T) { probe := &openAIWSStateStoreTimeoutProbeCache{} store := NewOpenAIWSStateStore(probe) diff --git a/backend/internal/service/ops_concurrency.go b/backend/internal/service/ops_concurrency.go index c03108c4b9..a571dd4df4 100644 --- a/backend/internal/service/ops_concurrency.go +++ b/backend/internal/service/ops_concurrency.go @@ -64,9 +64,12 @@ func (s *OpsService) getAccountsLoadMapBestEffort(ctx context.Context, accounts if acc.ID <= 0 { continue } - lf := acc.EffectiveLoadFactor() - if prev, ok := unique[acc.ID]; !ok || lf > prev { - unique[acc.ID] = lf + c := acc.Concurrency + if c <= 0 { + c = 1 + } + if prev, ok := unique[acc.ID]; !ok || c > prev { + unique[acc.ID] = c } } diff --git a/backend/internal/service/ops_metrics_collector.go b/backend/internal/service/ops_metrics_collector.go index 6c33707138..f93481e7ff 100644 --- a/backend/internal/service/ops_metrics_collector.go +++ b/backend/internal/service/ops_metrics_collector.go @@ -391,7 +391,7 @@ func (c *OpsMetricsCollector) collectConcurrencyQueueDepth(parentCtx context.Con } batch = append(batch, AccountWithConcurrency{ ID: acc.ID, - MaxConcurrency: acc.EffectiveLoadFactor(), + MaxConcurrency: acc.Concurrency, }) } if len(batch) == 0 { diff --git a/backend/internal/service/ops_system_log_sink_test.go b/backend/internal/service/ops_system_log_sink_test.go index 12a2ec0c7d..137ee33c72 100644 --- a/backend/internal/service/ops_system_log_sink_test.go +++ b/backend/internal/service/ops_system_log_sink_test.go @@ -183,6 +183,15 @@ func TestOpsSystemLogSink_StartStopAndFlushSuccess(t *testing.T) { if strings.TrimSpace(item.Message) == "" { t.Fatalf("message should not be empty") } + // writtenCount is incremented after BatchInsertSystemLogsFn returns, + // so poll briefly to avoid a race between the done signal and the atomic add. + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) { + if sink.Health().WrittenCount > 0 { + break + } + time.Sleep(time.Millisecond) + } health := sink.Health() if health.WrittenCount == 0 { t.Fatalf("written_count should be >0") diff --git a/backend/internal/service/pricing_service.go b/backend/internal/service/pricing_service.go index 41e8b5eb0c..897623d633 100644 --- a/backend/internal/service/pricing_service.go +++ b/backend/internal/service/pricing_service.go @@ -21,8 +21,19 @@ import ( ) var ( - openAIModelDatePattern = regexp.MustCompile(`-\d{8}$`) - openAIModelBasePattern = regexp.MustCompile(`^(gpt-\d+(?:\.\d+)?)(?:-|$)`) + openAIModelDatePattern = regexp.MustCompile(`-\d{8}$`) + openAIModelBasePattern = regexp.MustCompile(`^(gpt-\d+(?:\.\d+)?)(?:-|$)`) + openAIGPT54FallbackPricing = &LiteLLMModelPricing{ + InputCostPerToken: 2.5e-06, // $2.5 per MTok + OutputCostPerToken: 1.5e-05, // $15 per MTok + CacheReadInputTokenCost: 2.5e-07, // $0.25 per MTok + LongContextInputTokenThreshold: 272000, + LongContextInputCostMultiplier: 2.0, + LongContextOutputCostMultiplier: 1.5, + LiteLLMProvider: "openai", + Mode: "chat", + SupportsPromptCaching: true, + } ) // LiteLLMModelPricing LiteLLM价格数据结构 @@ -33,6 +44,9 @@ type LiteLLMModelPricing struct { CacheCreationInputTokenCost float64 `json:"cache_creation_input_token_cost"` CacheCreationInputTokenCostAbove1hr float64 `json:"cache_creation_input_token_cost_above_1hr"` CacheReadInputTokenCost float64 `json:"cache_read_input_token_cost"` + LongContextInputTokenThreshold int `json:"long_context_input_token_threshold,omitempty"` + LongContextInputCostMultiplier float64 `json:"long_context_input_cost_multiplier,omitempty"` + LongContextOutputCostMultiplier float64 `json:"long_context_output_cost_multiplier,omitempty"` LiteLLMProvider string `json:"litellm_provider"` Mode string `json:"mode"` SupportsPromptCaching bool `json:"supports_prompt_caching"` @@ -660,7 +674,8 @@ func (s *PricingService) matchByModelFamily(model string) *LiteLLMModelPricing { // 2. gpt-5.2-codex -> gpt-5.2(去掉后缀如 -codex, -mini, -max 等) // 3. gpt-5.2-20251222 -> gpt-5.2(去掉日期版本号) // 4. gpt-5.3-codex -> gpt-5.2-codex -// 5. 最终回退到 DefaultTestModel (gpt-5.1-codex) +// 5. gpt-5.4* -> 业务静态兜底价 +// 6. 最终回退到 DefaultTestModel (gpt-5.1-codex) func (s *PricingService) matchOpenAIModel(model string) *LiteLLMModelPricing { if strings.HasPrefix(model, "gpt-5.3-codex-spark") { if pricing, ok := s.pricingData["gpt-5.1-codex"]; ok { @@ -690,6 +705,12 @@ func (s *PricingService) matchOpenAIModel(model string) *LiteLLMModelPricing { } } + if strings.HasPrefix(model, "gpt-5.4") { + logger.With(zap.String("component", "service.pricing")). + Info(fmt.Sprintf("[Pricing] OpenAI fallback matched %s -> %s", model, "gpt-5.4(static)")) + return openAIGPT54FallbackPricing + } + // 最终回退到 DefaultTestModel defaultModel := strings.ToLower(openai.DefaultTestModel) if pricing, ok := s.pricingData[defaultModel]; ok { diff --git a/backend/internal/service/pricing_service_test.go b/backend/internal/service/pricing_service_test.go index 127ff342c6..6b67c55a06 100644 --- a/backend/internal/service/pricing_service_test.go +++ b/backend/internal/service/pricing_service_test.go @@ -51,3 +51,20 @@ func TestGetModelPricing_OpenAIFallbackMatchedLoggedAsInfo(t *testing.T) { require.True(t, logSink.ContainsMessageAtLevel("[Pricing] OpenAI fallback matched gpt-5.3-codex -> gpt-5.2-codex", "info")) require.False(t, logSink.ContainsMessageAtLevel("[Pricing] OpenAI fallback matched gpt-5.3-codex -> gpt-5.2-codex", "warn")) } + +func TestGetModelPricing_Gpt54UsesStaticFallbackWhenRemoteMissing(t *testing.T) { + svc := &PricingService{ + pricingData: map[string]*LiteLLMModelPricing{ + "gpt-5.1-codex": &LiteLLMModelPricing{InputCostPerToken: 1.25e-6}, + }, + } + + got := svc.GetModelPricing("gpt-5.4") + require.NotNil(t, got) + require.InDelta(t, 2.5e-6, got.InputCostPerToken, 1e-12) + require.InDelta(t, 1.5e-5, got.OutputCostPerToken, 1e-12) + require.InDelta(t, 2.5e-7, got.CacheReadInputTokenCost, 1e-12) + require.Equal(t, 272000, got.LongContextInputTokenThreshold) + require.InDelta(t, 2.0, got.LongContextInputCostMultiplier, 1e-12) + require.InDelta(t, 1.5, got.LongContextOutputCostMultiplier, 1e-12) +} diff --git a/backend/internal/service/ratelimit_service.go b/backend/internal/service/ratelimit_service.go index 9f16fb2b50..a888a2eeeb 100644 --- a/backend/internal/service/ratelimit_service.go +++ b/backend/internal/service/ratelimit_service.go @@ -128,8 +128,9 @@ func (s *RateLimitService) HandleUpstreamError(ctx context.Context, account *Acc } // 其他 400 错误(如参数问题)不处理,不禁用账号 case 401: - // 对所有 OAuth 账号在 401 错误时调用缓存失效并强制下次刷新 - if account.Type == AccountTypeOAuth { + // OpenAI OAuth 账号在 401 错误时临时不可调度(保持 active 让刷新服务恢复); + // 其他平台 OAuth 账号保持原有 SetError 行为(Antigravity 主流程不走此路径)。 + if account.Type == AccountTypeOAuth && account.Platform == PlatformOpenAI { // 1. 失效缓存 if s.tokenCacheInvalidator != nil { if err := s.tokenCacheInvalidator.InvalidateToken(ctx, account); err != nil { @@ -1093,7 +1094,8 @@ func (s *RateLimitService) tryTempUnschedulable(ctx context.Context, account *Ac } // 401 首次命中可临时不可调度(给 token 刷新窗口); // 若历史上已因 401 进入过临时不可调度,则本次应升级为 error(返回 false 交由默认错误逻辑处理)。 - if statusCode == http.StatusUnauthorized { + // Antigravity 跳过:其 401 由 applyErrorPolicy 的 temp_unschedulable_rules 自行控制,无需升级逻辑。 + if statusCode == http.StatusUnauthorized && account.Platform != PlatformAntigravity { reason := account.TempUnschedulableReason // 缓存可能没有 reason,从 DB 回退读取 if reason == "" { diff --git a/backend/internal/service/ratelimit_service_401_db_fallback_test.go b/backend/internal/service/ratelimit_service_401_db_fallback_test.go index e1611425f5..ad2c9c81e7 100644 --- a/backend/internal/service/ratelimit_service_401_db_fallback_test.go +++ b/backend/internal/service/ratelimit_service_401_db_fallback_test.go @@ -28,6 +28,7 @@ func (r *dbFallbackRepoStub) GetByID(ctx context.Context, id int64) (*Account, e func TestCheckErrorPolicy_401_DBFallback_Escalates(t *testing.T) { // Scenario: cache account has empty TempUnschedulableReason (cache miss), // but DB account has a previous 401 record → should escalate to ErrorPolicyNone. + // 注意:Antigravity 跳过 401 升级,此测试使用 Gemini 平台验证升级逻辑。 repo := &dbFallbackRepoStub{ dbAccount: &Account{ ID: 20, @@ -39,7 +40,7 @@ func TestCheckErrorPolicy_401_DBFallback_Escalates(t *testing.T) { account := &Account{ ID: 20, Type: AccountTypeOAuth, - Platform: PlatformAntigravity, + Platform: PlatformGemini, TempUnschedulableReason: "", // cache miss — reason is empty Credentials: map[string]any{ "temp_unschedulable_enabled": true, diff --git a/backend/internal/service/ratelimit_service_401_test.go b/backend/internal/service/ratelimit_service_401_test.go index 7bced46f4f..653e2fba4c 100644 --- a/backend/internal/service/ratelimit_service_401_test.go +++ b/backend/internal/service/ratelimit_service_401_test.go @@ -41,13 +41,37 @@ func (r *tokenCacheInvalidatorRecorder) InvalidateToken(ctx context.Context, acc return r.err } -func TestRateLimitService_HandleUpstreamError_OAuth401SetsTempUnschedulable(t *testing.T) { +// TestRateLimitService_HandleUpstreamError_OpenAIOAuth401SetsTempUnschedulable +// 仅 OpenAI OAuth 账号 401 使用 temp_unschedulable +func TestRateLimitService_HandleUpstreamError_OpenAIOAuth401SetsTempUnschedulable(t *testing.T) { + repo := &rateLimitAccountRepoStub{} + invalidator := &tokenCacheInvalidatorRecorder{} + service := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + service.SetTokenCacheInvalidator(invalidator) + account := &Account{ + ID: 100, + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + } + + shouldDisable := service.HandleUpstreamError(context.Background(), account, 401, http.Header{}, []byte("unauthorized")) + + require.True(t, shouldDisable) + require.Equal(t, 0, repo.setErrorCalls) + require.Equal(t, 1, repo.tempCalls) + require.Len(t, invalidator.accounts, 1) +} + +// TestRateLimitService_HandleUpstreamError_NonOpenAIOAuth401SetsError +// 非 OpenAI 的 OAuth 账号 401 走原有 SetError 行为 +func TestRateLimitService_HandleUpstreamError_NonOpenAIOAuth401SetsError(t *testing.T) { tests := []struct { name string platform string }{ {name: "gemini", platform: PlatformGemini}, {name: "antigravity", platform: PlatformAntigravity}, + {name: "anthropic", platform: PlatformAnthropic}, } for _, tt := range tests { @@ -60,29 +84,20 @@ func TestRateLimitService_HandleUpstreamError_OAuth401SetsTempUnschedulable(t *t ID: 100, Platform: tt.platform, Type: AccountTypeOAuth, - Credentials: map[string]any{ - "temp_unschedulable_enabled": true, - "temp_unschedulable_rules": []any{ - map[string]any{ - "error_code": 401, - "keywords": []any{"unauthorized"}, - "duration_minutes": 30, - "description": "custom rule", - }, - }, - }, } shouldDisable := service.HandleUpstreamError(context.Background(), account, 401, http.Header{}, []byte("unauthorized")) require.True(t, shouldDisable) - require.Equal(t, 0, repo.setErrorCalls) - require.Equal(t, 1, repo.tempCalls) - require.Len(t, invalidator.accounts, 1) + require.Equal(t, 1, repo.setErrorCalls) + require.Equal(t, 0, repo.tempCalls) + require.Contains(t, repo.lastErrorMsg, "Authentication failed (401)") }) } } +// TestRateLimitService_HandleUpstreamError_OAuth401InvalidatorError +// OpenAI OAuth 401 缓存失效出错时仍走 temp_unschedulable func TestRateLimitService_HandleUpstreamError_OAuth401InvalidatorError(t *testing.T) { repo := &rateLimitAccountRepoStub{} invalidator := &tokenCacheInvalidatorRecorder{err: errors.New("boom")} @@ -90,7 +105,7 @@ func TestRateLimitService_HandleUpstreamError_OAuth401InvalidatorError(t *testin service.SetTokenCacheInvalidator(invalidator) account := &Account{ ID: 101, - Platform: PlatformGemini, + Platform: PlatformOpenAI, Type: AccountTypeOAuth, } diff --git a/backend/internal/testutil/stubs.go b/backend/internal/testutil/stubs.go index 217a5f5699..669372ef51 100644 --- a/backend/internal/testutil/stubs.go +++ b/backend/internal/testutil/stubs.go @@ -97,6 +97,15 @@ func (c StubGatewayCache) RefreshSessionTTL(_ context.Context, _ int64, _ string func (c StubGatewayCache) DeleteSessionAccountID(_ context.Context, _ int64, _ string) error { return nil } +func (c StubGatewayCache) GetClientAffinityAccounts(_ context.Context, _ int64, _ string, _ time.Duration) ([]int64, error) { + return nil, nil +} +func (c StubGatewayCache) UpdateClientAffinity(_ context.Context, _ int64, _ string, _ int64, _ time.Duration) error { + return nil +} +func (c StubGatewayCache) RemoveClientAffinity(_ context.Context, _ int64, _ string, _ int64) error { + return nil +} // ============================================================ // StubSessionLimitCache — service.SessionLimitCache 的空实现 diff --git a/backend/resources/model-pricing/model_prices_and_context_window.json b/backend/resources/model-pricing/model_prices_and_context_window.json index 650e128e9a..72860bf9e7 100644 --- a/backend/resources/model-pricing/model_prices_and_context_window.json +++ b/backend/resources/model-pricing/model_prices_and_context_window.json @@ -5140,6 +5140,39 @@ "supports_vision": true, "supports_web_search": true }, + "gpt-5.4": { + "cache_read_input_token_cost": 2.5e-07, + "input_cost_per_token": 2.5e-06, + "litellm_provider": "openai", + "max_input_tokens": 1050000, + "max_output_tokens": 128000, + "max_tokens": 128000, + "mode": "chat", + "output_cost_per_token": 1.5e-05, + "supported_endpoints": [ + "/v1/chat/completions", + "/v1/responses" + ], + "supported_modalities": [ + "text", + "image" + ], + "supported_output_modalities": [ + "text", + "image" + ], + "supports_function_calling": true, + "supports_native_streaming": true, + "supports_parallel_function_calling": true, + "supports_pdf_input": true, + "supports_prompt_caching": true, + "supports_reasoning": true, + "supports_response_schema": true, + "supports_service_tier": true, + "supports_system_messages": true, + "supports_tool_choice": true, + "supports_vision": true + }, "gpt-5.3-codex": { "cache_read_input_token_cost": 1.75e-07, "cache_read_input_token_cost_priority": 3.5e-07, diff --git a/frontend/src/components/account/BulkEditAccountModal.vue b/frontend/src/components/account/BulkEditAccountModal.vue index 5e7d0ae4fe..4d26a48af9 100644 --- a/frontend/src/components/account/BulkEditAccountModal.vue +++ b/frontend/src/components/account/BulkEditAccountModal.vue @@ -496,6 +496,7 @@ class="input" :class="!enableConcurrency && 'cursor-not-allowed opacity-50'" aria-labelledby="bulk-edit-concurrency-label" + @input="concurrency = Math.max(1, concurrency || 1)" />
{{ t('admin.accounts.loadFactorHint') }}
{{ t('admin.accounts.loadFactorHint') }}
+ {{ t('admin.accounts.quotaControl.clientAffinity.hint') }} +
+{{ t('admin.accounts.loadFactorHint') }}
{{ t('admin.accounts.quotaLimitAmountHint') }}
{{ t('admin.accounts.quotaPeriodHint') }}
+