Merge remote-tracking branch 'pr/2131' into release/v0.1.133
# Conflicts: # backend/cmd/server/wire_gen.go # backend/internal/config/config.go # backend/internal/service/gateway_service.go # backend/internal/service/pricing_service.go # backend/internal/service/wire.go # deploy/config.example.yaml # frontend/src/views/admin/AccountsView.vue
This commit is contained in:
kone
@@ -27,6 +27,7 @@ import (
|
||||
"github.com/Wei-Shaw/sub2api/internal/config"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/claude"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/ctxkey"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/kirocooldown"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/logger"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/usagestats"
|
||||
"github.com/Wei-Shaw/sub2api/internal/util/responseheaders"
|
||||
@@ -56,6 +57,7 @@ const (
|
||||
defaultModelsListCacheTTL = 15 * time.Second
|
||||
postUsageBillingTimeout = 15 * time.Second
|
||||
debugGatewayBodyEnv = "SUB2API_DEBUG_GATEWAY_BODY"
|
||||
defaultKiroStreamKeepalive = 25 * time.Second
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -70,6 +72,7 @@ const (
|
||||
// ForceCacheBillingContextKey 强制缓存计费上下文键
|
||||
// 用于粘性会话切换时，将 input_tokens 转为 cache_read_input_tokens 计费
|
||||
type forceCacheBillingKeyType struct{}
|
||||
type kiroCooldownRecoveryAttemptedKeyType struct{}
|
||||
|
||||
// accountWithLoad 账号与负载信息的组合，用于负载感知调度
|
||||
type accountWithLoad struct {
|
||||
@@ -78,6 +81,7 @@ type accountWithLoad struct {
|
||||
}
|
||||
|
||||
var ForceCacheBillingContextKey = forceCacheBillingKeyType{}
|
||||
var kiroCooldownRecoveryAttemptedKey = kiroCooldownRecoveryAttemptedKeyType{}
|
||||
|
||||
var (
|
||||
windowCostPrefetchCacheHitTotal atomic.Int64
|
||||
@@ -554,6 +558,8 @@ type GatewayService struct {
|
||||
deferredService *DeferredService
|
||||
concurrencyService *ConcurrencyService
|
||||
claudeTokenProvider *ClaudeTokenProvider
|
||||
kiroTokenProvider *KiroTokenProvider
|
||||
kiroCooldownStore KiroCooldownStore
|
||||
sessionLimitCache SessionLimitCache // 会话数量限制缓存（仅 Anthropic OAuth/SetupToken）
|
||||
rpmCache RPMCache // RPM 计数缓存（仅 Anthropic OAuth/SetupToken）
|
||||
userGroupRateResolver *userGroupRateResolver
|
||||
@@ -592,6 +598,8 @@ func NewGatewayService(
|
||||
httpUpstream HTTPUpstream,
|
||||
deferredService *DeferredService,
|
||||
claudeTokenProvider *ClaudeTokenProvider,
|
||||
kiroTokenProvider *KiroTokenProvider,
|
||||
kiroCooldownStore KiroCooldownStore,
|
||||
sessionLimitCache SessionLimitCache,
|
||||
rpmCache RPMCache,
|
||||
digestStore *DigestSessionStore,
|
||||
@@ -624,6 +632,8 @@ func NewGatewayService(
|
||||
httpUpstream: httpUpstream,
|
||||
deferredService: deferredService,
|
||||
claudeTokenProvider: claudeTokenProvider,
|
||||
kiroTokenProvider: kiroTokenProvider,
|
||||
kiroCooldownStore: kiroCooldownStore,
|
||||
sessionLimitCache: sessionLimitCache,
|
||||
rpmCache: rpmCache,
|
||||
userGroupRateCache: gocache.New(userGroupRateTTL, time.Minute),
|
||||
@@ -902,6 +912,7 @@ type claudeOAuthNormalizeOptions struct {
|
||||
injectMetadata bool
|
||||
metadataUserID string
|
||||
stripSystemCacheControl bool
|
||||
preserveToolChoice bool
|
||||
}
|
||||
|
||||
// sanitizeSystemText rewrites only the fixed OpenCode identity sentence (if present).
|
||||
@@ -1116,6 +1127,12 @@ func normalizeClaudeOAuthRequestBody(body []byte, modelID string, opts claudeOAu
|
||||
modified = true
|
||||
}
|
||||
}
|
||||
if !opts.preserveToolChoice && gjson.GetBytes(out, "tool_choice").Exists() {
|
||||
if next, ok := deleteJSONPathBytes(out, "tool_choice"); ok {
|
||||
out = next
|
||||
modified = true
|
||||
}
|
||||
}
|
||||
|
||||
// max_tokens：真实 CLI 的默认值是 128000。缺失时补齐以对齐指纹。
|
||||
if !gjson.GetBytes(out, "max_tokens").Exists() {
|
||||
@@ -1967,6 +1984,10 @@ func (s *GatewayService) SelectAccountWithLoadAwareness(ctx context.Context, gro
|
||||
}
|
||||
|
||||
if len(candidates) == 0 {
|
||||
if s.tryRecoverKiroCooldownPool(ctx, accounts, requestedModel, excludedIDs, useMixed) {
|
||||
retryCtx := context.WithValue(ctx, kiroCooldownRecoveryAttemptedKey, true)
|
||||
return s.SelectAccountWithLoadAwareness(retryCtx, groupID, sessionHash, requestedModel, excludedIDs, metadataUserID, sub2apiUserID)
|
||||
}
|
||||
return nil, ErrNoAvailableAccounts
|
||||
}
|
||||
|
||||
@@ -2346,14 +2367,91 @@ func (s *GatewayService) isAccountSchedulableForSelection(account *Account) bool
|
||||
if account == nil {
|
||||
return false
|
||||
}
|
||||
return account.IsSchedulable()
|
||||
if !account.IsSchedulable() {
|
||||
return false
|
||||
}
|
||||
return s.isKiroRuntimeSchedulable(context.Background(), account)
|
||||
}
|
||||
|
||||
func (s *GatewayService) isAccountSchedulableForModelSelection(ctx context.Context, account *Account, requestedModel string) bool {
|
||||
if account == nil {
|
||||
return false
|
||||
}
|
||||
return account.IsSchedulableForModelWithContext(ctx, requestedModel)
|
||||
if !account.IsSchedulableForModelWithContext(ctx, requestedModel) {
|
||||
return false
|
||||
}
|
||||
return s.isKiroRuntimeSchedulable(ctx, account)
|
||||
}
|
||||
|
||||
func (s *GatewayService) isKiroRuntimeSchedulable(ctx context.Context, account *Account) bool {
|
||||
if account == nil || account.Platform != PlatformKiro || account.Type != AccountTypeOAuth || s == nil || s.kiroCooldownStore == nil {
|
||||
return true
|
||||
}
|
||||
state, err := s.getKiroCooldownState(ctx, buildKiroAccountKey(account))
|
||||
if err != nil {
|
||||
return true
|
||||
}
|
||||
return state == nil || !state.Active
|
||||
}
|
||||
|
||||
func (s *GatewayService) tryRecoverKiroCooldownPool(ctx context.Context, accounts []Account, requestedModel string, excludedIDs map[int64]struct{}, allowMixedScheduling bool) bool {
|
||||
if s == nil || s.kiroCooldownStore == nil || ctx.Value(kiroCooldownRecoveryAttemptedKey) == true {
|
||||
return false
|
||||
}
|
||||
tokenKeys := s.kiroTransientCooldownRecoveryKeys(ctx, accounts, requestedModel, excludedIDs, allowMixedScheduling)
|
||||
if len(tokenKeys) == 0 {
|
||||
return false
|
||||
}
|
||||
cleared, err := s.kiroCooldownStore.ClearEarliestTransientCooldown(ctx, tokenKeys)
|
||||
if err != nil {
|
||||
logger.LegacyPrintf("service.gateway", "Kiro cooldown pool recovery failed: %v", err)
|
||||
return false
|
||||
}
|
||||
if cleared {
|
||||
logger.LegacyPrintf("service.gateway", "Kiro cooldown pool recovery cleared one transient cooldown")
|
||||
}
|
||||
return cleared
|
||||
}
|
||||
|
||||
func (s *GatewayService) kiroTransientCooldownRecoveryKeys(ctx context.Context, accounts []Account, requestedModel string, excludedIDs map[int64]struct{}, allowMixedScheduling bool) []string {
|
||||
tokenKeys := make([]string, 0, len(accounts))
|
||||
eligible := 0
|
||||
for i := range accounts {
|
||||
acc := &accounts[i]
|
||||
if acc == nil || acc.Platform != PlatformKiro || acc.Type != AccountTypeOAuth {
|
||||
if allowMixedScheduling {
|
||||
continue
|
||||
}
|
||||
return nil
|
||||
}
|
||||
if _, excluded := excludedIDs[acc.ID]; excluded {
|
||||
continue
|
||||
}
|
||||
if !acc.IsSchedulable() {
|
||||
continue
|
||||
}
|
||||
if requestedModel != "" && !s.isModelSupportedByAccountWithContext(ctx, acc, requestedModel) {
|
||||
continue
|
||||
}
|
||||
if !s.isAccountSchedulableForQuota(acc) ||
|
||||
!s.isAccountSchedulableForWindowCost(ctx, acc, false) ||
|
||||
!s.isAccountSchedulableForRPM(ctx, acc, false) {
|
||||
continue
|
||||
}
|
||||
eligible++
|
||||
state, err := s.getKiroCooldownState(ctx, buildKiroAccountKey(acc))
|
||||
if err != nil || state == nil || !state.Active {
|
||||
return nil
|
||||
}
|
||||
if state.Reason != kirocooldown.CooldownReason429 {
|
||||
return nil
|
||||
}
|
||||
tokenKeys = append(tokenKeys, buildKiroAccountKey(acc))
|
||||
}
|
||||
if eligible == 0 || len(tokenKeys) != eligible {
|
||||
return nil
|
||||
}
|
||||
return tokenKeys
|
||||
}
|
||||
|
||||
// isAccountInGroup checks if the account belongs to the specified group.
|
||||
@@ -3232,6 +3330,10 @@ func (s *GatewayService) selectAccountForModelWithPlatform(ctx context.Context,
|
||||
|
||||
if selected == nil {
|
||||
stats := s.logDetailedSelectionFailure(ctx, groupID, sessionHash, requestedModel, platform, accounts, excludedIDs, false)
|
||||
if s.tryRecoverKiroCooldownPool(ctx, accounts, requestedModel, excludedIDs, false) {
|
||||
retryCtx := context.WithValue(ctx, kiroCooldownRecoveryAttemptedKey, true)
|
||||
return s.selectAccountForModelWithPlatform(retryCtx, groupID, sessionHash, requestedModel, excludedIDs, platform)
|
||||
}
|
||||
if requestedModel != "" {
|
||||
return nil, fmt.Errorf("%w supporting model: %s (%s)", ErrNoAvailableAccounts, requestedModel, summarizeSelectionFailureStats(stats))
|
||||
}
|
||||
@@ -3611,6 +3713,17 @@ func (s *GatewayService) diagnoseSelectionFailure(
|
||||
if _, excluded := excludedIDs[acc.ID]; excluded {
|
||||
return selectionFailureDiagnosis{Category: "excluded"}
|
||||
}
|
||||
if !acc.IsSchedulable() {
|
||||
return selectionFailureDiagnosis{Category: "unschedulable", Detail: "generic_unschedulable"}
|
||||
}
|
||||
if acc.Platform == PlatformKiro && acc.Type == AccountTypeOAuth {
|
||||
if state, err := s.getKiroCooldownState(ctx, buildKiroAccountKey(acc)); err == nil && state != nil && state.Active {
|
||||
return selectionFailureDiagnosis{
|
||||
Category: "unschedulable",
|
||||
Detail: fmt.Sprintf("kiro_runtime_%s remaining=%s", state.Reason, state.Remaining.Truncate(time.Second)),
|
||||
}
|
||||
}
|
||||
}
|
||||
if !s.isAccountSchedulableForSelection(acc) {
|
||||
return selectionFailureDiagnosis{Category: "unschedulable", Detail: "generic_unschedulable"}
|
||||
}
|
||||
@@ -3774,6 +3887,13 @@ func (s *GatewayService) getOAuthToken(ctx context.Context, account *Account) (s
|
||||
}
|
||||
return accessToken, "oauth", nil
|
||||
}
|
||||
if account.Platform == PlatformKiro && account.Type == AccountTypeOAuth && s.kiroTokenProvider != nil {
|
||||
accessToken, err := s.kiroTokenProvider.GetAccessToken(ctx, account)
|
||||
if err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
return accessToken, "oauth", nil
|
||||
}
|
||||
|
||||
// 其他情况（Gemini 有自己的 TokenProvider，setup-token 类型等）直接从账号读取
|
||||
accessToken := account.GetCredential("access_token")
|
||||
@@ -4343,11 +4463,6 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A
|
||||
return nil, fmt.Errorf("parse request: empty request")
|
||||
}
|
||||
|
||||
// Web Search 模拟：纯 web_search 请求时，直接调用搜索 API 构造响应
|
||||
if account != nil && s.shouldEmulateWebSearch(ctx, account, parsed.GroupID, parsed.Body) {
|
||||
return s.handleWebSearchEmulation(ctx, c, account, parsed)
|
||||
}
|
||||
|
||||
if account != nil && account.IsAnthropicAPIKeyPassthroughEnabled() {
|
||||
passthroughBody := parsed.Body
|
||||
passthroughModel := parsed.Model
|
||||
@@ -4371,6 +4486,15 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A
|
||||
return s.forwardBedrock(ctx, c, account, parsed, startTime)
|
||||
}
|
||||
|
||||
if account != nil && account.Platform == PlatformKiro && account.Type == AccountTypeOAuth {
|
||||
return s.forwardKiroMessages(ctx, c, account, parsed, startTime)
|
||||
}
|
||||
|
||||
// Web Search 模拟：纯 web_search 请求时，直接调用搜索 API 构造响应
|
||||
if account != nil && s.shouldEmulateWebSearch(ctx, account, parsed.GroupID, parsed.Body) {
|
||||
return s.handleWebSearchEmulation(ctx, c, account, parsed)
|
||||
}
|
||||
|
||||
// Beta policy: evaluate once; block check + cache filter set for buildUpstreamRequest.
|
||||
// Always overwrite the cache to prevent stale values from a previous retry with a different account.
|
||||
if account.Platform == PlatformAnthropic && c != nil {
|
||||
@@ -4425,7 +4549,10 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A
|
||||
// system 被重写时保留 CC prompt 的 cache_control: ephemeral（匹配真实 Claude Code 行为）；
|
||||
// 未重写时（haiku / 已含 CC 前缀）剥离客户端 cache_control，与原有行为一致。
|
||||
// 两种情况下 enforceCacheControlLimit 都会兜底处理上限。
|
||||
normalizeOpts := claudeOAuthNormalizeOptions{stripSystemCacheControl: !systemRewritten}
|
||||
normalizeOpts := claudeOAuthNormalizeOptions{
|
||||
stripSystemCacheControl: !systemRewritten,
|
||||
preserveToolChoice: account.Platform == PlatformKiro,
|
||||
}
|
||||
if s.identityService != nil {
|
||||
fp, err := s.identityService.GetOrCreateFingerprint(ctx, account.ID, c.Request.Header)
|
||||
if err == nil && fp != nil {
|
||||
@@ -4462,7 +4589,12 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A
|
||||
// - OAuth/SetupToken 账号：使用 Anthropic 标准映射（短ID → 长ID）
|
||||
mappedModel := reqModel
|
||||
mappingSource := ""
|
||||
if account.Type == AccountTypeAPIKey {
|
||||
if account.Platform == PlatformKiro {
|
||||
if next := account.GetMappedModel(reqModel); next != "" && next != reqModel {
|
||||
mappedModel = next
|
||||
mappingSource = "account"
|
||||
}
|
||||
} else if account.Type == AccountTypeAPIKey {
|
||||
mappedModel = account.GetMappedModel(reqModel)
|
||||
if mappedModel != reqModel {
|
||||
mappingSource = "account"
|
||||
@@ -5967,6 +6099,9 @@ func (s *GatewayService) buildUpstreamRequest(ctx context.Context, c *gin.Contex
|
||||
targetURL := claudeAPIURL
|
||||
if account.Type == AccountTypeAPIKey {
|
||||
baseURL := account.GetBaseURL()
|
||||
if baseURL == "" && account.Platform == PlatformKiro {
|
||||
return nil, fmt.Errorf("kiro api key account requires base_url")
|
||||
}
|
||||
if baseURL != "" {
|
||||
validatedURL, err := s.validateUpstreamBaseURL(baseURL)
|
||||
if err != nil {
|
||||
@@ -7228,10 +7363,7 @@ func (s *GatewayService) handleStreamingResponse(ctx context.Context, resp *http
|
||||
}
|
||||
|
||||
// 下游 keepalive：防止代理/Cloudflare Tunnel 因连接空闲而断开
|
||||
keepaliveInterval := time.Duration(0)
|
||||
if s.cfg != nil && s.cfg.Gateway.StreamKeepaliveInterval > 0 {
|
||||
keepaliveInterval = time.Duration(s.cfg.Gateway.StreamKeepaliveInterval) * time.Second
|
||||
}
|
||||
keepaliveInterval := s.streamKeepaliveIntervalForAccount(account)
|
||||
var keepaliveTicker *time.Ticker
|
||||
if keepaliveInterval > 0 {
|
||||
keepaliveTicker = time.NewTicker(keepaliveInterval)
|
||||
@@ -8277,6 +8409,9 @@ type recordUsageOpts struct {
|
||||
// 长上下文计费（仅 Gemini 路径需要）
|
||||
LongContextThreshold int
|
||||
LongContextMultiplier float64
|
||||
|
||||
// Kiro 账号在上游返回 auto 等无法定价模型时使用保守计费兜底。
|
||||
IsKiroAccount bool
|
||||
}
|
||||
|
||||
// RecordUsage 记录使用量并扣费（或更新订阅用量）
|
||||
@@ -8414,6 +8549,7 @@ func (s *GatewayService) recordUsageCore(ctx context.Context, input *recordUsage
|
||||
}
|
||||
|
||||
// 计算费用
|
||||
opts.IsKiroAccount = account != nil && account.Platform == PlatformKiro
|
||||
cost := s.calculateRecordUsageCost(ctx, result, apiKey, billingModel, multiplier, imageMultiplier, opts)
|
||||
|
||||
// 判断计费方式：订阅模式 vs 余额模式
|
||||
@@ -8492,6 +8628,28 @@ func (s *GatewayService) calculateRecordUsageCost(
|
||||
return s.calculateTokenCost(ctx, result, apiKey, billingModel, multiplier, opts)
|
||||
}
|
||||
|
||||
const kiroConservativeFallbackBillingModel = "claude-opus-4-6"
|
||||
|
||||
func shouldUseKiroConservativeBillingFallback(result *ForwardResult, billingModel string, opts *recordUsageOpts) bool {
|
||||
if result == nil {
|
||||
return false
|
||||
}
|
||||
|
||||
return opts != nil && opts.IsKiroAccount
|
||||
}
|
||||
|
||||
func (s *GatewayService) calculateKiroConservativeTokenCost(tokens UsageTokens, multiplier float64) *CostBreakdown {
|
||||
if s == nil || s.billingService == nil {
|
||||
return nil
|
||||
}
|
||||
cost, err := s.billingService.CalculateCost(kiroConservativeFallbackBillingModel, tokens, multiplier)
|
||||
if err != nil {
|
||||
logger.LegacyPrintf("service.gateway", "Calculate conservative Kiro fallback cost failed: %v", err)
|
||||
return nil
|
||||
}
|
||||
return cost
|
||||
}
|
||||
|
||||
// resolveChannelPricing 检查指定模型是否存在渠道级别定价。
|
||||
// 返回非 nil 的 ResolvedPricing 表示有渠道定价，nil 表示走默认定价路径。
|
||||
func (s *GatewayService) resolveChannelPricing(ctx context.Context, billingModel string, apiKey *APIKey) *ResolvedPricing {
|
||||
@@ -8596,6 +8754,12 @@ func (s *GatewayService) calculateTokenCost(
|
||||
}
|
||||
if err != nil {
|
||||
logger.LegacyPrintf("service.gateway", "Calculate cost failed: %v", err)
|
||||
if shouldUseKiroConservativeBillingFallback(result, billingModel, opts) {
|
||||
if fallback := s.calculateKiroConservativeTokenCost(tokens, multiplier); fallback != nil {
|
||||
logger.LegacyPrintf("service.gateway", "Using conservative Kiro fallback pricing for model=%s", billingModel)
|
||||
return fallback
|
||||
}
|
||||
}
|
||||
return &CostBreakdown{ActualCost: 0}
|
||||
}
|
||||
return cost
|
||||
@@ -8856,6 +9020,10 @@ func (s *GatewayService) ForwardCountTokens(ctx context.Context, c *gin.Context,
|
||||
s.countTokensError(c, http.StatusNotFound, "not_found_error", "count_tokens endpoint is not supported for this platform")
|
||||
return nil
|
||||
}
|
||||
if account.Platform == PlatformKiro && account.Type == AccountTypeOAuth {
|
||||
s.countTokensError(c, http.StatusNotFound, "not_found_error", "Token counting is not supported for this platform")
|
||||
return nil
|
||||
}
|
||||
|
||||
// 应用模型映射：
|
||||
// - APIKey 账号：使用账号级别的显式映射（如果配置），否则透传原始模型名
|
||||
@@ -9486,6 +9654,19 @@ func reconcileCachedTokens(usage map[string]any) bool {
|
||||
return true
|
||||
}
|
||||
|
||||
func (s *GatewayService) streamKeepaliveIntervalForAccount(account *Account) time.Duration {
|
||||
if account != nil && account.Platform == PlatformKiro {
|
||||
if s != nil && s.cfg != nil && s.cfg.Gateway.KiroStreamKeepaliveInterval > 0 {
|
||||
return time.Duration(s.cfg.Gateway.KiroStreamKeepaliveInterval) * time.Second
|
||||
}
|
||||
return defaultKiroStreamKeepalive
|
||||
}
|
||||
if s != nil && s.cfg != nil && s.cfg.Gateway.StreamKeepaliveInterval > 0 {
|
||||
return time.Duration(s.cfg.Gateway.StreamKeepaliveInterval) * time.Second
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
const debugGatewayBodyDefaultFilename = "gateway_debug.log"
|
||||
|
||||
// initDebugGatewayBodyFile 初始化网关调试日志文件。
|
||||
|
||||
Reference in New Issue
Block a user