feat(billing): 统一计价与候选冻结估算

新增 effective-pricing-v2、9 位十进制定点计算、显式免费校验和结构化缺价错误。估价与生产预处理覆盖全部可用候选，并按最大候选费用冻结；规则优先级为平台模型、平台、基准模型。\n\n同步计价契约和 OpenAPI，补充 Token 参数别名、视频五秒向上取整及缺价回归测试。\n\n验证：go test ./...、pnpm openapi、Web 测试与构建通过。
2026-07-20 23:22:45 +08:00
parent 01a013c809
commit 5114686c35
14 changed files with 1192 additions and 28 deletions
@@ -258,6 +258,44 @@ func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *aut
 			return Result{Task: failed, Output: failed.Result}, err
 		}
 	}
+	pricingByCandidate := map[string]resolvedPricing{}
+	reservationBillings := []any(nil)
+	if task.RunMode == "production" {
+		pricedCandidates := make([]store.RuntimeModelCandidate, 0, len(candidates))
+		estimates := make([]candidateEstimate, 0, len(candidates))
+		var pricingErr error
+		for _, candidate := range candidates {
+			pricingBody := preprocessRequest(task.Kind, cloneMap(body), candidate)
+			estimate, estimateErr := s.estimateCandidateV2(ctx, user, task.Kind, pricingBody, candidate)
+			if estimateErr != nil {
+				pricingErr = estimateErr
+				continue
+			}
+			pricedCandidates = append(pricedCandidates, candidate)
+			estimates = append(estimates, estimate)
+			pricingByCandidate[pricingCandidateKey(candidate)] = estimate.Pricing
+		}
+		if len(pricedCandidates) == 0 {
+			if pricingErr == nil {
+				pricingErr = &PricingUnavailableError{Reason: "no candidate has effective pricing"}
+			}
+			failed, finishErr := s.failTask(ctx, task.ID, "pricing_unavailable", pricingErr.Error(), false, pricingErr)
+			if finishErr != nil {
+				return Result{}, finishErr
+			}
+			return Result{Task: failed, Output: failed.Result}, pricingErr
+		}
+		maximumEstimate, estimateErr := maximumCandidateEstimate(estimates)
+		if estimateErr != nil {
+			failed, finishErr := s.failTask(ctx, task.ID, "pricing_unavailable", estimateErr.Error(), false, estimateErr)
+			if finishErr != nil {
+				return Result{}, finishErr
+			}
+			return Result{Task: failed, Output: failed.Result}, estimateErr
+		}
+		candidates = pricedCandidates
+		reservationBillings = maximumEstimate.Items
+	}
 	firstCandidateBody := body
 	normalizedModelType := modelType
 	attemptNo := task.AttemptCount
@@ -299,9 +337,8 @@ func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *aut
 		if err := s.store.MarkTaskRunning(ctx, task.ID, candidates[0].ModelType, s.slimTaskRequestSnapshot(task, firstCandidateBody)); err != nil {
 			return Result{}, err
 		}
-		estimatedBillings := s.estimatedBillings(ctx, user, task.Kind, firstCandidateBody, candidates[0])
 		var reserveErr error
-		walletReservations, reserveErr = s.store.ReserveTaskBilling(ctx, task, user, estimatedBillings)
+		walletReservations, reserveErr = s.store.ReserveTaskBilling(ctx, task, user, reservationBillings)
 		if reserveErr != nil {
 			if errors.Is(reserveErr, store.ErrInsufficientWalletBalance) {
 				attemptNo = s.recordFailedAttempt(ctx, failedAttemptRecord{
@@ -372,7 +409,19 @@ candidatesLoop:
 			response, err := s.runCandidate(ctx, task, user, candidateBody, preprocessing.Log, candidate, nextAttemptNo, onDelta, responseExecution, singleSourceProtected, runnerPolicy.CacheAffinityPolicy, cacheAffinityKeys.Record)
 			if err == nil {
 				attemptNo = nextAttemptNo
-				billings := s.billings(ctx, user, task.Kind, candidateBody, candidate, response, isSimulation(task, candidate))
+				var billings []any
+				if task.RunMode == "production" {
+					pricing := pricingByCandidate[pricingCandidateKey(candidate)]
+					var billingErr error
+					billings, _, _, billingErr = s.billingsWithResolvedPricingV2(ctx, user, task.Kind, candidateBody, candidate, response, false, pricing)
+					if billingErr != nil {
+						// The upstream result may already exist. Preserve the reservation for manual review.
+						walletReservationFinalized = true
+						return Result{}, billingErr
+					}
+				} else {
+					billings = s.billings(ctx, user, task.Kind, candidateBody, candidate, response, true)
+				}
 				record := buildSuccessRecord(task, user, candidateBody, candidate, response, billings, isSimulation(task, candidate))
 				record.Metrics = mergeMetrics(record.Metrics, candidateCapabilityFilterMetrics(candidateFilterSummary))
 				record.Metrics = mergeMetrics(record.Metrics, parameterPreprocessingMetrics(preprocessing.Log))
@@ -580,6 +629,10 @@ candidatesLoop:
 	return Result{Task: failed, Output: failed.Result}, lastErr
 }

+func pricingCandidateKey(candidate store.RuntimeModelCandidate) string {
+	return firstNonEmptyString(candidate.PlatformModelID, candidate.PlatformID+":"+candidate.ModelName)
+}
+
 func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user *auth.User, body map[string]any, preprocessing parameterPreprocessingLog, candidate store.RuntimeModelCandidate, attemptNo int, onDelta clients.StreamDelta, responseExecution responseExecutionContext, singleSourceProtected bool, cacheAffinityPolicy map[string]any, cacheAffinityRecordKeys []string) (clients.Response, error) {
 	simulated := isSimulation(task, candidate)
 	baseAttemptMetrics := mergeMetrics(attemptMetrics(candidate, attemptNo, simulated), parameterPreprocessingMetrics(preprocessing))