修复 Grok 错误分类与容量重试

backend/internal/handler/failover_loop.go

@@ -56,7 +56,13 @@ const (
 const profitVetoExhaustedMessage = "No available accounts: all candidates rejected by group profit control"
 
 func sameAccountRetryDelayFor(failoverErr *service.UpstreamFailoverError, retryCount int) time.Duration {
-	if failoverErr == nil || !failoverErr.RequestScopedTransient || retryCount <= 1 {
+	if failoverErr == nil {
+		return sameAccountRetryDelay
+	}
+	if failoverErr.SameAccountRetryDelay > 0 {
+		return failoverErr.SameAccountRetryDelay
+	}
+	if !failoverErr.RequestScopedTransient || retryCount <= 1 {
 		return sameAccountRetryDelay
 	}
 
@@ -158,14 +164,19 @@ func (s *FailoverState) HandleFailoverError(
 	}
 
 	// 同账号重试不算切换账号，粘性会话仅在实际切换时强制缓存计费。
-	sameAccountRetry := failoverErr.RetryableOnSameAccount && s.SameAccountRetryCount[accountID] < retryLimit
+	retryCount := s.SameAccountRetryCount[accountID]
+	sameAccountRetryAllowed := failoverErr.RetryableOnSameAccount && retryLimit > 0 && retryCount < retryLimit
+	if sameAccountRetryAllowed && !failoverErr.SameAccountRetryDeadline.IsZero() {
+		sameAccountRetryAllowed = time.Now().Before(failoverErr.SameAccountRetryDeadline)
+	}
+	sameAccountRetry := sameAccountRetryAllowed
 	if needForceCacheBilling(s.hasBoundSession, failoverErr, sameAccountRetry) {
 		s.ForceCacheBilling = true
 	}
 
 	// 同账号重试：对 RetryableOnSameAccount 的临时性错误，先在同一账号上重试。
 	// 重试次数上限 retryLimit 由调用方传入（账号级 pool_mode_retry_count 配置）。
-	if failoverErr.RetryableOnSameAccount && s.SameAccountRetryCount[accountID] < retryLimit {
+	if sameAccountRetryAllowed {
 		s.SameAccountRetryCount[accountID]++
 		retryDelay := sameAccountRetryDelayFor(failoverErr, s.SameAccountRetryCount[accountID])
 		logger.FromContext(ctx).Warn("gateway.failover_same_account_retry",

backend/internal/service/gateway_service.go

@@ -671,12 +671,14 @@ type GatewayFailureReason string
 // source-compatible and preserves their legacy retry-next-account behavior.
 type UpstreamFailoverError struct {
 	StatusCode               int
-	ResponseBody             []byte      // 上游响应体，用于错误透传规则匹配
-	ResponseHeaders          http.Header // 上游响应头，用于透传 cf-ray/cf-mitigated/content-type 等诊断信息
-	ForceCacheBilling        bool        // Antigravity 粘性会话切换时设为 true
-	RetryableOnSameAccount   bool        // 临时性错误（如 Google 间歇性 400、空响应），应在同一账号上重试 N 次再切换
-	RequestScopedTransient   bool        // 故障因素与账号无关（如上游按客户端身份/模型容量降载）：可同账号重试，但不得据此对账号做临时封禁
-	SafeToFailoverAfterWrite bool        // 仅写出 SSE 注释等非语义字节时，仍可在同一客户端流中切换账号
+	ResponseBody             []byte        // 上游响应体，用于错误透传规则匹配
+	ResponseHeaders          http.Header   // 上游响应头，用于透传 cf-ray/cf-mitigated/content-type 等诊断信息
+	ForceCacheBilling        bool          // Antigravity 粘性会话切换时设为 true
+	RetryableOnSameAccount   bool          // 临时性错误（如 Google 间歇性 400、空响应），应在同一账号上重试 N 次再切换
+	SameAccountRetryDelay    time.Duration // 同账号重试的最小间隔；零值使用 handler 默认值
+	SameAccountRetryDeadline time.Time     // 同账号重试截止时间；零值表示仅受 retryLimit 限制
+	RequestScopedTransient   bool          // 故障因素与账号无关（如上游按客户端身份/模型容量降载）：可同账号重试，但不得据此对账号做临时封禁
+	SafeToFailoverAfterWrite bool          // 仅写出 SSE 注释等非语义字节时，仍可在同一客户端流中切换账号
 	Stage                    GatewayFailureStage
 	Scope                    GatewayFailureScope
 	Reason                   GatewayFailureReason

backend/internal/service/grok_upstream_errors.go

@@ -199,7 +199,7 @@ func (s *OpenAIGatewayService) shouldFailoverGrokUpstreamError(statusCode int, r
 	}
 	decision := classifyGrokUpstreamFailure(statusCode, responseBody, "")
 	switch decision.Class {
-	case GrokFailureFreeUsage, GrokFailureEmptyUpstream, GrokFailureBilling, GrokFailureModelCapacity:
+	case GrokFailureFreeUsage, GrokFailureEmptyUpstream, GrokFailureBilling, GrokFailureModelCapacity, GrokFailureCompatibility:
 		return decision.ShouldFailover
 	}
 	return s.shouldFailoverUpstreamError(statusCode)

backend/internal/service/grok_upstream_failure.go

@@ -26,6 +26,10 @@ const (
 	GrokFailureRateLimit     GrokUpstreamFailureClass = "rate_limit"
 	GrokFailureAuth          GrokUpstreamFailureClass = "auth_error"
 	GrokFailureServer        GrokUpstreamFailureClass = "server_error"
+	// GrokFailureCompatibility represents a request-history/body-shape that is
+	// incompatible with the selected account or upstream replay contract. It
+	// is account-independent: fail over, but never quarantine the pool.
+	GrokFailureCompatibility GrokUpstreamFailureClass = "compatibility_error"
 )
 
 // GrokUpstreamFailureDecision is a pure classification result. Callers map it
@@ -111,6 +115,20 @@ func classifyGrokUpstreamFailure(statusCode int, responseBody []byte, requestedM
 		}
 	}
 
+	// Responses replay/compaction payloads can be rejected by one Grok
+	// deployment while the same request is valid on another account. Treat
+	// these precise decoder/content-shape failures as account compatibility
+	// errors, rather than durable account health failures.
+	if isGrokCompatibilityError(low, code) {
+		return GrokUpstreamFailureDecision{
+			Class:          GrokFailureCompatibility,
+			Model:          model,
+			ShouldFailover: true,
+			ShouldCooldown: false,
+			Reason:         firstNonEmpty(text, "grok response compatibility error"),
+		}
+	}
+
 	// Empty HTTP 200 / empty model output (often rewritten to synthetic 502).
 	if isGrokEmptyModelOutputText(low) || isGrokEmptyModelOutputCode(code) {
 		return GrokUpstreamFailureDecision{
@@ -355,7 +373,8 @@ func isGrokBillingQuotaText(low string) bool {
 	if strings.Contains(low, "payment") && (strings.Contains(low, "required") || strings.Contains(low, "fail")) {
 		return true
 	}
-	if strings.Contains(low, "spending limit") || strings.Contains(low, "run out of credits") || strings.Contains(low, "out of credits") {
+	if strings.Contains(low, "spending limit") || strings.Contains(low, "run out of credits") || strings.Contains(low, "out of credits") ||
+		(strings.Contains(low, "need a grok subscription") || strings.Contains(low, "need grok subscription")) {
 		return true
 	}
 	if strings.Contains(low, "余额不足") || strings.Contains(low, "欠费") || strings.Contains(low, "需要付费") {
@@ -364,6 +383,73 @@ func isGrokBillingQuotaText(low string) bool {
 	return false
 }
 
+// grokRetryableOnSameAccount marks transient 429 classes for the shared
+// failover loop. Capacity and ordinary throttles are request/model pressure,
+// not evidence that the credential is invalid, so a bounded retry on the same
+// account is preferable before switching accounts. Free-usage and billing
+// exhaustion deliberately skip same-account retry and fail over immediately.
+func grokRetryableOnSameAccount(account *Account, statusCode int, responseBody []byte) bool {
+	if account == nil || !account.IsGrok() {
+		return false
+	}
+	decision := classifyGrokUpstreamFailure(statusCode, responseBody, "")
+	switch decision.Class {
+	case GrokFailureFreeUsage, GrokFailureBilling, GrokFailureCompatibility:
+		// Quota/entitlement exhaustion is account state, not transient
+		// pressure. Retrying the same account only repeats the failure.
+		return false
+	case GrokFailureModelCapacity:
+		if statusCode == http.StatusTooManyRequests {
+			return true
+		}
+	}
+	return account.IsPoolMode() && account.IsPoolModeRetryableStatus(statusCode)
+}
+
+func grokSameAccountRetryMetadata(account *Account, statusCode int, responseBody []byte) (bool, time.Duration, time.Time) {
+	if !grokRetryableOnSameAccount(account, statusCode, responseBody) {
+		return false, 0, time.Time{}
+	}
+	decision := classifyGrokUpstreamFailure(statusCode, responseBody, "")
+	if decision.Class != GrokFailureModelCapacity {
+		return true, 0, time.Time{}
+	}
+	return true, 500 * time.Millisecond, time.Now().Add(30 * time.Second)
+}
+
+func isGrokCompatibilityError(low, code string) bool {
+	combined := strings.ToLower(strings.TrimSpace(low + " " + code))
+	// Compaction blobs are account/session-bound and frequently fail with 400
+	// or 422 after a reconnect. Also cover xAI's JSON decoder shape errors.
+	for _, phrase := range []string{
+		"could not decode the compaction blob",
+		"cannot decode the compaction blob",
+		"decode the compaction blob",
+		"ensure it is unmodified from the compact response",
+		"compaction blob",
+		"failed to deserialize the json body",
+		"data did not match any variant",
+		"untagged enum content",
+		"invalid content shape",
+		"invalid response history",
+	} {
+		if strings.Contains(combined, phrase) {
+			return true
+		}
+	}
+	for _, marker := range []string{
+		"invalid_compaction",
+		"compaction_decode_error",
+		"response_history_incompatible",
+		"invalid_content_shape",
+	} {
+		if strings.Contains(combined, marker) {
+			return true
+		}
+	}
+	return false
+}
+
 func isGrokModelCapacityText(low string) bool {
 	return strings.Contains(low, "capacity") ||
 		strings.Contains(low, "overloaded") ||
@@ -489,10 +575,11 @@ func (s *OpenAIGatewayService) applyGrokUpstreamFailureDecision(
 	case GrokFailureEmptyUpstream:
 		reason = "grok empty model output"
 	case GrokFailureModelCapacity:
-		if persistGrokTransientModelCooldown(account, decision) {
-			return true
-		}
-		reason = "grok model capacity"
+		// Capacity is scoped to the requested model. Never persist an account-wide
+		// unschedulable state for this transient class; the failover loop performs
+		// a bounded same-account retry before selecting another account.
+		_ = persistGrokTransientModelCooldown(account, decision)
+		return true
 	case GrokFailureRateLimit:
 		// Pure 429 without free-usage language keeps the existing rate-limit
 		// snapshot path (Retry-After / quota headers). Body-only rate-limit
@@ -501,6 +588,11 @@ func (s *OpenAIGatewayService) applyGrokUpstreamFailureDecision(
 		return false
 	case GrokFailureServer:
 		reason = "grok upstream temporary error"
+	case GrokFailureCompatibility:
+		// Deliberately no account mutation. The caller uses ShouldFailover to
+		// retry another account; cooling a pool for a request-shape mismatch
+		// would remove healthy accounts.
+		return true
 	default:
 		return false
 	}

backend/internal/service/grok_upstream_failure_test.go

@@ -61,6 +61,44 @@ func TestClassifyGrokUpstreamFailure_Billing(t *testing.T) {
 	require.True(t, d.ShouldFailover)
 }
 
+func TestClassifyGrokUpstreamFailure_GrokSubscriptionRequiredIsBilling(t *testing.T) {
+	d := classifyGrokUpstreamFailure(http.StatusPaymentRequired,
+		[]byte(`{"error":{"message":"You have run out of credits or need a Grok subscription"}}`), "grok-4.6")
+	require.Equal(t, GrokFailureBilling, d.Class)
+	require.True(t, d.ShouldFailover)
+	require.True(t, d.ShouldCooldown)
+}
+
+func TestGrokRetryableOnSameAccount_CapacityAndRateLimit(t *testing.T) {
+	account := &Account{ID: 9105, Platform: PlatformGrok, Type: AccountTypeOAuth}
+	require.True(t, grokRetryableOnSameAccount(account, http.StatusTooManyRequests,
+		[]byte(`{"error":{"message":"The model is currently at capacity due to high demand"}}`)))
+	require.False(t, grokRetryableOnSameAccount(account, http.StatusTooManyRequests,
+		[]byte(`{"error":{"message":"rate limit exceeded"}}`)))
+	require.False(t, grokRetryableOnSameAccount(account, http.StatusPaymentRequired,
+		[]byte(`{"error":{"message":"You have run out of credits or need a Grok subscription"}}`)))
+	require.False(t, grokRetryableOnSameAccount(account, http.StatusBadRequest,
+		[]byte(`{"error":{"message":"capacity field is invalid"}}`)))
+	nonGrok := &Account{ID: 9106, Platform: PlatformOpenAI, Type: AccountTypeOAuth}
+	require.False(t, grokRetryableOnSameAccount(nonGrok, http.StatusTooManyRequests,
+		[]byte(`{"error":{"message":"model at capacity"}}`)))
+}
+
+func TestGrokSameAccountRetryMetadata_CapacityDeadline(t *testing.T) {
+	account := &Account{ID: 9107, Platform: PlatformGrok, Type: AccountTypeOAuth}
+	retryable, delay, deadline := grokSameAccountRetryMetadata(account, http.StatusTooManyRequests,
+		[]byte(`{"error":{"message":"model capacity exceeded"}}`))
+	require.True(t, retryable)
+	require.Equal(t, 500*time.Millisecond, delay)
+	require.WithinDuration(t, time.Now().Add(30*time.Second), deadline, 2*time.Second)
+
+	retryable, delay, deadline = grokSameAccountRetryMetadata(account, http.StatusTooManyRequests,
+		[]byte(`{"error":{"message":"rate limit exceeded"}}`))
+	require.False(t, retryable)
+	require.Zero(t, delay)
+	require.True(t, deadline.IsZero())
+}
+
 func TestClassifyGrokUpstreamFailure_ValidationNoCool(t *testing.T) {
 	d := classifyGrokUpstreamFailure(http.StatusBadRequest, []byte(`{"error":{"message":"invalid tool schema"}}`), "")
 	require.Equal(t, GrokFailureNone, d.Class)
@@ -75,12 +113,33 @@ func TestClassifyGrokUpstreamFailure_FreeUsageWinsOver5xx(t *testing.T) {
 	require.NotEqual(t, GrokFailureServer, d.Class)
 }
 
+func TestClassifyGrokUpstreamFailure_CompatibilityDoesNotCooldown(t *testing.T) {
+	cases := []string{
+		`{"error":{"message":"Could not decode the compaction blob. Ensure it is unmodified from the compact response"}}`,
+		`{"error":{"message":"data did not match any variant of the untagged enum content"}}`,
+		`{"code":"compaction_decode_error","message":"invalid response history"}`,
+	}
+	for _, body := range cases {
+		d := classifyGrokUpstreamFailure(http.StatusUnprocessableEntity, []byte(body), "grok-4.6")
+		require.Equal(t, GrokFailureCompatibility, d.Class, body)
+		require.True(t, d.ShouldFailover, body)
+		require.False(t, d.ShouldCooldown, body)
+		require.Zero(t, d.Cooldown, body)
+	}
+}
+
 func TestShouldFailoverGrokUpstreamError_FreeUsageBody(t *testing.T) {
 	svc := &OpenAIGatewayService{}
 	body := []byte(`{"error":{"code":"subscription:free-usage-exhausted","message":"free usage exhausted"}}`)
 	require.True(t, svc.shouldFailoverGrokUpstreamError(http.StatusBadRequest, body))
 }
 
+func TestShouldFailoverGrokUpstreamError_CompatibilityBody(t *testing.T) {
+	svc := &OpenAIGatewayService{}
+	body := []byte(`{"error":{"message":"Could not decode the compaction blob"}}`)
+	require.True(t, svc.shouldFailoverGrokUpstreamError(http.StatusUnprocessableEntity, body))
+}
+
 func TestShouldFailoverGrokUpstreamError_ContentPolicyStillNoFailover(t *testing.T) {
 	svc := &OpenAIGatewayService{}
 	body := []byte(`{"error":{"code":"new_sensitive","message":"text is sensitive"}}`)
@@ -149,6 +208,19 @@ func TestHandleGrokAccountUpstreamError_MultiAgentCapacityBlocksOnlyThatModel(t
 	require.False(t, isGrokModelQuotaBlocked(account.ID, "grok-4.5", time.Now()))
 }
 
+func TestHandleGrokAccountUpstreamError_CapacityNeverCoolsAccount(t *testing.T) {
+	repo := &grokQuotaAccountRepo{}
+	svc := &OpenAIGatewayService{accountRepo: repo}
+	account := &Account{ID: 9121, Platform: PlatformGrok, Type: AccountTypeOAuth}
+	ctx := withGrokTeamRateLimitModel(context.Background(), "grok-4.6")
+
+	svc.handleGrokAccountUpstreamError(ctx, account, http.StatusTooManyRequests, nil,
+		[]byte(`{"error":{"message":"The model is currently at capacity due to high demand"}}`))
+
+	require.Zero(t, repo.tempUnschedCalls)
+	require.False(t, svc.isOpenAIAccountRuntimeBlocked(account))
+}
+
 func TestHandleGrokAccountUpstreamError_FreeUsageDoesNotCoolPoolMode(t *testing.T) {
 	repo := &grokQuotaAccountRepo{}
 	svc := &OpenAIGatewayService{accountRepo: repo}