fix(cn-providers): 修复 CN 分组五项功能缺陷(调度闸门/计费/断开漏记/count_tokens/403)

对已合并 PR #5666 + 分组入口放行后的全量功能审计发现的 P0/P1 修复,
全部先经代码与厂商文档实证再实施:

1. /v1/messages 调度闸门(P0):sanitizeGroupMessagesDispatchFields 对非
   openai 平台恒置 AllowMessagesDispatch=false,而闸门豁免名单只有 grok,
   CN 分组经正常途径创建后恒 403——原生 Anthropic 直通(Claude Code 主用例)
   完全不可达。修复:闸门对 CN 与 grok 同语义豁免;count_tokens 处的内联
   裸检查统一走同一 helper;ResolveMessagesDispatchModel 对 CN 早退,避免
   openai 专属的 gpt-5.x 默认映射发给 CN 上游。

2. 计费候选链(P0):候选链兜底含客户端原始模型名,配合 getFallbackPricing
   的 claude→Sonnet 统一兜底,映射的 CN 模型无价时 CN 流量会按 Claude 原价
   (数倍~数十倍)静默误计,且 usage 日志显示 claude-* 名无从察觉。修复:
   CN 账号的 claude-* 候选仅在显式分组/渠道定价时放行;候选全滤空时按
   ErrModelPricingUnavailable 走零成本+告警落账(顺带修复原空候选错误会
   丢弃整条 usage 记录的次生问题)。

