test(billing): 覆盖估价只读与请求重放契约

PostgreSQL HTTP 集成测试验证估价新增字段且不创建任务、冻结余额或写流水，并验证非流式重放只执行一次、异请求冲突及流式重复冲突。

apps/api/internal/httpapi/core_flow_integration_test.go

@@ -48,6 +48,7 @@ func TestCoreLocalFlow(t *testing.T) {
 		DatabaseURL:                 databaseURL,
 		IdentityMode:                "hybrid",
 		JWTSecret:                   "test-secret",
+		BillingEngineMode:           "enforce",
 		TaskProgressCallbackEnabled: true,
 		TaskProgressCallbackURL:     "http://callback.local/task-progress",
 		CORSAllowedOrigin:           "*",
@@ -814,6 +815,7 @@ LIMIT 1`).Scan(&gptImageModelTypesRaw); err != nil {
 		"currency":   "resource",
 		"rules": []map[string]any{
 			{"ruleKey": "text_input", "displayName": "Text Input", "resourceType": "text_input", "unit": "1k_tokens", "basePrice": 1},
+			{"ruleKey": "text_cached_input", "displayName": "Text Cached Input", "resourceType": "text_cached_input", "unit": "1k_tokens", "basePrice": 0.5},
 			{"ruleKey": "text_output", "displayName": "Text Output", "resourceType": "text_output", "unit": "1k_tokens", "basePrice": 2},
 			{"ruleKey": "image", "displayName": "Image", "resourceType": "image", "unit": "image", "basePrice": 7},
 			{"ruleKey": "image_edit", "displayName": "Image Edit", "resourceType": "image_edit", "unit": "image", "basePrice": 11},
@@ -845,6 +847,101 @@ WHERE gateway_user_id = $1::uuid
   AND currency = 'resource'`, smokeGatewayUserID).Scan(&walletBalanceBefore); err != nil {
 		t.Fatalf("read wallet balance before pricing task: %v", err)
 	}
