From 7a814682825941627ae446af7713f2523e44a711 Mon Sep 17 00:00:00 2001 From: IanShaw027 Date: Fri, 7 Aug 2026 17:15:43 +0800 Subject: [PATCH] =?UTF-8?q?fix(grok):=20=E6=8C=89=20review=20=E4=BC=98?= =?UTF-8?q?=E5=85=88=E7=BA=A7=E4=BF=AE=E5=A4=8D=E8=AE=A1=E8=B4=B9=E6=BC=8F?= =?UTF-8?q?=E6=89=A3=E3=80=81auth=20=E6=8A=95=E5=BD=B1=E4=B8=8E=20free=20?= =?UTF-8?q?=E9=97=A8=E7=A6=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P0: content 与 status 共用 claim 计费；稳定 grok-video request_id； auth 热路径投影 video_model_prices。 P1: claim 失败释放可重试；视频绑定 TTL≥24h；SSO 凭据白名单与脱敏； Token URL 校验；free soft-gate 默认 1M 并统一 free 判定； Voice 预检余额；STT 抗低报；Realtime 失败不计费；search 未定价告警。 --- backend/internal/config/config.go | 3 +- backend/internal/config/config_test.go | 2 +- .../handler/admin/grok_oauth_handler.go | 28 +++++++- .../internal/handler/gateway_web_search.go | 10 ++- backend/internal/handler/grok_audio.go | 26 ++++++-- backend/internal/handler/grok_media.go | 65 ++++++++++++++----- .../handler/openai_gateway_handler.go | 3 +- backend/internal/repository/api_key_repo.go | 1 + backend/internal/repository/gateway_cache.go | 11 ++++ .../internal/repository/grok_oauth_client.go | 8 ++- .../service/account_credentials_redact.go | 2 + .../internal/service/api_key_auth_cache.go | 17 ++--- .../service/api_key_auth_cache_impl.go | 18 ++--- .../gateway_hotpath_optimization_test.go | 5 ++ .../service/gateway_multiplatform_test.go | 5 ++ backend/internal/service/gateway_service.go | 2 + .../service/gemini_multiplatform_test.go | 5 ++ backend/internal/service/grok_audio.go | 38 +++++++++-- .../internal/service/grok_free_quota_gate.go | 28 +++----- backend/internal/service/grok_media.go | 58 +++++++++++++++-- .../service/openai_account_scheduler_test.go | 5 ++ .../openai_cyber_session_block_test.go | 5 ++ .../service/openai_gateway_grok_cache.go | 19 ++++-- .../service/openai_gateway_service_test.go | 5 ++ .../internal/service/openai_gateway_usage.go | 12 ++++ .../service/openai_ws_state_store_test.go | 5 ++ backend/internal/testutil/stubs.go | 5 ++ deploy/config.example.yaml | 2 +- 28 files changed, 310 insertions(+), 83 deletions(-) diff --git a/backend/internal/config/config.go b/backend/internal/config/config.go index ad9fe6c021..186e59d5b5 100644 --- a/backend/internal/config/config.go +++ b/backend/internal/config/config.go @@ -2343,7 +2343,8 @@ func setDefaults() { // Enabled by default because free detection requires an explicit free tier marker. viper.SetDefault("gateway.grok.free_quota_soft_gate_enabled", true) viper.SetDefault("gateway.grok.password_auth_enabled", false) - viper.SetDefault("gateway.grok.free_quota_token_limit", int64(2_000_000)) + // Align soft-gate nominal limit with current Free rolling quota (1M tokens / 24h). +viper.SetDefault("gateway.grok.free_quota_token_limit", int64(1_000_000)) viper.SetDefault("gateway.grok.free_quota_soft_gate_percent", 95) viper.SetDefault("gateway.grok.free_quota_window_hours", 24) viper.SetDefault("gateway.grok.free_quota_stats_cache_seconds", 5) diff --git a/backend/internal/config/config_test.go b/backend/internal/config/config_test.go index f6a4891f80..037c5cd135 100644 --- a/backend/internal/config/config_test.go +++ b/backend/internal/config/config_test.go @@ -545,7 +545,7 @@ func TestLoadDefaultGrokFreeQuotaSoftGate(t *testing.T) { require.NoError(t, err) require.False(t, cfg.Gateway.Grok.PasswordAuthEnabled) require.True(t, cfg.Gateway.Grok.FreeQuotaSoftGateEnabled) - require.Equal(t, int64(2_000_000), cfg.Gateway.Grok.FreeQuotaTokenLimit) + require.Equal(t, int64(1_000_000), cfg.Gateway.Grok.FreeQuotaTokenLimit) require.Equal(t, 95, cfg.Gateway.Grok.FreeQuotaSoftGatePercent) require.Equal(t, 24, cfg.Gateway.Grok.FreeQuotaWindowHours) require.Equal(t, 5, cfg.Gateway.Grok.FreeQuotaStatsCacheSeconds) diff --git a/backend/internal/handler/admin/grok_oauth_handler.go b/backend/internal/handler/admin/grok_oauth_handler.go index 64bcbf9115..7212b62e92 100644 --- a/backend/internal/handler/admin/grok_oauth_handler.go +++ b/backend/internal/handler/admin/grok_oauth_handler.go @@ -459,7 +459,33 @@ func (h *GrokOAuthHandler) createAccountFromSSOToken(ctx context.Context, req Gr // 配置且 Build 恒写官方地址，会吞掉导入时指定的自定义转发地址——与 // RefreshAccountToken 的保留逻辑对齐，请求显式提供时以请求为准。 func grokSSOImportCredentials(built map[string]any, reqCredentials map[string]any) map[string]any { - credentials := service.MergeCredentials(cloneGrokSSOMap(reqCredentials), built) + // Only merge operator config from the request — never free-form secrets + // (password / sso_token / cookie / etc.) into stored credentials. + allowedReqKeys := map[string]struct{}{ + "base_url": {}, "model_mapping": {}, "header_override": {}, + "header_overrides": {}, "custom_headers": {}, + } + ops := map[string]any{} + for k, v := range reqCredentials { + if _, ok := allowedReqKeys[k]; !ok { + continue + } + if service.IsSensitiveCredentialKey(k) { + continue + } + ops[k] = v + } + credentials := service.MergeCredentials(ops, built) + // Strip any sensitive keys that might have slipped in via older callers. + for k := range credentials { + if service.IsSensitiveCredentialKey(k) { + // Keep only keys produced by BuildAccountCredentials (tokens). + if k == "access_token" || k == "refresh_token" || k == "id_token" { + continue + } + delete(credentials, k) + } + } if reqBaseURL, ok := reqCredentials["base_url"].(string); ok && strings.TrimSpace(reqBaseURL) != "" { credentials["base_url"] = strings.TrimSpace(reqBaseURL) } diff --git a/backend/internal/handler/gateway_web_search.go b/backend/internal/handler/gateway_web_search.go index d5583bf444..039a9c2487 100644 --- a/backend/internal/handler/gateway_web_search.go +++ b/backend/internal/handler/gateway_web_search.go @@ -169,10 +169,18 @@ func (h *GatewayHandler) WebSearch(c *gin.Context) { upstreamEndpoint := GetUpstreamEndpoint(c, account.Platform) requestPayloadHash := service.HashUsageRequestPayload([]byte(req.Query)) quotaPlatform := service.QuotaPlatform(c.Request.Context(), apiKey) + // Unique per invocation (not hash(query) alone) so identical queries still bill. + searchRequestID := "web_search:" + service.HashUsageRequestPayload([]byte(req.Query+"|"+clientIP+"|"+userAgent)) + if apiKey.Group != nil && (apiKey.Group.GetSearchPricePer1k() == nil || *apiKey.Group.GetSearchPricePer1k() <= 0) { + logger.L().With( + zap.String("component", "handler.gateway.web_search"), + zap.Int64("group_id", apiKey.Group.ID), + ).Warn("gateway.web_search.search_price_per_1k_unset_free") + } h.submitUsageRecordTask(c.Request.Context(), func(ctx context.Context) { if err := h.gatewayService.RecordUsage(ctx, &service.RecordUsageInput{ Result: &service.ForwardResult{ - RequestID: "web_search:" + service.HashUsageRequestPayload([]byte(req.Query)), + RequestID: searchRequestID, Model: "grok-web-search", SearchCount: 1, Duration: 0, diff --git a/backend/internal/handler/grok_audio.go b/backend/internal/handler/grok_audio.go index 8f1c0c45cf..677bda053e 100644 --- a/backend/internal/handler/grok_audio.go +++ b/backend/internal/handler/grok_audio.go @@ -5,6 +5,7 @@ import ( "errors" "io" "net/http" + "strconv" "strings" "time" @@ -32,6 +33,15 @@ func (h *OpenAIGatewayHandler) GrokRealtime(c *gin.Context) { if !h.ensureResponsesDependencies(c, nil) { return } + subscription, _ := middleware2.GetSubscriptionFromContext(c) + if err := h.billingCacheService.CheckBillingEligibility(c.Request.Context(), apiKey.User, apiKey, apiKey.Group, subscription, service.QuotaPlatform(c.Request.Context(), apiKey)); err != nil { + status, code, message, retryAfter := billingErrorDetails(err) + if retryAfter > 0 { + c.Header("Retry-After", strconv.Itoa(retryAfter)) + } + h.errorResponse(c, status, code, message) + return + } selection, _, err := h.gatewayService.SelectAccountWithSchedulerForCapability( c.Request.Context(), @@ -83,11 +93,11 @@ func (h *OpenAIGatewayHandler) GrokRealtime(c *gin.Context) { if proxyErr != nil { reqLog.Info("grok_realtime.proxy_failed", zap.Error(proxyErr)) _ = conn.Close(coderws.StatusInternalError, "upstream realtime websocket failed") + // Do not bill failed realtime sessions (avoids charging idle/error wall-clock). + return } - // Bill completed (or partially completed) realtime sessions by elapsed minutes - // when group audio_realtime_price_per_min is configured. + // Bill successful realtime sessions by elapsed minutes when group price is set. if elapsed > 0 { - subscription, _ := middleware2.GetSubscriptionFromContext(c) result := &service.OpenAIForwardResult{ Model: model, Duration: elapsed, @@ -107,6 +117,15 @@ func (h *OpenAIGatewayHandler) GrokVoice(c *gin.Context, endpoint string) { if !h.ensureResponsesDependencies(c, nil) { return } + subscription, _ := middleware2.GetSubscriptionFromContext(c) + if err := h.billingCacheService.CheckBillingEligibility(c.Request.Context(), apiKey.User, apiKey, apiKey.Group, subscription, service.QuotaPlatform(c.Request.Context(), apiKey)); err != nil { + status, code, message, retryAfter := billingErrorDetails(err) + if retryAfter > 0 { + c.Header("Retry-After", strconv.Itoa(retryAfter)) + } + h.errorResponse(c, status, code, message) + return + } body, err := readGrokVoiceGatewayBody(c) if err != nil { @@ -167,7 +186,6 @@ func (h *OpenAIGatewayHandler) GrokVoice(c *gin.Context, endpoint string) { return h.gatewayService.ForwardGrokVoice(c.Request.Context(), c, account, endpoint, body, contentType) }() if forwardErr == nil { - subscription, _ := middleware2.GetSubscriptionFromContext(c) h.recordGrokVoiceUsage(c, apiKey, account, subscription, endpoint, body, result) return } diff --git a/backend/internal/handler/grok_media.go b/backend/internal/handler/grok_media.go index dc97875b90..8cd16924dc 100644 --- a/backend/internal/handler/grok_media.go +++ b/backend/internal/handler/grok_media.go @@ -433,9 +433,12 @@ func (h *OpenAIGatewayHandler) handleGrokMedia(c *gin.Context, endpoint service. ) } } - if endpoint == service.GrokMediaEndpointVideoStatus { - if billResult := prepareGrokVideoStatusBilling(requestCtx, h, reqLog, apiKey, subject, requestID, result); billResult != nil { - recordGrokMediaUsage(c, h, reqLog, apiKey, subject, subscription, account, billResult, billResult.Model, body, requestID) + // Status poll OR content download can observe official done+video.url. + // Both paths share the same claim key so the customer is charged once. + if endpoint == service.GrokMediaEndpointVideoStatus || endpoint == service.GrokMediaEndpointVideoContent { + taskID := strings.TrimSpace(requestID) + if billResult := prepareGrokVideoCompletionBilling(requestCtx, h, reqLog, apiKey, subject, taskID, result); billResult != nil { + recordGrokMediaUsage(c, h, reqLog, apiKey, subject, subscription, account, billResult, billResult.Model, body, taskID) } } else if shouldRecordGrokMediaUsage(endpoint, requestModel, result) { recordGrokMediaUsage(c, h, reqLog, apiKey, subject, subscription, account, result, requestModel, body, requestID) @@ -508,16 +511,16 @@ func shouldRecordGrokMediaUsage(endpoint service.GrokMediaEndpoint, requestModel return result.ImageCount > 0 } -// prepareGrokVideoStatusBilling claims one-shot billing for official done+video.url -// responses. Duration/model prefer status body; resolution uses create-time request -// (status response does not document resolution). -func prepareGrokVideoStatusBilling( +// prepareGrokVideoCompletionBilling claims one-shot billing for official done+video.url +// observations (status poll or content download). Duration/model prefer status body; +// resolution uses create-time request (status response does not document resolution). +func prepareGrokVideoCompletionBilling( ctx context.Context, h *OpenAIGatewayHandler, reqLog *zap.Logger, apiKey *service.APIKey, subject middleware2.AuthSubject, - requestID string, + taskRequestID string, statusResult *service.OpenAIForwardResult, ) *service.OpenAIForwardResult { if h == nil || h.gatewayService == nil || apiKey == nil || statusResult == nil { @@ -527,18 +530,22 @@ func prepareGrokVideoStatusBilling( if statusResult.VideoCount <= 0 { return nil } - claimed, err := h.gatewayService.ClaimGrokVideoBilling(ctx, requestID, subject.UserID, apiKey.ID) + taskRequestID = strings.TrimSpace(firstNonEmptyString(taskRequestID, statusResult.ResponseID)) + if taskRequestID == "" { + return nil + } + claimed, err := h.gatewayService.ClaimGrokVideoBilling(ctx, taskRequestID, subject.UserID, apiKey.ID) if err != nil { - reqLog.Warn("grok_media.video_billing_claim_failed", zap.String("request_id", requestID), zap.Error(err)) + reqLog.Warn("grok_media.video_billing_claim_failed", zap.String("request_id", taskRequestID), zap.Error(err)) return nil } if !claimed { - reqLog.Debug("grok_media.video_billing_already_claimed", zap.String("request_id", requestID)) + reqLog.Debug("grok_media.video_billing_already_claimed", zap.String("request_id", taskRequestID)) return nil } - pending, loadErr := h.gatewayService.LoadGrokVideoPendingBilling(ctx, requestID, subject.UserID, apiKey.ID) + pending, loadErr := h.gatewayService.LoadGrokVideoPendingBilling(ctx, taskRequestID, subject.UserID, apiKey.ID) if loadErr != nil { - reqLog.Warn("grok_media.video_pending_billing_load_failed", zap.String("request_id", requestID), zap.Error(loadErr)) + reqLog.Warn("grok_media.video_pending_billing_load_failed", zap.String("request_id", taskRequestID), zap.Error(loadErr)) } // Re-merge with pending: resolution is request-only; model/duration fill gaps. merged := *statusResult @@ -560,7 +567,7 @@ func prepareGrokVideoStatusBilling( merged.VideoDurationSeconds = pending.VideoDurationSeconds } if strings.TrimSpace(merged.ResponseID) == "" { - merged.ResponseID = requestID + merged.ResponseID = taskRequestID } } if strings.TrimSpace(merged.Model) == "" { @@ -569,11 +576,13 @@ func prepareGrokVideoStatusBilling( if strings.TrimSpace(merged.BillingModel) == "" { merged.BillingModel = merged.Model } - // Stable dedup key: same video task bills once across status polls. - if strings.TrimSpace(merged.RequestID) == "" { - merged.RequestID = "grok-video:" + strings.TrimSpace(firstNonEmptyString(merged.ResponseID, requestID)) - } + // Always force durable task id so usage_billing_dedup survives multi-poll + + // context-local request ids (do not prefer empty-only fill). + merged.RequestID = service.StableGrokVideoBillingRequestID(firstNonEmptyString(merged.ResponseID, taskRequestID)) + merged.ResponseID = firstNonEmptyString(merged.ResponseID, taskRequestID) merged.VideoCount = 1 + // Pure video: do not keep legacy ImageCount (avoids image-path heuristics). + merged.ImageCount = 0 // Official default resolution is 480p when the create request omitted it. merged.VideoResolution = service.NormalizeVideoBillingResolutionOrDefault(merged.VideoResolution) // Official default duration is 8s when neither status nor create provided it. @@ -620,6 +629,18 @@ func recordGrokMediaUsage( OriginalModel: clientRequestedModel(c, requestModel), ChannelMappedModel: requestModel, } + // Async video: force durable task request id and release claim if billing fails. + videoTaskID := "" + if result != nil && result.VideoCount > 0 { + videoTaskID = strings.TrimSpace(firstNonEmptyString(requestID, result.ResponseID)) + if stable := service.StableGrokVideoBillingRequestID(firstNonEmptyString(result.ResponseID, requestID)); stable != "" { + result.RequestID = stable + } + // Prefer task id hash for payload fingerprint stability across status/content. + if len(body) == 0 && videoTaskID != "" { + payloadForHash = []byte(videoTaskID) + } + } h.submitOpenAIUsageRecordTask(c.Request.Context(), result, func(ctx context.Context) { if err := h.gatewayService.RecordUsage(ctx, &service.OpenAIRecordUsageInput{ Result: result, @@ -637,6 +658,14 @@ func recordGrokMediaUsage( SessionID: sessionID, ChannelUsageFields: channelUsageFields, }); err != nil { + if videoTaskID != "" { + if releaseErr := h.gatewayService.ReleaseGrokVideoBilling(ctx, videoTaskID, subject.UserID, apiKey.ID); releaseErr != nil { + reqLog.Warn("grok_media.video_billing_claim_release_failed", + zap.String("request_id", videoTaskID), + zap.Error(releaseErr), + ) + } + } logger.L().With( zap.String("component", "handler.openai_gateway.grok_media"), zap.Int64("user_id", subject.UserID), diff --git a/backend/internal/handler/openai_gateway_handler.go b/backend/internal/handler/openai_gateway_handler.go index b787d0f976..9e0f253059 100644 --- a/backend/internal/handler/openai_gateway_handler.go +++ b/backend/internal/handler/openai_gateway_handler.go @@ -2415,7 +2415,8 @@ func (h *OpenAIGatewayHandler) submitUsageRecordTask(parent context.Context, tas } func (h *OpenAIGatewayHandler) submitOpenAIUsageRecordTask(parent context.Context, result *service.OpenAIForwardResult, task service.UsageRecordTask) { - if result != nil && result.ImageCount > 0 { + // Image and async video bills are money-critical: never drop on pool overflow. + if result != nil && (result.ImageCount > 0 || result.VideoCount > 0) { h.submitMandatoryUsageRecordTask(parent, task) return } diff --git a/backend/internal/repository/api_key_repo.go b/backend/internal/repository/api_key_repo.go index 30e3c27844..a1c7cad801 100644 --- a/backend/internal/repository/api_key_repo.go +++ b/backend/internal/repository/api_key_repo.go @@ -190,6 +190,7 @@ func (r *apiKeyRepository) GetByKeyForAuth(ctx context.Context, key string) (*se group.FieldVideoPrice480p, group.FieldVideoPrice720p, group.FieldVideoPrice1080p, + group.FieldVideoModelPrices, group.FieldWebSearchPricePerCall, group.FieldSearchPricePer1k, group.FieldAudioRealtimePricePerMin, diff --git a/backend/internal/repository/gateway_cache.go b/backend/internal/repository/gateway_cache.go index 18dcd8d117..dc3d4a980a 100644 --- a/backend/internal/repository/gateway_cache.go +++ b/backend/internal/repository/gateway_cache.go @@ -116,6 +116,17 @@ func (c *gatewayCache) ClaimGrokVideoBilled(ctx context.Context, key string, ttl return c.rdb.SetNX(ctx, grokVideoBilledPrefix+key, "1", ttl).Result() } +func (c *gatewayCache) ReleaseGrokVideoBilled(ctx context.Context, key string) error { + if c == nil || c.rdb == nil { + return errors.New("gateway cache unavailable") + } + key = strings.TrimSpace(key) + if key == "" { + return errors.New("invalid grok video billed key") + } + return c.rdb.Del(ctx, grokVideoBilledPrefix+key).Err() +} + // Compile-time assertion: gatewayCache must implement CyberSessionBlockStore. var _ service.CyberSessionBlockStore = (*gatewayCache)(nil) var _ service.LiveCallStore = (*gatewayCache)(nil) diff --git a/backend/internal/repository/grok_oauth_client.go b/backend/internal/repository/grok_oauth_client.go index 8911722429..0b709535b6 100644 --- a/backend/internal/repository/grok_oauth_client.go +++ b/backend/internal/repository/grok_oauth_client.go @@ -35,7 +35,13 @@ const ( ) func NewGrokOAuthClient() service.GrokOAuthClient { - return &grokOAuthClient{tokenURL: xai.EffectiveTokenURL()} + // Prefer validated official/allowlisted token URL so env misconfig cannot + // exfiltrate auth codes / refresh tokens to an arbitrary host. + tokenURL, err := xai.ValidatedTokenURL() + if err != nil || strings.TrimSpace(tokenURL) == "" { + tokenURL = xai.EffectiveTokenURL() + } + return &grokOAuthClient{tokenURL: tokenURL} } func (c *grokOAuthClient) ExchangeCode(ctx context.Context, code, codeVerifier, redirectURI, proxyURL, clientID string) (*xai.TokenResponse, error) { diff --git a/backend/internal/service/account_credentials_redact.go b/backend/internal/service/account_credentials_redact.go index 8eb70513ba..1e32483817 100644 --- a/backend/internal/service/account_credentials_redact.go +++ b/backend/internal/service/account_credentials_redact.go @@ -7,6 +7,8 @@ var SensitiveCredentialKeys = []string{ "access_token", "refresh_token", "id_token", "agent_private_key", // API Key 类 "api_key", "session_key", "cookie", + // Grok Web SSO / password (must never persist or echo after Build OAuth) + "password", "sso_token", "sso", "sso-rw", "clearTextPassword", // 云服务凭据 "aws_secret_access_key", "aws_session_token", "service_account_json", "service_account", "private_key", diff --git a/backend/internal/service/api_key_auth_cache.go b/backend/internal/service/api_key_auth_cache.go index cbfdf3338b..53647e4025 100644 --- a/backend/internal/service/api_key_auth_cache.go +++ b/backend/internal/service/api_key_auth_cache.go @@ -75,14 +75,15 @@ type APIKeyAuthGroupSnapshot struct { ImagePrice4K *float64 `json:"image_price_4k,omitempty"` VideoRateIndependent bool `json:"video_rate_independent"` VideoRateMultiplier float64 `json:"video_rate_multiplier"` - VideoPrice480P *float64 `json:"video_price_480p,omitempty"` - VideoPrice720P *float64 `json:"video_price_720p,omitempty"` - VideoPrice1080P *float64 `json:"video_price_1080p,omitempty"` - WebSearchPricePerCall *float64 `json:"web_search_price_per_call,omitempty"` - SearchPricePer1k *float64 `json:"search_price_per_1k,omitempty"` - AudioRealtimePricePerMin *float64 `json:"audio_realtime_price_per_min,omitempty"` - AudioTTSPricePerMillionChars *float64 `json:"audio_tts_price_per_million_chars,omitempty"` - AudioSTTPricePerHour *float64 `json:"audio_stt_price_per_hour,omitempty"` + VideoPrice480P *float64 `json:"video_price_480p,omitempty"` + VideoPrice720P *float64 `json:"video_price_720p,omitempty"` + VideoPrice1080P *float64 `json:"video_price_1080p,omitempty"` + VideoModelPrices map[string]map[string]float64 `json:"video_model_prices,omitempty"` + WebSearchPricePerCall *float64 `json:"web_search_price_per_call,omitempty"` + SearchPricePer1k *float64 `json:"search_price_per_1k,omitempty"` + AudioRealtimePricePerMin *float64 `json:"audio_realtime_price_per_min,omitempty"` + AudioTTSPricePerMillionChars *float64 `json:"audio_tts_price_per_million_chars,omitempty"` + AudioSTTPricePerHour *float64 `json:"audio_stt_price_per_hour,omitempty"` ClaudeCodeOnly bool `json:"claude_code_only"` FallbackGroupID *int64 `json:"fallback_group_id,omitempty"` FallbackGroupIDOnInvalidRequest *int64 `json:"fallback_group_id_on_invalid_request,omitempty"` diff --git a/backend/internal/service/api_key_auth_cache_impl.go b/backend/internal/service/api_key_auth_cache_impl.go index dd6adfb131..125b593386 100644 --- a/backend/internal/service/api_key_auth_cache_impl.go +++ b/backend/internal/service/api_key_auth_cache_impl.go @@ -400,11 +400,12 @@ func (s *APIKeyService) snapshotFromAPIKey(ctx context.Context, apiKey *APIKey) VideoPrice480P: apiKey.Group.VideoPrice480P, VideoPrice720P: apiKey.Group.VideoPrice720P, VideoPrice1080P: apiKey.Group.VideoPrice1080P, + VideoModelPrices: NormalizeVideoModelPrices(apiKey.Group.VideoModelPrices), WebSearchPricePerCall: apiKey.Group.WebSearchPricePerCall, - SearchPricePer1k: apiKey.Group.SearchPricePer1k, - AudioRealtimePricePerMin: apiKey.Group.AudioRealtimePricePerMin, - AudioTTSPricePerMillionChars: apiKey.Group.AudioTTSPricePerMillionChars, - AudioSTTPricePerHour: apiKey.Group.AudioSTTPricePerHour, + SearchPricePer1k: apiKey.Group.SearchPricePer1k, + AudioRealtimePricePerMin: apiKey.Group.AudioRealtimePricePerMin, + AudioTTSPricePerMillionChars: apiKey.Group.AudioTTSPricePerMillionChars, + AudioSTTPricePerHour: apiKey.Group.AudioSTTPricePerHour, ClaudeCodeOnly: apiKey.Group.ClaudeCodeOnly, FallbackGroupID: apiKey.Group.FallbackGroupID, FallbackGroupIDOnInvalidRequest: apiKey.Group.FallbackGroupIDOnInvalidRequest, @@ -494,11 +495,12 @@ func (s *APIKeyService) snapshotToAPIKey(key string, snapshot *APIKeyAuthSnapsho VideoPrice480P: snapshot.Group.VideoPrice480P, VideoPrice720P: snapshot.Group.VideoPrice720P, VideoPrice1080P: snapshot.Group.VideoPrice1080P, + VideoModelPrices: NormalizeVideoModelPrices(snapshot.Group.VideoModelPrices), WebSearchPricePerCall: snapshot.Group.WebSearchPricePerCall, - SearchPricePer1k: snapshot.Group.SearchPricePer1k, - AudioRealtimePricePerMin: snapshot.Group.AudioRealtimePricePerMin, - AudioTTSPricePerMillionChars: snapshot.Group.AudioTTSPricePerMillionChars, - AudioSTTPricePerHour: snapshot.Group.AudioSTTPricePerHour, + SearchPricePer1k: snapshot.Group.SearchPricePer1k, + AudioRealtimePricePerMin: snapshot.Group.AudioRealtimePricePerMin, + AudioTTSPricePerMillionChars: snapshot.Group.AudioTTSPricePerMillionChars, + AudioSTTPricePerHour: snapshot.Group.AudioSTTPricePerHour, ClaudeCodeOnly: snapshot.Group.ClaudeCodeOnly, FallbackGroupID: snapshot.Group.FallbackGroupID, FallbackGroupIDOnInvalidRequest: snapshot.Group.FallbackGroupIDOnInvalidRequest, diff --git a/backend/internal/service/gateway_hotpath_optimization_test.go b/backend/internal/service/gateway_hotpath_optimization_test.go index ec06568e9e..80c41925aa 100644 --- a/backend/internal/service/gateway_hotpath_optimization_test.go +++ b/backend/internal/service/gateway_hotpath_optimization_test.go @@ -154,6 +154,11 @@ func (s *stickyGatewayCacheHotpathStub) ClaimGrokVideoBilled(_ context.Context, return true, nil } +func (s *stickyGatewayCacheHotpathStub) ReleaseGrokVideoBilled(_ context.Context, _ string) 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 9dc6fef4c3..866328dd3b 100644 --- a/backend/internal/service/gateway_multiplatform_test.go +++ b/backend/internal/service/gateway_multiplatform_test.go @@ -287,6 +287,11 @@ func (m *mockGatewayCacheForPlatform) ClaimGrokVideoBilled(_ context.Context, _ return true, nil } +func (m *mockGatewayCacheForPlatform) ReleaseGrokVideoBilled(_ context.Context, _ string) 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 7998568c2c..0b48c00bb0 100644 --- a/backend/internal/service/gateway_service.go +++ b/backend/internal/service/gateway_service.go @@ -485,6 +485,8 @@ type GatewayCache interface { // ClaimGrokVideoBilled atomically marks a video request as billed (SetNX). // Returns true when this caller won the claim; false when already billed or claim unavailable. ClaimGrokVideoBilled(ctx context.Context, key string, ttl time.Duration) (bool, error) + // ReleaseGrokVideoBilled clears a claim so a failed RecordUsage can retry billing. + ReleaseGrokVideoBilled(ctx context.Context, key string) error } // derefGroupID safely dereferences *int64 to int64, returning 0 if nil diff --git a/backend/internal/service/gemini_multiplatform_test.go b/backend/internal/service/gemini_multiplatform_test.go index c75eab035a..47f2bd1be1 100644 --- a/backend/internal/service/gemini_multiplatform_test.go +++ b/backend/internal/service/gemini_multiplatform_test.go @@ -315,6 +315,11 @@ func (m *mockGatewayCacheForGemini) ClaimGrokVideoBilled(_ context.Context, _ st return true, nil } +func (m *mockGatewayCacheForGemini) ReleaseGrokVideoBilled(_ context.Context, _ string) error { + return nil +} + + // TestGeminiMessagesCompatService_SelectAccountForModelWithExclusions_GeminiPlatform 测试 Gemini 单平台选择 func TestGeminiMessagesCompatService_SelectAccountForModelWithExclusions_GeminiPlatform(t *testing.T) { ctx := context.Background() diff --git a/backend/internal/service/grok_audio.go b/backend/internal/service/grok_audio.go index 498edc3301..2a9fa9030d 100644 --- a/backend/internal/service/grok_audio.go +++ b/backend/internal/service/grok_audio.go @@ -198,20 +198,46 @@ func estimateGrokVoiceAudioUsage(endpoint string, reqBody []byte, contentType st } return &AudioUsage{Mode: "tts", DurationOrUnits: float64(chars) / 1_000_000.0} case "stt": - // Prefer explicit duration_seconds; else estimate from elapsed with a floor. + // Prefer response duration when present; do not trust client duration_seconds alone + // (under-report would underbill). Floor against body-size heuristic and elapsed. secs := 0.0 + if gjson.ValidBytes(respBody) { + for _, path := range []string{"duration", "duration_seconds", "audio_duration", "usage.seconds"} { + if v := gjson.GetBytes(respBody, path); v.Exists() && v.Type == gjson.Number && v.Float() > 0 { + secs = v.Float() + break + } + } + } + // Multipart / body size heuristic: ~16KB/s for compressed speech (lower bound). + sizeFloor := 0.0 + if len(reqBody) > 0 { + sizeFloor = float64(len(reqBody)) / 16000.0 + } + clientSecs := 0.0 if gjson.ValidBytes(reqBody) { - if v := gjson.GetBytes(reqBody, "duration_seconds"); v.Exists() { - secs = v.Float() + if v := gjson.GetBytes(reqBody, "duration_seconds"); v.Exists() && v.Type == gjson.Number { + clientSecs = v.Float() } } if secs <= 0 { secs = elapsed.Seconds() } if secs <= 0 { - // Multipart audio size heuristic: ~16KB/s for compressed speech. - if len(reqBody) > 0 { - secs = float64(len(reqBody)) / 16000.0 + secs = clientSecs + } + if secs <= 0 { + secs = sizeFloor + } + // Cap untrusted client under-report: if client duration is much smaller than + // size/elapsed floors, bill the larger of floors (anti underbill). + if clientSecs > 0 && secs == clientSecs { + floor := sizeFloor + if elapsed.Seconds() > floor { + floor = elapsed.Seconds() + } + if floor > 0 && clientSecs < floor*0.5 { + secs = floor } } if secs <= 0 { diff --git a/backend/internal/service/grok_free_quota_gate.go b/backend/internal/service/grok_free_quota_gate.go index 77f8a5885b..affd3e5fcb 100644 --- a/backend/internal/service/grok_free_quota_gate.go +++ b/backend/internal/service/grok_free_quota_gate.go @@ -3,7 +3,6 @@ package service import ( "context" "log/slog" - "strings" "sync/atomic" "time" @@ -15,7 +14,7 @@ import ( // // Config keys (gateway.grok.*): // - free_quota_soft_gate_enabled (bool, default true) — only applied when free-tier detection is strict -// - free_quota_token_limit (int64, default 2_000_000) — nominal rolling-window token allowance +// - free_quota_token_limit (int64, default = xai free rolling limit 1_000_000) // - free_quota_soft_gate_percent (int, default 95) — stop scheduling before the nominal limit // - free_quota_window_hours (int, default 24) — local usage rolling window // - free_quota_stats_cache_seconds (int, default 5) — bound hot-path aggregate query frequency @@ -24,7 +23,8 @@ import ( // filterGrokFreeQuotaAccounts; only the OpenAI-compatible account scheduler filter does. const ( - defaultGrokFreeQuotaTokenLimit int64 = 2_000_000 + // Match current Free rolling limit (pkg/xai). A 2M default never pre-empts real exhaustion. + defaultGrokFreeQuotaTokenLimit int64 = 1_000_000 defaultGrokFreeQuotaSoftGatePercent = 95 defaultGrokFreeQuotaWindowHours = 24 ) @@ -83,24 +83,12 @@ func calculateGrokFreeQuotaSoftGateTokens(limit int64, percent int) int64 { return (limit/100)*int64(percent) + (limit%100)*int64(percent)/100 } -// isExplicitGrokFreeOAuthAccount is intentionally strict: only OAuth accounts with an -// explicit free marker (subscription_tier / plan_type on credentials or extra) are gated. -// Unknown/empty tier, paid tiers, and API-key accounts fail open (not gated). +// isExplicitGrokFreeOAuthAccount decides whether the free soft-gate applies. +// Uses the same free/paid evidence as media eligibility (isKnownGrokFreeAccount): +// free/basic tiers, billing-inferred free, while paid plan/monthly evidence wins. +// Unknown/empty tier without free evidence and API-key accounts are not gated. func isExplicitGrokFreeOAuthAccount(account *Account) bool { - if account == nil || !account.IsGrokOAuth() { - return false - } - for _, tier := range []string{ - account.GetCredential("subscription_tier"), - account.GetCredential("plan_type"), - account.GetExtraString("subscription_tier"), - account.GetExtraString("plan_type"), - } { - if strings.EqualFold(strings.TrimSpace(tier), "free") { - return true - } - } - return false + return isKnownGrokFreeAccount(account) } // filterGrokFreeQuotaAccounts applies a local, rolling soft gate only to diff --git a/backend/internal/service/grok_media.go b/backend/internal/service/grok_media.go index ff3d89058e..ad5316b1b9 100644 --- a/backend/internal/service/grok_media.go +++ b/backend/internal/service/grok_media.go @@ -293,9 +293,13 @@ func (s *OpenAIGatewayService) BindGrokMediaVideoRequestAccount( if cacheKey == "" || accountID <= 0 { return fmt.Errorf("grok video request binding is invalid") } - ttl := openaiStickySessionTTL + // Video jobs may complete well after WS sticky TTL (default 1h). Bind at least + // as long as the pending-billing snapshot so late status/content polls resolve. + ttl := grokVideoPendingBillingTTL(s.cfg) if s.cfg != nil && s.cfg.Gateway.OpenAIWS.StickySessionTTLSeconds > 0 { - ttl = time.Duration(s.cfg.Gateway.OpenAIWS.StickySessionTTLSeconds) * time.Second + if sticky := time.Duration(s.cfg.Gateway.OpenAIWS.StickySessionTTLSeconds) * time.Second; sticky > ttl { + ttl = sticky + } } return s.cache.SetSessionAccountID(ctx, derefGroupID(groupID), cacheKey, accountID, ttl) } @@ -419,6 +423,36 @@ func (s *OpenAIGatewayService) ClaimGrokVideoBilling( return s.cache.ClaimGrokVideoBilled(ctx, key, grokVideoBilledClaimTTL(s.cfg)) } +// ReleaseGrokVideoBilling clears a claim after a failed durable RecordUsage so a +// later status/content poll can retry billing. +func (s *OpenAIGatewayService) ReleaseGrokVideoBilling( + ctx context.Context, + requestID string, + userID, apiKeyID int64, +) error { + if s == nil || s.cache == nil { + return fmt.Errorf("grok video billing claim cache is unavailable") + } + key := grokVideoPendingBillingKey(requestID, userID, apiKeyID) + if key == "" { + return fmt.Errorf("grok video billing claim key is invalid") + } + return s.cache.ReleaseGrokVideoBilled(ctx, key) +} + +// StableGrokVideoBillingRequestID is the durable usage_logs / dedup key for one +// async video task (not the per-poll gateway request id). +func StableGrokVideoBillingRequestID(taskRequestID string) string { + taskRequestID = strings.TrimSpace(taskRequestID) + if taskRequestID == "" { + return "" + } + if strings.HasPrefix(taskRequestID, "grok-video:") { + return taskRequestID + } + return "grok-video:" + taskRequestID +} + // Official xAI async video status success shape (docs.x.ai Video Generation): // // {"status":"done","model":"grok-imagine-video-1.5","video":{"url":"...","duration":8,"respect_moderation":true}} @@ -524,8 +558,6 @@ func ExtractGrokVideoBillingFromStatusBody(statusBody []byte, pending *GrokVideo VideoCount: 1, VideoResolution: resolution, VideoDurationSeconds: durationSeconds, - // Keep legacy media-unit counter for existing usage displays. - ImageCount: 1, } } @@ -793,11 +825,24 @@ func (s *OpenAIGatewayService) forwardGrokMediaVideoContent( if err := writeGrokMediaContentResponse(c, contentResp); err != nil { return nil, err } - return &OpenAIForwardResult{ + // Content download is an alternate completion observation: when status body is + // official done+video.url, attach billable units so the handler can claim once + // (same path as status polling). Pending snapshot is merged in the handler. + result := &OpenAIForwardResult{ RequestID: contentRequestID, ResponseHeaders: contentResp.Header.Clone(), Duration: time.Since(startTime), - }, nil + } + if billed := ExtractGrokVideoBillingFromStatusBody(statusBody, nil, requestID); billed != nil { + result.ResponseID = firstNonEmpty(billed.ResponseID, strings.TrimSpace(requestID)) + result.Model = billed.Model + result.BillingModel = billed.BillingModel + result.UpstreamModel = billed.UpstreamModel + result.VideoCount = billed.VideoCount + result.VideoResolution = billed.VideoResolution + result.VideoDurationSeconds = billed.VideoDurationSeconds + } + return result, nil } func grokMediaSignedVideoContentURL(body []byte, requestID string) (string, error) { @@ -1045,7 +1090,6 @@ func grokMediaUsageFromResponse(endpoint GrokMediaEndpoint, requestInfo GrokMedi meta.VideoCount = billed.VideoCount meta.VideoResolution = billed.VideoResolution meta.VideoDurationSeconds = billed.VideoDurationSeconds - meta.ImageCount = billed.ImageCount } } } diff --git a/backend/internal/service/openai_account_scheduler_test.go b/backend/internal/service/openai_account_scheduler_test.go index c84463f288..d5b8f39382 100644 --- a/backend/internal/service/openai_account_scheduler_test.go +++ b/backend/internal/service/openai_account_scheduler_test.go @@ -192,6 +192,11 @@ func (c *schedulerTestGatewayCache) ClaimGrokVideoBilled(_ context.Context, _ st return true, nil } +func (c *schedulerTestGatewayCache) ReleaseGrokVideoBilled(_ context.Context, _ string) error { + return nil +} + + func newSchedulerTestOpenAIWSV2Config() *config.Config { cfg := &config.Config{} cfg.Gateway.OpenAIWS.Enabled = true diff --git a/backend/internal/service/openai_cyber_session_block_test.go b/backend/internal/service/openai_cyber_session_block_test.go index 279c635d47..5afd59c37b 100644 --- a/backend/internal/service/openai_cyber_session_block_test.go +++ b/backend/internal/service/openai_cyber_session_block_test.go @@ -144,6 +144,11 @@ func (c *comboCacheAndStore) ClaimGrokVideoBilled(_ context.Context, _ string, _ return true, nil } +func (c *comboCacheAndStore) ReleaseGrokVideoBilled(_ context.Context, _ string) error { + return nil +} + + func (c *comboCacheAndStore) SetCyberSessionBlocked(ctx context.Context, key string, ttl time.Duration) error { return c.store.SetCyberSessionBlocked(ctx, key, ttl) } diff --git a/backend/internal/service/openai_gateway_grok_cache.go b/backend/internal/service/openai_gateway_grok_cache.go index 696e7352ec..bd3924edf8 100644 --- a/backend/internal/service/openai_gateway_grok_cache.go +++ b/backend/internal/service/openai_gateway_grok_cache.go @@ -346,11 +346,20 @@ func isKnownGrokFreeAccount(account *Account) bool { inferredFreeSignal = true } } - if tier := strings.TrimSpace(account.GetCredential("subscription_tier")); tier != "" { - if isGrokFreeSubscriptionTier(tier) { - freeSignal = true - } else if !isGrokUnknownSubscriptionTier(tier) { - paidSignal = true + for _, key := range []string{"subscription_tier", "plan_type"} { + if tier := strings.TrimSpace(account.GetCredential(key)); tier != "" { + if isGrokFreeSubscriptionTier(tier) { + freeSignal = true + } else if !isGrokUnknownSubscriptionTier(tier) { + paidSignal = true + } + } + if tier := strings.TrimSpace(account.GetExtraString(key)); tier != "" { + if isGrokFreeSubscriptionTier(tier) { + freeSignal = true + } else if !isGrokUnknownSubscriptionTier(tier) { + paidSignal = true + } } } // Explicit paid evidence always wins over an inferred Free signal. This diff --git a/backend/internal/service/openai_gateway_service_test.go b/backend/internal/service/openai_gateway_service_test.go index 3b893b7b7f..9783ac14d6 100644 --- a/backend/internal/service/openai_gateway_service_test.go +++ b/backend/internal/service/openai_gateway_service_test.go @@ -674,6 +674,11 @@ func (c *stubGatewayCache) ClaimGrokVideoBilled(_ context.Context, _ string, _ t return true, nil } +func (c *stubGatewayCache) ReleaseGrokVideoBilled(_ context.Context, _ string) 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_gateway_usage.go b/backend/internal/service/openai_gateway_usage.go index 7bc11321da..0ad8cfed8b 100644 --- a/backend/internal/service/openai_gateway_usage.go +++ b/backend/internal/service/openai_gateway_usage.go @@ -257,6 +257,18 @@ func (s *OpenAIGatewayService) RecordUsage(ctx context.Context, input *OpenAIRec requestID = upstreamRequestID } } + // Async Grok video: always use the stable task id for dedup (status + content polls + // share one bill). Context-local client/local IDs would otherwise create a new row + // per poll if Redis claim is lost. + if result.VideoCount > 0 { + if stable := StableGrokVideoBillingRequestID(firstNonEmpty( + strings.TrimPrefix(strings.TrimSpace(result.RequestID), "grok-video:"), + strings.TrimSpace(result.ResponseID), + strings.TrimPrefix(strings.TrimSpace(requestID), "grok-video:"), + )); stable != "" { + requestID = stable + } + } // 确定 RequestedModel（渠道映射前的原始模型） requestedModel := result.Model diff --git a/backend/internal/service/openai_ws_state_store_test.go b/backend/internal/service/openai_ws_state_store_test.go index e0b7f5f6f9..a092175504 100644 --- a/backend/internal/service/openai_ws_state_store_test.go +++ b/backend/internal/service/openai_ws_state_store_test.go @@ -203,6 +203,11 @@ func (c *openAIWSStateStoreTimeoutProbeCache) ClaimGrokVideoBilled(_ context.Con return true, nil } +func (c *openAIWSStateStoreTimeoutProbeCache) ReleaseGrokVideoBilled(_ context.Context, _ string) error { + return nil +} + + func TestOpenAIWSStateStore_RedisOpsUseShortTimeout(t *testing.T) { probe := &openAIWSStateStoreTimeoutProbeCache{} store := NewOpenAIWSStateStore(probe) diff --git a/backend/internal/testutil/stubs.go b/backend/internal/testutil/stubs.go index b8023b95bb..fd2280c0ee 100644 --- a/backend/internal/testutil/stubs.go +++ b/backend/internal/testutil/stubs.go @@ -114,6 +114,11 @@ func (c StubGatewayCache) ClaimGrokVideoBilled(_ context.Context, _ string, _ ti return true, nil } +func (c StubGatewayCache) ReleaseGrokVideoBilled(_ context.Context, _ string) error { + return nil +} + + // ============================================================ // StubSessionLimitCache — service.SessionLimitCache 的空实现 diff --git a/deploy/config.example.yaml b/deploy/config.example.yaml index c1e51a8c41..d9d68f4659 100644 --- a/deploy/config.example.yaml +++ b/deploy/config.example.yaml @@ -449,7 +449,7 @@ gateway: # TurnstileTaskProxyless uses solver egress independent from the optional account proxy. password_auth_enabled: false free_quota_soft_gate_enabled: true - free_quota_token_limit: 2000000 + free_quota_token_limit: 1000000 free_quota_soft_gate_percent: 95 free_quota_window_hours: 24 free_quota_stats_cache_seconds: 5