3. 断开/中断漏记(P0,#5148 对齐,惠及 openai 平台):messages/responses/
   chat_completions 三个 handler 的错误路径此前在 err!=nil 时丢弃携带的
   部分 result——客户端断开排水后的完整 usage 被丢,payg 上游照常计费而
   平台漏记(anthropic 网关早有同修复,openai 网关缺失)。修复:错误路径
   result 非空时照常提交 usage;failover 错误恒 result=nil 无重复计费。
   Responses×anthropic 流式转换器同时改为断开后继续排水至流自然结束
   (末尾 message_delta 的 output_tokens 不再丢),finalize 帧补工具名反转
   与客户端工具还原、仅在客户端仍连接时写出。

4. count_tokens(P1,证据修正):经实证三家 Anthropic 兼容层均无
   /v1/messages/count_tokens(DeepSeek 官方文档无此端点且注明
   anthropic-version 被忽略;OpenModel 标注该端点 Anthropic only),
   anthropic 协议转发上游=常态 404,且错误处置缺模型上下文会把不计费的
   探测放大成整账号停调。修复:CN 全协议一律本地 tiktoken 估算(与 Grok
   同方案),删除上游转发死代码。

5. 403 处置(P1):CN 此前落入通用 handleAuthError,单次 HTML 403(CDN/
   代理拦截页)即永久禁用,且 403 在 failover 集里会逐账号重放连环禁用
   整组。修复:CN 与 openai 同口径——HTML 豁免 + 3 次累计 + 临时冷却。

新增回归测试 8 项:闸门豁免(含 openai 仍受控断言)、CN 调度映射空返回、
候选过滤三态、空候选零成本落账、断开排水 usage 完整性、HTML-403 零处罚、
结构化 403 首次临时停调。handler/service 全包测试通过。

backend/internal/handler/openai_chat_completions.go

@@ -255,6 +255,48 @@ func (h *OpenAIGatewayHandler) ChatCompletions(c *gin.Context) {
 		if err == nil && result != nil && result.FirstTokenMs != nil {
 			service.SetOpsLatencyMs(c, service.OpsTimeToFirstTokenMsKey, int64(*result.FirstTokenMs))
 		}
+		// #5148 对齐：错误返回携带的部分 result（流中断前上游已计量的 usage）照常
+		// 入账；failover 错误恒定 result=nil，不会重复计费。
+		submitChatUsage := func(res *service.OpenAIForwardResult) {
+			if res == nil {
+				return
+			}
+			userAgent := c.GetHeader("User-Agent")
+			clientIP := ip.GetClientIP(c)
+			inboundEndpoint := GetInboundEndpoint(c)
+			upstreamEndpoint := resolveOpenAIUpstreamEndpoint(c, account, res)
+			quotaPlatform := service.QuotaPlatform(c.Request.Context(), apiKey)
+			sessionID := service.ExtractClientSessionID(c)
+			cyberBlocked := service.GetOpsCyberPolicy(c) != nil
+			h.submitOpenAIUsageRecordTask(c.Request.Context(), res, func(ctx context.Context) {
+				if err := h.gatewayService.RecordUsage(ctx, &service.OpenAIRecordUsageInput{
+					Result:             res,
+					APIKey:             apiKey,
+					User:               apiKey.User,
+					Account:            account,
+					Subscription:       subscription,
+					InboundEndpoint:    inboundEndpoint,
+					UpstreamEndpoint:   upstreamEndpoint,
+					UserAgent:          userAgent,
+					IPAddress:          clientIP,
+					APIKeyService:      h.apiKeyService,
+					QuotaPlatform:      quotaPlatform,
+					SessionID:          sessionID,
+					ChannelUsageFields: clientRequestedUsageFields(c, channelMapping, reqModel, res.UpstreamModel),
+					PricingAt:          pricingAt,
+					CyberBlocked:       cyberBlocked,
+				}); err != nil {
+					logger.L().With(
+						zap.String("component", "handler.openai_gateway.chat_completions"),
+						zap.Int64("user_id", subject.UserID),
+						zap.Int64("api_key_id", apiKey.ID),
+						zap.Any("group_id", apiKey.GroupID),
+						zap.String("model", reqModel),
+						zap.Int64("account_id", account.ID),
+					).Error("openai_chat_completions.record_usage_failed", zap.Error(err))
+				}
+			})
+		}
 		if err != nil {
 			if result != nil && result.ImageCount > 0 {
 				reqLog.Warn("openai_chat_completions.forward_partial_error_with_image_result",
@@ -339,6 +381,7 @@ func (h *OpenAIGatewayHandler) ChatCompletions(c *gin.Context) {
 					zap.Bool("upstream_error_response_already_written", upstreamErrorAlreadyCommunicated),
 					zap.Error(err),
 				)
+				submitChatUsage(result)
 				return
 			}
 		}
@@ -348,42 +391,7 @@ func (h *OpenAIGatewayHandler) ChatCompletions(c *gin.Context) {
 			h.gatewayService.ReportOpenAIAccountScheduleResult(account.ID, account.GetMappedModel(reqModel), true, nil)
 		}
 
-		userAgent := c.GetHeader("User-Agent")
-		clientIP := ip.GetClientIP(c)
-		inboundEndpoint := GetInboundEndpoint(c)
-		upstreamEndpoint := resolveOpenAIUpstreamEndpoint(c, account, result)
-		quotaPlatform := service.QuotaPlatform(c.Request.Context(), apiKey)
-		sessionID := service.ExtractClientSessionID(c)
-
-		cyberBlocked := service.GetOpsCyberPolicy(c) != nil
-		h.submitOpenAIUsageRecordTask(c.Request.Context(), result, func(ctx context.Context) {
-			if err := h.gatewayService.RecordUsage(ctx, &service.OpenAIRecordUsageInput{
-				Result:             result,
-				APIKey:             apiKey,
-				User:               apiKey.User,
-				Account:            account,
-				Subscription:       subscription,
-				InboundEndpoint:    inboundEndpoint,
-				UpstreamEndpoint:   upstreamEndpoint,
-				UserAgent:          userAgent,
-				IPAddress:          clientIP,
-				APIKeyService:      h.apiKeyService,
-				QuotaPlatform:      quotaPlatform,
-				SessionID:          sessionID,
-				ChannelUsageFields: clientRequestedUsageFields(c, channelMapping, reqModel, result.UpstreamModel),
-				PricingAt:          pricingAt,
-				CyberBlocked:       cyberBlocked,
-			}); err != nil {
-				logger.L().With(
-					zap.String("component", "handler.openai_gateway.chat_completions"),
-					zap.Int64("user_id", subject.UserID),
-					zap.Int64("api_key_id", apiKey.ID),
-					zap.Any("group_id", apiKey.GroupID),
-					zap.String("model", reqModel),
-					zap.Int64("account_id", account.ID),
-				).Error("openai_chat_completions.record_usage_failed", zap.Error(err))
-			}
-		})
+		submitChatUsage(result)
 		reqLog.Debug("openai_chat_completions.request_completed",
 			zap.Int64("account_id", account.ID),
 			zap.Int("switch_count", switchCount),

backend/internal/handler/openai_gateway_cn_dispatch_test.go

@@ -0,0 +1,29 @@
+package handler
+
+// CN 分组 /v1/messages 调度闸门回归（修复:正常途径创建的 CN 分组曾恒 403）：
+// sanitizeGroupMessagesDispatchFields 对非 openai 平台强制 AllowMessagesDispatch
+// =false，故 CN 分组必须与 grok 一样在闸门处豁免，否则原生 Anthropic 直通
+//（Claude Code 主用例）永远不可达。
+
+import (
+	"testing"
+
+	"github.com/Wei-Shaw/sub2api/internal/service"
+	"github.com/stretchr/testify/require"
+)
+
+func TestAllowOpenAICompatibleMessagesDispatch_CNProvidersExempt(t *testing.T) {
+	require.True(t, allowOpenAICompatibleMessagesDispatch(nil), "无 key 保持放行")
+
+	for _, platform := range []string{service.PlatformKimi, service.PlatformZhipu, service.PlatformDeepseek, service.PlatformGrok} {
+		apiKey := &service.APIKey{Group: &service.Group{Platform: platform, AllowMessagesDispatch: false}}
+		require.True(t, allowOpenAICompatibleMessagesDispatch(apiKey),
+			"%s 分组必须豁免 allow_messages_dispatch 闸门", platform)
+	}
+
+	// 非回归：openai 分组仍受开关控制。
+	openaiOff := &service.APIKey{Group: &service.Group{Platform: service.PlatformOpenAI, AllowMessagesDispatch: false}}
+	require.False(t, allowOpenAICompatibleMessagesDispatch(openaiOff))
+	openaiOn := &service.APIKey{Group: &service.Group{Platform: service.PlatformOpenAI, AllowMessagesDispatch: true}}
+	require.True(t, allowOpenAICompatibleMessagesDispatch(openaiOn))
+}

backend/internal/handler/openai_gateway_count_tokens.go

@@ -78,7 +78,7 @@ func (h *OpenAIGatewayHandler) CountTokens(c *gin.Context) {
 		zap.Any("group_id", apiKey.GroupID),
 	)
 
-	if apiKey.Group != nil && !apiKey.Group.AllowMessagesDispatch {
+	if !allowOpenAICompatibleMessagesDispatch(apiKey) {
 		h.anthropicErrorResponse(c, http.StatusForbidden, "permission_error",
 			"This group does not allow /v1/messages dispatch")
 		return

backend/internal/handler/openai_gateway_handler.go

@@ -197,6 +197,13 @@ func allowOpenAICompatibleMessagesDispatch(apiKey *service.APIKey) bool {
 	if apiKey.Group.Platform == service.PlatformGrok {
 		return true
 	}
+	// 国产供应商分组与 grok 同语义:/v1/messages 就是其主要服务形态(anthropic
+	// 协议账号原生直通 Claude Code),无需 allow_messages_dispatch 开关授权——
+	// 该开关对非 openai 平台恒被 sanitizeGroupMessagesDispatchFields 置 false,
+	// 若不豁免,CN 分组将永远 403。
+	if service.IsCNProvider(apiKey.Group.Platform) {
+		return true
+	}
 	return apiKey.Group.AllowMessagesDispatch
 }
 
@@ -597,6 +604,50 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) {
 		if err == nil && result != nil && result.FirstTokenMs != nil {
 			service.SetOpsLatencyMs(c, service.OpsTimeToFirstTokenMsKey, int64(*result.FirstTokenMs))
 		}
+		// #5148 对齐：错误返回携带的部分 result（流中断前上游已计量的 usage）照常
+		// 入账；failover 错误恒定 result=nil，不会重复计费。
+		submitResponsesUsage := func(res *service.OpenAIForwardResult) {
+			if res == nil {
+				return
+			}
+			userAgent := c.GetHeader("User-Agent")
+			clientIP := ip.GetClientIP(c)
+			requestPayloadHash := service.HashUsageRequestPayload(body)
+			inboundEndpoint := GetInboundEndpoint(c)
+			upstreamEndpoint := resolveOpenAIUpstreamEndpoint(c, account, res)
+			quotaPlatform := service.QuotaPlatform(c.Request.Context(), apiKey)
+			sessionID := service.ExtractClientSessionID(c)
+			cyberBlocked := service.GetOpsCyberPolicy(c) != nil
+			h.submitOpenAIUsageRecordTask(c.Request.Context(), res, func(ctx context.Context) {
+				if err := h.gatewayService.RecordUsage(ctx, &service.OpenAIRecordUsageInput{
+					Result:             res,
+					APIKey:             apiKey,
+					User:               apiKey.User,
+					Account:            account,
+					Subscription:       subscription,
+					InboundEndpoint:    inboundEndpoint,
+					UpstreamEndpoint:   upstreamEndpoint,
+					UserAgent:          userAgent,
+					IPAddress:          clientIP,
+					RequestPayloadHash: requestPayloadHash,
+					APIKeyService:      h.apiKeyService,
+					QuotaPlatform:      quotaPlatform,
+					SessionID:          sessionID,
+					ChannelUsageFields: clientRequestedUsageFields(c, channelMapping, reqModel, res.UpstreamModel),
+					PricingAt:          pricingAt,
+					CyberBlocked:       cyberBlocked,
+				}); err != nil {
+					logger.L().With(
+						zap.String("component", "handler.openai_gateway.responses"),
+						zap.Int64("user_id", subject.UserID),
+						zap.Int64("api_key_id", apiKey.ID),
+						zap.Any("group_id", apiKey.GroupID),
+						zap.String("model", reqModel),
+						zap.Int64("account_id", account.ID),
+					).Error("openai.record_usage_failed", zap.Error(err))
+				}
+			})
+		}
 		if err != nil {
 			if result != nil && result.ImageCount > 0 {
 				reqLog.Warn("openai.forward_partial_error_with_image_result",
@@ -696,6 +747,7 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) {
 					zap.Bool("upstream_error_response_already_written", upstreamErrorAlreadyCommunicated),
 					zap.Error(err),
 				}
+				submitResponsesUsage(result)
 				if shouldLogOpenAIForwardFailureAsWarn(c, wroteFallback) {
 					reqLog.Warn("openai.forward_failed", fields...)
 					return
@@ -714,46 +766,8 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) {
 			h.gatewayService.ReportOpenAIAccountScheduleResult(account.ID, account.GetMappedModel(reqModel), openAIForwardSucceededForScheduling(result), nil)
 		}
 
-		// 捕获请求信息（用于异步记录，避免在 goroutine 中访问 gin.Context）
-		userAgent := c.GetHeader("User-Agent")
-		clientIP := ip.GetClientIP(c)
-		requestPayloadHash := service.HashUsageRequestPayload(body)
-		inboundEndpoint := GetInboundEndpoint(c)
-		upstreamEndpoint := resolveOpenAIUpstreamEndpoint(c, account, result)
-		quotaPlatform := service.QuotaPlatform(c.Request.Context(), apiKey)
-		sessionID := service.ExtractClientSessionID(c)
-
 		// 使用量记录通过有界 worker 池提交，避免请求热路径创建无界 goroutine。
-		cyberBlocked := service.GetOpsCyberPolicy(c) != nil
-		h.submitOpenAIUsageRecordTask(c.Request.Context(), result, func(ctx context.Context) {
-			if err := h.gatewayService.RecordUsage(ctx, &service.OpenAIRecordUsageInput{
-				Result:             result,
-				APIKey:             apiKey,
-				User:               apiKey.User,
-				Account:            account,
-				Subscription:       subscription,
-				InboundEndpoint:    inboundEndpoint,
-				UpstreamEndpoint:   upstreamEndpoint,
-				UserAgent:          userAgent,
-				IPAddress:          clientIP,
-				RequestPayloadHash: requestPayloadHash,
-				APIKeyService:      h.apiKeyService,
-				QuotaPlatform:      quotaPlatform,
-				SessionID:          sessionID,
-				ChannelUsageFields: clientRequestedUsageFields(c, channelMapping, reqModel, result.UpstreamModel),
-				PricingAt:          pricingAt,
-				CyberBlocked:       cyberBlocked,
-			}); err != nil {
-				logger.L().With(
-					zap.String("component", "handler.openai_gateway.responses"),
-					zap.Int64("user_id", subject.UserID),
-					zap.Int64("api_key_id", apiKey.ID),
-					zap.Any("group_id", apiKey.GroupID),
-					zap.String("model", reqModel),
-					zap.Int64("account_id", account.ID),
-				).Error("openai.record_usage_failed", zap.Error(err))
-			}
-		})
+		submitResponsesUsage(result)
 		reqLog.Debug("openai.request_completed",
 			zap.Int64("account_id", account.ID),
 			zap.Int("switch_count", switchCount),
@@ -1145,6 +1159,51 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) {
 		if err == nil && result != nil && result.FirstTokenMs != nil {
 			service.SetOpsLatencyMs(c, service.OpsTimeToFirstTokenMsKey, int64(*result.FirstTokenMs))
 		}
+		// Forward 与错误一起返回的部分结果：流中断/客户端断开排水前上游已计量的
+		// usage 照常入账，避免上游已产生消耗的请求完全漏记（#5148，对齐 anthropic
+		// 网关同名修复）。failover 错误恒定 result=nil，不会重复计费。
+		submitMessagesUsage := func(res *service.OpenAIForwardResult) {
+			if res == nil {
+				return
+			}
+			userAgent := c.GetHeader("User-Agent")
+			clientIP := ip.GetClientIP(c)
+			requestPayloadHash := service.HashUsageRequestPayload(body)
+			inboundEndpoint := GetInboundEndpoint(c)
+			upstreamEndpoint := resolveOpenAIUpstreamEndpoint(c, account, res)
+			quotaPlatform := service.QuotaPlatform(c.Request.Context(), apiKey)
+			sessionID := service.ExtractClientSessionID(c)
+			cyberBlocked := service.GetOpsCyberPolicy(c) != nil
+			h.submitOpenAIUsageRecordTask(c.Request.Context(), res, func(ctx context.Context) {
+				if err := h.gatewayService.RecordUsage(ctx, &service.OpenAIRecordUsageInput{
+					Result:             res,
+					APIKey:             apiKey,
+					User:               apiKey.User,
+					Account:            account,
+					Subscription:       subscription,
+					InboundEndpoint:    inboundEndpoint,
+					UpstreamEndpoint:   upstreamEndpoint,
+					UserAgent:          userAgent,
+					IPAddress:          clientIP,
+					RequestPayloadHash: requestPayloadHash,
+					APIKeyService:      h.apiKeyService,
+					QuotaPlatform:      quotaPlatform,
+					SessionID:          sessionID,
+					ChannelUsageFields: clientRequestedUsageFields(c, channelMappingMsg, reqModel, res.UpstreamModel),
+					PricingAt:          pricingAt,
+					CyberBlocked:       cyberBlocked,
+				}); err != nil {
+					logger.L().With(
+						zap.String("component", "handler.openai_gateway.messages"),
+						zap.Int64("user_id", subject.UserID),
+						zap.Int64("api_key_id", apiKey.ID),
+						zap.Any("group_id", apiKey.GroupID),
+						zap.String("model", reqModel),
+						zap.Int64("account_id", account.ID),
+					).Error("openai_messages.record_usage_failed", zap.Error(err))
+				}
+			})
+		}
 		if err != nil {
 			if result != nil && result.ImageCount > 0 {
 				reqLog.Warn("openai_messages.forward_partial_error_with_image_result",
@@ -1219,6 +1278,9 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) {
 						zap.Int64("account_id", account.ID),
 						zap.Error(err),
 					)
+					// 断开排水期间上游已计量的 usage 必须入账（此前直接 return 丢弃，
+					// payg 上游照常计费而平台漏记）。
+					submitMessagesUsage(result)
 					return
 				}
 				h.gatewayService.ReportOpenAIAccountScheduleResult(account.ID, account.GetMappedModel(currentRoutingModel), false, nil)
@@ -1228,6 +1290,7 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) {
 					zap.Bool("fallback_error_response_written", wroteFallback),
 					zap.Error(err),
 				)
+				submitMessagesUsage(result)
 				return
 			}
 		}
@@ -1237,44 +1300,7 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) {
 			h.gatewayService.ReportOpenAIAccountScheduleResult(account.ID, account.GetMappedModel(currentRoutingModel), true, nil)
 		}
 
-		userAgent := c.GetHeader("User-Agent")
-		clientIP := ip.GetClientIP(c)
-		requestPayloadHash := service.HashUsageRequestPayload(body)
-		inboundEndpoint := GetInboundEndpoint(c)
-		upstreamEndpoint := resolveOpenAIUpstreamEndpoint(c, account, result)
-		quotaPlatform := service.QuotaPlatform(c.Request.Context(), apiKey)
-		sessionID := service.ExtractClientSessionID(c)
-
-		cyberBlocked := service.GetOpsCyberPolicy(c) != nil
-		h.submitOpenAIUsageRecordTask(c.Request.Context(), result, func(ctx context.Context) {
-			if err := h.gatewayService.RecordUsage(ctx, &service.OpenAIRecordUsageInput{
-				Result:             result,
-				APIKey:             apiKey,
-				User:               apiKey.User,
-				Account:            account,
-				Subscription:       subscription,
-				InboundEndpoint:    inboundEndpoint,
-				UpstreamEndpoint:   upstreamEndpoint,
-				UserAgent:          userAgent,
-				IPAddress:          clientIP,
-				RequestPayloadHash: requestPayloadHash,
-				APIKeyService:      h.apiKeyService,
-				QuotaPlatform:      quotaPlatform,
-				SessionID:          sessionID,
-				ChannelUsageFields: clientRequestedUsageFields(c, channelMappingMsg, reqModel, result.UpstreamModel),
-				PricingAt:          pricingAt,
-				CyberBlocked:       cyberBlocked,
-			}); err != nil {
-				logger.L().With(
-					zap.String("component", "handler.openai_gateway.messages"),
-					zap.Int64("user_id", subject.UserID),
-					zap.Int64("api_key_id", apiKey.ID),
-					zap.Any("group_id", apiKey.GroupID),
-					zap.String("model", reqModel),
-					zap.Int64("account_id", account.ID),
-				).Error("openai_messages.record_usage_failed", zap.Error(err))
-			}
-		})
+		submitMessagesUsage(result)
 		reqLog.Debug("openai_messages.request_completed",
 			zap.Int64("account_id", account.ID),
 			zap.Int("switch_count", switchCount),

backend/internal/service/openai_gateway_cn_fixes_test.go

@@ -0,0 +1,144 @@
+//go:build unit
+
+package service
+
+// 国产供应商功能修复回归测试：
+//  1. CN 分组不适用 /v1/messages 调度级模型映射（openai 的 gpt-5.x 默认值发给
+//     CN 上游必错）；
+//  2. 计费候选链对 CN 账号过滤 claude-* 兜底候选（防按 Claude 原价误计 CN 流量）；
+//  3. 空候选按 ErrModelPricingUnavailable 处理（零成本落账而非丢弃 usage 记录）；
+//  4. Responses×anthropic 流式转换器客户端断开后继续排水、usage 汇总完整。
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/Wei-Shaw/sub2api/internal/config"
+	"github.com/Wei-Shaw/sub2api/internal/pkg/apicompat"
+	"github.com/gin-gonic/gin"
+	"github.com/stretchr/testify/require"
+)
+
+func TestResolveMessagesDispatchModel_CNProvidersNoDispatchMapping(t *testing.T) {
+	for _, platform := range []string{PlatformKimi, PlatformZhipu, PlatformDeepseek} {
+		g := &Group{Platform: platform}
+		require.Empty(t, g.ResolveMessagesDispatchModel("claude-sonnet-4-5"),
+			"CN 分组(%s)不得返回调度级映射模型（openai 默认值会发给 CN 上游）", platform)
+		require.Empty(t, g.ResolveMessagesDispatchModel("claude-opus-4-1"), platform)
+	}
+	// 非回归：openai 分组保持原有默认映射行为。
+	openaiGroup := &Group{Platform: PlatformOpenAI}
+	require.NotEmpty(t, openaiGroup.ResolveMessagesDispatchModel("claude-sonnet-4-5"),
+		"openai 分组的调度默认映射不应受 CN 修复影响")
+}
+
+func TestFilterCNProviderBillingModelCandidates(t *testing.T) {
+	svc := &OpenAIGatewayService{} // resolver 为 nil → 无显式分组/渠道定价
+	apiKey := &APIKey{Group: &Group{ID: 1, Platform: PlatformKimi}}
+
+	cnAccount := &Account{ID: 1, Platform: PlatformKimi}
+	filtered := svc.filterCNProviderBillingModelCandidates(context.Background(), cnAccount, apiKey,
+		[]string{"kimi-k2-0905-preview", "claude-sonnet-4-5", "moonshot-v1-8k"})
+	require.Equal(t, []string{"kimi-k2-0905-preview", "moonshot-v1-8k"}, filtered,
+		"无显式定价时 claude-* 候选必须被过滤")
+
+	allClaude := svc.filterCNProviderBillingModelCandidates(context.Background(), cnAccount, apiKey,
+		[]string{"claude-sonnet-4-5", "claude-sonnet-4-5"})
+	require.Empty(t, allClaude, "全 claude 候选应被清空（上层走零成本+告警落账）")
+
+	// 非 CN 账号完全不受影响。
+	openaiAccount := &Account{ID: 2, Platform: PlatformOpenAI}
+	passthrough := svc.filterCNProviderBillingModelCandidates(context.Background(), openaiAccount, apiKey,
+		[]string{"claude-sonnet-4-5", "gpt-5.4"})
+	require.Equal(t, []string{"claude-sonnet-4-5", "gpt-5.4"}, passthrough)
+
+	require.Nil(t, svc.filterCNProviderBillingModelCandidates(context.Background(), nil, apiKey, nil))
+}
+
+func TestCalculateOpenAIRecordUsageCost_EmptyCandidatesIsPricingUnavailable(t *testing.T) {
+	svc := &OpenAIGatewayService{}
+	apiKey := &APIKey{Group: &Group{ID: 1, Platform: PlatformKimi}}
+
+	_, err := svc.calculateOpenAIRecordUsageCost(
+		context.Background(), nil, apiKey, nil,
+		1.0, 1.0, 1.0, 1.0, UsageTokens{InputTokens: 100}, "", nil,
+	)
+	require.Error(t, err)
+	require.True(t, isUsagePricingUnavailableError(err),
+		"空候选必须按无价可循处理（上层零成本落账），而不是丢弃整条 usage 记录: %v", err)
+}
+
+func TestResponsesStreamingFromNativeAnthropic_ClientDisconnectDrainsUsage(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	svc := newNativeAnthropicHangTestService(5)
+
+	rec := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(rec)
+	c.Request = httptest.NewRequest(http.MethodPost, "/", nil)
+	// failAfter=0：首次写出即失败，模拟客户端断开（复用测试包既有 failingGinWriter）。
+	failWriter := &failingGinWriter{ResponseWriter: c.Writer, failAfter: 0}
+	c.Writer = failWriter
+
+	resp, pr, pw := newHangingUpstreamResponse()
+	go func() {
+		// 首事件触发客户端写失败后，末尾 message_delta 才携带最终 output_tokens：
+		// 断开即弃会把整段生成记成 1 token。
+		_, _ = pw.Write([]byte(miniAnthropicSSEStream()))
+		_ = pw.Close()
+	}()
+	defer func() { _ = pr.Close() }()
+
+	res, err := svc.handleResponsesStreamingFromNativeAnthropic(
+		resp, c, "glm-4.7", "glm-4.7", "glm-4.7", nil, time.Now(), apicompat.ResponsesClientToolMapping{})
+
+	require.NoError(t, err, "断开排水至上游自然结束应返回 nil error（usage 走成功路径落账）")
+	require.NotNil(t, res)
+	require.True(t, res.ClientDisconnect)
+	require.Equal(t, 10, res.Usage.InputTokens, "input_tokens 应来自 message_start")
+	require.Equal(t, 5, res.Usage.OutputTokens,
+		"output_tokens 必须来自排水读到的末尾 message_delta（断开即弃时会是 1）")
+}
+
+func TestHandle403_CNProviderHTMLBodySkipsAccountPenalty(t *testing.T) {
+	for _, platform := range []string{PlatformKimi, PlatformZhipu, PlatformDeepseek} {
+		repo := &rateLimitAccountRepoStub{}
+		service := NewRateLimitService(repo, nil, &config.Config{}, nil, nil)
+		account := &Account{ID: 401, Platform: platform, Type: AccountTypeAPIKey}
+
+		shouldDisable := service.HandleUpstreamError(
+			context.Background(),
+			account,
+			http.StatusForbidden,
+			http.Header{},
+			[]byte("<html><body>Access denied by CDN</body></html>"),
+		)
+
+		require.False(t, shouldDisable, "%s: HTML 403（CDN/代理拦截页）不得作为账号失效证据", platform)
+		require.Equal(t, 0, repo.setErrorCalls, "%s: 不得永久禁用账号", platform)
+		require.Equal(t, 0, repo.tempCalls, "%s: 不得临时停调账号", platform)
+	}
+}
+
+func TestHandle403_CNProviderStructured403TempUnschedulableFirstHit(t *testing.T) {
+	repo := &rateLimitAccountRepoStub{}
+	counter := &openAI403CounterCacheStub{counts: []int64{1}}
+	service := NewRateLimitService(repo, nil, &config.Config{}, nil, nil)
+	service.SetOpenAI403CounterCache(counter)
+	account := &Account{ID: 402, Platform: PlatformKimi, Type: AccountTypeAPIKey}
+
+	shouldDisable := service.HandleUpstreamError(
+		context.Background(),
+		account,
+		http.StatusForbidden,
+		http.Header{},
+		[]byte(`{"error":{"message":"forbidden"}}`),
+	)
+
+	require.True(t, shouldDisable)
+	require.Equal(t, 0, repo.setErrorCalls, "首次结构化 403 应临时停调而非永久禁用")
+	require.Equal(t, 1, repo.tempCalls)
+	require.Contains(t, repo.lastTempReason, "(1/3)")
+}

backend/internal/service/openai_gateway_count_tokens.go

@@ -92,15 +92,13 @@ func (s *OpenAIGatewayService) ForwardCountTokensAsAnthropic(
 		return fmt.Errorf("count_tokens: missing account")
 	}
 
-	// 国产供应商 Anthropic 协议：上游有原生 /v1/messages/count_tokens 端点，
-	// 直接透传（仅模型名映射），不走 /v1/responses/input_tokens 估算。
-	if account.IsAnthropicProtocol() {
-		return s.forwardCountTokensViaNativeAnthropic(ctx, c, account, body, defaultMappedModel)
-	}
-
-	// 国产供应商其余协议（chat_completions / responses）：三家上游均无
-	// OpenAI 兼容的 /v1/responses/input_tokens 端点，与 Grok 一样本地估算，
-	// 不发上游请求（Claude Code 客户端会高频调用 count_tokens）。
+	// 国产供应商（全部协议，含 anthropic）：一律本地估算，不发上游请求。
+	// 依据（2026-08 核实）：三家的 Anthropic 兼容层均未提供
+	// /v1/messages/count_tokens——DeepSeek 官方 anthropic_api 文档无此端点
+	// （且注明 anthropic-version 头被忽略），聚合网关 OpenModel 明确标注
+	// count_tokens 为 "Anthropic only"，Kimi/智谱亦无任何文档承诺。转发上游
+	// 只会常态 404，且错误还会流入账号处置逻辑误伤整账号调度；Claude Code
+	// 高频调用此端点，本地 tiktoken 估算是与 Grok 一致的既有方案。
 	if account.IsCNProvider() {
 		estimated, err := estimateAnthropicCountTokensLocally(body)
 		if err != nil {

backend/internal/service/openai_gateway_messages_anthropic_native.go

@@ -515,95 +515,6 @@ func (s *OpenAIGatewayService) nativeAnthropicStreamResult(
 	}
 }
 
-// forwardCountTokensViaNativeAnthropic 把 Anthropic count_tokens 请求透传到
-// 国产供应商原生 Anthropic 端点（{base}/v1/messages/count_tokens），
-// 仅做模型名映射，不做协议转换。
-func (s *OpenAIGatewayService) forwardCountTokensViaNativeAnthropic(
-	ctx context.Context,
-	c *gin.Context,
-	account *Account,
-	body []byte,
-	defaultMappedModel string,
-) error {
-	originalModel := strings.TrimSpace(gjson.GetBytes(body, "model").String())
-	if originalModel == "" {
-		writeAnthropicCountTokensError(c, http.StatusBadRequest, "invalid_request_error", "model is required")
-		return fmt.Errorf("count_tokens: missing model in request")
-	}
-	billingModel := resolveOpenAIForwardModel(account, originalModel, strings.TrimSpace(defaultMappedModel))
-	upstreamModel := normalizeOpenAIModelForUpstream(account, billingModel)
-	if upstreamModel != originalModel {
-		rewritten, err := sjson.SetBytes(body, "model", upstreamModel)
-		if err != nil {
-			return fmt.Errorf("count_tokens: rewrite model: %w", err)
-		}
-		body = rewritten
-	}
-
-	apiKey := strings.TrimSpace(account.GetOpenAIProtocolAPIKey())
-	if apiKey == "" {
-		writeAnthropicCountTokensError(c, http.StatusBadGateway, "upstream_error", "Account api_key is missing")
-		return fmt.Errorf("count_tokens: account %d missing api_key", account.ID)
-	}
-	targetURL, err := s.nativeAnthropicTargetURL(account)
-	if err != nil {
-		return fmt.Errorf("count_tokens: %w", err)
-	}
-	targetURL = strings.TrimSuffix(targetURL, "/v1/messages") + "/v1/messages/count_tokens"
-
-	upstreamReq, err := http.NewRequestWithContext(ctx, http.MethodPost, targetURL, bytes.NewReader(body))
-	if err != nil {
-		writeAnthropicCountTokensError(c, http.StatusInternalServerError, "api_error", "Failed to build request")
-		return fmt.Errorf("count_tokens: build request: %w", err)
-	}
-	reqHeader := upstreamReq.Header
-	reqHeader.Del("authorization")
-	reqHeader.Del("x-api-key")
-	setAnthropicAPIKeyAuthHeader(reqHeader, account, apiKey)
-	reqHeader.Set("content-type", "application/json")
-	reqHeader.Set("accept", "application/json")
-	account.ApplyHeaderOverrides(reqHeader)
-
-	proxyURL := ""
-	if account.Proxy != nil {
-		proxyURL = account.Proxy.URL()
-	}
-	resp, err := s.httpUpstream.Do(upstreamReq, proxyURL, account.ID, account.Concurrency)
-	if err != nil {
-		safeErr := sanitizeUpstreamErrorMessage(err.Error())
-		setOpsUpstreamError(c, 0, safeErr, "")
-		writeAnthropicCountTokensError(c, http.StatusBadGateway, "upstream_error", "Upstream request failed")
-		return fmt.Errorf("count_tokens: upstream request failed: %s", safeErr)
-	}
-	defer func() { _ = resp.Body.Close() }()
-
-	// count_tokens 响应体极小；与其他探测路径一致加 256KB 上限防异常上游放大内存。
-	respBody, err := io.ReadAll(io.LimitReader(resp.Body, cnQuotaMaxBodyBytes))
-	if err != nil {
-		writeAnthropicCountTokensError(c, http.StatusBadGateway, "upstream_error", "Failed to read response")
-		return fmt.Errorf("count_tokens: read response: %w", err)
-	}
-	if resp.StatusCode >= 400 {
-		if s.rateLimitService != nil {
-			s.rateLimitService.HandleUpstreamError(ctx, account, resp.StatusCode, resp.Header, respBody)
-		}
-		upstreamMsg := sanitizeUpstreamErrorMessage(strings.TrimSpace(extractUpstreamErrorMessage(respBody)))
-		setOpsUpstreamError(c, resp.StatusCode, upstreamMsg, "")
-		writeAnthropicCountTokensError(c, resp.StatusCode, "upstream_error", "Upstream request failed")
-		return fmt.Errorf("count_tokens: upstream error: %d", resp.StatusCode)
-	}
-
-	inputTokens := gjson.GetBytes(respBody, "input_tokens")
-	if !inputTokens.Exists() {
-		writeAnthropicCountTokensError(c, http.StatusBadGateway, "upstream_error", "Upstream response missing input_tokens")
-		return fmt.Errorf("count_tokens: response missing input_tokens field")
-	}
-	c.JSON(http.StatusOK, gin.H{
-		"input_tokens": int(inputTokens.Int()),
-	})
-	return nil
-}
-
 // claudeUsageToOpenAIUsage 把 Anthropic 格式 usage 映射到 OpenAI 网关统一的
 // 用量结构（字段一一对应）。
 func claudeUsageToOpenAIUsage(u *ClaudeUsage) OpenAIUsage {

backend/internal/service/openai_gateway_responses_anthropic_native.go

@@ -377,7 +377,11 @@ func (s *OpenAIGatewayService) handleResponsesStreamingFromNativeAnthropic(
 		return resultWithUsage(), fmt.Errorf("stream data interval timeout")
 	}
 
-	processAnthropicEvent := func(event *apicompat.AnthropicStreamEvent) bool {
+	// 与 CC 姊妹路径（handleCCStreamingFromNativeAnthropic.writeChunk）同语义：
+	// 客户端断开后不再写出，但继续排水上游至流自然结束——Anthropic 的最终
+	// output_tokens 只在末尾 message_delta 携带，提前退出会把整段生成记成 ~1
+	// token，payg 上游照常计费而平台漏记。状态机照常推进以保证 finalize 一致。
+	processAnthropicEvent := func(event *apicompat.AnthropicStreamEvent) {
 		if firstChunk {
 			firstChunk = false
 			ms := int(time.Since(startTime).Milliseconds())
@@ -392,6 +396,9 @@ func (s *OpenAIGatewayService) handleResponsesStreamingFromNativeAnthropic(
 		}
 
 		events := apicompat.AnthropicEventToResponsesEvents(event, state)
+		if clientDisconnected {
+			return
+		}
 		for _, evt := range events {
 			payload, err := json.Marshal(evt)
 			if err != nil {
@@ -406,14 +413,13 @@ func (s *OpenAIGatewayService) handleResponsesStreamingFromNativeAnthropic(
 				eventType := gjson.GetBytes(restored, "type").String()
 				if _, err := fmt.Fprintf(c.Writer, "event: %s\ndata: %s\n\n", eventType, restored); err != nil {
 					clientDisconnected = true
-					return true
+					return
 				}
 			}
 		}
 		if len(events) > 0 {
 			c.Writer.Flush()
 		}
-		return false
 	}
 
 	for {
@@ -447,21 +453,39 @@ func (s *OpenAIGatewayService) handleResponsesStreamingFromNativeAnthropic(
 			continue
 		}
 
-		if processAnthropicEvent(&event) {
-			return resultWithUsage(), nil
-		}
+		processAnthropicEvent(&event)
 	}
 
-	// Finalize state machine（客户端已断开时仍执行，保证 usage 汇总完整）。
-	if finalEvents := apicompat.FinalizeAnthropicResponsesStream(state); len(finalEvents) > 0 {
+	// Finalize state machine（客户端已断开时仍推进，保证 usage 汇总完整；仅在
+	// 客户端仍连接时写出）。终态帧与逐事件路径一致过工具名反转与客户端工具还原，
+	// 避免流截断时终态帧携带改写后的工具名。
+	if finalEvents := apicompat.FinalizeAnthropicResponsesStream(state); len(finalEvents) > 0 && !clientDisconnected {
+		wrote := false
 		for _, evt := range finalEvents {
-			sse, err := apicompat.ResponsesEventToSSE(evt)
+			payload, err := json.Marshal(evt)
 			if err != nil {
 				continue
 			}
-			fmt.Fprint(c.Writer, sse) //nolint:errcheck
+			payload = reverseToolNamesIfPresent(c, payload)
+			payloads, _, err := clientToolRestorer.RestoreEvent(payload)
+			if err != nil {
+				continue
+			}
+			for _, restored := range payloads {
+				eventType := gjson.GetBytes(restored, "type").String()
+				if _, err := fmt.Fprintf(c.Writer, "event: %s\ndata: %s\n\n", eventType, restored); err != nil {
+					clientDisconnected = true
+					break
+				}
+				wrote = true
+			}
+			if clientDisconnected {
+				break
+			}
+		}
+		if wrote {
+			c.Writer.Flush()
 		}
-		c.Writer.Flush()
 	}
 
 	return resultWithUsage(), nil

backend/internal/service/openai_gateway_usage.go

@@ -200,6 +200,7 @@ func (s *OpenAIGatewayService) RecordUsage(ctx context.Context, input *OpenAIRec
 		result.UpstreamModel,
 		result.Model,
 	)
+	billingModels = s.filterCNProviderBillingModelCandidates(ctx, account, apiKey, billingModels)
 	serviceTier := ""
 	if result.ServiceTier != nil {
 		serviceTier = strings.TrimSpace(*result.ServiceTier)
@@ -253,7 +254,7 @@ func (s *OpenAIGatewayService) RecordUsage(ctx context.Context, input *OpenAIRec
 			result.AudioUsage != nil || result.SearchCount > 0,
 	); responseModel != "" && !strings.EqualFold(responseModel, baselineBillingModel) {
 		if identified, responseChannelPriced := s.hasIdentifiedOpenAIResponsePricing(ctx, responseModel, apiKey); identified {
-			responseModels := usageBillingModelCandidates(responseModel)
+			responseModels := s.filterCNProviderBillingModelCandidates(ctx, account, apiKey, usageBillingModelCandidates(responseModel))
 			responseCost, responseErr := s.calculateOpenAIRecordUsageCost(
 				ctx, result, apiKey, responseModels, multiplier, imageMultiplier,
 				videoMultiplier, baseMultiplier, tokens, serviceTier, longContextBillingGate,
@@ -585,7 +586,7 @@ func (s *OpenAIGatewayService) calculateOpenAIRecordUsageCost(
 	if tokenCost == nil {
 		if tokenBillingAttempted {
 			if lastErr == nil {
-				lastErr = errors.New("no non-empty billing model candidates")
+				lastErr = fmt.Errorf("%w: no non-empty billing model candidates", ErrModelPricingUnavailable)
 			}
 			return nil, fmt.Errorf("calculate OpenAI usage cost failed for billing models %s: %w", strings.Join(billingModels, ","), lastErr)
 		}
@@ -593,8 +594,11 @@ func (s *OpenAIGatewayService) calculateOpenAIRecordUsageCost(
 		if searchCost != nil {
 			return searchCost, nil
 		}
+		// 空候选按「无价可循」处理并携带 ErrModelPricingUnavailable：上层据此走
+		// 零成本+告警落账，而不是丢弃整条 usage 记录。CN 账号的 claude-* 候选被
+		// filterCNProviderBillingModelCandidates 全数过滤后即落到这里。
 		if lastErr == nil {
-			lastErr = errors.New("openai usage billing model is empty")
+			lastErr = fmt.Errorf("%w: openai usage billing model is empty", ErrModelPricingUnavailable)
 		}
 		return nil, fmt.Errorf("calculate OpenAI usage cost failed for billing models %s: %w", strings.Join(billingModels, ","), lastErr)
 	}
@@ -841,6 +845,36 @@ func groupMediaPricingLooksIncomplete(group *Group) bool {
 		group.VideoPrice480P == nil && group.VideoPrice720P == nil && group.VideoPrice1080P == nil
 }
 
+// filterCNProviderBillingModelCandidates 过滤国产供应商（kimi/zhipu/deepseek）
+// 账号的计费候选模型名：claude-* 候选仅在运营者显式配置了分组/渠道定价时保留。
+//
+// 背景：候选链的兜底候选含客户端请求的原始模型名。CN 上游的 Anthropic 兼容端点
+// 接受 claude-* 模型名但从不真正服务 Claude 模型；若放行，目录里的 Claude 价卡
+// 与 getFallbackPricing 的 "claude"→Sonnet 统一兜底会把 CN 流量按 Claude 原价
+// （数倍～数十倍）静默误计，且 usage 日志显示的正是 claude-* 名，无从察觉。
+// 候选全部落空时走既有的零成本+告警路径（openai_usage.pricing_missing_record_
+// zero_cost），与定价层「未知型号不回退以避免误计价」的既有设计意图一致；
+// 运营者的修复手段是配置账号级 model_mapping（映射到已定价的 CN 模型）或
+// 分组/渠道显式定价。
+func (s *OpenAIGatewayService) filterCNProviderBillingModelCandidates(ctx context.Context, account *Account, apiKey *APIKey, candidates []string) []string {
+	if account == nil || !account.IsCNProvider() {
+		return candidates
+	}
+	out := make([]string, 0, len(candidates))
+	for _, candidate := range candidates {
+		trimmed := strings.TrimSpace(candidate)
+		if trimmed == "" {
+			continue
+		}
+		if strings.Contains(strings.ToLower(trimmed), "claude") &&
+			s.resolveOpenAIChannelPricing(ctx, trimmed, apiKey) == nil {
+			continue
+		}
+		out = append(out, candidate)
+	}
+	return out
+}
+
 func (s *OpenAIGatewayService) resolveOpenAIChannelPricing(ctx context.Context, billingModel string, apiKey *APIKey) *ResolvedPricing {
 	if s.resolver == nil || apiKey == nil || apiKey.Group == nil {
 		return nil

backend/internal/service/openai_messages_dispatch.go

@@ -79,6 +79,13 @@ func (g *Group) ResolveMessagesDispatchModel(requestedModel string) string {
 		return xai.ModelMappingWithOptions(opts)["claude-*"]
 	}
 
+	// 国产供应商分组:调度级模型映射不适用(其配置被 sanitize 置空,且下方的
+	// gpt-5.x 默认值是 openai 专属,发给 CN 上游必错)。模型改写完全交给账号级
+	// model_mapping;anthropic 协议上游本身接受 claude-* 模型名。
+	if IsCNProvider(g.Platform) {
+		return ""
+	}
+
 	cfg := normalizeOpenAIMessagesDispatchModelConfig(g.MessagesDispatchModelConfig)
 	if mappedModel := strings.TrimSpace(cfg.ExactModelMappings[requestedModel]); mappedModel != "" {
 		return mappedModel

backend/internal/service/ratelimit_service.go

@@ -906,7 +906,10 @@ func (s *RateLimitService) handle403(ctx context.Context, account *Account, upst
 	if account.Platform == PlatformAntigravity {
 		return s.handleAntigravity403(ctx, account, upstreamMsg, responseBody)
 	}
-	if account.Platform == PlatformOpenAI {
+	// 国产供应商与 openai 同口径:HTML 403(CDN/代理拦截页)不构成账号失效证据,
+	// 且 403 在 failover 状态集里会被逐账号重放——直接 SetError 会让一个坏请求/
+	// 一层坏代理连环永久禁用整组账号。走 HTML 豁免 + N 次累计 + 临时冷却。
+	if account.Platform == PlatformOpenAI || IsCNProvider(account.Platform) {
 		return s.handleOpenAI403(ctx, account, upstreamMsg, responseBody)
 	}
 	// 非 Antigravity 平台：保持原有行为