+	var estimateSideEffectsBefore struct {
+		Tasks        int
+		Transactions int
+		Frozen       string
+	}
+	if err := testPool.QueryRow(ctx, `
+SELECT (SELECT count(*) FROM gateway_tasks),
+       (SELECT count(*) FROM gateway_wallet_transactions WHERE gateway_user_id=$1::uuid),
+       (SELECT frozen_balance::text FROM gateway_wallet_accounts WHERE gateway_user_id=$1::uuid AND currency='resource')`,
+		smokeGatewayUserID,
+	).Scan(&estimateSideEffectsBefore.Tasks, &estimateSideEffectsBefore.Transactions, &estimateSideEffectsBefore.Frozen); err != nil {
+		t.Fatalf("read estimate side effects before request: %v", err)
+	}
+	var pricingEstimate struct {
+		TotalAmount        float64 `json:"totalAmount"`
+		ReservationAmount  float64 `json:"reservationAmount"`
+		CandidateCount     int     `json:"candidateCount"`
+		PricingVersion     string  `json:"pricingVersion"`
+		RequestFingerprint string  `json:"requestFingerprint"`
+	}
+	doJSON(t, server.URL, http.MethodPost, "/api/v1/pricing/estimate", apiKeyResponse.Secret, map[string]any{
+		"kind": "chat.completions", "model": pricingModel, "max_completion_tokens": 32,
+		"messages": []map[string]any{{"role": "user", "content": "estimate only"}},
+	}, http.StatusOK, &pricingEstimate)
+	if pricingEstimate.TotalAmount <= 0 || pricingEstimate.ReservationAmount < pricingEstimate.TotalAmount ||
+		pricingEstimate.CandidateCount != 1 || pricingEstimate.PricingVersion != "effective-pricing-v2" || pricingEstimate.RequestFingerprint == "" {
+		t.Fatalf("unexpected effective pricing estimate: %+v", pricingEstimate)
+	}
+	var estimateSideEffectsAfter struct {
+		Tasks        int
+		Transactions int
+		Frozen       string
+	}
+	if err := testPool.QueryRow(ctx, `
+SELECT (SELECT count(*) FROM gateway_tasks),
+       (SELECT count(*) FROM gateway_wallet_transactions WHERE gateway_user_id=$1::uuid),
+       (SELECT frozen_balance::text FROM gateway_wallet_accounts WHERE gateway_user_id=$1::uuid AND currency='resource')`,
+		smokeGatewayUserID,
+	).Scan(&estimateSideEffectsAfter.Tasks, &estimateSideEffectsAfter.Transactions, &estimateSideEffectsAfter.Frozen); err != nil {
+		t.Fatalf("read estimate side effects after request: %v", err)
+	}
+	if estimateSideEffectsAfter != estimateSideEffectsBefore {
+		t.Fatalf("pricing estimate must be read-only, before=%+v after=%+v", estimateSideEffectsBefore, estimateSideEffectsAfter)
+	}
+
+	idempotencyPayload := map[string]any{
+		"model": pricingModel, "runMode": "simulation", "simulation": true, "simulationDurationMs": 5,
+		"messages": []map[string]any{{"role": "user", "content": "idempotent replay"}},
+	}
+	idempotencyHeaders := map[string]string{"Idempotency-Key": "chat-replay-" + suffixText}
+	var idempotentFirst map[string]any
+	firstReplayHeaders := doJSONWithHeaders(t, server.URL, http.MethodPost, "/api/v1/chat/completions", apiKeyResponse.Secret, idempotencyPayload, idempotencyHeaders, http.StatusOK, &idempotentFirst)
+	var idempotentSecond map[string]any
+	secondReplayHeaders := doJSONWithHeaders(t, server.URL, http.MethodPost, "/api/v1/chat/completions", apiKeyResponse.Secret, idempotencyPayload, idempotencyHeaders, http.StatusOK, &idempotentSecond)
+	idempotentTaskID := firstReplayHeaders.Get("X-Gateway-Task-Id")
+	if idempotentTaskID == "" || secondReplayHeaders.Get("X-Gateway-Task-Id") != idempotentTaskID || secondReplayHeaders.Get("Idempotent-Replayed") != "true" {
+		t.Fatalf("non-stream replay headers first=%v second=%v", firstReplayHeaders, secondReplayHeaders)
+	}
+	var idempotentAttempts int
+	if err := testPool.QueryRow(ctx, `SELECT count(*) FROM gateway_task_attempts WHERE task_id=$1::uuid`, idempotentTaskID).Scan(&idempotentAttempts); err != nil {
+		t.Fatalf("count idempotent task attempts: %v", err)
+	}
+	if idempotentAttempts != 1 {
+		t.Fatalf("idempotent request called upstream %d times, want 1", idempotentAttempts)
+	}
+	idempotencyPayload["messages"] = []map[string]any{{"role": "user", "content": "different request"}}
+	doJSONWithHeaders(t, server.URL, http.MethodPost, "/api/v1/chat/completions", apiKeyResponse.Secret, idempotencyPayload, idempotencyHeaders, http.StatusConflict, nil)
+
+	streamPayload := map[string]any{
+		"model": pricingModel, "runMode": "simulation", "simulation": true, "simulationDurationMs": 5, "stream": true,
+		"messages": []map[string]any{{"role": "user", "content": "stream replay"}},
+	}
+	streamRaw, err := json.Marshal(streamPayload)
+	if err != nil {
+		t.Fatal(err)
+	}
+	streamKey := "stream-replay-" + suffixText
+	streamRequest, err := http.NewRequest(http.MethodPost, server.URL+"/api/v1/chat/completions", bytes.NewReader(streamRaw))
+	if err != nil {
+		t.Fatal(err)
+	}
+	streamRequest.Header.Set("Authorization", "Bearer "+apiKeyResponse.Secret)
+	streamRequest.Header.Set("Content-Type", "application/json")
+	streamRequest.Header.Set("Idempotency-Key", streamKey)
+	streamResponse, err := http.DefaultClient.Do(streamRequest)
+	if err != nil {
+		t.Fatalf("execute first idempotent stream: %v", err)
+	}
+	_, _ = io.ReadAll(streamResponse.Body)
+	_ = streamResponse.Body.Close()
+	if streamResponse.StatusCode != http.StatusOK || streamResponse.Header.Get("X-Gateway-Task-Id") == "" {
+		t.Fatalf("first idempotent stream status=%d headers=%v", streamResponse.StatusCode, streamResponse.Header)
+	}
+	doJSONWithHeaders(t, server.URL, http.MethodPost, "/api/v1/chat/completions", apiKeyResponse.Secret, streamPayload,
+		map[string]string{"Idempotency-Key": streamKey}, http.StatusConflict, nil)
 	var pricingTask struct {
 		Task struct {
 			ID                string         `json:"id"`
@@ -1588,6 +1685,7 @@ WHERE m.platform_id = $1::uuid
 		DatabaseURL:                 databaseURL,
 		IdentityMode:                "hybrid",
 		JWTSecret:                   "test-secret",
+		BillingEngineMode:           "enforce",
 		TaskProgressCallbackEnabled: true,
 		TaskProgressCallbackURL:     "http://callback.local/task-progress",
 		CORSAllowedOrigin:           "*",