fix(billing): 封住发布前计费竞态

阻止上游提交状态不明的任务被租约接管后重复执行，并为人工复核保留可操作的结算记录。

将生产提交绑定到当前估价签名，统一复用预处理快照，并补强规则形状、定点溢出与历史规则兼容校验。

已通过 PostgreSQL 16 集成测试、Go 全量测试与静态检查、前端测试与构建、OpenAPI、依赖审计、镜像、迁移、流水线和 SemVer 门禁。

apps/api/internal/store/billing_v2_integration_test.go

@@ -62,6 +62,149 @@ func TestTaskIdempotencyAndExecutionLease(t *testing.T) {
 	if err != nil || finished.ErrorCode != "new_worker" {
 		t.Fatalf("new worker terminal task=%+v err=%v", finished, err)
 	}
+	if err := db.RenewTaskExecutionLease(ctx, created.Task.ID, secondToken, 5*time.Minute); !errors.Is(err, ErrTaskExecutionFinished) {
+		t.Fatalf("terminal task renewal error=%v", err)
+	}
+}
+
+func TestExpiredExecutionLeaseDoesNotReplayAmbiguousUpstreamSubmission(t *testing.T) {
+	db := billingV2IntegrationStore(t)
+	ctx := context.Background()
+	tenantID, gatewayUserID := seedWalletReservationUser(t, ctx, db)
+	user := &auth.User{ID: "billing-review-" + uuid.NewString(), GatewayUserID: gatewayUserID, GatewayTenantID: tenantID}
+	created, err := db.CreateTask(ctx, CreateTaskInput{
+		Kind: "images.generations", Model: "billing-v2-model", RunMode: "production",
+		Request: map[string]any{"model": "billing-v2-model"},
+	}, user)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		_, _ = db.pool.Exec(context.Background(), `DELETE FROM gateway_tasks WHERE id=$1::uuid`, created.ID)
+	})
+
+	firstToken := uuid.NewString()
+	if _, err := db.ClaimTaskExecution(ctx, created.ID, firstToken, 5*time.Minute); err != nil {
+		t.Fatal(err)
+	}
+	attemptID, err := db.CreateTaskAttempt(ctx, CreateTaskAttemptInput{
+		TaskID: created.ID, AttemptNo: 1, Status: "running", RequestSnapshot: map[string]any{"model": created.Model},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := db.SetAttemptUpstreamSubmissionStatus(ctx, attemptID, "submitting"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.pool.Exec(ctx, `UPDATE gateway_tasks SET execution_lease_expires_at=now()-interval '1 second' WHERE id=$1::uuid`, created.ID); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.ClaimTaskExecution(ctx, created.ID, uuid.NewString(), 5*time.Minute); !errors.Is(err, ErrTaskExecutionManualReview) {
+		t.Fatalf("ambiguous submission takeover error=%v", err)
+	}
+
+	review, err := db.GetTask(ctx, created.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if review.Status != "failed" || review.BillingStatus != "manual_review" || review.ErrorCode != "upstream_submission_unknown" {
+		t.Fatalf("manual review task=%+v", review)
+	}
+	var outboxStatus string
+	var outboxAction string
+	var reviewReason string
+	if err := db.pool.QueryRow(ctx, `
+SELECT status, action, COALESCE(manual_review_reason, '')
+FROM settlement_outbox
+WHERE task_id=$1::uuid AND event_type='task.billing.review'`, created.ID).Scan(&outboxStatus, &outboxAction, &reviewReason); err != nil {
+		t.Fatal(err)
+	}
+	if outboxStatus != "manual_review" || outboxAction != "release" || reviewReason != "upstream_submission_unknown" {
+		t.Fatalf("review outbox status=%s action=%s reason=%s", outboxStatus, outboxAction, reviewReason)
+	}
+}
+
+func TestExpiredExecutionLeaseCanResumeAfterKnownRejectedResponse(t *testing.T) {
+	db := billingV2IntegrationStore(t)
+	ctx := context.Background()
+	_, gatewayUserID := seedWalletReservationUser(t, ctx, db)
+	user := &auth.User{ID: "billing-known-response-" + uuid.NewString(), GatewayUserID: gatewayUserID}
+	created, err := db.CreateTask(ctx, CreateTaskInput{
+		Kind: "images.generations", Model: "billing-v2-model", RunMode: "production",
+		Request: map[string]any{"model": "billing-v2-model"},
+	}, user)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		_, _ = db.pool.Exec(context.Background(), `DELETE FROM gateway_tasks WHERE id=$1::uuid`, created.ID)
+	})
+
+	if _, err := db.ClaimTaskExecution(ctx, created.ID, uuid.NewString(), 5*time.Minute); err != nil {
+		t.Fatal(err)
+	}
+	attemptID, err := db.CreateTaskAttempt(ctx, CreateTaskAttemptInput{TaskID: created.ID, AttemptNo: 1, Status: "running"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := db.SetAttemptUpstreamSubmissionStatus(ctx, attemptID, "response_received"); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.FinishTaskAttempt(ctx, FinishTaskAttemptInput{AttemptID: attemptID, Status: "failed", ErrorCode: "upstream_rejected"}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.pool.Exec(ctx, `UPDATE gateway_tasks SET execution_lease_expires_at=now()-interval '1 second' WHERE id=$1::uuid`, created.ID); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.ClaimTaskExecution(ctx, created.ID, uuid.NewString(), 5*time.Minute); err != nil {
+		t.Fatalf("known rejected response should remain retryable: %v", err)
+	}
+}
+
+func TestFinishTaskManualReviewCreatesVisibleBillingRecord(t *testing.T) {
+	db := billingV2IntegrationStore(t)
+	ctx := context.Background()
+	_, gatewayUserID := seedWalletReservationUser(t, ctx, db)
+	user := &auth.User{ID: "billing-direct-review-" + uuid.NewString(), GatewayUserID: gatewayUserID}
+	created, err := db.CreateTask(ctx, CreateTaskInput{
+		Kind: "images.generations", Model: "billing-v2-model", RunMode: "production",
+		Request: map[string]any{"model": "billing-v2-model"},
+	}, user)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		_, _ = db.pool.Exec(context.Background(), `DELETE FROM gateway_tasks WHERE id=$1::uuid`, created.ID)
+	})
+	token := uuid.NewString()
+	if _, err := db.ClaimTaskExecution(ctx, created.ID, token, 5*time.Minute); err != nil {
+		t.Fatal(err)
+	}
+	attemptID, err := db.CreateTaskAttempt(ctx, CreateTaskAttemptInput{TaskID: created.ID, AttemptNo: 1, Status: "running"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := db.SetAttemptUpstreamSubmissionStatus(ctx, attemptID, "submitting"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.FinishTaskManualReview(ctx, FinishTaskManualReviewInput{
+		TaskID: created.ID, ExecutionToken: token, AttemptID: attemptID, TaskStatus: "failed",
+		Code: "upstream_submission_unknown", Message: "upstream submission result is unknown",
+	}); err != nil {
+		t.Fatal(err)
+	}
+	var visible bool
+	if err := db.pool.QueryRow(ctx, `
+SELECT EXISTS (
+  SELECT 1 FROM settlement_outbox
+  WHERE task_id=$1::uuid AND status='manual_review'
+    AND action='release' AND manual_review_reason='upstream_submission_unknown'
+)`, created.ID).Scan(&visible); err != nil {
+		t.Fatal(err)
+	}
+	if !visible {
+		t.Fatal("manual review billing record is not visible in settlement outbox")
+	}
 }
 
 func TestBillingSettlementStaleTakeoverDebitsExactlyOnce(t *testing.T) {

apps/api/internal/store/postgres.go

@@ -59,6 +59,8 @@ var (
 	ErrIdempotencyKeyReused          = errors.New("idempotency key was reused for a different request")
 	ErrTaskExecutionLeaseUnavailable = errors.New("task execution lease is unavailable")
 	ErrTaskExecutionLeaseLost        = errors.New("task execution lease was lost")
+	ErrTaskExecutionFinished         = errors.New("task execution already finished")
+	ErrTaskExecutionManualReview     = errors.New("task execution requires manual review")
 	ErrProtectedDefault              = errors.New("protected default resource cannot be deleted")
 	ErrUserAlreadyExists             = errors.New("user already exists")
 	ErrWeakPassword                  = errors.New("password must be at least 8 characters")

apps/api/internal/store/pricing_rules.go

@@ -335,6 +335,9 @@ ORDER BY priority ASC, resource_type ASC, rule_key ASC`, id)
 		default:
 			return EffectivePricingConfig{}, fmt.Errorf("pricing rule %s uses unsupported calculator %s", ruleKey, calculatorType)
 		}
+		if err := ValidateEffectivePricingRuleShape(resourceType, unit, calculatorType); err != nil {
+			return EffectivePricingConfig{}, fmt.Errorf("pricing rule %s: %w", ruleKey, err)
+		}
 		if strings.HasPrefix(strings.TrimSpace(basePrice), "-") {
 			return EffectivePricingConfig{}, fmt.Errorf("pricing rule %s has a negative base price", ruleKey)
 		}
@@ -372,6 +375,51 @@ ORDER BY priority ASC, resource_type ASC, rule_key ASC`, id)
 	}, nil
 }
 
+func ValidateEffectivePricingRuleShape(resourceType string, unit string, calculatorType string) error {
+	unit = NormalizeEffectivePricingRuleUnit(resourceType, unit)
+	if strings.TrimSpace(calculatorType) == "" {
+		calculatorType = DefaultEffectivePricingCalculator(resourceType)
+	}
+	allowed := false
+	switch resourceType {
+	case "text_input", "text_cached_input", "text_output", "text_total":
+		allowed = unit == "1k_tokens" && calculatorType == "token_usage"
+	case "image", "image_edit":
+		allowed = unit == "image" && calculatorType == "unit_weight"
+	case "video":
+		allowed = unit == "5s" && calculatorType == "duration_weight"
+	case "music":
+		allowed = (unit == "song" || unit == "item") && calculatorType == "unit_weight"
+	case "audio":
+		allowed = unit == "character" && calculatorType == "unit_weight"
+	default:
+		return nil
+	}
+	if !allowed {
+		return fmt.Errorf("unit %q and calculator %q do not match resource %q", unit, calculatorType, resourceType)
+	}
+	return nil
+}
+
+func NormalizeEffectivePricingRuleUnit(resourceType string, unit string) string {
+	unit = strings.TrimSpace(unit)
+	if resourceType == "video" && unit == "video" {
+		return "5s"
+	}
+	return unit
+}
+
+func DefaultEffectivePricingCalculator(resourceType string) string {
+	switch strings.TrimSpace(resourceType) {
+	case "text_input", "text_cached_input", "text_output", "text_total":
+		return "token_usage"
+	case "video":
+		return "duration_weight"
+	default:
+		return "unit_weight"
+	}
+}
+
 func addPricingRuleToConfig(config map[string]any, resourceType string, basePrice string, dynamicWeight map[string]any, formulaConfig map[string]any) {
 	resourceConfig := map[string]any{"basePrice": basePrice}
 	if len(dynamicWeight) > 0 {
@@ -591,7 +639,7 @@ func normalizePricingRule(input PricingRuleInput, index int, defaultCurrency str
 	input.RuleKey = strings.TrimSpace(input.RuleKey)
 	input.DisplayName = strings.TrimSpace(input.DisplayName)
 	input.ResourceType = strings.TrimSpace(input.ResourceType)
-	input.Unit = strings.TrimSpace(input.Unit)
+	input.Unit = NormalizeEffectivePricingRuleUnit(input.ResourceType, input.Unit)
 	input.Currency = strings.TrimSpace(input.Currency)
 	input.CalculatorType = strings.TrimSpace(input.CalculatorType)
 	input.Status = strings.TrimSpace(input.Status)
@@ -608,7 +656,7 @@ func normalizePricingRule(input PricingRuleInput, index int, defaultCurrency str
 		input.Currency = defaultCurrency
 	}
 	if input.CalculatorType == "" {
-		input.CalculatorType = "unit_weight"
+		input.CalculatorType = DefaultEffectivePricingCalculator(input.ResourceType)
 	}
 	if input.Priority == 0 {
 		input.Priority = (index + 1) * 10

apps/api/internal/store/pricing_rules_test.go

@@ -0,0 +1,27 @@
+package store
+
+import "testing"
+
+func TestValidateEffectivePricingRuleShape(t *testing.T) {
+	tests := []struct {
+		name       string
+		resource   string
+		unit       string
+		calculator string
+		valid      bool
+	}{
+		{name: "text", resource: "text_input", unit: "1k_tokens", calculator: "token_usage", valid: true},
+		{name: "image", resource: "image", unit: "image", calculator: "unit_weight", valid: true},
+		{name: "video", resource: "video", unit: "5s", calculator: "duration_weight", valid: true},
+		{name: "wrong video unit", resource: "video", unit: "second", calculator: "duration_weight"},
+		{name: "wrong image calculator", resource: "image", unit: "image", calculator: "token_usage"},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			err := ValidateEffectivePricingRuleShape(test.resource, test.unit, test.calculator)
+			if (err == nil) != test.valid {
+				t.Fatalf("valid=%v err=%v", test.valid, err)
+			}
+		})
+	}
+}

apps/api/internal/store/tasks_runtime.go

@@ -157,7 +157,83 @@ func (s *Store) ClaimTaskExecution(ctx context.Context, taskID string, execution
 	if leaseTTL <= 0 {
 		leaseTTL = 5 * time.Minute
 	}
-	task, err := scanGatewayTask(s.pool.QueryRow(ctx, `
+	var task GatewayTask
+	manualReview := false
+	err := pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error {
+		var queuedReady bool
+		var runningExpired bool
+		var production bool
+		var hasGatewayUser bool
+		if err := tx.QueryRow(ctx, `
+SELECT status = 'queued' AND next_run_at <= now(),
+       status = 'running' AND (execution_lease_expires_at IS NULL OR execution_lease_expires_at <= now()),
+       run_mode = 'production',
+       gateway_user_id IS NOT NULL
+FROM gateway_tasks
+WHERE id = $1::uuid
+FOR UPDATE`, taskID).Scan(&queuedReady, &runningExpired, &production, &hasGatewayUser); err != nil {
+			return err
+		}
+		if !queuedReady && !runningExpired {
+			return ErrTaskExecutionLeaseUnavailable
+		}
+		if runningExpired && production {
+			var submissionAmbiguous bool
+			if err := tx.QueryRow(ctx, `
+SELECT COALESCE((
+  SELECT (
+    (status = 'running' AND upstream_submission_status IN ('submitting', 'response_received'))
+    OR (status = 'failed' AND upstream_submission_status = 'submitting')
+  )
+  FROM gateway_task_attempts
+  WHERE task_id = $1::uuid
+  ORDER BY attempt_no DESC, started_at DESC
+  LIMIT 1
+), false)`, taskID).Scan(&submissionAmbiguous); err != nil {
+				return err
+			}
+			if submissionAmbiguous {
+				if _, err := tx.Exec(ctx, `
+UPDATE gateway_tasks
+SET status = 'failed',
+    billing_status = CASE WHEN $2 THEN 'manual_review' ELSE 'not_required' END,
+    billing_updated_at = now(),
+    error = 'upstream submission result is unknown',
+    error_code = 'upstream_submission_unknown',
+    error_message = 'upstream submission result is unknown',
+    locked_by = NULL,
+    locked_at = NULL,
+    heartbeat_at = NULL,
+    execution_token = NULL,
+    execution_lease_expires_at = NULL,
+    finished_at = now(),
+    updated_at = now()
+WHERE id = $1::uuid`, taskID, hasGatewayUser); err != nil {
+					return err
+				}
+				if hasGatewayUser {
+					payloadJSON, _ := json.Marshal(map[string]any{
+						"taskId": taskID, "classification": "upstream_submission_unknown",
+					})
+					if _, err := tx.Exec(ctx, `
+INSERT INTO settlement_outbox (
+  task_id, event_type, action, amount, currency, pricing_snapshot, payload,
+  status, next_attempt_at, manual_review_reason
+)
+SELECT id, 'task.billing.review', 'release', reservation_amount, billing_currency,
+       pricing_snapshot, $2::jsonb, 'manual_review', now(), 'upstream_submission_unknown'
+FROM gateway_tasks
+WHERE id = $1::uuid
+ON CONFLICT (task_id, event_type) DO NOTHING`, taskID, string(payloadJSON)); err != nil {
+						return err
+					}
+				}
+				manualReview = true
+				return nil
+			}
+		}
+		var err error
+		task, err = scanGatewayTask(tx.QueryRow(ctx, `
 UPDATE gateway_tasks
 SET status = 'running',
     execution_token = $2::uuid,
@@ -171,10 +247,18 @@ WHERE id = $1::uuid
     OR (status = 'running' AND (execution_lease_expires_at IS NULL OR execution_lease_expires_at <= now()))
   )
 RETURNING `+gatewayTaskColumns, taskID, executionToken, int(leaseTTL/time.Second)))
-	if errors.Is(err, pgx.ErrNoRows) {
-		return GatewayTask{}, ErrTaskExecutionLeaseUnavailable
+		return err
+	})
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return GatewayTask{}, ErrTaskExecutionLeaseUnavailable
+		}
+		return GatewayTask{}, err
 	}
-	return task, err
+	if manualReview {
+		return GatewayTask{}, ErrTaskExecutionManualReview
+	}
+	return task, nil
 }
 
 func (s *Store) RenewTaskExecutionLease(ctx context.Context, taskID string, executionToken string, leaseTTL time.Duration) error {
@@ -193,6 +277,14 @@ WHERE id = $1::uuid
 		return err
 	}
 	if tag.RowsAffected() != 1 {
+		var stillRunning bool
+		if err := s.pool.QueryRow(ctx, `
+SELECT COALESCE((SELECT status = 'running' FROM gateway_tasks WHERE id = $1::uuid), false)`, taskID).Scan(&stillRunning); err != nil {
+			return err
+		}
+		if !stillRunning {
+			return ErrTaskExecutionFinished
+		}
 		return ErrTaskExecutionLeaseLost
 	}
 	return nil
@@ -902,8 +994,7 @@ SET status = $2,
     response_started_at = $6::timestamptz,
     response_finished_at = $7::timestamptz,
     response_duration_ms = $8,
-    finished_at = now(),
-    updated_at = now()
+    finished_at = now()
 WHERE id = $1::uuid`, input.AttemptID, attemptStatus, input.RequestID, input.Code, input.Message,
 				nullableTime(input.ResponseStartedAt), nullableTime(input.ResponseFinishedAt), input.ResponseDurationMS); err != nil {
 				return err
@@ -942,7 +1033,22 @@ WHERE id = $1::uuid
 		if tag.RowsAffected() != 1 {
 			return ErrTaskExecutionLeaseLost
 		}
-		return nil
+		payloadJSON, _ := json.Marshal(map[string]any{
+			"taskId": input.TaskID, "classification": input.Code,
+		})
+		_, err = tx.Exec(ctx, `
+INSERT INTO settlement_outbox (
+  task_id, event_type, action, amount, currency, pricing_snapshot, payload,
+  status, next_attempt_at, manual_review_reason
+)
+SELECT id, 'task.billing.review', 'release', reservation_amount, billing_currency,
+       pricing_snapshot, $2::jsonb, 'manual_review', now(), $3
+FROM gateway_tasks
+WHERE id = $1::uuid
+  AND run_mode = 'production'
+  AND gateway_user_id IS NOT NULL
+ON CONFLICT (task_id, event_type) DO NOTHING`, input.TaskID, string(payloadJSON), input.Code)
+		return err
 	})
 	if err != nil {
 		return GatewayTask{}, err