mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-20 00:18:55 +00:00
fix(provider): 修复 Claude 计费与钱包冷却并隔离模型目录缓存
This commit is contained in:
@@ -29,6 +29,7 @@ func TestGroupPlatformBinding_AllowedPlatforms(t *testing.T) {
|
||||
"anthropic", "openai", "gemini", "antigravity", "grok",
|
||||
"kimi", "zhipu", "deepseek", "minimax", "opencode_go", "composite",
|
||||
"typesafe",
|
||||
"command_code", "cline",
|
||||
}
|
||||
for _, platform := range allowed {
|
||||
t.Run("create_"+platform, func(t *testing.T) {
|
||||
@@ -73,7 +74,7 @@ func TestGroupPlatformBinding_RejectsInvalidPlatforms(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestCompositeRouteTargetPlatform_AllowsConcreteProviders(t *testing.T) {
|
||||
for _, platform := range []string{"kimi", "zhipu", "deepseek", "minimax", "opencode_go", "typesafe"} {
|
||||
for _, platform := range []string{"kimi", "zhipu", "deepseek", "minimax", "opencode_go", "typesafe", "command_code", "cline"} {
|
||||
var req CompositeRouteRequest
|
||||
body := fmt.Sprintf(`{"public_model":"m","target_platform":%q}`, platform)
|
||||
require.NoError(t, bindGroupPlatformJSON(t, &req, body))
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
package admin
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"github.com/Wei-Shaw/sub2api/internal/domain"
|
||||
"github.com/gin-gonic/gin/binding"
|
||||
"github.com/go-playground/validator/v10"
|
||||
@@ -13,12 +15,16 @@ import (
|
||||
func init() {
|
||||
engine, ok := binding.Validator.Engine().(*validator.Validate)
|
||||
if !ok {
|
||||
return
|
||||
panic("platform validation requires a go-playground validator engine")
|
||||
}
|
||||
_ = engine.RegisterValidation("group_platform", func(fl validator.FieldLevel) bool {
|
||||
if err := engine.RegisterValidation("group_platform", func(fl validator.FieldLevel) bool {
|
||||
return domain.IsGroupPlatform(fl.Field().String())
|
||||
})
|
||||
_ = engine.RegisterValidation("concrete_platform", func(fl validator.FieldLevel) bool {
|
||||
}); err != nil {
|
||||
panic(fmt.Errorf("register group_platform validation: %w", err))
|
||||
}
|
||||
if err := engine.RegisterValidation("concrete_platform", func(fl validator.FieldLevel) bool {
|
||||
return domain.IsConcretePlatform(fl.Field().String())
|
||||
})
|
||||
}); err != nil {
|
||||
panic(fmt.Errorf("register concrete_platform validation: %w", err))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -450,7 +450,7 @@ func (s *AccountTestService) testModelRoutedAccountConnection(c *gin.Context, ac
|
||||
// 与网关同一判定(含上游模型目录);测试没有入站协议,取模型的首选协议。
|
||||
protocol := account.resolveModelRoutedProtocol(testModelID)
|
||||
if s.openaiGatewayService != nil {
|
||||
protocol = s.openaiGatewayService.resolveUpstreamProtocolFor(account, "", testModelID)
|
||||
protocol = s.openaiGatewayService.resolveUpstreamProtocolFor(c.Request.Context(), account, "", testModelID)
|
||||
}
|
||||
switch protocol {
|
||||
case APIProtocolAnthropic:
|
||||
|
||||
@@ -152,7 +152,8 @@ func TestClassifyClineError(t *testing.T) {
|
||||
`{"error":{"message":"You have reached your limit. Please try again later."}}`: clineErrorNone,
|
||||
}
|
||||
for body, want := range cases {
|
||||
require.Equal(t, want, classifyClineError([]byte(body)), body)
|
||||
kind, _ := parseClineError([]byte(body))
|
||||
require.Equal(t, want, kind, body)
|
||||
}
|
||||
|
||||
require.Equal(t, "5h", clinePassWindow([]byte(clinePass5hBody)))
|
||||
|
||||
@@ -98,7 +98,7 @@ func TestAccountTestService_CommandCodeUsesModelCatalog(t *testing.T) {
|
||||
APIProtocolResponses: base,
|
||||
}
|
||||
url := buildOpenAIModelsURL(base)
|
||||
upstreamModelProtocols.store(modelProtocolCatalogKey(account, base, url),
|
||||
upstreamModelProtocols.store(modelProtocolCatalogKey(account, url),
|
||||
map[string][]string{"vendor/responses-only": {APIProtocolResponses}}, nil, time.Now())
|
||||
svc, upstream := adaptiveCNAccountTestService(account, adaptiveCNResponsesTestResponse())
|
||||
svc.openaiGatewayService = &OpenAIGatewayService{cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream}
|
||||
|
||||
@@ -23,12 +23,12 @@ import (
|
||||
// 按模型分流时据此判断模型是否支持入站协议:支持就同协议直通,省去一次协议转换
|
||||
// (见 modelProtocolSet)。
|
||||
//
|
||||
// 缓存范围:官方默认地址的目录与账号无关,按地址全进程共享;自定义上游的目录可能随
|
||||
// Key、租户或请求头不同,按账号隔离(见 modelProtocolCatalogKey)。
|
||||
// 缓存范围:所有目录均按账号和地址隔离,包括官方默认地址。目录可能随 Key、租户
|
||||
// 或请求头不同,鉴权失败的刷新退避也不能影响其他账号(见 modelProtocolCatalogKey)。
|
||||
//
|
||||
// 刷新在后台进行:过期时继续使用旧目录。目录从未加载成功时,首批请求最多等待
|
||||
// modelProtocolCatalogFirstLoadWait,让冷启动也按目录分流,而不是先走一次协议转换;
|
||||
// 超时或失败则回落内置规则,失败退避期内不再等待。
|
||||
// modelProtocolCatalogFirstLoadWait,各请求共用本轮刷新开始时的截止时间,并响应请求
|
||||
// 取消。超时后同一轮刷新中的后续请求立即回落内置规则,失败退避期内不再等待。
|
||||
|
||||
const (
|
||||
modelProtocolCatalogTTL = 30 * time.Minute
|
||||
@@ -46,13 +46,13 @@ type modelProtocolCatalogEntry struct {
|
||||
refreshing bool
|
||||
// done 在进行中的刷新结束时关闭。
|
||||
done chan struct{}
|
||||
// firstLoadDeadline 是本轮冷启动刷新共享的等待截止时间。
|
||||
firstLoadDeadline time.Time
|
||||
}
|
||||
|
||||
type modelProtocolCatalog struct {
|
||||
mu sync.Mutex
|
||||
entries map[string]*modelProtocolCatalogEntry
|
||||
// firstLoadWait 覆盖 modelProtocolCatalogFirstLoadWait(测试用)。
|
||||
firstLoadWait time.Duration
|
||||
}
|
||||
|
||||
// upstreamModelProtocols 是全进程共享的模型协议目录。
|
||||
@@ -72,8 +72,11 @@ func (c *modelProtocolCatalog) entryLocked(key string) *modelProtocolCatalogEntr
|
||||
|
||||
// lookup 返回 key 目录中 model 支持的协议;目录尚未就绪或模型不在目录中时返回 nil。
|
||||
// 目录缺失或过期、且不在刷新中也不在失败退避期时,异步调用 refresh(由其调用 store);
|
||||
// 目录从未加载成功且刷新进行中时,最多等待 firstLoadWait。
|
||||
func (c *modelProtocolCatalog) lookup(key, model string, now time.Time, refresh func()) []string {
|
||||
// 目录从未加载成功且刷新进行中时,只等待本轮冷启动窗口的剩余时间。
|
||||
func (c *modelProtocolCatalog) lookup(ctx context.Context, key, model string, now time.Time, refresh func()) []string {
|
||||
if ctx.Err() != nil {
|
||||
return nil
|
||||
}
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
entry := c.entryLocked(key)
|
||||
@@ -81,24 +84,25 @@ func (c *modelProtocolCatalog) lookup(key, model string, now time.Time, refresh
|
||||
if stale && !entry.refreshing && !now.Before(entry.retryAt) && refresh != nil {
|
||||
entry.refreshing = true
|
||||
entry.done = make(chan struct{})
|
||||
if entry.models == nil {
|
||||
entry.firstLoadDeadline = now.Add(modelProtocolCatalogFirstLoadWait)
|
||||
}
|
||||
go refresh()
|
||||
}
|
||||
if entry.models == nil && entry.refreshing {
|
||||
wait := c.firstLoadWait
|
||||
if wait <= 0 {
|
||||
wait = modelProtocolCatalogFirstLoadWait
|
||||
}
|
||||
if entry.models == nil && entry.refreshing && now.Before(entry.firstLoadDeadline) {
|
||||
done := entry.done
|
||||
deadline := entry.firstLoadDeadline
|
||||
c.mu.Unlock()
|
||||
timer := time.NewTimer(wait)
|
||||
timer := time.NewTimer(time.Until(deadline))
|
||||
select {
|
||||
case <-done:
|
||||
case <-timer.C:
|
||||
case <-ctx.Done():
|
||||
}
|
||||
timer.Stop()
|
||||
c.mu.Lock()
|
||||
}
|
||||
if entry.models == nil {
|
||||
if entry.models == nil || ctx.Err() != nil {
|
||||
return nil
|
||||
}
|
||||
return entry.models[strings.ToLower(strings.TrimSpace(model))]
|
||||
@@ -170,7 +174,7 @@ func parseModelProtocolCatalog(body []byte) (map[string][]string, error) {
|
||||
|
||||
// modelCatalogProtocols 返回上游模型目录中 model 支持的协议;平台不提供目录、账号固定了
|
||||
// 上游协议或目录尚未就绪时返回 nil。
|
||||
func (s *OpenAIGatewayService) modelCatalogProtocols(account *Account, model string) []string {
|
||||
func (s *OpenAIGatewayService) modelCatalogProtocols(ctx context.Context, account *Account, model string) []string {
|
||||
if s == nil || s.httpUpstream == nil || strings.TrimSpace(model) == "" || !account.routesByModel() {
|
||||
return nil
|
||||
}
|
||||
@@ -185,14 +189,14 @@ func (s *OpenAIGatewayService) modelCatalogProtocols(account *Account, model str
|
||||
return nil
|
||||
}
|
||||
url := buildOpenAIModelsURL(base)
|
||||
key := modelProtocolCatalogKey(account, base, url)
|
||||
key := modelProtocolCatalogKey(account, url)
|
||||
headers := modelProtocolCatalogHeaders(account, url)
|
||||
proxyURL := ""
|
||||
if account.ProxyID != nil && account.Proxy != nil {
|
||||
proxyURL = account.Proxy.URL()
|
||||
}
|
||||
accountID, concurrency := account.ID, account.Concurrency
|
||||
return upstreamModelProtocols.lookup(key, model, time.Now(), func() {
|
||||
return upstreamModelProtocols.lookup(ctx, key, model, time.Now(), func() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), modelProtocolCatalogTimeout)
|
||||
defer cancel()
|
||||
models, err := s.fetchModelProtocolCatalog(ctx, url, headers, proxyURL, accountID, concurrency)
|
||||
@@ -205,13 +209,9 @@ func (s *OpenAIGatewayService) modelCatalogProtocols(account *Account, model str
|
||||
})
|
||||
}
|
||||
|
||||
// modelProtocolCatalogKey 返回目录的缓存键:账号使用平台官方默认地址时按地址共享,
|
||||
// 自定义上游按账号隔离,避免一个账号的目录或鉴权失败影响同地址的其他账号。
|
||||
func modelProtocolCatalogKey(account *Account, base, url string) string {
|
||||
official := account.providerProfile().DefaultBaseURL(account.GetCredential("account_mode"), APIProtocolChatCompletions)
|
||||
if official != "" && strings.EqualFold(strings.TrimRight(base, "/"), strings.TrimRight(official, "/")) {
|
||||
return url
|
||||
}
|
||||
// modelProtocolCatalogKey 按账号和地址隔离目录及刷新退避,避免一个账号的目录或
|
||||
// 鉴权失败影响同地址的其他账号。
|
||||
func modelProtocolCatalogKey(account *Account, url string) string {
|
||||
return fmt.Sprintf("%s#account=%d", url, account.ID)
|
||||
}
|
||||
|
||||
@@ -258,6 +258,6 @@ func (s *OpenAIGatewayService) fetchModelProtocolCatalog(ctx context.Context, ur
|
||||
|
||||
// resolveUpstreamProtocolFor 是网关入口使用的 resolveUpstreamProtocol:按模型分流且平台
|
||||
// 提供模型目录时,带上目录中该模型支持的协议。
|
||||
func (s *OpenAIGatewayService) resolveUpstreamProtocolFor(account *Account, inbound, model string) string {
|
||||
return resolveUpstreamProtocol(account, inbound, model, s.modelCatalogProtocols(account, model))
|
||||
func (s *OpenAIGatewayService) resolveUpstreamProtocolFor(ctx context.Context, account *Account, inbound, model string) string {
|
||||
return resolveUpstreamProtocol(account, inbound, model, s.modelCatalogProtocols(ctx, account, model))
|
||||
}
|
||||
|
||||
@@ -3,11 +3,15 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"testing/synctest"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
@@ -41,74 +45,155 @@ func TestParseModelProtocolCatalog(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestModelProtocolCatalogLookupRefreshesInBackground(t *testing.T) {
|
||||
catalog := modelProtocolCatalog{firstLoadWait: 20 * time.Millisecond}
|
||||
const url = "https://catalog.example/v1/models"
|
||||
now := time.Date(2026, 9, 26, 12, 0, 0, 0, time.UTC)
|
||||
var refreshes atomic.Int32
|
||||
started := make(chan struct{}, 4)
|
||||
refresh := func() { refreshes.Add(1); started <- struct{}{} }
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
var catalog modelProtocolCatalog
|
||||
const key = "https://catalog.example/v1/models"
|
||||
now := time.Now()
|
||||
models, err := parseModelProtocolCatalog([]byte(commandCodeModelsSample))
|
||||
require.NoError(t, err)
|
||||
catalog.store(key, models, nil, now)
|
||||
var refreshes atomic.Int32
|
||||
refresh := func() { refreshes.Add(1) }
|
||||
|
||||
// 未就绪且刷新迟迟不返回:等待上限后返回 nil;刷新期间不重复触发。
|
||||
require.Nil(t, catalog.lookup(url, "gpt-5.5", now, refresh))
|
||||
<-started
|
||||
require.Nil(t, catalog.lookup(url, "gpt-5.5", now, refresh))
|
||||
require.EqualValues(t, 1, refreshes.Load())
|
||||
require.Equal(t, []string{APIProtocolChatCompletions, APIProtocolResponses}, catalog.lookup(t.Context(), key, "ZAI-ORG/GLM-5.3", now, refresh))
|
||||
require.Nil(t, catalog.lookup(t.Context(), key, "unknown-model", now, refresh))
|
||||
require.Zero(t, refreshes.Load())
|
||||
|
||||
models, err := parseModelProtocolCatalog([]byte(commandCodeModelsSample))
|
||||
require.NoError(t, err)
|
||||
catalog.store(url, models, nil, now)
|
||||
require.Equal(t, []string{APIProtocolChatCompletions, APIProtocolResponses}, catalog.lookup(url, "ZAI-ORG/GLM-5.3", now, refresh))
|
||||
require.Nil(t, catalog.lookup(url, "unknown-model", now, refresh))
|
||||
require.EqualValues(t, 1, refreshes.Load())
|
||||
// A stale catalog is returned immediately while one refresh runs in the background.
|
||||
time.Sleep(modelProtocolCatalogTTL) // synctest advances virtual time.
|
||||
later := time.Now()
|
||||
require.NotNil(t, catalog.lookup(t.Context(), key, "gpt-5.5", later, refresh))
|
||||
require.Equal(t, later, time.Now(), "a stale result must not wait for refresh")
|
||||
synctest.Wait()
|
||||
require.EqualValues(t, 1, refreshes.Load())
|
||||
require.NotNil(t, catalog.lookup(t.Context(), key, "gpt-5.5", later, refresh))
|
||||
require.EqualValues(t, 1, refreshes.Load(), "concurrent stale reads share one refresh")
|
||||
catalog.store(key, nil, errors.New("boom"), later)
|
||||
require.NotNil(t, catalog.lookup(t.Context(), key, "gpt-5.5", later.Add(time.Minute), refresh))
|
||||
require.EqualValues(t, 1, refreshes.Load(), "failure backoff preserves the old catalog")
|
||||
require.NotNil(t, catalog.lookup(t.Context(), key, "gpt-5.5", later.Add(modelProtocolCatalogRetryBackoff), refresh))
|
||||
synctest.Wait()
|
||||
require.EqualValues(t, 2, refreshes.Load())
|
||||
})
|
||||
}
|
||||
|
||||
// 过期后继续返回旧目录,同时后台刷新;失败时保留旧目录并退避。
|
||||
later := now.Add(modelProtocolCatalogTTL)
|
||||
require.NotNil(t, catalog.lookup(url, "gpt-5.5", later, refresh))
|
||||
<-started
|
||||
require.EqualValues(t, 2, refreshes.Load())
|
||||
catalog.store(url, nil, errors.New("boom"), later)
|
||||
require.NotNil(t, catalog.lookup(url, "gpt-5.5", later.Add(time.Minute), refresh))
|
||||
require.EqualValues(t, 2, refreshes.Load(), "failure backoff suppresses refresh")
|
||||
require.NotNil(t, catalog.lookup(url, "gpt-5.5", later.Add(modelProtocolCatalogRetryBackoff), refresh))
|
||||
<-started
|
||||
require.EqualValues(t, 3, refreshes.Load())
|
||||
func TestModelProtocolCatalogFirstLoadSharesWaitDeadline(t *testing.T) {
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
var catalog modelProtocolCatalog
|
||||
const key = "https://catalog.example/v1/models"
|
||||
models := map[string][]string{"gpt-5.5": {APIProtocolResponses}}
|
||||
release := make(chan struct{})
|
||||
var refreshes atomic.Int32
|
||||
refresh := func() {
|
||||
refreshes.Add(1)
|
||||
<-release
|
||||
catalog.store(key, models, nil, time.Now())
|
||||
}
|
||||
type result struct {
|
||||
protocols []string
|
||||
elapsed time.Duration
|
||||
}
|
||||
first, second := make(chan result, 1), make(chan result, 1)
|
||||
lookup := func(done chan<- result) {
|
||||
start := time.Now()
|
||||
protocols := catalog.lookup(t.Context(), key, "gpt-5.5", start, refresh)
|
||||
done <- result{protocols, time.Since(start)}
|
||||
}
|
||||
start := time.Now()
|
||||
go lookup(first)
|
||||
synctest.Wait()
|
||||
time.Sleep(modelProtocolCatalogFirstLoadWait / 2) // Virtual time only.
|
||||
go lookup(second)
|
||||
synctest.Wait()
|
||||
time.Sleep(modelProtocolCatalogFirstLoadWait / 2)
|
||||
synctest.Wait()
|
||||
|
||||
r1, r2 := <-first, <-second
|
||||
require.Nil(t, r1.protocols)
|
||||
require.Nil(t, r2.protocols)
|
||||
require.Equal(t, modelProtocolCatalogFirstLoadWait, r1.elapsed)
|
||||
require.Equal(t, modelProtocolCatalogFirstLoadWait/2, r2.elapsed, "a later caller waits only for the remaining shared window")
|
||||
require.Equal(t, modelProtocolCatalogFirstLoadWait, time.Since(start))
|
||||
require.Nil(t, catalog.lookup(t.Context(), key, "gpt-5.5", time.Now(), refresh))
|
||||
require.Equal(t, modelProtocolCatalogFirstLoadWait, time.Since(start), "later calls fall back immediately while the same refresh continues")
|
||||
require.EqualValues(t, 1, refreshes.Load())
|
||||
|
||||
close(release)
|
||||
synctest.Wait()
|
||||
require.Equal(t, models["gpt-5.5"], catalog.lookup(t.Context(), key, "gpt-5.5", time.Now(), refresh), "a late successful refresh still populates the cache")
|
||||
})
|
||||
}
|
||||
|
||||
// 冷启动:目录从未加载时首批请求等待进行中的刷新,按目录分流;刷新失败则立即回落,
|
||||
// 退避期内不再等待。
|
||||
func TestModelProtocolCatalogFirstLoadWaitsForRefresh(t *testing.T) {
|
||||
const key = "https://catalog.example/v1/models"
|
||||
now := time.Date(2026, 9, 26, 12, 0, 0, 0, time.UTC)
|
||||
models, err := parseModelProtocolCatalog([]byte(commandCodeModelsSample))
|
||||
require.NoError(t, err)
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
const key = "https://catalog.example/v1/models"
|
||||
models, err := parseModelProtocolCatalog([]byte(commandCodeModelsSample))
|
||||
require.NoError(t, err)
|
||||
var loaded modelProtocolCatalog
|
||||
release := make(chan struct{})
|
||||
result := make(chan []string, 1)
|
||||
go func() {
|
||||
result <- loaded.lookup(t.Context(), key, "gpt-5.5", time.Now(), func() {
|
||||
<-release
|
||||
loaded.store(key, models, nil, time.Now())
|
||||
})
|
||||
}()
|
||||
synctest.Wait()
|
||||
close(release)
|
||||
synctest.Wait()
|
||||
require.Equal(t, []string{APIProtocolChatCompletions, APIProtocolResponses}, <-result)
|
||||
|
||||
loaded := modelProtocolCatalog{firstLoadWait: 5 * time.Second}
|
||||
require.Equal(t, []string{APIProtocolChatCompletions, APIProtocolResponses}, loaded.lookup(key, "gpt-5.5", now, func() {
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
loaded.store(key, models, nil, now)
|
||||
}))
|
||||
|
||||
failing := modelProtocolCatalog{firstLoadWait: 5 * time.Second}
|
||||
var refreshes atomic.Int32
|
||||
refresh := func() {
|
||||
refreshes.Add(1)
|
||||
failing.store(key, nil, errors.New("boom"), now)
|
||||
}
|
||||
start := time.Now()
|
||||
require.Nil(t, failing.lookup(key, "gpt-5.5", now, refresh))
|
||||
require.Nil(t, failing.lookup(key, "gpt-5.5", now.Add(time.Minute), refresh))
|
||||
require.Less(t, time.Since(start), time.Second, "a failed first load must not hold requests")
|
||||
require.EqualValues(t, 1, refreshes.Load())
|
||||
var failing modelProtocolCatalog
|
||||
var refreshes atomic.Int32
|
||||
refresh := func() {
|
||||
refreshes.Add(1)
|
||||
failing.store(key, nil, errors.New("boom"), time.Now())
|
||||
}
|
||||
start := time.Now()
|
||||
require.Nil(t, failing.lookup(t.Context(), key, "gpt-5.5", start, refresh))
|
||||
require.Nil(t, failing.lookup(t.Context(), key, "gpt-5.5", start.Add(time.Minute), refresh))
|
||||
require.Equal(t, start, time.Now(), "a failed first load and its backoff must not hold requests")
|
||||
require.EqualValues(t, 1, refreshes.Load())
|
||||
})
|
||||
}
|
||||
|
||||
// 缓存范围与请求头:官方默认地址按地址共享,自定义上游按账号隔离;请求头与转发一致。
|
||||
func TestModelProtocolCatalogFirstLoadWaitRespondsToCancellation(t *testing.T) {
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
var catalog modelProtocolCatalog
|
||||
const key = "https://catalog.example/v1/models"
|
||||
ctx, cancel := context.WithCancel(t.Context())
|
||||
defer cancel()
|
||||
release := make(chan struct{})
|
||||
models := map[string][]string{"gpt-5.5": {APIProtocolResponses}}
|
||||
result := make(chan []string, 1)
|
||||
start := time.Now()
|
||||
go func() {
|
||||
result <- catalog.lookup(ctx, key, "gpt-5.5", start, func() {
|
||||
<-release
|
||||
catalog.store(key, models, nil, time.Now())
|
||||
})
|
||||
}()
|
||||
synctest.Wait()
|
||||
cancel()
|
||||
synctest.Wait()
|
||||
require.Nil(t, <-result)
|
||||
require.Equal(t, start, time.Now(), "cancellation releases the caller without waiting for the cold-start deadline")
|
||||
|
||||
// The caller's cancellation must not cancel the background refresh for future requests.
|
||||
close(release)
|
||||
synctest.Wait()
|
||||
require.Equal(t, models["gpt-5.5"], catalog.lookup(t.Context(), key, "gpt-5.5", time.Now(), nil))
|
||||
})
|
||||
}
|
||||
|
||||
// 官方默认地址和自定义上游均按账号隔离;请求头与转发一致。
|
||||
func TestModelProtocolCatalogKeyAndHeaders(t *testing.T) {
|
||||
official := commandCodeTestAccount(21)
|
||||
base := official.GetCNProtocolBaseURL(APIProtocolChatCompletions)
|
||||
url := buildOpenAIModelsURL(base)
|
||||
require.Equal(t, "https://api.commandcode.ai/provider/v1/models", url)
|
||||
require.Equal(t, url, modelProtocolCatalogKey(official, base, url))
|
||||
require.Equal(t, url, modelProtocolCatalogKey(commandCodeTestAccount(22), base+"/", url))
|
||||
require.Equal(t, url+"#account=21", modelProtocolCatalogKey(official, url))
|
||||
require.Equal(t, url+"#account=22", modelProtocolCatalogKey(commandCodeTestAccount(22), url))
|
||||
|
||||
headers := modelProtocolCatalogHeaders(official, url)
|
||||
require.Equal(t, "Bearer user_test_key", headers.Get("Authorization"))
|
||||
@@ -120,12 +205,79 @@ func TestModelProtocolCatalogKeyAndHeaders(t *testing.T) {
|
||||
custom.Credentials["header_overrides"] = map[string]any{"X-Tenant": "t-1"}
|
||||
customBase := custom.GetCNProtocolBaseURL(APIProtocolChatCompletions)
|
||||
customURL := buildOpenAIModelsURL(customBase)
|
||||
require.Equal(t, customURL+"#account=23", modelProtocolCatalogKey(custom, customBase, customURL))
|
||||
require.Equal(t, customURL+"#account=23", modelProtocolCatalogKey(custom, customURL))
|
||||
headers = modelProtocolCatalogHeaders(custom, customURL)
|
||||
require.Equal(t, "t-1", getHeaderRaw(headers, "x-tenant"))
|
||||
require.Empty(t, headers.Get("User-Agent"))
|
||||
}
|
||||
|
||||
type modelCatalogAccountUpstream struct {
|
||||
HTTPUpstream
|
||||
mu sync.Mutex
|
||||
requests []*http.Request
|
||||
failureStatus int
|
||||
failureErr error
|
||||
}
|
||||
|
||||
func (u *modelCatalogAccountUpstream) Do(req *http.Request, _ string, _ int64, _ int) (*http.Response, error) {
|
||||
u.mu.Lock()
|
||||
u.requests = append(u.requests, req)
|
||||
u.mu.Unlock()
|
||||
if req.Header.Get("Authorization") == "Bearer invalid-key" {
|
||||
if u.failureErr != nil {
|
||||
return nil, u.failureErr
|
||||
}
|
||||
return newJSONResponse(u.failureStatus, `{"error":"invalid credentials"}`), nil
|
||||
}
|
||||
endpoint := "/chat/completions"
|
||||
if getHeaderRaw(req.Header, "x-tenant") == "responses-tenant" {
|
||||
endpoint = "/responses"
|
||||
}
|
||||
return newJSONResponse(http.StatusOK, fmt.Sprintf(`{"data":[{"id":"vendor/model","supported_endpoints":[%q]}]}`, endpoint)), nil
|
||||
}
|
||||
|
||||
func TestModelProtocolCatalogOfficialAccountsAreIsolated(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
status int
|
||||
err error
|
||||
}{
|
||||
{name: "unauthorized", status: http.StatusUnauthorized},
|
||||
{name: "forbidden", status: http.StatusForbidden},
|
||||
{name: "transport", err: errors.New("upstream connection failed")},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
upstream := &modelCatalogAccountUpstream{failureStatus: tc.status, failureErr: tc.err}
|
||||
svc := &OpenAIGatewayService{cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream}
|
||||
id := time.Now().UnixNano()
|
||||
bad, chat, responses := commandCodeTestAccount(id), commandCodeTestAccount(id+1), commandCodeTestAccount(id+2)
|
||||
bad.Credentials["api_key"] = "invalid-key"
|
||||
responses.Credentials["header_override_enabled"] = true
|
||||
responses.Credentials["header_overrides"] = map[string]any{"X-Tenant": "responses-tenant"}
|
||||
url := buildOpenAIModelsURL(bad.GetCNProtocolBaseURL(APIProtocolChatCompletions))
|
||||
t.Cleanup(func() {
|
||||
upstreamModelProtocols.mu.Lock()
|
||||
defer upstreamModelProtocols.mu.Unlock()
|
||||
for _, account := range []*Account{bad, chat, responses} {
|
||||
delete(upstreamModelProtocols.entries, modelProtocolCatalogKey(account, url))
|
||||
}
|
||||
})
|
||||
|
||||
require.Nil(t, svc.modelCatalogProtocols(t.Context(), bad, "vendor/model"))
|
||||
require.Equal(t, APIProtocolChatCompletions, svc.resolveUpstreamProtocolFor(t.Context(), chat, APIProtocolResponses, "vendor/model"), "a healthy account fetches its own catalog despite another account's backoff")
|
||||
require.Equal(t, APIProtocolResponses, svc.resolveUpstreamProtocolFor(t.Context(), responses, APIProtocolChatCompletions, "vendor/model"), "successful catalogs stay isolated when headers select different tenants")
|
||||
require.Equal(t, []string{APIProtocolChatCompletions}, svc.modelCatalogProtocols(t.Context(), chat, "vendor/model"))
|
||||
require.Nil(t, svc.modelCatalogProtocols(t.Context(), bad, "vendor/model"))
|
||||
upstream.mu.Lock()
|
||||
defer upstream.mu.Unlock()
|
||||
require.Len(t, upstream.requests, 3, "each healthy account fetches once; the failed account remains in its own backoff")
|
||||
for _, req := range upstream.requests {
|
||||
require.Equal(t, url, req.URL.String(), "all accounts use the same official model-list URL")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestCommandCodeModelProtocolSetUsesCatalog(t *testing.T) {
|
||||
account := commandCodeTestAccount(11)
|
||||
both := []string{APIProtocolChatCompletions, APIProtocolResponses}
|
||||
@@ -190,7 +342,7 @@ func TestCommandCodeGatewayPassesThroughCatalogProtocols(t *testing.T) {
|
||||
APIProtocolAnthropic: base + "/provider",
|
||||
}
|
||||
url := buildOpenAIModelsURL(base + "/provider/v1")
|
||||
key := modelProtocolCatalogKey(account, base+"/provider/v1", url)
|
||||
key := modelProtocolCatalogKey(account, url)
|
||||
if catalog != nil {
|
||||
upstreamModelProtocols.store(key, catalog, nil, time.Now())
|
||||
} else {
|
||||
@@ -252,9 +404,9 @@ func TestCommandCodeGatewayFetchesModelCatalog(t *testing.T) {
|
||||
both := []string{APIProtocolChatCompletions, APIProtocolResponses}
|
||||
|
||||
first, second := newAccount(13), newAccount(14)
|
||||
require.Equal(t, both, svc.modelCatalogProtocols(first, "deepseek/deepseek-v4-flash"))
|
||||
require.Equal(t, both, svc.modelCatalogProtocols(first, "deepseek/deepseek-v4-flash"))
|
||||
require.Equal(t, both, svc.modelCatalogProtocols(second, "deepseek/deepseek-v4-flash"))
|
||||
require.Equal(t, both, svc.modelCatalogProtocols(t.Context(), first, "deepseek/deepseek-v4-flash"))
|
||||
require.Equal(t, both, svc.modelCatalogProtocols(t.Context(), first, "deepseek/deepseek-v4-flash"))
|
||||
require.Equal(t, both, svc.modelCatalogProtocols(t.Context(), second, "deepseek/deepseek-v4-flash"))
|
||||
|
||||
upstream.mu.Lock()
|
||||
defer upstream.mu.Unlock()
|
||||
|
||||
@@ -128,7 +128,7 @@ func (s *OpenAIGatewayService) forwardAsChatCompletions(
|
||||
if account.IsOpenCodeGo() && IsOpenCodeUnsupportedModel(routingModel) {
|
||||
return nil, writeOpenCodeUnsupportedModelError(c, false, routingModel)
|
||||
}
|
||||
switch s.resolveUpstreamProtocolFor(account, inbound, routingModel) {
|
||||
switch s.resolveUpstreamProtocolFor(ctx, account, inbound, routingModel) {
|
||||
case APIProtocolAnthropic:
|
||||
if convertResponsesShape {
|
||||
return s.forwardResponsesViaNativeAnthropic(ctx, c, account, body, "")
|
||||
|
||||
@@ -183,7 +183,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
|
||||
if account.IsOpenCodeGo() && IsOpenCodeUnsupportedModel(routingModel) {
|
||||
return nil, writeOpenCodeUnsupportedModelError(c, false, routingModel)
|
||||
}
|
||||
switch s.resolveUpstreamProtocolFor(account, APIProtocolResponses, routingModel) {
|
||||
switch s.resolveUpstreamProtocolFor(ctx, account, APIProtocolResponses, routingModel) {
|
||||
case APIProtocolAnthropic:
|
||||
// Responses 客户端 × Anthropic 上游:转成 Anthropic 请求走原生端点。不能落到
|
||||
// raw-CC 分支——其 URL 构造会把 anthropic base 当 CC base 用。
|
||||
|
||||
@@ -60,7 +60,7 @@ func (s *OpenAIGatewayService) ForwardAsAnthropic(
|
||||
if account.IsOpenCodeGo() && IsOpenCodeUnsupportedModel(routingModel) {
|
||||
return nil, writeOpenCodeUnsupportedModelError(c, true, routingModel)
|
||||
}
|
||||
switch s.resolveUpstreamProtocolFor(account, APIProtocolAnthropic, routingModel) {
|
||||
switch s.resolveUpstreamProtocolFor(ctx, account, APIProtocolAnthropic, routingModel) {
|
||||
case APIProtocolAnthropic:
|
||||
// 上游为供应商原生 Anthropic 端点:/v1/messages 零转换直通(仅模型名映射 +
|
||||
// 少量 body 清洗),完整保留 thinking / tool_use / cache 语义。
|
||||
|
||||
@@ -0,0 +1,120 @@
|
||||
//go:build unit
|
||||
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestOpenAIGatewayServiceRecordUsage_AggregatorClaudeUsesDefaultPricing(t *testing.T) {
|
||||
for _, platform := range []string{PlatformCommandCode, PlatformCline} {
|
||||
for _, model := range []string{"claude-sonnet-4-6", "anthropic/claude-sonnet-4-6"} {
|
||||
t.Run(platform+"/"+model, func(t *testing.T) {
|
||||
usageRepo := &openAIRecordUsageLogRepoStub{inserted: true}
|
||||
userRepo := &openAIRecordUsageUserRepoStub{}
|
||||
svc := newOpenAIRecordUsageServiceForTest(usageRepo, userRepo, &openAIRecordUsageSubRepoStub{}, nil)
|
||||
svc.resolver = NewModelPricingResolver(nil, svc.billingService)
|
||||
group := &Group{ID: 1, Platform: platform, RateMultiplier: 1.1}
|
||||
usage := OpenAIUsage{InputTokens: 1000, OutputTokens: 200}
|
||||
expected := expectedOpenAICost(t, svc, model, usage, group.RateMultiplier)
|
||||
require.Greater(t, expected.ActualCost, 0.0)
|
||||
|
||||
err := svc.RecordUsage(context.Background(), providerClaudeRecordUsageInput(platform, model, group, usage))
|
||||
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, 1, usageRepo.calls)
|
||||
require.NotNil(t, usageRepo.lastLog)
|
||||
require.Equal(t, usage.InputTokens, usageRepo.lastLog.InputTokens)
|
||||
require.Equal(t, usage.OutputTokens, usageRepo.lastLog.OutputTokens)
|
||||
require.InDelta(t, expected.TotalCost, usageRepo.lastLog.TotalCost, 1e-12)
|
||||
require.InDelta(t, expected.ActualCost, usageRepo.lastLog.ActualCost, 1e-12)
|
||||
require.Equal(t, 1, userRepo.deductCalls)
|
||||
require.InDelta(t, expected.ActualCost, userRepo.lastAmount, 1e-12)
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpenAIGatewayServiceRecordUsage_CNAndOpenCodeClaudeRequireExplicitPricing(t *testing.T) {
|
||||
for _, platform := range []string{PlatformKimi, PlatformZhipu, PlatformDeepseek, PlatformMiniMax, PlatformOpenCodeGo} {
|
||||
t.Run(platform, func(t *testing.T) {
|
||||
usageRepo := &openAIRecordUsageLogRepoStub{inserted: true}
|
||||
userRepo := &openAIRecordUsageUserRepoStub{}
|
||||
svc := newOpenAIRecordUsageServiceForTest(usageRepo, userRepo, &openAIRecordUsageSubRepoStub{}, nil)
|
||||
svc.resolver = NewModelPricingResolver(nil, svc.billingService)
|
||||
group := &Group{ID: 1, Platform: platform, RateMultiplier: 1.1}
|
||||
usage := OpenAIUsage{InputTokens: 1000, OutputTokens: 200}
|
||||
|
||||
err := svc.RecordUsage(context.Background(), providerClaudeRecordUsageInput(platform, "claude-sonnet-4-6", group, usage))
|
||||
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, 1, usageRepo.calls)
|
||||
require.NotNil(t, usageRepo.lastLog)
|
||||
require.Equal(t, usage.InputTokens, usageRepo.lastLog.InputTokens)
|
||||
require.Equal(t, usage.OutputTokens, usageRepo.lastLog.OutputTokens)
|
||||
require.Zero(t, usageRepo.lastLog.TotalCost)
|
||||
require.Zero(t, usageRepo.lastLog.ActualCost)
|
||||
require.Zero(t, userRepo.deductCalls)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpenAIGatewayServiceRecordUsage_CNAndOpenCodeClaudeHonorExplicitPricing(t *testing.T) {
|
||||
const model = "claude-sonnet-4-6"
|
||||
for _, platform := range []string{PlatformKimi, PlatformOpenCodeGo} {
|
||||
for _, source := range []string{PricingSourceGroup, PricingSourceChannel} {
|
||||
t.Run(platform+"/"+source, func(t *testing.T) {
|
||||
usageRepo := &openAIRecordUsageLogRepoStub{inserted: true}
|
||||
userRepo := &openAIRecordUsageUserRepoStub{}
|
||||
svc := newOpenAIRecordUsageServiceForTest(usageRepo, userRepo, &openAIRecordUsageSubRepoStub{}, nil)
|
||||
group := &Group{ID: 1, Platform: platform, RateMultiplier: 1.25}
|
||||
inputPrice, outputPrice := 1e-6, 2e-6
|
||||
pricing := ChannelModelPricing{
|
||||
Models: []string{model}, BillingMode: BillingModeToken,
|
||||
InputPrice: &inputPrice, OutputPrice: &outputPrice,
|
||||
}
|
||||
if source == PricingSourceGroup {
|
||||
group.ModelPricing = []ChannelModelPricing{pricing}
|
||||
} else {
|
||||
cache := newEmptyChannelCache()
|
||||
cache.pricingByGroupModel[channelModelKey{groupID: group.ID, platform: platform, model: model}] = &pricing
|
||||
cache.channelByGroupID[group.ID] = &Channel{ID: 1, Status: StatusActive}
|
||||
cache.groupPlatform[group.ID] = platform
|
||||
cache.loadedAt = time.Now()
|
||||
svc.channelService = &ChannelService{}
|
||||
svc.channelService.cache.Store(cache)
|
||||
}
|
||||
svc.resolver = NewModelPricingResolver(svc.channelService, svc.billingService)
|
||||
usage := OpenAIUsage{InputTokens: 1000, OutputTokens: 200}
|
||||
expectedTotal := float64(usage.InputTokens)*inputPrice + float64(usage.OutputTokens)*outputPrice
|
||||
expectedActual := expectedTotal * group.RateMultiplier
|
||||
|
||||
err := svc.RecordUsage(context.Background(), providerClaudeRecordUsageInput(platform, model, group, usage))
|
||||
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, usageRepo.lastLog)
|
||||
require.InDelta(t, expectedTotal, usageRepo.lastLog.TotalCost, 1e-12)
|
||||
require.InDelta(t, expectedActual, usageRepo.lastLog.ActualCost, 1e-12)
|
||||
require.Equal(t, 1, userRepo.deductCalls)
|
||||
require.InDelta(t, expectedActual, userRepo.lastAmount, 1e-12)
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func providerClaudeRecordUsageInput(platform, model string, group *Group, usage OpenAIUsage) *OpenAIRecordUsageInput {
|
||||
return &OpenAIRecordUsageInput{
|
||||
Result: &OpenAIForwardResult{
|
||||
RequestID: "provider_claude_" + platform,
|
||||
Model: model, BillingModel: model, UpstreamModel: model,
|
||||
Usage: usage, Duration: time.Second,
|
||||
},
|
||||
APIKey: &APIKey{ID: 10, GroupID: &group.ID, Group: group},
|
||||
User: &User{ID: 20},
|
||||
Account: &Account{ID: 30, Platform: platform, Type: AccountTypeAPIKey},
|
||||
}
|
||||
}
|
||||
@@ -930,7 +930,7 @@ func groupMediaPricingLooksIncomplete(group *Group) bool {
|
||||
// 运营者的修复手段是配置账号级 model_mapping(映射到已定价的 CN 模型)或
|
||||
// 分组/渠道显式定价。
|
||||
func (s *OpenAIGatewayService) filterCNProviderBillingModelCandidates(ctx context.Context, account *Account, apiKey *APIKey, candidates []string) []string {
|
||||
if account == nil || !account.IsMultiProtocolAPIKey() {
|
||||
if account == nil || (!account.IsCNProvider() && !account.IsOpenCodeGo()) {
|
||||
return candidates
|
||||
}
|
||||
out := make([]string, 0, len(candidates))
|
||||
|
||||
@@ -7,6 +7,8 @@ import (
|
||||
"regexp"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/tidwall/gjson"
|
||||
)
|
||||
|
||||
// Cline 上游错误的响应式处理。依据官方错误说明(https://docs.cline.bot/api/errors)与
|
||||
@@ -22,8 +24,8 @@ import (
|
||||
// cannot use individual model inference subscriptions":没有可用的 ClinePass 订阅;
|
||||
// - "Free limit reached on model … try again in …":免费模型的每日额度,只限该模型。
|
||||
//
|
||||
// 错误码只出现在响应体里(状态码对应关系文档没有写全),按文案分类;未识别的错误交回
|
||||
// 默认逻辑(普通频率限制走默认 429)。
|
||||
// 只读取结构化 error.code 与错误消息;ClinePass / 免费模型限额兼容官方固定文案。
|
||||
// metadata 等字段可能回显请求内容,不能据此冷却钱包;未识别的错误交回默认逻辑。
|
||||
//
|
||||
// 作用范围:积分与 ClinePass 是同一个 Key 下互不影响的两个钱包,一边超限只冷却该边的
|
||||
// 全部模型(模型级冷却键,见 clineWalletRateLimitKey),不停整个账号;钱包按请求模型
|
||||
@@ -75,24 +77,59 @@ var (
|
||||
clineTryAgainPattern = regexp.MustCompile(`(?i)\btry\s+again\s+in\s+`)
|
||||
)
|
||||
|
||||
func classifyClineError(body []byte) clineErrorKind {
|
||||
text := strings.ToLower(string(body))
|
||||
// parseClineError 返回错误分类与命中的限额消息,后者用于解析窗口和重试时间。
|
||||
func parseClineError(body []byte) (clineErrorKind, string) {
|
||||
if !gjson.ValidBytes(body) {
|
||||
return clineErrorNone, ""
|
||||
}
|
||||
err := gjson.GetBytes(body, "error")
|
||||
if !err.IsObject() {
|
||||
return clineErrorNone, ""
|
||||
}
|
||||
if code := err.Get("code"); code.Type == gjson.String {
|
||||
if kind := classifyClineErrorCode(code.String()); kind != clineErrorNone {
|
||||
return kind, ""
|
||||
}
|
||||
}
|
||||
// 官方客户端也识别 details.code 中的组织花费上限。
|
||||
if code := err.Get("details.code"); code.Type == gjson.String && classifyClineErrorCode(code.String()) == clineSpendLimit {
|
||||
return clineSpendLimit, ""
|
||||
}
|
||||
for _, path := range []string{"message", "details.message"} {
|
||||
message := err.Get(path)
|
||||
if message.Type != gjson.String {
|
||||
continue
|
||||
}
|
||||
if kind := classifyClineLimitMessage(message.String()); kind != clineErrorNone {
|
||||
return kind, message.String()
|
||||
}
|
||||
}
|
||||
return clineErrorNone, ""
|
||||
}
|
||||
|
||||
func classifyClineErrorCode(code string) clineErrorKind {
|
||||
switch strings.ToLower(strings.TrimSpace(code)) {
|
||||
case "insufficient_credits":
|
||||
return clineCreditsExhausted
|
||||
case "spend_limit_exceeded":
|
||||
return clineSpendLimit
|
||||
case "inference_cap_error":
|
||||
return clineInferenceCap
|
||||
default:
|
||||
return clineErrorNone
|
||||
}
|
||||
}
|
||||
|
||||
func classifyClineLimitMessage(message string) clineErrorKind {
|
||||
text := strings.ToLower(message)
|
||||
switch {
|
||||
case strings.Contains(text, "organization accounts cannot use individual model inference subscriptions"),
|
||||
strings.Contains(text, "the user is not subscribed to required model plan"):
|
||||
return clinePassUnavailable
|
||||
case strings.Contains(text, "free limit reached on model"):
|
||||
return clineFreeModelLimit
|
||||
case clinePassLimitPattern.Match(body):
|
||||
case clinePassLimitPattern.MatchString(message):
|
||||
return clinePassLimit
|
||||
case strings.Contains(text, "spend_limit_exceeded"):
|
||||
return clineSpendLimit
|
||||
case strings.Contains(text, "inference_cap_error"):
|
||||
return clineInferenceCap
|
||||
case strings.Contains(text, "insufficient_credits"),
|
||||
strings.Contains(text, "insufficient credits"),
|
||||
cnProviderResponseIndicatesInsufficientBalance(body):
|
||||
return clineCreditsExhausted
|
||||
default:
|
||||
return clineErrorNone
|
||||
}
|
||||
@@ -174,7 +211,7 @@ func (s *RateLimitService) handleClineError(
|
||||
responseBody []byte,
|
||||
upstreamMsg string,
|
||||
) bool {
|
||||
kind := classifyClineError(responseBody)
|
||||
kind, limitMessage := parseClineError(responseBody)
|
||||
if kind == clineErrorNone && statusCode == http.StatusPaymentRequired {
|
||||
// 402 只会来自积分不足(ClinePass 超限走 429 等其他状态码)。
|
||||
kind = clineCreditsExhausted
|
||||
@@ -208,11 +245,11 @@ func (s *RateLimitService) handleClineError(
|
||||
cooldown = clinePassLimitCooldown(account, "5h", now)
|
||||
}
|
||||
case clinePassLimit:
|
||||
cooldown, reason = clinePassLimitCooldown(account, clinePassWindow(responseBody), now), clinePassLimitReason
|
||||
cooldown, reason = clinePassLimitCooldown(account, clinePassWindow([]byte(limitMessage)), now), clinePassLimitReason
|
||||
case clinePassUnavailable:
|
||||
cooldown, reason = clineLimitRecheck, clinePassUnavailableReason
|
||||
case clineFreeModelLimit:
|
||||
cooldown, reason = clineFreeModelResetAfter(responseBody), clineFreeModelLimitReason
|
||||
cooldown, reason = clineFreeModelResetAfter([]byte(limitMessage)), clineFreeModelLimitReason
|
||||
if cooldown <= 0 {
|
||||
cooldown = clineLimitRecheck
|
||||
}
|
||||
|
||||
@@ -0,0 +1,146 @@
|
||||
//go:build unit
|
||||
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/Wei-Shaw/sub2api/internal/config"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestClassifyClineErrorRequiresStructuredFields(t *testing.T) {
|
||||
for name, body := range map[string]string{
|
||||
"code in model message": `{"error":{"code":"invalid_request_error","message":"Model INFERENCE_CAP_ERROR not found"}}`,
|
||||
"code suffix": `{"error":{"code":"model-SPEND_LIMIT_EXCEEDED","message":"Invalid model"}}`,
|
||||
"credits message without code": `{"error":{"message":"Insufficient credits"}}`,
|
||||
"spend message without code": `{"error":{"message":"SPEND_LIMIT_EXCEEDED"}}`,
|
||||
"metadata code": `{"error":{"code":429,"message":"Rate limit exceeded","metadata":{"code":"SPEND_LIMIT_EXCEEDED"}}}`,
|
||||
"metadata pass message": `{"error":{"message":"Invalid model","metadata":{"message":"You have reached your monthly ClinePass limit. Please try again later."}}}`,
|
||||
"metadata free message": `{"error":{"message":"Invalid model","metadata":{"message":"Free limit reached on model x. Try again in 3h 20m"}}}`,
|
||||
"root request code": `{"error":{"message":"Invalid model"},"request":{"model":"insufficient_credits"}}`,
|
||||
"root message": `{"message":"You have reached your monthly ClinePass limit. Please try again later."}`,
|
||||
"error string": `{"error":"SPEND_LIMIT_EXCEEDED"}`,
|
||||
"plain code": `Model SPEND_LIMIT_EXCEEDED not found`,
|
||||
"plain pass message": `You have reached your weekly ClinePass limit. Please try again later.`,
|
||||
"malformed json": `{"error":{"code":"insufficient_credits"}`,
|
||||
} {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
kind, _ := parseClineError([]byte(body))
|
||||
require.Equal(t, clineErrorNone, kind)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleUpstreamError_Cline400DoesNotCoolWallet(t *testing.T) {
|
||||
for name, body := range map[string]string{
|
||||
"credits model echo": `{"error":{"code":"invalid_request_error","message":"Model insufficient_credits not found"}}`,
|
||||
"spend model echo": `{"error":{"message":"Model SPEND_LIMIT_EXCEEDED not found"}}`,
|
||||
"pass model echo": `{"error":{"message":"Model cline-pass/INFERENCE_CAP_ERROR not found"}}`,
|
||||
"metadata echo": `{"error":{"code":400,"message":"Invalid model","metadata":{"model":"SPEND_LIMIT_EXCEEDED"}}}`,
|
||||
"plain echo": `Model SPEND_LIMIT_EXCEEDED not found`,
|
||||
"business code": clineInferenceCapBody,
|
||||
"pass message": clinePassMonthlyBody,
|
||||
} {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
account := clineTestAccount(400)
|
||||
repo := &commandCodeRateLimitRepo{}
|
||||
shouldDisable := NewRateLimitService(repo, nil, &config.Config{}, nil, nil).HandleUpstreamError(
|
||||
context.Background(), account, http.StatusBadRequest, http.Header{}, []byte(body), "cline-pass/INFERENCE_CAP_ERROR")
|
||||
|
||||
require.False(t, shouldDisable)
|
||||
require.Empty(t, repo.modelLimits)
|
||||
require.Zero(t, repo.setErrorCalls)
|
||||
require.Zero(t, repo.tempCalls)
|
||||
require.False(t, account.isModelRateLimitedWithContext(context.Background(), "cline-pass/glm-5.3"))
|
||||
require.False(t, account.isModelRateLimitedWithContext(context.Background(), "anthropic/claude-sonnet-4-6"))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleUpstreamError_Cline429EchoedKeywordsDoNotCoolWallet(t *testing.T) {
|
||||
for name, body := range map[string]string{
|
||||
"message without code": `{"error":{"message":"Model INFERENCE_CAP_ERROR not found"}}`,
|
||||
"metadata model": `{"error":{"code":429,"message":"Rate limit exceeded","metadata":{"model":"SPEND_LIMIT_EXCEEDED"}}}`,
|
||||
"request model": `{"error":{"code":429,"message":"Rate limit exceeded"},"request":{"model":"insufficient_credits"}}`,
|
||||
"plain text": `Model SPEND_LIMIT_EXCEEDED not found`,
|
||||
} {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
repo := &commandCodeRateLimitRepo{}
|
||||
NewRateLimitService(repo, nil, &config.Config{}, nil, nil).HandleUpstreamError(context.Background(),
|
||||
clineTestAccount(429), http.StatusTooManyRequests, http.Header{}, []byte(body), "deepseek/deepseek-v4-flash")
|
||||
require.Empty(t, repo.modelLimits)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleUpstreamError_ClineStructuredErrorsPreserveCooldown(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name, model, body, scope, reason string
|
||||
status int
|
||||
wait time.Duration
|
||||
}{
|
||||
{
|
||||
name: "credits code before echoed metadata", model: "deepseek/deepseek-v4-flash",
|
||||
body: `{"error":{"code":"insufficient_credits","message":"Insufficient credits","metadata":{"model":"SPEND_LIMIT_EXCEEDED"}}}`,
|
||||
scope: clineCreditsRateLimitKey, reason: clineCreditsReason, status: http.StatusPaymentRequired, wait: cnBalanceCheckCooldown(nil),
|
||||
},
|
||||
{
|
||||
name: "spend code before pass wording", model: "deepseek/deepseek-v4-flash",
|
||||
body: `{"error":{"code":"SPEND_LIMIT_EXCEEDED","message":"You have reached your monthly ClinePass limit. Please try again later."}}`,
|
||||
scope: clineCreditsRateLimitKey, reason: clineSpendLimitReason, status: http.StatusTooManyRequests, wait: clineLimitRecheck,
|
||||
},
|
||||
{
|
||||
name: "details spend code", model: "deepseek/deepseek-v4-flash",
|
||||
body: `{"error":{"code":429,"message":"Rate limit exceeded","details":{"code":"SPEND_LIMIT_EXCEEDED"}}}`,
|
||||
scope: clineCreditsRateLimitKey, reason: clineSpendLimitReason, status: http.StatusTooManyRequests, wait: clineLimitRecheck,
|
||||
},
|
||||
{
|
||||
name: "inference code", model: "cline-pass/glm-5.3", body: clineInferenceCapBody,
|
||||
scope: clinePassRateLimitKey, reason: clineInferenceCapReason, status: http.StatusTooManyRequests, wait: clineLimitRecheck,
|
||||
},
|
||||
{
|
||||
name: "pass message ignores metadata window", model: "cline-pass/glm-5.3",
|
||||
body: `{"error":{"code":429,"metadata":{"message":"You have reached your monthly ClinePass limit. Please try again later."},"message":"You have reached your 5-hour ClinePass limit. Please try again later."}}`,
|
||||
scope: clinePassRateLimitKey, reason: clinePassLimitReason, status: http.StatusTooManyRequests, wait: clineLimitRecheck,
|
||||
},
|
||||
{
|
||||
name: "details pass message", model: "cline-pass/glm-5.3",
|
||||
body: `{"error":{"code":429,"message":"Rate limit exceeded","details":{"message":"You have reached your monthly ClinePass limit. Please try again later."}}}`,
|
||||
scope: clinePassRateLimitKey, reason: clinePassLimitReason, status: http.StatusTooManyRequests, wait: clinePassMonthlyRecheck,
|
||||
},
|
||||
{
|
||||
name: "details entitlement message", model: "cline-pass/glm-5.3",
|
||||
body: `{"error":{"code":403,"message":"Forbidden","details":{"message":"The user is not subscribed to required model plan"}}}`,
|
||||
scope: clinePassRateLimitKey, reason: clinePassUnavailableReason, status: http.StatusForbidden, wait: clineLimitRecheck,
|
||||
},
|
||||
{
|
||||
name: "free message ignores metadata retry", model: "cline-free/kimi-k3",
|
||||
body: `{"error":{"code":429,"metadata":{"message":"Try again in 20h"},"message":"Free limit reached on model cline-free/kimi-k3. Try again in 5m"}}`,
|
||||
scope: "cline-free/kimi-k3", reason: clineFreeModelLimitReason, status: http.StatusTooManyRequests, wait: 5 * time.Minute,
|
||||
},
|
||||
{
|
||||
name: "details free message", model: "cline-free/kimi-k3",
|
||||
body: `{"error":{"code":429,"message":"Rate limit exceeded","details":{"message":"Free limit reached on model cline-free/kimi-k3. Try again in 3h 20m"}}}`,
|
||||
scope: "cline-free/kimi-k3", reason: clineFreeModelLimitReason, status: http.StatusTooManyRequests, wait: 3*time.Hour + 20*time.Minute,
|
||||
},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
repo := &commandCodeRateLimitRepo{}
|
||||
account := clineTestAccount(430)
|
||||
shouldDisable := NewRateLimitService(repo, nil, &config.Config{}, nil, nil).HandleUpstreamError(
|
||||
context.Background(), account, tc.status, http.Header{}, []byte(tc.body), tc.model)
|
||||
|
||||
require.False(t, shouldDisable)
|
||||
require.Len(t, repo.modelLimits, 1)
|
||||
require.WithinDuration(t, time.Now().Add(tc.wait), repo.modelLimits[tc.scope], time.Second)
|
||||
require.True(t, strings.HasPrefix(repo.reasons[tc.scope], tc.reason), repo.reasons[tc.scope])
|
||||
require.Zero(t, repo.setErrorCalls)
|
||||
require.Zero(t, repo.tempCalls)
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -407,9 +407,9 @@ func (s *RateLimitService) HandleUpstreamError(ctx context.Context, account *Acc
|
||||
return disable
|
||||
}
|
||||
}
|
||||
// Cline 积分、花费上限与 ClinePass 超限同样按文案区分,只冷却对应的钱包
|
||||
// (见 ratelimit_cline.go)。
|
||||
if account.IsCline() && (statusCode == http.StatusBadRequest || statusCode == http.StatusPaymentRequired ||
|
||||
// Cline 积分、花费上限与 ClinePass 超限按结构化错误区分,只冷却对应的钱包。
|
||||
// 400 是请求错误,不能因回显的模型名或请求内容触发钱包冷却。
|
||||
if account.IsCline() && (statusCode == http.StatusPaymentRequired ||
|
||||
statusCode == http.StatusForbidden || statusCode == http.StatusTooManyRequests) {
|
||||
if s.handleClineError(ctx, account, statusCode, responseBody, upstreamMsg) {
|
||||
return false
|
||||
|
||||
Reference in New Issue
Block a user