package httpapi

import (
	"bytes"
	"context"
	"encoding/json"
	"io"
	"log/slog"
	"math"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"runtime"
	"sort"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/easyai/easyai-ai-gateway/apps/api/internal/auth"
	"github.com/easyai/easyai-ai-gateway/apps/api/internal/config"
	"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
	"github.com/jackc/pgx/v5/pgxpool"
)

func TestCoreLocalFlow(t *testing.T) {
	databaseURL := strings.TrimSpace(os.Getenv("AI_GATEWAY_TEST_DATABASE_URL"))
	if databaseURL == "" {
		t.Skip("set AI_GATEWAY_TEST_DATABASE_URL to run the PostgreSQL integration flow")
	}
	ctx := context.Background()
	applyMigration(t, ctx, databaseURL)

	db, err := store.Connect(ctx, databaseURL)
	if err != nil {
		t.Fatalf("connect store: %v", err)
	}
	defer db.Close()

	assertRuntimeRecoveryReleasesPendingRateReservations(t, ctx, db)

	serverCtx, cancelServer := context.WithCancel(ctx)
	defer cancelServer()
	handler := NewServerWithContext(serverCtx, config.Config{
		AppEnv:                            "test",
		HTTPAddr:                          ":0",
		DatabaseURL:                       databaseURL,
		IdentityMode:                      "hybrid",
		JWTSecret:                         "test-secret",
		BillingEngineMode:                 "enforce",
		TaskProgressCallbackEnabled:       true,
		TaskProgressCallbackURL:           "http://callback.local/task-progress",
		CORSAllowedOrigin:                 "*",
		AsyncQueueWorkerEnabled:           true,
		AsyncWorkerHardLimit:              32,
		AsyncWorkerInstanceHardLimit:      32,
		AsyncWorkerRefreshIntervalSeconds: 1,
	}, db, slog.New(slog.NewTextHandler(io.Discard, nil)))
	server := httptest.NewServer(handler)
	defer server.Close()

	suffix := time.Now().UnixNano()
	suffixText := strconv.FormatInt(suffix, 10)
	username := "smoke_admin_" + suffixText
	password := "password123"

	var registerResponse struct {
		AccessToken string `json:"accessToken"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/v1/auth/register", "", map[string]any{
		"username":   username,
		"email":      username + "@example.com",
		"password":   password,
		"tenantKey":  "manual-" + suffixText,
		"tenantName": "Manual Tenant",
	}, http.StatusCreated, &registerResponse)
	if registerResponse.AccessToken == "" {
		t.Fatal("register did not return access token")
	}
	ordinaryUsername := "smoke_user_" + suffixText
	doJSON(t, server.URL, http.MethodPost, "/api/v1/auth/register", "", map[string]any{
		"username": ordinaryUsername,
		"email":    ordinaryUsername + "@example.com",
		"password": password,
	}, http.StatusCreated, &struct{}{})

	var duplicateResponse map[string]any
	doJSON(t, server.URL, http.MethodPost, "/api/v1/auth/register", "", map[string]any{
		"username": username,
		"email":    username + "@example.com",
		"password": password,
	}, http.StatusConflict, &duplicateResponse)
	if errorBody, ok := duplicateResponse["error"].(map[string]any); !ok || errorBody["message"] != "user already exists" {
		t.Fatalf("unexpected duplicate response: %+v", duplicateResponse)
	}

	var loginResponse struct {
		AccessToken string `json:"accessToken"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/v1/auth/login", "", map[string]any{
		"account":  username,
		"password": password,
	}, http.StatusOK, &loginResponse)
	if loginResponse.AccessToken == "" {
		t.Fatal("login did not return access token")
	}

	var apiKeyResponse struct {
		Secret string `json:"secret"`
		APIKey struct {
			ID        string `json:"id"`
			Name      string `json:"name"`
			KeyPrefix string `json:"keyPrefix"`
			Status    string `json:"status"`
		} `json:"apiKey"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/v1/api-keys", loginResponse.AccessToken, map[string]any{
		"name": "smoke key",
	}, http.StatusCreated, &apiKeyResponse)
	if !strings.HasPrefix(apiKeyResponse.Secret, "sk-gw-") || apiKeyResponse.APIKey.Status != "active" {
		t.Fatalf("unexpected api key response: %+v", apiKeyResponse)
	}

	var me map[string]any
	doJSON(t, server.URL, http.MethodGet, "/api/v1/me", apiKeyResponse.Secret, nil, http.StatusOK, &me)
	if me["apiKeyId"] == "" {
		t.Fatalf("api key auth did not expose apiKeyId: %+v", me)
	}
	if me["tenantKey"] != "default" {
		t.Fatalf("register should ignore public tenant fields and use default tenant: %+v", me)
	}

	testPool, err := pgxpool.New(ctx, databaseURL)
	if err != nil {
		t.Fatalf("connect test pool: %v", err)
	}
	defer testPool.Close()
	assertSeedreamProCatalogMigration(t, ctx, testPool)
	if _, err := testPool.Exec(ctx, `UPDATE gateway_users SET roles = '["admin"]'::jsonb WHERE username = $1`, username); err != nil {
		t.Fatalf("promote smoke user: %v", err)
	}
	serverMainCtx, cancelServerMain := context.WithCancel(ctx)
	serverMain := httptest.NewServer(NewServerWithContext(serverMainCtx, config.Config{
		AppEnv: "test", HTTPAddr: ":0", DatabaseURL: databaseURL, IdentityMode: "server-main",
		JWTSecret: "test-secret", CORSAllowedOrigin: "*",
	}, db, slog.New(slog.NewTextHandler(io.Discard, nil))))
	var breakGlassLogin struct {
		AccessToken string `json:"accessToken"`
	}
	doJSON(t, serverMain.URL, http.MethodPost, "/api/v1/auth/login", "", map[string]any{
		"account": username, "password": password,
	}, http.StatusOK, &breakGlassLogin)
	if breakGlassLogin.AccessToken == "" {
		t.Fatal("server-main break-glass manager login did not return access token")
	}
	doJSON(t, serverMain.URL, http.MethodPost, "/api/v1/auth/login", "", map[string]any{
		"account": ordinaryUsername, "password": password,
	}, http.StatusForbidden, &map[string]any{})
	serverMain.Close()
	cancelServerMain()
	doJSON(t, server.URL, http.MethodPost, "/api/v1/auth/login", "", map[string]any{
		"account":  username,
		"password": password,
	}, http.StatusOK, &loginResponse)
	if loginResponse.AccessToken == "" {
		t.Fatal("admin login did not return access token")
	}
	var smokeGatewayUserID string
	if err := testPool.QueryRow(ctx, `SELECT id::text FROM gateway_users WHERE username = $1`, username).Scan(&smokeGatewayUserID); err != nil {
		t.Fatalf("read smoke gateway user id: %v", err)
	}
	var walletAdjustment struct {
		Account struct {
			Balance float64 `json:"balance"`
		} `json:"account"`
		AuditLog struct {
			Action string `json:"action"`
		} `json:"auditLog"`
	}
	doJSON(t, server.URL, http.MethodPatch, "/api/admin/users/"+smokeGatewayUserID+"/wallet", loginResponse.AccessToken, map[string]any{
		"currency": "resource",
		"balance":  1000,
		"reason":   "seed integration wallet",
	}, http.StatusOK, &walletAdjustment)
	if !floatNear(walletAdjustment.Account.Balance, 1000) || walletAdjustment.AuditLog.Action != "wallet.balance.set" {
		t.Fatalf("wallet adjustment did not update balance and audit log: %+v", walletAdjustment)
	}
	doJSON(t, server.URL, http.MethodPost, "/api/admin/users/"+smokeGatewayUserID+"/wallet/recharge", loginResponse.AccessToken, map[string]any{
		"currency": "resource",
		"amount":   125,
		"reason":   "seed integration wallet recharge",
	}, http.StatusOK, &walletAdjustment)
	if !floatNear(walletAdjustment.Account.Balance, 1125) || walletAdjustment.AuditLog.Action != "wallet.balance.recharge" {
		t.Fatalf("wallet recharge did not add balance and audit log: %+v", walletAdjustment)
	}
	var auditResponse struct {
		Items []struct {
			Action string `json:"action"`
		} `json:"items"`
	}
	doJSON(t, server.URL, http.MethodGet, "/api/admin/audit-logs?category=billing&limit=5", loginResponse.AccessToken, nil, http.StatusOK, &auditResponse)
	if len(auditResponse.Items) == 0 || auditResponse.Items[0].Action != "wallet.balance.recharge" {
		t.Fatalf("wallet recharge audit log not found: %+v", auditResponse)
	}
	doJSON(t, server.URL, http.MethodGet, "/api/admin/models", apiKeyResponse.Secret, nil, http.StatusForbidden, nil)

	var chatOnlyAPIKeyResponse struct {
		Secret string `json:"secret"`
		APIKey struct {
			ID string `json:"id"`
		} `json:"apiKey"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/v1/api-keys", loginResponse.AccessToken, map[string]any{
		"name":   "chat only key",
		"scopes": []string{"chat"},
	}, http.StatusCreated, &chatOnlyAPIKeyResponse)
	var taskCountBefore int
	if err := testPool.QueryRow(ctx, `SELECT count(*) FROM gateway_tasks`).Scan(&taskCountBefore); err != nil {
		t.Fatalf("count tasks before scoped request: %v", err)
	}
	defaultImageModel := "openai:gpt-image-1"
	doJSON(t, server.URL, http.MethodPost, "/api/v1/images/generations", chatOnlyAPIKeyResponse.Secret, map[string]any{
		"model":  defaultImageModel,
		"prompt": "scope should block this",
	}, http.StatusForbidden, nil)
	doJSON(t, server.URL, http.MethodPost, "/api/v1/pricing/estimate", chatOnlyAPIKeyResponse.Secret, map[string]any{
		"kind":   "images.generations",
		"model":  defaultImageModel,
		"prompt": "scope should block this estimate",
	}, http.StatusForbidden, nil)
	var taskCountAfter int
	if err := testPool.QueryRow(ctx, `SELECT count(*) FROM gateway_tasks`).Scan(&taskCountAfter); err != nil {
		t.Fatalf("count tasks after scoped request: %v", err)
	}
	if taskCountAfter != taskCountBefore {
		t.Fatalf("scoped API key rejection should happen before task creation, before=%d after=%d", taskCountBefore, taskCountAfter)
	}

	inviteCode := "INVITE-" + suffixText
	if _, err := testPool.Exec(ctx, `
INSERT INTO gateway_invitations (invite_code, max_uses, metadata)
VALUES ($1, 5, '{"purpose":"core-flow"}'::jsonb)`, inviteCode); err != nil {
		t.Fatalf("insert invitation: %v", err)
	}
	invitedUsername := "smoke_invited_" + suffixText
	var invitedRegisterResponse struct {
		AccessToken string `json:"accessToken"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/v1/auth/register", "", map[string]any{
		"username":       invitedUsername,
		"email":          invitedUsername + "@example.com",
		"password":       password,
		"tenantKey":      "manual-invited-" + suffixText,
		"tenantName":     "Manual Invited Tenant",
		"invitationCode": inviteCode,
	}, http.StatusCreated, &invitedRegisterResponse)
	var invitedMe map[string]any
	doJSON(t, server.URL, http.MethodGet, "/api/v1/me", invitedRegisterResponse.AccessToken, nil, http.StatusOK, &invitedMe)
	if invitedMe["tenantKey"] != "default" {
		t.Fatalf("invitation should not change tenant context: %+v", invitedMe)
	}
	var usedCount int
	if err := testPool.QueryRow(ctx, `SELECT used_count FROM gateway_invitations WHERE invite_code = $1`, inviteCode).Scan(&usedCount); err != nil {
		t.Fatalf("read invitation used_count: %v", err)
	}
	if usedCount != 1 {
		t.Fatalf("invitation used_count = %d, want 1", usedCount)
	}
	var userMetadata []byte
	if err := testPool.QueryRow(ctx, `SELECT metadata FROM gateway_users WHERE username = $1`, invitedUsername).Scan(&userMetadata); err != nil {
		t.Fatalf("read invited user metadata: %v", err)
	}
	var metadata map[string]any
	if err := json.Unmarshal(userMetadata, &metadata); err != nil {
		t.Fatalf("decode invited user metadata: %v", err)
	}
	registration, ok := metadata["registration"].(map[string]any)
	if !ok || registration["invitationCode"] != inviteCode {
		t.Fatalf("invitation relationship was not recorded: %+v", metadata)
	}

	var platform struct {
		ID          string `json:"id"`
		Provider    string `json:"provider"`
		PlatformKey string `json:"platformKey"`
		Name        string `json:"name"`
		Status      string `json:"status"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/admin/platforms", loginResponse.AccessToken, map[string]any{
		"provider":    "openai",
		"platformKey": "openai-smoke-" + suffixText,
		"name":        "OpenAI Smoke",
		"baseUrl":     "https://api.openai.com/v1",
		"authType":    "bearer",
		"credentials": map[string]any{"mode": "simulation"},
		"config":      map[string]any{"testMode": true},
	}, http.StatusCreated, &platform)
	if platform.ID == "" || platform.Status != "enabled" {
		t.Fatalf("unexpected platform response: %+v", platform)
	}

	var baseModels struct {
		Items []struct {
			ID                string   `json:"id"`
			CanonicalModelKey string   `json:"canonicalModelKey"`
			InvocationName    string   `json:"invocationName"`
			ProviderModelName string   `json:"providerModelName"`
			ModelType         []string `json:"modelType"`
			ModelAlias        string   `json:"modelAlias"`
			DisplayName       string   `json:"displayName"`
			Status            string   `json:"status"`
		} `json:"items"`
	}
	doJSON(t, server.URL, http.MethodGet, "/api/admin/catalog/base-models", loginResponse.AccessToken, nil, http.StatusOK, &baseModels)
	if len(baseModels.Items) == 0 {
		t.Fatal("migrated base model catalog must not be empty")
	}
	requiredLifecycleModels := map[string]string{
		"gemini:gemini-3.6-flash": "active",
		"aliyun:qwen3.7-max":      "active",
	}
	for _, model := range baseModels.Items {
		if expectedStatus, ok := requiredLifecycleModels[model.CanonicalModelKey]; ok {
			if model.InvocationName == "" || model.ModelAlias != model.InvocationName || model.Status != expectedStatus {
				t.Fatalf("base model identity/lifecycle mismatch: %+v", model)
			}
			delete(requiredLifecycleModels, model.CanonicalModelKey)
		}
	}
	if len(requiredLifecycleModels) != 0 {
		t.Fatalf("missing lifecycle-managed base models: %+v", requiredLifecycleModels)
	}
	requiredMiniMaxModels := map[string]struct {
		providerModelName string
		modelType         string
		alias             string
	}{
		"minimax:speech-2.8-hd":    {providerModelName: "speech-2.8-hd", modelType: "text_to_speech", alias: "speech-2.8-hd"},
		"minimax:speech-2.8-turbo": {providerModelName: "speech-2.8-turbo", modelType: "text_to_speech", alias: "speech-2.8-turbo"},
		"minimax:voice-clone":      {providerModelName: "voice_clone", modelType: "voice_clone", alias: "voice_clone"},
	}
	seenMiniMaxModels := map[string]bool{}
	for _, model := range baseModels.Items {
		required, ok := requiredMiniMaxModels[model.CanonicalModelKey]
		if !ok {
			continue
		}
		seenMiniMaxModels[model.CanonicalModelKey] = true
		if model.ProviderModelName != required.providerModelName || model.ModelAlias != required.alias || !stringSliceContains(model.ModelType, required.modelType) {
			t.Fatalf("MiniMax base model is not aligned with server-main: %+v", model)
		}
		if strings.Contains(model.ModelAlias, " ") {
			t.Fatalf("MiniMax Agent-facing alias should not contain spaces: %+v", model)
		}
	}
	for canonicalKey := range requiredMiniMaxModels {
		if !seenMiniMaxModels[canonicalKey] {
			t.Fatalf("missing MiniMax base model %s in catalog", canonicalKey)
		}
	}

	var modelCatalog struct {
		Items []miniMaxModelCatalogItem `json:"items"`
	}
	doJSON(t, server.URL, http.MethodGet, "/api/v1/model-catalog", apiKeyResponse.Secret, nil, http.StatusOK, &modelCatalog)
	for _, required := range requiredMiniMaxModels {
		item := findModelCatalogItem(modelCatalog.Items, required.alias)
		if item == nil {
			t.Fatalf("missing MiniMax model %s in development model catalog", required.alias)
		}
		if item.ModelName != required.providerModelName || !stringSliceContains(item.ModelType, required.modelType) {
			t.Fatalf("MiniMax model catalog item is not aligned with base model: %+v", *item)
		}
		if strings.Contains(item.Alias, " ") {
			t.Fatalf("MiniMax development catalog alias should not contain spaces: %+v", *item)
		}
		if !modelCatalogItemHasEnabledSource(item.Sources, required.alias, required.providerModelName, required.modelType) {
			t.Fatalf("MiniMax development catalog item has no enabled source: %+v", *item)
		}
	}

	baseModelInput := map[string]any{
		"providerKey":       "openai",
		"canonicalModelKey": "openai:smoke-base-" + suffixText,
		"invocationName":    "smoke-base-" + suffixText,
		"providerModelName": "smoke-base-" + suffixText,
		"modelType":         []string{"text_generate"},
		"displayName":       "Smoke Base Model",
		"legacyAliases":     []string{"Smoke Base Model Legacy"},
		"capabilities":      map[string]any{"originalTypes": []string{"text_generate"}},
		"metadata":          map[string]any{"source": "test"},
	}
	var createdBaseModel struct {
		ID                string `json:"id"`
		CanonicalModelKey string `json:"canonicalModelKey"`
		ModelAlias        string `json:"modelAlias"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/admin/catalog/base-models", loginResponse.AccessToken, baseModelInput, http.StatusCreated, &createdBaseModel)
	if createdBaseModel.ID == "" || createdBaseModel.CanonicalModelKey != baseModelInput["canonicalModelKey"] {
		t.Fatalf("unexpected created base model: %+v", createdBaseModel)
	}
	baseModelInput["displayName"] = "Smoke Base Model Updated"
	var updatedBaseModel struct {
		ModelAlias  string `json:"modelAlias"`
		DisplayName string `json:"displayName"`
	}
	doJSON(t, server.URL, http.MethodPatch, "/api/admin/catalog/base-models/"+createdBaseModel.ID, loginResponse.AccessToken, baseModelInput, http.StatusOK, &updatedBaseModel)
	if updatedBaseModel.ModelAlias != baseModelInput["invocationName"] || updatedBaseModel.DisplayName != "Smoke Base Model Updated" {
		t.Fatalf("unexpected updated base model: %+v", updatedBaseModel)
	}
	doJSON(t, server.URL, http.MethodDelete, "/api/admin/catalog/base-models/"+createdBaseModel.ID, loginResponse.AccessToken, nil, http.StatusNoContent, nil)

	var taskResponse struct {
		Task struct {
			ID                 string         `json:"id"`
			Status             string         `json:"status"`
			RunMode            string         `json:"runMode"`
			APIKeyID           string         `json:"apiKeyId"`
			APIKeyName         string         `json:"apiKeyName"`
			RequestID          string         `json:"requestId"`
			ResolvedModel      string         `json:"resolvedModel"`
			Usage              map[string]any `json:"usage"`
			Metrics            map[string]any `json:"metrics"`
			BillingSummary     map[string]any `json:"billingSummary"`
			FinalChargeAmount  float64        `json:"finalChargeAmount"`
			ResponseStartedAt  string         `json:"responseStartedAt"`
			ResponseFinishedAt string         `json:"responseFinishedAt"`
			ResponseDurationMS int64          `json:"responseDurationMs"`
			Result             map[string]any `json:"result"`
		} `json:"task"`
	}
	defaultTextModel := "openai:gpt-4o-mini"
	var apiV1Chat map[string]any
	doAPIV1ChatCompletionAndLoadTask(t, ctx, testPool, server.URL, apiKeyResponse.Secret, map[string]any{
		"model":                defaultTextModel,
		"runMode":              "simulation",
		"simulation":           true,
		"simulationDurationMs": 5,
		"messages":             []map[string]any{{"role": "user", "content": "ping"}},
	}, "default-chat-"+suffixText, http.StatusOK, &apiV1Chat, &taskResponse.Task)
	if apiV1Chat["object"] != "chat.completion" {
		t.Fatalf("unexpected api v1 chat response: %+v", apiV1Chat)
	}
	if taskResponse.Task.ID == "" || taskResponse.Task.Status != "succeeded" || taskResponse.Task.RunMode != "simulation" {
		t.Fatalf("unexpected task response: %+v", taskResponse.Task)
	}
	if taskResponse.Task.APIKeyID != apiKeyResponse.APIKey.ID || taskResponse.Task.APIKeyName != apiKeyResponse.APIKey.Name {
		t.Fatalf("task should record full api key identity: %+v key=%+v", taskResponse.Task, apiKeyResponse.APIKey)
	}
	if taskResponse.Task.RequestID == "" || taskResponse.Task.ResolvedModel == "" || taskResponse.Task.ResponseStartedAt == "" || taskResponse.Task.ResponseFinishedAt == "" {
		t.Fatalf("task should record provider request and response timing: %+v", taskResponse.Task)
	}
	if taskResponse.Task.Usage["totalTokens"] == nil || taskResponse.Task.FinalChargeAmount <= 0 {
		t.Fatalf("task should record token usage and final charge: %+v", taskResponse.Task)
	}
	if taskResponse.Task.BillingSummary["finalCharge"] == nil || taskResponse.Task.Metrics["requestedModel"] == nil {
		t.Fatalf("task should record billing summary and task metrics: %+v", taskResponse.Task)
	}

	var compatChat map[string]any
	doJSON(t, server.URL, http.MethodPost, "/v1/chat/completions", apiKeyResponse.Secret, map[string]any{
		"model":                defaultTextModel,
		"runMode":              "simulation",
		"messages":             []map[string]any{{"role": "user", "content": "ping"}},
		"simulation":           true,
		"simulationDurationMs": 5,
	}, http.StatusOK, &compatChat)
	if compatChat["object"] != "chat.completion" {
		t.Fatalf("unexpected compatible chat response: %+v", compatChat)
	}

	cancelCtx, cancelRequest := context.WithCancel(context.Background())
	cancelPayload := map[string]any{
		"model":                defaultTextModel,
		"runMode":              "simulation",
		"messages":             []map[string]any{{"role": "user", "content": "cancelled stream"}},
		"stream":               true,
		"simulation":           true,
		"simulationDurationMs": 250,
	}
	cancelRaw, err := json.Marshal(cancelPayload)
	if err != nil {
		t.Fatalf("marshal cancelled stream payload: %v", err)
	}
	cancelReq, err := http.NewRequestWithContext(cancelCtx, http.MethodPost, server.URL+"/v1/chat/completions", bytes.NewReader(cancelRaw))
	if err != nil {
		t.Fatalf("build cancelled stream request: %v", err)
	}
	cancelReq.Header.Set("Authorization", "Bearer "+apiKeyResponse.Secret)
	cancelReq.Header.Set("Content-Type", "application/json")
	cancelErrCh := make(chan error, 1)
	cancelTaskIDCh := make(chan string, 1)
	cancelRequestStartedAt := time.Now().UTC().Add(-time.Second)
	go func() {
		resp, err := http.DefaultClient.Do(cancelReq)
		if resp != nil {
			cancelTaskIDCh <- strings.TrimSpace(resp.Header.Get("X-Gateway-Task-Id"))
			_, _ = io.ReadAll(resp.Body)
			_ = resp.Body.Close()
		} else {
			cancelTaskIDCh <- ""
		}
		cancelErrCh <- err
	}()
	var cancelTaskID string
	select {
	case cancelTaskID = <-cancelTaskIDCh:
	case <-time.After(2 * time.Second):
		t.Fatal("cancelled stream did not return response headers")
	}
	if cancelTaskID == "" {
		cancelTaskID = waitForLatestTaskIDSince(t, ctx, testPool, "chat.completions", cancelRequestStartedAt, 2*time.Second)
	}
	cancelRequest()
	select {
	case <-cancelErrCh:
	case <-time.After(time.Second):
		t.Fatal("cancelled stream request did not return after client cancellation")
	}
	cancelledStreamTask := waitForTaskStatus(t, server.URL, apiKeyResponse.Secret, cancelTaskID, []string{"succeeded"}, 2*time.Second)
	if cancelledStreamTask.Status != "succeeded" {
		t.Fatalf("client-cancelled compatible stream should keep backend task running to success, got %+v", cancelledStreamTask)
	}

	var imageResponse struct {
		Status       string `json:"status"`
		LegacyTaskID string `json:"task_id"`
		TaskID       string `json:"taskId"`
		QueryURL     string `json:"query_url"`
		Task         struct {
			ID     string         `json:"id"`
			Status string         `json:"status"`
			Result map[string]any `json:"result"`
		} `json:"task"`
		Next struct {
			Detail string `json:"detail"`
			Events string `json:"events"`
			Result string `json:"result"`
		} `json:"next"`
	}
	doJSONWithHeaders(t, server.URL, http.MethodPost, "/api/v1/images/generations", apiKeyResponse.Secret, map[string]any{
		"model":                defaultImageModel,
		"runMode":              "simulation",
		"prompt":               "a tiny gateway console",
		"size":                 "1024x1024",
		"quality":              "medium",
		"simulation":           true,
		"simulationDurationMs": 5,
	}, map[string]string{"X-Async": "true"}, http.StatusAccepted, &imageResponse)
	if imageResponse.Status != "submitted" ||
		imageResponse.TaskID == "" ||
		imageResponse.LegacyTaskID != imageResponse.TaskID ||
		imageResponse.Task.ID != imageResponse.TaskID ||
		imageResponse.QueryURL != "/api/v1/ai/result/"+imageResponse.TaskID ||
		imageResponse.Next.Result != imageResponse.QueryURL {
		t.Fatalf("unexpected EasyAI-compatible image submission: %+v", imageResponse)
	}
	waitForTaskStatus(t, server.URL, apiKeyResponse.Secret, imageResponse.Task.ID, []string{"succeeded"}, 10*time.Second)
	doJSON(t, server.URL, http.MethodGet, "/api/v1/tasks/"+imageResponse.Task.ID, apiKeyResponse.Secret, nil, http.StatusOK, &imageResponse.Task)
	if imageResponse.Task.Status != "succeeded" || imageResponse.Task.Result["id"] == "" {
		t.Fatalf("unexpected image generation task: %+v", imageResponse.Task)
	}
	var imageEasyAIResult struct {
		Status        string           `json:"status"`
		TaskID        string           `json:"task_id"`
		Data          []map[string]any `json:"data"`
		Output        []string         `json:"output"`
		OutputContent []map[string]any `json:"output_content"`
	}
	doJSON(t, server.URL, http.MethodGet, imageResponse.QueryURL, apiKeyResponse.Secret, nil, http.StatusOK, &imageEasyAIResult)
	if imageEasyAIResult.Status != "success" ||
		imageEasyAIResult.TaskID != imageResponse.TaskID ||
		len(imageEasyAIResult.Data) == 0 ||
		len(imageEasyAIResult.Output) == 0 ||
		len(imageEasyAIResult.OutputContent) != len(imageEasyAIResult.Data) ||
		imageEasyAIResult.Data[0]["type"] != "image" {
		t.Fatalf("unexpected EasyAI-compatible image result: %+v", imageEasyAIResult)
	}
	var ordinaryLoginResponse struct {
		AccessToken string `json:"accessToken"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/v1/auth/login", "", map[string]any{
		"account":  ordinaryUsername,
		"password": password,
	}, http.StatusOK, &ordinaryLoginResponse)
	var inaccessibleImageResult struct {
		Status string `json:"status"`
		Code   string `json:"code"`
		Data   []any  `json:"data"`
	}
	doJSON(t, server.URL, http.MethodGet, imageResponse.QueryURL, ordinaryLoginResponse.AccessToken, nil, http.StatusNotFound, &inaccessibleImageResult)
	if inaccessibleImageResult.Status != "failed" ||
		inaccessibleImageResult.Code != "not_found" ||
		len(inaccessibleImageResult.Data) != 0 {
		t.Fatalf("EasyAI task result must stay owner-isolated: %+v", inaccessibleImageResult)
	}

	var imageEditResponse struct {
		Status       string `json:"status"`
		LegacyTaskID string `json:"task_id"`
		TaskID       string `json:"taskId"`
		Task         struct {
			ID     string         `json:"id"`
			Status string         `json:"status"`
			Result map[string]any `json:"result"`
		} `json:"task"`
	}
	doJSONWithHeaders(t, server.URL, http.MethodPost, "/api/v1/images/edits", apiKeyResponse.Secret, map[string]any{
		"model":                defaultImageModel,
		"runMode":              "simulation",
		"prompt":               "replace background with clean studio light",
		"image":                server.URL + "/static/simulation/image.png",
		"mask":                 server.URL + "/static/simulation/image-edit.png",
		"simulation":           true,
		"simulationDurationMs": 5,
	}, map[string]string{"X-Async": "true"}, http.StatusAccepted, &imageEditResponse)
	if imageEditResponse.Status != "submitted" ||
		imageEditResponse.TaskID == "" ||
		imageEditResponse.LegacyTaskID != imageEditResponse.TaskID ||
		imageEditResponse.Task.ID != imageEditResponse.TaskID {
		t.Fatalf("unexpected EasyAI-compatible image edit submission: %+v", imageEditResponse)
	}
	waitForTaskStatus(t, server.URL, apiKeyResponse.Secret, imageEditResponse.Task.ID, []string{"succeeded"}, 10*time.Second)
	doJSON(t, server.URL, http.MethodGet, "/api/v1/tasks/"+imageEditResponse.Task.ID, apiKeyResponse.Secret, nil, http.StatusOK, &imageEditResponse.Task)
	if imageEditResponse.Task.Status != "succeeded" || imageEditResponse.Task.Result["id"] == "" {
		t.Fatalf("unexpected image edit task: %+v", imageEditResponse.Task)
	}

	songMarker := "song-simulation-" + suffixText
	var songResult map[string]any
	doJSON(t, server.URL, http.MethodPost, "/api/v1/song/generations", apiKeyResponse.Secret, map[string]any{
		"model":                 "chirp-v5-0",
		"runMode":               "simulation",
		"prompt":                "city lights and soft drums",
		"tags":                  "pop, synth",
		"negativeTags":          "noise",
		"simulation":            true,
		"simulationDurationMs":  5,
		"integrationTestMarker": songMarker,
	}, http.StatusOK, &songResult)
	songData, _ := songResult["data"].([]any)
	if songResult["status"] != "success" || len(songData) == 0 {
		t.Fatalf("unexpected song generation compatible result: %+v", songResult)
	}
	songItem, _ := songData[0].(map[string]any)
	if songItem["type"] != "audio" || songItem["audio_url"] != "/static/simulation/audio.wav" {
		t.Fatalf("song simulation should return audio asset data: %+v", songItem)
	}
	var songTaskDetail struct {
		Status            string         `json:"status"`
		ModelType         string         `json:"modelType"`
		Result            map[string]any `json:"result"`
		FinalChargeAmount float64        `json:"finalChargeAmount"`
	}
	songTaskID := waitForTaskIDByRequestField(t, ctx, testPool, "integrationTestMarker", songMarker, 2*time.Second)
	doJSON(t, server.URL, http.MethodGet, "/api/v1/tasks/"+songTaskID, apiKeyResponse.Secret, nil, http.StatusOK, &songTaskDetail)
	if songTaskDetail.Status != "succeeded" || songTaskDetail.ModelType != "audio_generate" || songTaskDetail.FinalChargeAmount <= 0 {
		t.Fatalf("song simulation task should succeed with audio_generate billing: %+v", songTaskDetail)
	}

	speechMarker := "speech-simulation-" + suffixText
	var speechResult map[string]any
	doJSON(t, server.URL, http.MethodPost, "/api/v1/speech/generations", apiKeyResponse.Secret, map[string]any{
		"model":                 "MiniMax-Speech-2.8-HD",
		"runMode":               "simulation",
		"text":                  "hello gateway speech",
		"voice_id":              "female-shaonv",
		"speed":                 1,
		"vol":                   1,
		"pitch":                 0,
		"simulation":            true,
		"simulationDurationMs":  5,
		"integrationTestMarker": speechMarker,
	}, http.StatusOK, &speechResult)
	speechData, _ := speechResult["data"].([]any)
	if speechResult["status"] != "success" || len(speechData) == 0 {
		t.Fatalf("unexpected speech generation compatible result: %+v", speechResult)
	}
	speechItem, _ := speechData[0].(map[string]any)
	if speechItem["type"] != "audio" || speechItem["audio_url"] != "/static/simulation/audio.wav" || speechItem["revised_text"] != "hello gateway speech" {
		t.Fatalf("speech simulation should return audio asset data: %+v", speechItem)
	}
	var speechTaskDetail struct {
		Status            string  `json:"status"`
		ModelType         string  `json:"modelType"`
		FinalChargeAmount float64 `json:"finalChargeAmount"`
	}
	speechTaskID := waitForTaskIDByRequestField(t, ctx, testPool, "integrationTestMarker", speechMarker, 2*time.Second)
	doJSON(t, server.URL, http.MethodGet, "/api/v1/tasks/"+speechTaskID, apiKeyResponse.Secret, nil, http.StatusOK, &speechTaskDetail)
	if speechTaskDetail.Status != "succeeded" || speechTaskDetail.ModelType != "text_to_speech" || speechTaskDetail.FinalChargeAmount <= 0 {
		t.Fatalf("speech simulation task should succeed with text_to_speech billing: %+v", speechTaskDetail)
	}

	voiceCloneMarker := "voice-clone-simulation-" + suffixText
	voiceID := "voice_clone_" + suffixText
	var voiceCloneResult map[string]any
	doJSON(t, server.URL, http.MethodPost, "/api/v1/voice_clone", apiKeyResponse.Secret, map[string]any{
		"model":                 "MiniMax-Voice-Clone",
		"runMode":               "simulation",
		"voice_id":              voiceID,
		"audio_url":             "/static/simulation/audio.wav",
		"prompt_audio_url":      "/static/simulation/audio.wav",
		"prompt_text":           "hello prompt audio",
		"text":                  "hello voice clone preview",
		"preview_model":         "speech-2.8-hd",
		"simulation":            true,
		"simulationDurationMs":  5,
		"integrationTestMarker": voiceCloneMarker,
	}, http.StatusOK, &voiceCloneResult)
	if voiceCloneResult["status"] != "success" || voiceCloneResult["voice_id"] != voiceID {
		t.Fatalf("unexpected voice clone compatible result: %+v", voiceCloneResult)
	}
	clonedVoice, _ := voiceCloneResult["cloned_voice"].(map[string]any)
	if clonedVoice == nil || clonedVoice["voiceId"] != voiceID || clonedVoice["previewModel"] != "speech-2.8-hd" || clonedVoice["platformId"] == "" {
		t.Fatalf("voice clone should persist a platform-bound cloned voice: %+v", voiceCloneResult)
	}
	var voiceCloneTaskDetail struct {
		Status    string `json:"status"`
		ModelType string `json:"modelType"`
		Model     string `json:"model"`
		Attempts  []struct {
			PlatformID string `json:"platformId"`
		} `json:"attempts"`
	}
	voiceCloneTaskID := waitForTaskIDByRequestField(t, ctx, testPool, "integrationTestMarker", voiceCloneMarker, 2*time.Second)
	doJSON(t, server.URL, http.MethodGet, "/api/v1/tasks/"+voiceCloneTaskID, apiKeyResponse.Secret, nil, http.StatusOK, &voiceCloneTaskDetail)
	if voiceCloneTaskDetail.Status != "succeeded" || voiceCloneTaskDetail.ModelType != "voice_clone" || voiceCloneTaskDetail.Model != "MiniMax-Voice-Clone" || len(voiceCloneTaskDetail.Attempts) == 0 {
		t.Fatalf("voice clone simulation task should succeed through the development interface: %+v", voiceCloneTaskDetail)
	}
	if voiceCloneTaskDetail.Attempts[0].PlatformID != clonedVoice["platformId"] {
		t.Fatalf("voice clone persisted platform should match task attempt: task=%+v voice=%+v", voiceCloneTaskDetail, clonedVoice)
	}

	asyncImageEditModel := "async-image-edit-smoke-" + suffixText
	asyncImageEditCanonicalKey := "openai:" + asyncImageEditModel
	var asyncImageEditBaseModel struct {
		ID string `json:"id"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/admin/catalog/base-models", loginResponse.AccessToken, map[string]any{
		"providerKey":       "openai",
		"canonicalModelKey": asyncImageEditCanonicalKey,
		"invocationName":    asyncImageEditModel,
		"providerModelName": asyncImageEditModel,
		"modelType":         []string{"image_edit", "image_generate"},
		"displayName":       "Async Image Edit Smoke",
		"capabilities":      map[string]any{"originalTypes": []string{"image_edit", "image_generate"}},
		"metadata":          map[string]any{"source": "test"},
	}, http.StatusCreated, &asyncImageEditBaseModel)
	if asyncImageEditBaseModel.ID == "" {
		t.Fatal("async image edit smoke base model was not created")
	}
	var asyncImageEditPlatformModel struct {
		ID string `json:"id"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/admin/platforms/"+platform.ID+"/models", loginResponse.AccessToken, map[string]any{
		"canonicalModelKey": asyncImageEditCanonicalKey,
		"modelName":         asyncImageEditModel,
		"modelAlias":        asyncImageEditModel,
		"modelType":         []string{"image_edit", "image_generate"},
		"displayName":       "Async Image Edit Smoke",
	}, http.StatusCreated, &asyncImageEditPlatformModel)
	var asyncImageEdit struct {
		TaskID string `json:"taskId"`
		Task   struct {
			ID        string `json:"id"`
			Status    string `json:"status"`
			AsyncMode bool   `json:"asyncMode"`
		} `json:"task"`
	}
	doJSONWithHeaders(t, server.URL, http.MethodPost, "/api/v1/images/edits", apiKeyResponse.Secret, map[string]any{
		"model":                asyncImageEditModel,
		"runMode":              "simulation",
		"prompt":               "turn the attached bright desktop object into a clean product-style render",
		"image":                server.URL + "/static/simulation/image.png",
		"size":                 "2048x2048",
		"simulation":           true,
		"simulationDurationMs": 5,
	}, map[string]string{"X-Async": "true"}, http.StatusAccepted, &asyncImageEdit)
	if asyncImageEdit.TaskID == "" || !asyncImageEdit.Task.AsyncMode {
		t.Fatalf("async image edit task should be accepted: %+v", asyncImageEdit)
	}
	asyncImageEditDone := waitForTaskStatus(t, server.URL, apiKeyResponse.Secret, asyncImageEdit.TaskID, []string{"succeeded"}, 5*time.Second)
	if asyncImageEditDone.Status != "succeeded" {
		t.Fatalf("async image edit task should succeed through river queue, got %+v", asyncImageEditDone)
	}

	var gptImageModelTypesRaw []byte
	if err := testPool.QueryRow(ctx, `
SELECT model_type
FROM platform_models
WHERE model_name = 'gpt-image-1'
  AND model_type @> '["image_generate","image_edit"]'::jsonb
LIMIT 1`).Scan(&gptImageModelTypesRaw); err != nil {
		t.Fatalf("gpt-image-1 platform model should store both image model types: %v", err)
	}
	var gptImageModelTypes []string
	if err := json.Unmarshal(gptImageModelTypesRaw, &gptImageModelTypes); err != nil {
		t.Fatalf("decode gpt-image-1 model_type: %v raw=%s", err, string(gptImageModelTypesRaw))
	}
	if !stringSliceContains(gptImageModelTypes, "image_generate") || !stringSliceContains(gptImageModelTypes, "image_edit") {
		t.Fatalf("gpt-image-1 model_type should include generation and edit types: %+v", gptImageModelTypes)
	}

	deniedModel := "permission-deny-smoke-" + suffixText
	createSimulationTextBaseModel(t, server.URL, loginResponse.AccessToken, deniedModel)
	var deniedPlatformModel struct {
		ID string `json:"id"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/admin/platforms/"+platform.ID+"/models", loginResponse.AccessToken, map[string]any{
		"canonicalModelKey": "openai:" + deniedModel,
		"modelName":         deniedModel,
		"modelAlias":        deniedModel,
		"modelType":         []string{"text_generate"},
		"displayName":       "Permission Deny Smoke",
	}, http.StatusCreated, &deniedPlatformModel)
	doJSON(t, server.URL, http.MethodPost, "/api/admin/access-rules", loginResponse.AccessToken, map[string]any{
		"subjectType":        "api_key",
		"subjectId":          chatOnlyAPIKeyResponse.APIKey.ID,
		"resourceType":       "platform_model",
		"resourceId":         deniedPlatformModel.ID,
		"effect":             "deny",
		"priority":           10,
		"minPermissionLevel": 0,
		"status":             "active",
	}, http.StatusCreated, nil)
	var deniedTaskCountBefore, deniedTaskCountAfter int
	if err := testPool.QueryRow(ctx, `SELECT count(*) FROM gateway_tasks`).Scan(&deniedTaskCountBefore); err != nil {
		t.Fatalf("count tasks before denied model request: %v", err)
	}
	doJSON(t, server.URL, http.MethodPost, "/api/v1/chat/completions", chatOnlyAPIKeyResponse.Secret, map[string]any{
		"model":                deniedModel,
		"runMode":              "simulation",
		"simulation":           true,
		"simulationDurationMs": 5,
		"messages":             []map[string]any{{"role": "user", "content": "permission deny"}},
	}, http.StatusNotFound, nil)
	if err := testPool.QueryRow(ctx, `SELECT count(*) FROM gateway_tasks`).Scan(&deniedTaskCountAfter); err != nil {
		t.Fatalf("count tasks after denied model request: %v", err)
	}
	if deniedTaskCountAfter != deniedTaskCountBefore {
		t.Fatalf("deny access rule rejection created a task: before=%d after=%d", deniedTaskCountBefore, deniedTaskCountAfter)
	}
	var restrictedModels struct {
		Items []struct {
			ID        string `json:"id"`
			ModelName string `json:"modelName"`
		} `json:"items"`
	}
	doJSON(t, server.URL, http.MethodGet, "/api/v1/playground/models", chatOnlyAPIKeyResponse.Secret, nil, http.StatusOK, &restrictedModels)
	if modelListContains(restrictedModels.Items, deniedPlatformModel.ID) {
		t.Fatalf("deny access rule should hide denied model from playable list: %+v", restrictedModels.Items)
	}

	controlledModel := "permission-allow-smoke-" + suffixText
	createSimulationTextBaseModel(t, server.URL, loginResponse.AccessToken, controlledModel)
	var controlledPlatformModel struct {
		ID string `json:"id"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/admin/platforms/"+platform.ID+"/models", loginResponse.AccessToken, map[string]any{
		"canonicalModelKey": "openai:" + controlledModel,
		"modelName":         controlledModel,
		"modelAlias":        controlledModel,
		"modelType":         []string{"text_generate"},
		"displayName":       "Permission Allow Smoke",
	}, http.StatusCreated, &controlledPlatformModel)
	doJSON(t, server.URL, http.MethodPost, "/api/admin/access-rules", loginResponse.AccessToken, map[string]any{
		"subjectType":        "api_key",
		"subjectId":          apiKeyResponse.APIKey.ID,
		"resourceType":       "platform_model",
		"resourceId":         controlledPlatformModel.ID,
		"effect":             "allow",
		"priority":           10,
		"minPermissionLevel": 0,
		"status":             "active",
	}, http.StatusCreated, nil)
	var siblingControlledTask struct {
		Task struct {
			Status string `json:"status"`
		} `json:"task"`
	}
	doAPIV1ChatCompletionAndLoadTask(t, ctx, testPool, server.URL, chatOnlyAPIKeyResponse.Secret, map[string]any{
		"model":                controlledModel,
		"runMode":              "simulation",
		"simulation":           true,
		"simulationDurationMs": 5,
		"messages":             []map[string]any{{"role": "user", "content": "another key whitelist must not affect this key"}},
	}, "permission-allow-sibling-"+suffixText, http.StatusOK, nil, &siblingControlledTask.Task)
	if siblingControlledTask.Task.Status != "succeeded" {
		t.Fatalf("another key's allow rule must not restrict this key: %+v", siblingControlledTask.Task)
	}
	var allowedControlledTask struct {
		Task struct {
			Status string `json:"status"`
		} `json:"task"`
	}
	doAPIV1ChatCompletionAndLoadTask(t, ctx, testPool, server.URL, apiKeyResponse.Secret, map[string]any{
		"model":                controlledModel,
		"runMode":              "simulation",
		"simulation":           true,
		"simulationDurationMs": 5,
		"messages":             []map[string]any{{"role": "user", "content": "allow should pass"}},
	}, "permission-allow-pass-"+suffixText, http.StatusOK, nil, &allowedControlledTask.Task)
	if allowedControlledTask.Task.Status != "succeeded" {
		t.Fatalf("matching current-key allow rule should make the controlled model usable: %+v", allowedControlledTask.Task)
	}

	var customPricingRuleSet struct {
		ID string `json:"id"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/admin/pricing/rule-sets", loginResponse.AccessToken, map[string]any{
		"ruleSetKey": "smoke-pricing-" + suffixText,
		"name":       "Smoke Pricing",
		"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},
			{"ruleKey": "video", "displayName": "Video", "resourceType": "video", "unit": "video", "basePrice": 13},
		},
	}, http.StatusCreated, &customPricingRuleSet)
	pricingModel := "pricing-smoke-" + suffixText
	createSimulationTextBaseModel(t, server.URL, loginResponse.AccessToken, pricingModel)
	var pricingPlatformModel map[string]any
	doJSON(t, server.URL, http.MethodPost, "/api/admin/platforms/"+platform.ID+"/models", loginResponse.AccessToken, map[string]any{
		"canonicalModelKey": "openai:" + pricingModel,
		"modelName":         pricingModel,
		"modelAlias":        pricingModel,
		"modelType":         []string{"text_generate"},
		"displayName":       "Pricing Smoke",
		"pricingRuleSetId":  customPricingRuleSet.ID,
	}, http.StatusCreated, &pricingPlatformModel)
	if _, err := testPool.Exec(ctx, `
UPDATE gateway_wallet_accounts
SET balance = 100, total_spent = 0, updated_at = now()
WHERE gateway_user_id = $1::uuid
  AND currency = 'resource'`, smokeGatewayUserID); err != nil {
		t.Fatalf("seed wallet balance: %v", err)
	}
	var walletBalanceBefore float64
	if err := testPool.QueryRow(ctx, `
SELECT balance::float8
FROM gateway_wallet_accounts
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"`
			Status            string         `json:"status"`
			BillingSummary    map[string]any `json:"billingSummary"`
			FinalChargeAmount float64        `json:"finalChargeAmount"`
		} `json:"task"`
	}
	doAPIV1ChatCompletionAndLoadTask(t, ctx, testPool, server.URL, apiKeyResponse.Secret, map[string]any{
		"model":                pricingModel,
		"runMode":              "production",
		"simulation":           true,
		"simulationDurationMs": 5,
		"messages":             []map[string]any{{"role": "user", "content": "priced ping"}},
	}, "pricing-chat-"+suffixText, http.StatusOK, nil, &pricingTask.Task)
	if pricingTask.Task.Status != "succeeded" || !floatNear(pricingTask.Task.FinalChargeAmount, 0.028) {
		t.Fatalf("custom pricing rule set should drive text billing, got task=%+v", pricingTask.Task)
	}
	settlementDeadline := time.Now().Add(3 * time.Second)
	for {
		var billingStatus string
		if err := testPool.QueryRow(ctx, `SELECT billing_status FROM gateway_tasks WHERE id = $1::uuid`, pricingTask.Task.ID).Scan(&billingStatus); err != nil {
			t.Fatalf("read pricing task billing status: %v", err)
		}
		if billingStatus == "settled" {
			break
		}
		if time.Now().After(settlementDeadline) {
			t.Fatalf("pricing task billing did not settle before deadline, status=%s", billingStatus)
		}
		time.Sleep(25 * time.Millisecond)
	}
	var walletBalanceAfter float64
	var walletSpentAfter float64
	if err := testPool.QueryRow(ctx, `
SELECT balance::float8, total_spent::float8
FROM gateway_wallet_accounts
WHERE gateway_user_id = $1::uuid
  AND currency = 'resource'`, smokeGatewayUserID).Scan(&walletBalanceAfter, &walletSpentAfter); err != nil {
		t.Fatalf("read wallet balance after pricing task: %v", err)
	}
	if !floatNear(walletBalanceAfter, walletBalanceBefore-pricingTask.Task.FinalChargeAmount) || !floatNear(walletSpentAfter, pricingTask.Task.FinalChargeAmount) {
		t.Fatalf("task billing should debit wallet balance and spent totals, before=%f after=%f spent=%f task=%+v", walletBalanceBefore, walletBalanceAfter, walletSpentAfter, pricingTask.Task)
	}
	var walletTransactionAmount float64
	if err := testPool.QueryRow(ctx, `
SELECT amount::float8
FROM gateway_wallet_transactions
WHERE reference_type = 'gateway_task'
  AND reference_id = $1
  AND transaction_type = 'task_billing'`, pricingTask.Task.ID).Scan(&walletTransactionAmount); err != nil {
		t.Fatalf("read task billing wallet transaction: %v", err)
	}
	if !floatNear(walletTransactionAmount, pricingTask.Task.FinalChargeAmount) {
		t.Fatalf("task billing transaction amount=%f want=%f", walletTransactionAmount, pricingTask.Task.FinalChargeAmount)
	}
	var walletSummary struct {
		Accounts []struct {
			Currency   string  `json:"currency"`
			Balance    float64 `json:"balance"`
			TotalSpent float64 `json:"totalSpent"`
		} `json:"accounts"`
		PrimaryAccount struct {
			Currency string  `json:"currency"`
			Balance  float64 `json:"balance"`
		} `json:"primaryAccount"`
	}
	doJSON(t, server.URL, http.MethodGet, "/api/workspace/wallet?currency=resource", loginResponse.AccessToken, nil, http.StatusOK, &walletSummary)
	if walletSummary.PrimaryAccount.Currency != "resource" || !floatNear(walletSummary.PrimaryAccount.Balance, walletBalanceAfter) || len(walletSummary.Accounts) == 0 {
		t.Fatalf("workspace wallet should expose current resource balance, got %+v want balance=%f", walletSummary, walletBalanceAfter)
	}
	var desktopConfig struct {
		Billing struct {
			GatewayBaseURL     string `json:"gatewayBaseUrl"`
			GatewayBillingPath string `json:"gatewayBillingPath"`
		} `json:"billing"`
	}
	doJSON(t, server.URL, http.MethodGet, "/api/workspace/desktop-config", loginResponse.AccessToken, nil, http.StatusOK, &desktopConfig)
	if desktopConfig.Billing.GatewayBillingPath != "/workspace/billing" || desktopConfig.Billing.GatewayBaseURL == "" {
		t.Fatalf("desktop config should expose gateway billing route, got %+v", desktopConfig)
	}
	var walletTransactions struct {
		Items []struct {
			TransactionType string  `json:"transactionType"`
			Direction       string  `json:"direction"`
			ReferenceID     string  `json:"referenceId"`
			Currency        string  `json:"currency"`
			Amount          float64 `json:"amount"`
		} `json:"items"`
		Total int `json:"total"`
	}
	doJSON(t, server.URL, http.MethodGet, "/api/workspace/wallet/transactions?direction=debit&pageSize=20", loginResponse.AccessToken, nil, http.StatusOK, &walletTransactions)
	if walletTransactions.Total == 0 || !walletTransactionListContains(walletTransactions.Items, pricingTask.Task.ID) {
		t.Fatalf("workspace wallet transactions should include task billing debit, got %+v", walletTransactions)
	}
	var filteredWalletTransactions struct {
		Items []struct {
			TransactionType string         `json:"transactionType"`
			Direction       string         `json:"direction"`
			ReferenceID     string         `json:"referenceId"`
			Currency        string         `json:"currency"`
			Amount          float64        `json:"amount"`
			Metadata        map[string]any `json:"metadata"`
		} `json:"items"`
		Total int `json:"total"`
	}
	createdFrom := time.Now().Add(-1 * time.Hour).UTC().Format(time.RFC3339)
	doJSON(t, server.URL, http.MethodGet, "/api/workspace/wallet/transactions?direction=debit&pageSize=1&q="+pricingModel+"&createdFrom="+createdFrom, loginResponse.AccessToken, nil, http.StatusOK, &filteredWalletTransactions)
	if filteredWalletTransactions.Total == 0 || !walletTransactionListContainsWithMetadata(filteredWalletTransactions.Items, pricingTask.Task.ID, pricingModel, platform.Name, apiKeyResponse.APIKey.Name, pricingTask.Task.FinalChargeAmount) {
		t.Fatalf("workspace wallet transaction filters should match model and expose task metadata, got %+v", filteredWalletTransactions)
	}

	rateLimitedModel := "rate-limit-smoke-" + suffixText
	rateLimitWindowSeconds := 3
	var rateLimitPolicySet struct {
		ID string `json:"id"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/admin/runtime/policy-sets", loginResponse.AccessToken, map[string]any{
		"policyKey": "smoke-rate-limit-" + suffixText,
		"name":      "Smoke Rate Limit",
		"retryPolicy": map[string]any{
			"enabled":     false,
			"maxAttempts": 1,
		},
		"rateLimitPolicy": map[string]any{
			"rules": []map[string]any{{"metric": "rpm", "limit": 1, "windowSeconds": rateLimitWindowSeconds}},
		},
	}, http.StatusCreated, &rateLimitPolicySet)
	createSimulationBaseModel(
		t,
		server.URL,
		loginResponse.AccessToken,
		rateLimitedModel,
		[]string{"text_generate", "image_generate"},
	)
	var rateLimitPlatformModel map[string]any
	doJSON(t, server.URL, http.MethodPost, "/api/admin/platforms/"+platform.ID+"/models", loginResponse.AccessToken, map[string]any{
		"canonicalModelKey":     "openai:" + rateLimitedModel,
		"modelName":             rateLimitedModel,
		"modelAlias":            rateLimitedModel,
		"modelType":             []string{"text_generate", "image_generate"},
		"displayName":           "Rate Limit Smoke",
		"runtimePolicySetId":    rateLimitPolicySet.ID,
		"runtimePolicyOverride": map[string]any{},
	}, http.StatusCreated, &rateLimitPlatformModel)
	var rateLimitFailedTask struct {
		Task struct {
			Status    string `json:"status"`
			ErrorCode string `json:"errorCode"`
		} `json:"task"`
	}
	doAPIV1ChatCompletionAndLoadTask(t, ctx, testPool, server.URL, apiKeyResponse.Secret, map[string]any{
		"model":                rateLimitedModel,
		"runMode":              "simulation",
		"simulation":           true,
		"simulationDurationMs": 5,
		"simulationProfile":    "non_retryable_failure",
		"messages":             []map[string]any{{"role": "user", "content": "failed first"}},
	}, "rate-limit-failed-first-"+suffixText, http.StatusBadRequest, nil, &rateLimitFailedTask.Task)
	if rateLimitFailedTask.Task.Status != "failed" || rateLimitFailedTask.Task.ErrorCode != "bad_request" {
		t.Fatalf("failed rate-limited task should fail before consuming rpm: %+v", rateLimitFailedTask.Task)
	}
	waitForRateLimitWindowHead(t, rateLimitWindowSeconds)
	var rateLimitTaskOne struct {
		Task struct {
			Status string `json:"status"`
		} `json:"task"`
	}
	doAPIV1ChatCompletionAndLoadTask(t, ctx, testPool, server.URL, apiKeyResponse.Secret, map[string]any{
		"model":                rateLimitedModel,
		"runMode":              "simulation",
		"simulation":           true,
		"simulationDurationMs": 5,
		"messages":             []map[string]any{{"role": "user", "content": "first"}},
	}, "rate-limit-first-"+suffixText, http.StatusOK, nil, &rateLimitTaskOne.Task)
	if rateLimitTaskOne.Task.Status != "succeeded" {
		t.Fatalf("first rate-limited task should succeed: %+v", rateLimitTaskOne.Task)
	}
	var rateLimitTaskTwo struct {
		Task struct {
			Status    string `json:"status"`
			ErrorCode string `json:"errorCode"`
		} `json:"task"`
	}
	doAPIV1ChatCompletionAndLoadTask(t, ctx, testPool, server.URL, apiKeyResponse.Secret, map[string]any{
		"model":                rateLimitedModel,
		"runMode":              "simulation",
		"simulation":           true,
		"simulationDurationMs": 5,
		"messages":             []map[string]any{{"role": "user", "content": "second"}},
	}, "rate-limit-second-"+suffixText, http.StatusTooManyRequests, nil, &rateLimitTaskTwo.Task)
	if rateLimitTaskTwo.Task.Status != "failed" || rateLimitTaskTwo.Task.ErrorCode != "gateway_rate_limited" {
		t.Fatalf("runtime policy rate limit should fail second task with gateway_rate_limited: %+v", rateLimitTaskTwo.Task)
	}
	var asyncRateLimitTask struct {
		TaskID string `json:"taskId"`
		Task   struct {
			ID        string `json:"id"`
			Status    string `json:"status"`
			AsyncMode bool   `json:"asyncMode"`
		} `json:"task"`
	}
	doJSONWithHeaders(t, server.URL, http.MethodPost, "/api/v1/images/generations", apiKeyResponse.Secret, map[string]any{
		"model":                rateLimitedModel,
		"runMode":              "simulation",
		"simulation":           true,
		"simulationDurationMs": 5,
		"prompt":               "async queued",
	}, map[string]string{"X-Async": "true"}, http.StatusAccepted, &asyncRateLimitTask)
	if asyncRateLimitTask.TaskID == "" || asyncRateLimitTask.Task.ID != asyncRateLimitTask.TaskID || !asyncRateLimitTask.Task.AsyncMode {
		t.Fatalf("async task response should expose task id and async mode: %+v", asyncRateLimitTask)
	}
	asyncRateLimitDetail := waitForTaskStatus(t, server.URL, apiKeyResponse.Secret, asyncRateLimitTask.TaskID, []string{"queued"}, 2*time.Second)
	if asyncRateLimitDetail.Status != "queued" {
		t.Fatalf("async rate-limited task should return to queued state, got %+v", asyncRateLimitDetail)
	}
	if len(asyncRateLimitDetail.Attempts) != 0 {
		t.Fatalf("async rate-limited task should wait in queue without recording a failed attempt: %+v", asyncRateLimitDetail)
	}
	var modelRateLimits struct {
		Items []struct {
			ModelName  string `json:"modelName"`
			ModelAlias string `json:"modelAlias"`
			Concurrent struct {
				CurrentValue float64 `json:"currentValue"`
			} `json:"concurrent"`
			QueuedTasks float64 `json:"queuedTasks"`
		} `json:"items"`
	}
	doJSON(t, server.URL, http.MethodGet, "/api/admin/runtime/model-rate-limits", loginResponse.AccessToken, nil, http.StatusOK, &modelRateLimits)
	var queuedTasks float64
	var runningTasks float64
	for _, item := range modelRateLimits.Items {
		if item.ModelName == rateLimitedModel || item.ModelAlias == rateLimitedModel {
			queuedTasks = item.QueuedTasks
			runningTasks = item.Concurrent.CurrentValue
			break
		}
	}
	if queuedTasks+runningTasks < 1 && asyncRateLimitDetail.Status != "queued" {
		t.Fatalf("realtime load should count async rate-limited task as queued or running, got queued=%v running=%v in %+v", queuedTasks, runningTasks, modelRateLimits.Items)
	}
	asyncRateLimitCompleted := waitForTaskStatus(t, server.URL, apiKeyResponse.Secret, asyncRateLimitTask.TaskID, []string{"succeeded"}, time.Duration(rateLimitWindowSeconds+3)*time.Second)
	if asyncRateLimitCompleted.Status != "succeeded" {
		t.Fatalf("async rate-limited task should be pulled from queue after the limit window resets, got %+v", asyncRateLimitCompleted)
	}
	if len(asyncRateLimitCompleted.Attempts) != 1 || asyncRateLimitCompleted.Attempts[0].Status != "succeeded" {
		t.Fatalf("async rate-limited task should record only the real provider attempt after requeue: %+v", asyncRateLimitCompleted)
	}

	videoRouteModel := "video-route-smoke-" + suffixText
	createSimulationBaseModel(
		t,
		server.URL,
		loginResponse.AccessToken,
		videoRouteModel,
		[]string{"video_generate", "image_to_video"},
	)
	var videoRoutePlatformModel map[string]any
	doJSON(t, server.URL, http.MethodPost, "/api/admin/platforms/"+platform.ID+"/models", loginResponse.AccessToken, map[string]any{
		"canonicalModelKey": "openai:" + videoRouteModel,
		"modelName":         videoRouteModel,
		"modelAlias":        videoRouteModel,
		"modelType":         []string{"video_generate", "image_to_video"},
		"displayName":       "Video Route Smoke",
	}, http.StatusCreated, &videoRoutePlatformModel)
	var textToVideoTask struct {
		Task struct {
			ID        string `json:"id"`
			Status    string `json:"status"`
			ModelType string `json:"modelType"`
		} `json:"task"`
	}
	doJSONWithHeaders(t, server.URL, http.MethodPost, "/api/v1/videos/generations", apiKeyResponse.Secret, map[string]any{
		"model":                videoRouteModel,
		"runMode":              "simulation",
		"simulation":           true,
		"simulationDurationMs": 5,
		"prompt":               "text to video route",
	}, map[string]string{"X-Async": "true"}, http.StatusAccepted, &textToVideoTask)
	waitForTaskStatus(t, server.URL, apiKeyResponse.Secret, textToVideoTask.Task.ID, []string{"succeeded"}, 10*time.Second)
	doJSON(t, server.URL, http.MethodGet, "/api/v1/tasks/"+textToVideoTask.Task.ID, apiKeyResponse.Secret, nil, http.StatusOK, &textToVideoTask.Task)
	if textToVideoTask.Task.Status != "succeeded" || textToVideoTask.Task.ModelType != "video_generate" {
		t.Fatalf("text-to-video request should use video_generate model_type: %+v", textToVideoTask.Task)
	}
	var imageToVideoTask struct {
		Task struct {
			ID        string         `json:"id"`
			Status    string         `json:"status"`
			ModelType string         `json:"modelType"`
			Metrics   map[string]any `json:"metrics"`
		} `json:"task"`
	}
	doJSONWithHeaders(t, server.URL, http.MethodPost, "/api/v1/videos/generations", apiKeyResponse.Secret, map[string]any{
		"model":                videoRouteModel,
		"runMode":              "simulation",
		"simulation":           true,
		"simulationDurationMs": 5,
		"prompt":               "image to video route",
		"image":                server.URL + "/static/simulation/image.png",
	}, map[string]string{"X-Async": "true"}, http.StatusAccepted, &imageToVideoTask)
	waitForTaskStatus(t, server.URL, apiKeyResponse.Secret, imageToVideoTask.Task.ID, []string{"succeeded"}, 10*time.Second)
	doJSON(t, server.URL, http.MethodGet, "/api/v1/tasks/"+imageToVideoTask.Task.ID, apiKeyResponse.Secret, nil, http.StatusOK, &imageToVideoTask.Task)
	if imageToVideoTask.Task.Status != "succeeded" || imageToVideoTask.Task.ModelType != "image_to_video" {
		t.Fatalf("image-to-video request should use image_to_video model_type: %+v", imageToVideoTask.Task)
	}
	if _, ok := imageToVideoTask.Task.Metrics["parameterPreprocessing"]; ok {
		t.Fatalf("task metrics should not embed full parameter preprocessing log: %+v", imageToVideoTask.Task.Metrics)
	}
	if imageToVideoTask.Task.Metrics["parameterPreprocessingSummary"] == nil {
		t.Fatalf("task metrics should keep lightweight preprocessing summary: %+v", imageToVideoTask.Task.Metrics)
	}
	var preprocessingDetail struct {
		Items []map[string]any `json:"items"`
	}
	doJSON(t, server.URL, http.MethodGet, "/api/v1/tasks/"+imageToVideoTask.Task.ID+"/param-preprocessing", apiKeyResponse.Secret, nil, http.StatusOK, &preprocessingDetail)
	if len(preprocessingDetail.Items) == 0 {
		t.Fatalf("task preprocessing endpoint should expose persisted preprocessing logs: %+v", preprocessingDetail)
	}
	if preprocessingDetail.Items[0]["changes"] == nil {
		t.Fatalf("preprocessing log should retain its compact change summary: %+v", preprocessingDetail.Items)
	}
	if preprocessingDetail.Items[0]["actualInput"] != nil || preprocessingDetail.Items[0]["convertedOutput"] != nil {
		t.Fatalf("history compaction should remove full preprocessing payloads: %+v", preprocessingDetail.Items)
	}
	var preprocessingRows int
	if err := testPool.QueryRow(ctx, `SELECT count(*) FROM gateway_task_param_preprocessing_logs WHERE task_id = $1::uuid`, imageToVideoTask.Task.ID).Scan(&preprocessingRows); err != nil {
		t.Fatalf("count preprocessing logs: %v", err)
	}
	if preprocessingRows == 0 {
		t.Fatalf("expected preprocessing logs in dedicated table for task %s", imageToVideoTask.Task.ID)
	}

	failoverModel := "phase1-failover-" + suffixText
	createSimulationTextBaseModel(t, server.URL, loginResponse.AccessToken, failoverModel)
	unrelatedPriorityModel := "priority-demote-unrelated-" + suffixText
	createSimulationTextBaseModel(t, server.URL, loginResponse.AccessToken, unrelatedPriorityModel)
	var failedPlatform struct {
		ID string `json:"id"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/admin/platforms", loginResponse.AccessToken, map[string]any{
		"provider":    "openai",
		"platformKey": "openai-fail-" + suffixText,
		"name":        "OpenAI Retryable Failure",
		"baseUrl":     "https://api.openai.com/v1",
		"authType":    "bearer",
		"credentials": map[string]any{"mode": "simulation", "simulationFailure": "retryable_failure"},
		"priority":    10,
	}, http.StatusCreated, &failedPlatform)
	var successPlatform struct {
		ID string `json:"id"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/admin/platforms", loginResponse.AccessToken, map[string]any{
		"provider":    "openai",
		"platformKey": "openai-success-" + suffixText,
		"name":        "OpenAI Retry Success",
		"baseUrl":     "https://api.openai.com/v1",
		"authType":    "bearer",
		"credentials": map[string]any{"mode": "simulation"},
		"priority":    20,
	}, http.StatusCreated, &successPlatform)
	var unrelatedPriorityPlatform struct {
		ID string `json:"id"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/admin/platforms", loginResponse.AccessToken, map[string]any{
		"provider":    "openai",
		"platformKey": "openai-unrelated-priority-" + suffixText,
		"name":        "OpenAI Unrelated Priority",
		"baseUrl":     "https://api.openai.com/v1",
		"authType":    "bearer",
		"credentials": map[string]any{"mode": "simulation"},
		"priority":    900,
	}, http.StatusCreated, &unrelatedPriorityPlatform)
	var unrelatedPriorityPlatformModel map[string]any
	doJSON(t, server.URL, http.MethodPost, "/api/admin/platforms/"+unrelatedPriorityPlatform.ID+"/models", loginResponse.AccessToken, map[string]any{
		"canonicalModelKey": "openai:" + unrelatedPriorityModel,
		"modelName":         unrelatedPriorityModel,
		"modelAlias":        unrelatedPriorityModel,
		"modelType":         []string{"text_generate"},
		"displayName":       "Unrelated Priority",
	}, http.StatusCreated, &unrelatedPriorityPlatformModel)
	for _, platformID := range []string{failedPlatform.ID, successPlatform.ID} {
		var platformModel map[string]any
		doJSON(t, server.URL, http.MethodPost, "/api/admin/platforms/"+platformID+"/models", loginResponse.AccessToken, map[string]any{
			"canonicalModelKey": "openai:" + failoverModel,
			"modelName":         failoverModel,
			"modelAlias":        failoverModel,
			"modelType":         []string{"text_generate"},
			"displayName":       "Failover Smoke",
			"retryPolicy":       map[string]any{"enabled": true, "maxAttempts": 2},
		}, http.StatusCreated, &platformModel)
		if platformModel["id"] == "" {
			t.Fatalf("platform model was not created: %+v", platformModel)
		}
	}
	var failoverTask struct {
		Task struct {
			ID     string `json:"id"`
			Status string `json:"status"`
		} `json:"task"`
	}
	doAPIV1ChatCompletionAndLoadTask(t, ctx, testPool, server.URL, apiKeyResponse.Secret, map[string]any{
		"model":    failoverModel,
		"runMode":  "simulation",
		"messages": []map[string]any{{"role": "user", "content": "retry please"}},
	}, "failover-chat-"+suffixText, http.StatusOK, nil, &failoverTask.Task)
	if failoverTask.Task.Status != "succeeded" {
		t.Fatalf("failover task should succeed through second client: %+v", failoverTask.Task)
	}
	var failoverDetail struct {
		Attempts []struct {
			AttemptNo    int    `json:"attemptNo"`
			PlatformName string `json:"platformName"`
			Status       string `json:"status"`
			Retryable    bool   `json:"retryable"`
			ErrorCode    string `json:"errorCode"`
			ErrorMessage string `json:"errorMessage"`
			ResponseMS   int64  `json:"responseDurationMs"`
		} `json:"attempts"`
		Metrics map[string]any `json:"metrics"`
	}
	doJSON(t, server.URL, http.MethodGet, "/api/v1/tasks/"+failoverTask.Task.ID, apiKeyResponse.Secret, nil, http.StatusOK, &failoverDetail)
	if len(failoverDetail.Attempts) != 3 {
		t.Fatalf("failover task history should include two failed retries plus one successful failover attempt, got %+v", failoverDetail.Attempts)
	}
	if failoverDetail.Attempts[0].PlatformName != "OpenAI Retryable Failure" || failoverDetail.Attempts[0].Status != "failed" || !failoverDetail.Attempts[0].Retryable || failoverDetail.Attempts[0].ErrorCode == "" {
		t.Fatalf("first failover attempt should preserve failed platform and reason: %+v", failoverDetail.Attempts[0])
	}
	if failoverDetail.Attempts[1].PlatformName != "OpenAI Retryable Failure" || failoverDetail.Attempts[1].Status != "failed" || !failoverDetail.Attempts[1].Retryable {
		t.Fatalf("second failover attempt should preserve the same-client retry failure: %+v", failoverDetail.Attempts[1])
	}
	if failoverDetail.Attempts[2].PlatformName != "OpenAI Retry Success" || failoverDetail.Attempts[2].Status != "succeeded" || failoverDetail.Attempts[2].ResponseMS <= 0 {
		t.Fatalf("third failover attempt should preserve successful platform: %+v", failoverDetail.Attempts[2])
	}
	if _, ok := failoverDetail.Metrics["attempts"]; ok {
		t.Fatalf("task metrics should not duplicate the dedicated attempt chain: %+v", failoverDetail.Metrics)
	}
	var demotedDynamicPriority int
	var successEffectivePriority int
	var unrelatedEffectivePriority int
	if err := testPool.QueryRow(ctx, `
SELECT COALESCE(failed.dynamic_priority, -1),
       COALESCE(success.dynamic_priority, success.priority),
       COALESCE(unrelated.dynamic_priority, unrelated.priority)
FROM integration_platforms failed
JOIN integration_platforms success ON success.id = $2::uuid
JOIN integration_platforms unrelated ON unrelated.id = $3::uuid
WHERE failed.id = $1::uuid`, failedPlatform.ID, successPlatform.ID, unrelatedPriorityPlatform.ID).Scan(&demotedDynamicPriority, &successEffectivePriority, &unrelatedEffectivePriority); err != nil {
		t.Fatalf("read priority demotion state: %v", err)
	}
	expectedDemotedPriority := successEffectivePriority + 1
	if unrelatedEffectivePriority <= expectedDemotedPriority {
		t.Fatalf("unrelated priority fixture should sit beyond same-model peers: unrelated=%d expected=%d", unrelatedEffectivePriority, expectedDemotedPriority)
	}
	if demotedDynamicPriority != expectedDemotedPriority {
		t.Fatalf("priority demotion should use only same-model peer clients, got dynamic_priority=%d want %d", demotedDynamicPriority, expectedDemotedPriority)
	}

	var degradePolicySet struct {
		ID string `json:"id"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/admin/runtime/policy-sets", loginResponse.AccessToken, map[string]any{
		"policyKey": "smoke-degrade-" + suffixText,
		"name":      "Smoke Degrade",
		"degradePolicy": map[string]any{
			"enabled":         true,
			"keywords":        []string{"rate_limit"},
			"cooldownSeconds": 600,
		},
	}, http.StatusCreated, &degradePolicySet)
	degradeModel := "degrade-smoke-" + suffixText
	createSimulationTextBaseModel(t, server.URL, loginResponse.AccessToken, degradeModel)
	var degradedPlatform struct {
		ID string `json:"id"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/admin/platforms", loginResponse.AccessToken, map[string]any{
		"provider":    "openai",
		"platformKey": "openai-degrade-" + suffixText,
		"name":        "OpenAI Degrade Failure",
		"baseUrl":     "https://api.openai.com/v1",
		"authType":    "bearer",
		"credentials": map[string]any{"mode": "simulation", "simulationFailure": "rate_limit"},
		"priority":    30,
	}, http.StatusCreated, &degradedPlatform)
	var degradeSuccessPlatform struct {
		ID string `json:"id"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/admin/platforms", loginResponse.AccessToken, map[string]any{
		"provider":    "openai",
		"platformKey": "openai-degrade-success-" + suffixText,
		"name":        "OpenAI Degrade Success",
		"baseUrl":     "https://api.openai.com/v1",
		"authType":    "bearer",
		"credentials": map[string]any{"mode": "simulation"},
		"priority":    40,
	}, http.StatusCreated, &degradeSuccessPlatform)
	for _, item := range []struct {
		platformID         string
		runtimePolicySetID string
	}{
		{platformID: degradedPlatform.ID, runtimePolicySetID: degradePolicySet.ID},
		{platformID: degradeSuccessPlatform.ID},
	} {
		var platformModel struct {
			ID string `json:"id"`
		}
		doJSON(t, server.URL, http.MethodPost, "/api/admin/platforms/"+item.platformID+"/models", loginResponse.AccessToken, map[string]any{
			"canonicalModelKey":  "openai:" + degradeModel,
			"modelName":          degradeModel,
			"modelAlias":         degradeModel,
			"modelType":          []string{"text_generate"},
			"displayName":        "Degrade Smoke",
			"runtimePolicySetId": item.runtimePolicySetID,
		}, http.StatusCreated, &platformModel)
	}
	var degradeTask struct {
		Task struct {
			ID     string `json:"id"`
			Status string `json:"status"`
		} `json:"task"`
	}
	doAPIV1ChatCompletionAndLoadTask(t, ctx, testPool, server.URL, apiKeyResponse.Secret, map[string]any{
		"model":                degradeModel,
		"runMode":              "simulation",
		"simulation":           true,
		"simulationDurationMs": 5,
		"messages":             []map[string]any{{"role": "user", "content": "degrade please"}},
	}, "degrade-chat-"+suffixText, http.StatusOK, nil, &degradeTask.Task)
	if degradeTask.Task.Status != "succeeded" {
		t.Fatalf("degrade task should fail over after cooling down failed model: %+v", degradeTask.Task)
	}
	var cooledDownModel bool
	var cooledDownPlatform bool
	if err := testPool.QueryRow(ctx, `
SELECT COALESCE(m.cooldown_until > now(), false), COALESCE(p.cooldown_until > now(), false)
FROM platform_models m
JOIN integration_platforms p ON p.id = m.platform_id
WHERE m.platform_id = $1::uuid
  AND m.model_name = $2`, degradedPlatform.ID, degradeModel).Scan(&cooledDownModel, &cooledDownPlatform); err != nil {
		t.Fatalf("read degraded model cooldown: %v", err)
	}
	if !cooledDownModel {
		t.Fatal("degrade policy should set platform model cooldown_until")
	}
	if cooledDownPlatform {
		t.Fatal("degrade policy should not cool down the entire platform")
	}

	var autoDisablePolicySet struct {
		ID string `json:"id"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/admin/runtime/policy-sets", loginResponse.AccessToken, map[string]any{
		"policyKey": "smoke-auto-disable-" + suffixText,
		"name":      "Smoke Auto Disable",
		"autoDisablePolicy": map[string]any{
			"enabled":   true,
			"keywords":  []string{"invalid_api_key"},
			"threshold": 1,
		},
	}, http.StatusCreated, &autoDisablePolicySet)
	autoDisableModel := "auto-disable-smoke-" + suffixText
	createSimulationTextBaseModel(t, server.URL, loginResponse.AccessToken, autoDisableModel)
	var invalidKeyPlatform struct {
		ID string `json:"id"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/admin/platforms", loginResponse.AccessToken, map[string]any{
		"provider":    "openai",
		"platformKey": "openai-invalid-key-" + suffixText,
		"name":        "OpenAI Invalid Key",
		"baseUrl":     "https://api.openai.com/v1",
		"authType":    "bearer",
		"credentials": map[string]any{"mode": "simulation", "simulationFailure": "invalid_api_key"},
		"priority":    50,
	}, http.StatusCreated, &invalidKeyPlatform)
	var invalidKeyPlatformModel map[string]any
	doJSON(t, server.URL, http.MethodPost, "/api/admin/platforms/"+invalidKeyPlatform.ID+"/models", loginResponse.AccessToken, map[string]any{
		"canonicalModelKey":  "openai:" + autoDisableModel,
		"modelName":          autoDisableModel,
		"modelAlias":         autoDisableModel,
		"modelType":          []string{"text_generate"},
		"displayName":        "Auto Disable Smoke",
		"runtimePolicySetId": autoDisablePolicySet.ID,
	}, http.StatusCreated, &invalidKeyPlatformModel)
	var autoDisableTask struct {
		Task struct {
			Status    string `json:"status"`
			ErrorCode string `json:"errorCode"`
		} `json:"task"`
	}
	doAPIV1ChatCompletionAndLoadTask(t, ctx, testPool, server.URL, apiKeyResponse.Secret, map[string]any{
		"model":                autoDisableModel,
		"runMode":              "simulation",
		"simulation":           true,
		"simulationDurationMs": 5,
		"messages":             []map[string]any{{"role": "user", "content": "disable please"}},
	}, "auto-disable-chat-"+suffixText, http.StatusBadGateway, nil, &autoDisableTask.Task)
	if autoDisableTask.Task.Status != "failed" || autoDisableTask.Task.ErrorCode != "invalid_api_key" {
		t.Fatalf("auto disable task should fail with invalid_api_key: %+v", autoDisableTask.Task)
	}
	var invalidKeyPlatformStatus string
	if err := testPool.QueryRow(ctx, `SELECT status FROM integration_platforms WHERE id = $1::uuid`, invalidKeyPlatform.ID).Scan(&invalidKeyPlatformStatus); err != nil {
		t.Fatalf("read invalid key platform status: %v", err)
	}
	if invalidKeyPlatformStatus != "enabled" {
		t.Fatalf("single source protection should keep the only source enabled, got %q", invalidKeyPlatformStatus)
	}
	var validKeyPlatform struct {
		ID string `json:"id"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/admin/platforms", loginResponse.AccessToken, map[string]any{
		"provider":    "openai",
		"platformKey": "openai-auto-disable-success-" + suffixText,
		"name":        "OpenAI Auto Disable Success",
		"baseUrl":     "https://api.openai.com/v1",
		"authType":    "bearer",
		"credentials": map[string]any{"mode": "simulation"},
		"priority":    60,
	}, http.StatusCreated, &validKeyPlatform)
	doJSON(t, server.URL, http.MethodPost, "/api/admin/platforms/"+validKeyPlatform.ID+"/models", loginResponse.AccessToken, map[string]any{
		"canonicalModelKey": "openai:" + autoDisableModel,
		"modelName":         autoDisableModel,
		"modelAlias":        autoDisableModel,
		"modelType":         []string{"text_generate"},
		"displayName":       "Auto Disable Smoke Success",
	}, http.StatusCreated, nil)
	doAPIV1ChatCompletionAndLoadTask(t, ctx, testPool, server.URL, apiKeyResponse.Secret, map[string]any{
		"model":                autoDisableModel,
		"runMode":              "simulation",
		"simulation":           true,
		"simulationDurationMs": 5,
		"messages":             []map[string]any{{"role": "user", "content": "disable with fallback please"}},
	}, "auto-disable-with-fallback-chat-"+suffixText, http.StatusOK, nil, &autoDisableTask.Task)
	if autoDisableTask.Task.Status != "succeeded" {
		t.Fatalf("auto disable task should fail over after disabling the bad source: %+v", autoDisableTask.Task)
	}
	if err := testPool.QueryRow(ctx, `SELECT status FROM integration_platforms WHERE id = $1::uuid`, invalidKeyPlatform.ID).Scan(&invalidKeyPlatformStatus); err != nil {
		t.Fatalf("read invalid key platform status after fallback: %v", err)
	}
	if invalidKeyPlatformStatus != "disabled" {
		t.Fatalf("auto disable policy should disable platform, got %q", invalidKeyPlatformStatus)
	}

	var taskDetail struct {
		ID                string         `json:"id"`
		Status            string         `json:"status"`
		APIKeyName        string         `json:"apiKeyName"`
		RequestID         string         `json:"requestId"`
		Usage             map[string]any `json:"usage"`
		BillingSummary    map[string]any `json:"billingSummary"`
		FinalChargeAmount float64        `json:"finalChargeAmount"`
		Result            map[string]any `json:"result"`
	}
	doJSON(t, server.URL, http.MethodGet, "/api/v1/tasks/"+taskResponse.Task.ID, apiKeyResponse.Secret, nil, http.StatusOK, &taskDetail)
	if taskDetail.Status != "succeeded" || taskDetail.Result["id"] == "" {
		t.Fatalf("unexpected task detail: %+v", taskDetail)
	}
	if taskDetail.APIKeyName != apiKeyResponse.APIKey.Name || taskDetail.RequestID == "" || taskDetail.Usage["totalTokens"] == nil || taskDetail.FinalChargeAmount <= 0 {
		t.Fatalf("task detail should expose enriched record fields: %+v", taskDetail)
	}
	var taskList struct {
		Items []struct {
			ID           string  `json:"id"`
			Status       string  `json:"status"`
			APIKeyName   string  `json:"apiKeyName"`
			ModelType    string  `json:"modelType"`
			FinalCharge  float64 `json:"finalChargeAmount"`
			ErrorCode    string  `json:"errorCode"`
			ErrorMessage string  `json:"errorMessage"`
		} `json:"items"`
	}
	doJSON(t, server.URL, http.MethodGet, "/api/v1/tasks?limit=50", loginResponse.AccessToken, nil, http.StatusOK, &taskList)
	if !taskListContains(taskList.Items, taskResponse.Task.ID) || !taskListContains(taskList.Items, pricingTask.Task.ID) {
		t.Fatalf("task list should include persisted task records, got %+v", taskList.Items)
	}
	var workspaceTaskList struct {
		Items []struct {
			ID           string  `json:"id"`
			Status       string  `json:"status"`
			APIKeyName   string  `json:"apiKeyName"`
			ModelType    string  `json:"modelType"`
			FinalCharge  float64 `json:"finalChargeAmount"`
			ErrorCode    string  `json:"errorCode"`
			ErrorMessage string  `json:"errorMessage"`
		} `json:"items"`
	}
	doJSON(t, server.URL, http.MethodGet, "/api/workspace/tasks?limit=50", loginResponse.AccessToken, nil, http.StatusOK, &workspaceTaskList)
	if !taskListContains(workspaceTaskList.Items, taskResponse.Task.ID) || !taskListContains(workspaceTaskList.Items, pricingTask.Task.ID) {
		t.Fatalf("workspace task list should include persisted task records, got %+v", workspaceTaskList.Items)
	}
	doJSON(t, server.URL, http.MethodGet, "/api/admin/tasks", ordinaryLoginResponse.AccessToken, nil, http.StatusForbidden, nil)
	if _, err := testPool.Exec(ctx, `
UPDATE gateway_tasks
SET request = request || '{"diagnostic":{"apiKey":"private-admin-task-test-key"}}'::jsonb
WHERE id = $1::uuid`, taskResponse.Task.ID); err != nil {
		t.Fatalf("seed admin task redaction payload: %v", err)
	}
	var adminTaskList struct {
		Items []struct {
			ID      string         `json:"id"`
			Request map[string]any `json:"request"`
			Context struct {
				User struct {
					ID       string `json:"id"`
					Username string `json:"username"`
				} `json:"user"`
				Tenant struct {
					ID   string `json:"id"`
					Name string `json:"name"`
				} `json:"tenant"`
			} `json:"adminContext"`
		} `json:"items"`
		Total int `json:"total"`
	}
	var taskPlatformID string
	var taskBillingStatus string
	if err := testPool.QueryRow(ctx, `
SELECT platform_id::text
FROM gateway_task_attempts
WHERE task_id = $1::uuid
  AND status = 'succeeded'
ORDER BY attempt_no DESC
LIMIT 1`, taskResponse.Task.ID).Scan(&taskPlatformID); err != nil {
		t.Fatalf("read task platform for admin filters: %v", err)
	}
	if err := testPool.QueryRow(ctx, `
SELECT billing_status
FROM gateway_tasks
WHERE id = $1::uuid`, taskResponse.Task.ID).Scan(&taskBillingStatus); err != nil {
		t.Fatalf("read task billing status for admin filters: %v", err)
	}
	adminTaskQuery := "/api/admin/tasks?q=" + taskResponse.Task.ID +
		"&userId=" + smokeGatewayUserID +
		"&status=succeeded" +
		"&platformId=" + taskPlatformID +
		"&model=" + defaultTextModel +
		"&modelType=text_generate" +
		"&runMode=simulation" +
		"&billingStatus=" + taskBillingStatus +
		"&apiKey=smoke" +
		"&pageSize=10"
	doJSON(t, server.URL, http.MethodGet, adminTaskQuery, loginResponse.AccessToken, nil, http.StatusOK, &adminTaskList)
	if len(adminTaskList.Items) != 1 || adminTaskList.Total != 1 || adminTaskList.Items[0].Context.User.Username != username {
		t.Fatalf("admin task list should expose the matching task and identity summary: %+v", adminTaskList)
	}
	maskedDiagnostic, _ := adminTaskList.Items[0].Request["diagnostic"].(map[string]any)
	if maskedDiagnostic["apiKey"] != "***" {
		t.Fatalf("admin task list must mask nested secrets: %+v", adminTaskList.Items[0].Request)
	}
	var maskedAdminTask struct {
		Request map[string]any `json:"request"`
	}
	doJSON(t, server.URL, http.MethodGet, "/api/admin/tasks/"+taskResponse.Task.ID, loginResponse.AccessToken, nil, http.StatusOK, &maskedAdminTask)
	maskedDiagnostic, _ = maskedAdminTask.Request["diagnostic"].(map[string]any)
	if maskedDiagnostic["apiKey"] != "***" {
		t.Fatalf("admin task detail must be masked by default: %+v", maskedAdminTask.Request)
	}
	var fullAdminTask struct {
		Request map[string]any `json:"request"`
	}
	doJSON(t, server.URL, http.MethodGet, "/api/admin/tasks/"+taskResponse.Task.ID+"?sensitive=full", loginResponse.AccessToken, nil, http.StatusOK, &fullAdminTask)
	fullDiagnostic, _ := fullAdminTask.Request["diagnostic"].(map[string]any)
	if fullDiagnostic["apiKey"] != "private-admin-task-test-key" {
		t.Fatalf("explicit full admin task detail should return stored content: %+v", fullAdminTask.Request)
	}

	req, err := http.NewRequest(http.MethodGet, server.URL+"/api/v1/tasks/"+pricingTask.Task.ID+"/events", nil)
	if err != nil {
		t.Fatalf("build events request: %v", err)
	}
	req.Header.Set("Authorization", "Bearer "+apiKeyResponse.Secret)
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatalf("events request: %v", err)
	}
	defer resp.Body.Close()
	body, _ := io.ReadAll(resp.Body)
	if resp.StatusCode != http.StatusOK || !bytes.Contains(body, []byte("task.completed")) {
		t.Fatalf("unexpected events response status=%d body=%s", resp.StatusCode, string(body))
	}
	if !bytes.Contains(body, []byte("task.running")) {
		t.Fatalf("events response should include running transition event body=%s", string(body))
	}
	if !bytes.Contains(body, []byte("task.billing.settled")) {
		t.Fatalf("events response should include billing settlement event body=%s", string(body))
	}

	req, err = http.NewRequest(http.MethodGet, server.URL+"/api/v1/tasks/"+failoverTask.Task.ID+"/events", nil)
	if err != nil {
		t.Fatalf("build failover events request: %v", err)
	}
	req.Header.Set("Authorization", "Bearer "+apiKeyResponse.Secret)
	resp, err = http.DefaultClient.Do(req)
	if err != nil {
		t.Fatalf("failover events request: %v", err)
	}
	defer resp.Body.Close()
	body, _ = io.ReadAll(resp.Body)
	if resp.StatusCode != http.StatusOK ||
		!bytes.Contains(body, []byte("task.attempt.failed")) ||
		!bytes.Contains(body, []byte("task.policy.priority_demoted")) {
		t.Fatalf("failover events should include the failed attempt and applied policy status=%d body=%s", resp.StatusCode, string(body))
	}

	assertLoadAvoidanceSimulatedRetryChain(t, ctx, testPool, server.URL, loginResponse.AccessToken, apiKeyResponse.Secret, suffixText)

	var callbackRows int
	if err := testPool.QueryRow(ctx, `SELECT count(*) FROM gateway_task_callback_outbox WHERE task_id = $1::uuid`, taskResponse.Task.ID).Scan(&callbackRows); err != nil {
		t.Fatalf("read callback outbox: %v", err)
	}
	if callbackRows == 0 {
		t.Fatal("task progress callback outbox should receive events")
	}

	var restartAsyncTask struct {
		TaskID string `json:"taskId"`
		Task   struct {
			ID        string `json:"id"`
			Status    string `json:"status"`
			AsyncMode bool   `json:"asyncMode"`
		} `json:"task"`
	}
	doJSONWithHeaders(t, server.URL, http.MethodPost, "/api/v1/images/generations", apiKeyResponse.Secret, map[string]any{
		"model":                defaultImageModel,
		"runMode":              "simulation",
		"simulation":           true,
		"simulationDurationMs": 2000,
		"prompt":               "river worker restart",
		"size":                 "1024x1024",
	}, map[string]string{"X-Async": "true"}, http.StatusAccepted, &restartAsyncTask)
	if restartAsyncTask.TaskID == "" || !restartAsyncTask.Task.AsyncMode {
		t.Fatalf("restart async task should be accepted as async: %+v", restartAsyncTask)
	}
	restartRunning := waitForTaskStatus(t, server.URL, apiKeyResponse.Secret, restartAsyncTask.TaskID, []string{"running"}, 3*time.Second)
	if restartRunning.Status != "running" {
		t.Fatalf("restart async task should be running before worker restart, got %+v", restartRunning)
	}
	cancelServer()
	serverCtx2, cancelServer2 := context.WithCancel(ctx)
	defer cancelServer2()
	server2 := httptest.NewServer(NewServerWithContext(serverCtx2, config.Config{
		AppEnv:                            "test",
		HTTPAddr:                          ":0",
		DatabaseURL:                       databaseURL,
		IdentityMode:                      "hybrid",
		JWTSecret:                         "test-secret",
		BillingEngineMode:                 "enforce",
		TaskProgressCallbackEnabled:       true,
		TaskProgressCallbackURL:           "http://callback.local/task-progress",
		CORSAllowedOrigin:                 "*",
		AsyncQueueWorkerEnabled:           true,
		AsyncWorkerHardLimit:              32,
		AsyncWorkerInstanceHardLimit:      32,
		AsyncWorkerRefreshIntervalSeconds: 1,
	}, db, slog.New(slog.NewTextHandler(io.Discard, nil))))
	defer server2.Close()
	restartRecovered := waitForTaskStatus(t, server2.URL, apiKeyResponse.Secret, restartAsyncTask.TaskID, []string{"succeeded"}, 8*time.Second)
	if restartRecovered.Status != "succeeded" {
		t.Fatalf("river worker restart should recover and finish async task, got %+v", restartRecovered)
	}
}

func TestCacheAffinityRoutingFlow(t *testing.T) {
	databaseURL := strings.TrimSpace(os.Getenv("AI_GATEWAY_TEST_DATABASE_URL"))
	if databaseURL == "" {
		t.Skip("set AI_GATEWAY_TEST_DATABASE_URL to run the PostgreSQL integration flow")
	}
	ctx := context.Background()
	applyMigration(t, ctx, databaseURL)

	db, err := store.Connect(ctx, databaseURL)
	if err != nil {
		t.Fatalf("connect store: %v", err)
	}
	defer db.Close()

	serverCtx, cancelServer := context.WithCancel(ctx)
	defer cancelServer()
	server := httptest.NewServer(NewServerWithContext(serverCtx, config.Config{
		AppEnv:            "test",
		HTTPAddr:          ":0",
		DatabaseURL:       databaseURL,
		IdentityMode:      "hybrid",
		JWTSecret:         "test-secret",
		CORSAllowedOrigin: "*",
	}, db, slog.New(slog.NewTextHandler(io.Discard, nil))))
	defer server.Close()

	suffixText := strconv.FormatInt(time.Now().UnixNano(), 10)
	username := "cache_affinity_admin_" + suffixText
	password := "password123"
	var registerResponse struct {
		AccessToken string `json:"accessToken"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/v1/auth/register", "", map[string]any{
		"username": username,
		"email":    username + "@example.com",
		"password": password,
	}, http.StatusCreated, &registerResponse)

	testPool, err := pgxpool.New(ctx, databaseURL)
	if err != nil {
		t.Fatalf("connect test pool: %v", err)
	}
	defer testPool.Close()
	if _, err := testPool.Exec(ctx, `UPDATE gateway_users SET roles = '["admin"]'::jsonb WHERE username = $1`, username); err != nil {
		t.Fatalf("promote cache affinity user: %v", err)
	}
	var loginResponse struct {
		AccessToken string `json:"accessToken"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/v1/auth/login", "", map[string]any{
		"account":  username,
		"password": password,
	}, http.StatusOK, &loginResponse)
	var apiKeyResponse struct {
		Secret string `json:"secret"`
		APIKey struct {
			ID string `json:"id"`
		} `json:"apiKey"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/v1/api-keys", loginResponse.AccessToken, map[string]any{
		"name": "cache affinity key",
	}, http.StatusCreated, &apiKeyResponse)

	model := "cache-affinity-smoke-" + suffixText
	cacheKey := "cache-affinity-key-" + suffixText
	createSimulationTextBaseModel(t, server.URL, loginResponse.AccessToken, model)
	lowPlatform, lowPlatformModelID := createSimulationTextPlatformModel(t, server.URL, loginResponse.AccessToken, "cache-low-"+suffixText, "Cache Affinity Low Priority", model, 20, map[string]any{
		"rules": []map[string]any{
			{"metric": "concurrent", "limit": 1, "leaseTtlSeconds": 120},
		},
	})
	for index := 0; index < 3; index++ {
		detail := runCacheAffinityChatTask(t, ctx, testPool, server.URL, apiKeyResponse.Secret, model, cacheKey, "prime-"+strconv.Itoa(index)+"-"+suffixText, map[string]any{
			"inputTokens":       1000,
			"cachedInputTokens": 800,
			"outputTokens":      20,
		})
		if detail.Status != "succeeded" || len(detail.Attempts) != 1 || detail.Attempts[0].PlatformName != "Cache Affinity Low Priority" {
			t.Fatalf("priming request should use the only low-priority platform, got %+v", detail)
		}
	}

	tools := []any{map[string]any{
		"type": "function",
		"function": map[string]any{
			"name":        "lookup_weather",
			"description": "Look up current weather",
			"parameters":  map[string]any{"type": "object", "properties": map[string]any{"city": map[string]any{"type": "string"}}},
		},
	}}
	initialToolMessages := []any{
		map[string]any{"role": "system", "content": "Use tools when helpful."},
		map[string]any{"role": "user", "content": "Weather in Hangzhou?"},
	}
	for index := 0; index < 3; index++ {
		detail := runCacheAffinityChatTaskWithBody(t, ctx, testPool, server.URL, apiKeyResponse.Secret, model, "tool-prime-"+strconv.Itoa(index)+"-"+suffixText, map[string]any{
			"tools":           tools,
			"messages":        initialToolMessages,
			"simulationUsage": map[string]any{"inputTokens": 1000, "cachedInputTokens": 750, "outputTokens": 20},
		})
		if detail.Status != "succeeded" || len(detail.Attempts) != 1 || detail.Attempts[0].PlatformName != "Cache Affinity Low Priority" {
			t.Fatalf("tool-history priming request should use the only platform, got %+v", detail)
		}
	}

	peerPlatform, _ := createSimulationTextPlatformModel(t, server.URL, loginResponse.AccessToken, "cache-peer-"+suffixText, "Cache Affinity Peer", model, 20, nil)
	_ = peerPlatform
	sameKeyDetail := runCacheAffinityChatTask(t, ctx, testPool, server.URL, apiKeyResponse.Secret, model, cacheKey, "same-key-"+suffixText, map[string]any{
		"inputTokens":       1000,
		"cachedInputTokens": 700,
		"outputTokens":      20,
	})
	if sameKeyDetail.Status != "succeeded" || len(sameKeyDetail.Attempts) != 1 || sameKeyDetail.Attempts[0].PlatformName != "Cache Affinity Low Priority" {
		t.Fatalf("same cache key should prefer cached low-priority platform, got %+v", sameKeyDetail)
	}
	if !boolFromTestMap(sameKeyDetail.Attempts[0].Metrics, "cacheAffinityApplied") || floatFromTestAny(sameKeyDetail.Attempts[0].Metrics["cacheAffinityBoost"]) <= 0 {
		t.Fatalf("same cache key attempt should expose cache affinity metrics: %+v", sameKeyDetail.Attempts[0].Metrics)
	}
	if intFromTestAny(sameKeyDetail.Usage["inputTokens"]) != 1000 ||
		intFromTestAny(sameKeyDetail.Usage["cachedInputTokens"]) != 700 ||
		!boolFromTestMap(sameKeyDetail.Usage, "cachedInputTokensKnown") {
		t.Fatalf("same cache key task detail should expose simulated usage, got %+v", sameKeyDetail.Usage)
	}

	toolHistoryDetail := runCacheAffinityChatTaskWithBody(t, ctx, testPool, server.URL, apiKeyResponse.Secret, model, "tool-history-"+suffixText, map[string]any{
		"tools": tools,
		"messages": append(append([]any{}, initialToolMessages...),
			map[string]any{"role": "assistant", "tool_calls": []any{map[string]any{
				"id":   "call_weather",
				"type": "function",
				"function": map[string]any{
					"name":      "lookup_weather",
					"arguments": `{"city":"Hangzhou"}`,
				},
			}}},
			map[string]any{"role": "tool", "tool_call_id": "call_weather", "content": `{"temperature":31}`},
		),
		"simulationUsage": map[string]any{"inputTokens": 1400, "cachedInputTokens": 900, "outputTokens": 30},
	})
	if toolHistoryDetail.Status != "succeeded" || len(toolHistoryDetail.Attempts) != 1 || toolHistoryDetail.Attempts[0].PlatformName != "Cache Affinity Low Priority" {
		t.Fatalf("tool continuation should retain the cached platform, got %+v", toolHistoryDetail)
	}
	if !boolFromTestMap(toolHistoryDetail.Attempts[0].Metrics, "cacheAffinityMatched") ||
		intFromTestAny(toolHistoryDetail.Attempts[0].Metrics["cacheAffinityMatchedPrefixDepth"]) < 2 ||
		intFromTestAny(toolHistoryDetail.Attempts[0].Metrics["cacheAffinityCandidateCount"]) < 1 {
		t.Fatalf("tool continuation should expose prefix and candidate metrics, got %+v", toolHistoryDetail.Attempts[0].Metrics)
	}

	newKeyDetail := runCacheAffinityChatTask(t, ctx, testPool, server.URL, apiKeyResponse.Secret, model, "cache-affinity-new-"+suffixText, "new-key-"+suffixText, map[string]any{
		"inputTokens":       1000,
		"cachedInputTokens": 0,
		"outputTokens":      20,
	})
	if newKeyDetail.Status != "succeeded" || len(newKeyDetail.Attempts) != 1 || newKeyDetail.Attempts[0].PlatformName != "Cache Affinity Peer" {
		t.Fatalf("new cache key should fall back to base priority, got %+v", newKeyDetail)
	}

	seedQueuedConcurrencyLoad(t, ctx, testPool, lowPlatform.PlatformKey, model, lowPlatformModelID)
	fullAvoidedDetail := runCacheAffinityChatTask(t, ctx, testPool, server.URL, apiKeyResponse.Secret, model, cacheKey, "full-avoid-"+suffixText, map[string]any{
		"inputTokens":       1000,
		"cachedInputTokens": 0,
		"outputTokens":      20,
	})
	if fullAvoidedDetail.Status != "succeeded" || len(fullAvoidedDetail.Attempts) != 1 || fullAvoidedDetail.Attempts[0].PlatformName != "Cache Affinity Peer" {
		t.Fatalf("full cached platform should be avoided before cache affinity boost, got %+v", fullAvoidedDetail)
	}
	if fullAvoidedDetail.Attempts[0].Metrics["cacheAffinityOverrideReason"] != "capacity_tier_unavailable" {
		t.Fatalf("full cached platform override reason should be visible, got %+v", fullAvoidedDetail.Attempts[0].Metrics)
	}

	var requestCount int
	var inputTokens int
	var cachedTokens int
	var emaHitRatio float64
	if err := testPool.QueryRow(ctx, `
	SELECT request_count::int, input_tokens::int, cached_input_tokens::int, ema_hit_ratio::float8
	FROM gateway_cache_affinity_stats
	WHERE client_id = $1
	ORDER BY request_count DESC, cached_input_tokens DESC, updated_at DESC
	LIMIT 1`, lowPlatform.PlatformKey+":text_generate:"+model).Scan(&requestCount, &inputTokens, &cachedTokens, &emaHitRatio); err != nil {
		t.Fatalf("read cache affinity stats: %v", err)
	}
	if requestCount < 4 || inputTokens < 4000 || cachedTokens < 3100 || emaHitRatio <= 0 {
		t.Fatalf("cache affinity stats should aggregate low-priority platform observations, count=%d input=%d cached=%d ema=%v", requestCount, inputTokens, cachedTokens, emaHitRatio)
	}
}

func TestOriginAllowedSupportsCommaSeparatedOrigins(t *testing.T) {
	allowed := "http://localhost:5178, http://127.0.0.1:5178"
	if !originAllowed("http://localhost:5178", allowed) {
		t.Fatal("localhost origin should be allowed")
	}
	if !originAllowed("http://127.0.0.1:5178", allowed) {
		t.Fatal("127.0.0.1 origin should be allowed")
	}
	if originAllowed("http://127.0.0.1:5179", allowed) {
		t.Fatal("unexpected origin should not be allowed")
	}
	if originAllowed("https://evil.example.com", "*") {
		t.Fatal("credentialed wildcard CORS origin should not be allowed")
	}
}

func assertRuntimeRecoveryReleasesPendingRateReservations(t *testing.T, ctx context.Context, db *store.Store) {
	t.Helper()
	suffix := strconv.FormatInt(time.Now().UnixNano(), 10)
	task, err := db.CreateTask(ctx, store.CreateTaskInput{
		Kind:    "chat.completions",
		Model:   "recovery-reservation-smoke-" + suffix,
		RunMode: "test",
		Async:   true,
		Request: map[string]any{"messages": []any{}},
	}, &auth.User{ID: "recovery-user-" + suffix})
	if err != nil {
		t.Fatalf("create recovery reservation task: %v", err)
	}
	scopeKey := "recovery-client-" + suffix
	if _, err := db.ReserveRateLimits(ctx, task.ID, "", []store.RateLimitReservation{{
		ScopeType:     "client",
		ScopeKey:      scopeKey,
		Metric:        "rpm",
		Limit:         10,
		Amount:        3,
		WindowSeconds: 60,
	}}); err != nil {
		t.Fatalf("reserve recovery rate limit: %v", err)
	}
	if _, err := db.Pool().Exec(ctx, `
UPDATE gateway_tasks
SET status = 'failed',
    error_code = 'integration_test',
    error_message = 'terminal recovery fixture',
    finished_at = now()
WHERE id = $1::uuid`, task.ID); err != nil {
		t.Fatalf("mark recovery reservation task terminal: %v", err)
	}
	recovery, err := db.RecoverInterruptedRuntimeState(ctx)
	if err != nil {
		t.Fatalf("recover interrupted runtime state with pending reservation: %v", err)
	}
	if recovery.ReleasedRateReservations == 0 {
		t.Fatalf("recovery should release pending rate reservation, got %+v", recovery)
	}
	var reservedValue float64
	if err := db.Pool().QueryRow(ctx, `
SELECT COALESCE(MAX(reserved_value), 0)::float8
FROM gateway_rate_limit_counters
WHERE scope_type = 'client'
  AND scope_key = $1
  AND metric = 'rpm'`, scopeKey).Scan(&reservedValue); err != nil {
		t.Fatalf("query recovered rate limit counter: %v", err)
	}
	if reservedValue != 0 {
		t.Fatalf("recovery should release reserved counter value, got %f", reservedValue)
	}
}

func applyMigration(t *testing.T, ctx context.Context, databaseURL string) {
	t.Helper()
	_, filename, _, _ := runtime.Caller(0)
	migrationFiles, err := filepath.Glob(filepath.Join(filepath.Dir(filename), "..", "..", "migrations", "*.sql"))
	if err != nil {
		t.Fatalf("read migration files: %v", err)
	}
	sort.Strings(migrationFiles)
	pool, err := pgxpool.New(ctx, databaseURL)
	if err != nil {
		t.Fatalf("connect migration db: %v", err)
	}
	defer pool.Close()
	requireDedicatedIntegrationTestDatabase(t, ctx, pool)
	if _, err := pool.Exec(ctx, `
CREATE TABLE IF NOT EXISTS schema_migrations (
  version text PRIMARY KEY,
  applied_at timestamptz NOT NULL DEFAULT now()
);`); err != nil {
		t.Fatalf("ensure schema migrations: %v", err)
	}
	for _, migrationPath := range migrationFiles {
		version := strings.TrimSuffix(filepath.Base(migrationPath), filepath.Ext(migrationPath))
		var exists bool
		if err := pool.QueryRow(ctx, "SELECT EXISTS (SELECT 1 FROM schema_migrations WHERE version = $1)", version).Scan(&exists); err != nil {
			t.Fatalf("check migration %s: %v", filepath.Base(migrationPath), err)
		}
		if exists {
			continue
		}
		migration, err := os.ReadFile(migrationPath)
		if err != nil {
			t.Fatalf("read migration %s: %v", filepath.Base(migrationPath), err)
		}
		migrationSQL := string(migration)
		const noTransactionMarker = "-- easyai:migration:no-transaction"
		if strings.HasPrefix(strings.TrimSpace(migrationSQL), noTransactionMarker) {
			migrationSQL = strings.TrimSpace(strings.TrimPrefix(strings.TrimSpace(migrationSQL), noTransactionMarker))
			for _, statement := range strings.Split(migrationSQL, "-- easyai:migration:statement") {
				if statement = strings.TrimSpace(statement); statement != "" {
					if _, err := pool.Exec(ctx, statement); err != nil {
						t.Fatalf("apply non-transaction migration %s: %v", filepath.Base(migrationPath), err)
					}
				}
			}
			if _, err := pool.Exec(ctx, "INSERT INTO schema_migrations(version) VALUES($1)", version); err != nil {
				t.Fatalf("record non-transaction migration %s: %v", filepath.Base(migrationPath), err)
			}
			continue
		}
		tx, err := pool.Begin(ctx)
		if err != nil {
			t.Fatalf("begin migration %s: %v", filepath.Base(migrationPath), err)
		}
		if _, err := tx.Exec(ctx, string(migration)); err != nil {
			_ = tx.Rollback(ctx)
			t.Fatalf("apply migration %s: %v", filepath.Base(migrationPath), err)
		}
		if _, err := tx.Exec(ctx, "INSERT INTO schema_migrations(version) VALUES($1)", version); err != nil {
			_ = tx.Rollback(ctx)
			t.Fatalf("record migration %s: %v", filepath.Base(migrationPath), err)
		}
		if err := tx.Commit(ctx); err != nil {
			t.Fatalf("commit migration %s: %v", filepath.Base(migrationPath), err)
		}
	}
}

func assertSeedreamProCatalogMigration(t *testing.T, ctx context.Context, pool *pgxpool.Pool) {
	t.Helper()
	var valid bool
	if err := pool.QueryRow(ctx, `
SELECT EXISTS (
  SELECT 1
  FROM base_model_catalog model
  WHERE model.provider_key = 'volces'
    AND model.canonical_model_key = 'volces:doubao-seedream-5-0-pro-260628'
    AND model.invocation_name = 'doubao-seedream-5-0-pro-260628'
    AND model.provider_model_name = 'doubao-seedream-5-0-pro-260628'
    AND model.display_name = 'Seedream 5.0 Pro'
    AND model.model_type = '["image_edit","image_generate"]'::jsonb
    AND model.capabilities->'image_edit'->>'input_multiple_images' = 'true'
    AND model.capabilities->'image_edit'->>'input_max_images_count' = '10'
    AND model.capabilities->'image_edit'->>'input_max_file_size_bytes' = '31457280'
    AND model.capabilities->'image_edit'->'output_resolutions' = '["1K","2K"]'::jsonb
    AND model.capabilities->'image_edit'->'output_size_range' = '[921600,4624220]'::jsonb
    AND model.capabilities->'image_edit'->'aspect_ratio_range' = '[0.0625,16]'::jsonb
    AND model.capabilities->'image_edit'->>'output_multiple_images' = 'false'
    AND model.capabilities->'image_edit'->>'output_max_images_count' = '1'
    AND model.capabilities->'image_generate'->>'allow_custom_width_height_size' = 'true'
    AND model.capabilities->'image_generate'->>'output_multiple_images' = 'false'
    AND model.capabilities->>'stream' = 'false'
    AND model.capabilities->>'supportWebSearch' = 'false'
    AND model.default_rate_limit_policy->'rules'->0->>'limit' = '500'
    AND model.default_snapshot->>'modelAlias' = 'doubao-seedream-5-0-pro-260628'
    AND model.default_snapshot->>'displayName' = 'Seedream 5.0 Pro'
)`).Scan(&valid); err != nil {
		t.Fatalf("query Seedream 5.0 Pro catalog migration: %v", err)
	}
	if !valid {
		t.Fatal("Seedream 5.0 Pro catalog migration does not match the expected alias, upstream model, capabilities, or rate limit")
	}
}

func doJSON(t *testing.T, baseURL string, method string, path string, token string, payload any, expectedStatus int, out any) {
	t.Helper()
	doJSONWithHeaders(t, baseURL, method, path, token, payload, nil, expectedStatus, out)
}

func doJSONWithHeaders(t *testing.T, baseURL string, method string, path string, token string, payload any, headers map[string]string, expectedStatus int, out any) http.Header {
	t.Helper()
	var body io.Reader
	if payload != nil {
		raw, err := json.Marshal(payload)
		if err != nil {
			t.Fatalf("marshal payload: %v", err)
		}
		body = bytes.NewReader(raw)
	}
	req, err := http.NewRequest(method, baseURL+path, body)
	if err != nil {
		t.Fatalf("build request: %v", err)
	}
	if payload != nil {
		req.Header.Set("Content-Type", "application/json")
	}
	if token != "" {
		req.Header.Set("Authorization", "Bearer "+token)
	}
	for key, value := range headers {
		req.Header.Set(key, value)
	}
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatalf("%s %s: %v", method, path, err)
	}
	defer resp.Body.Close()
	raw, _ := io.ReadAll(resp.Body)
	if resp.StatusCode != expectedStatus {
		t.Fatalf("%s %s status=%d want=%d body=%s", method, path, resp.StatusCode, expectedStatus, string(raw))
	}
	if out != nil && len(raw) > 0 {
		if err := json.Unmarshal(raw, out); err != nil {
			t.Fatalf("decode %s %s response: %v body=%s", method, path, err, string(raw))
		}
	}
	return resp.Header.Clone()
}

func doAPIV1ChatCompletionAndLoadTask(t *testing.T, ctx context.Context, pool *pgxpool.Pool, baseURL string, token string, payload map[string]any, marker string, expectedStatus int, responseOut any, taskDetailOut any) string {
	t.Helper()
	if responseOut == nil {
		responseOut = &map[string]any{}
	}
	requestStartedAt := time.Now().UTC().Add(-time.Second)
	responseHeaders := doJSONWithHeaders(t, baseURL, http.MethodPost, "/api/v1/chat/completions", token, payload, nil, expectedStatus, responseOut)
	taskID := strings.TrimSpace(responseHeaders.Get("X-Gateway-Task-Id"))
	if taskID == "" {
		taskID = waitForLatestTaskIDSince(t, ctx, pool, "chat.completions", requestStartedAt, 2*time.Second)
	}
	if taskID == "" {
		t.Fatalf("load chat completion task %s failed: headers=%v body=%+v", marker, responseHeaders, responseOut)
	}
	if taskDetailOut != nil {
		doJSON(t, baseURL, http.MethodGet, "/api/v1/tasks/"+taskID, token, nil, http.StatusOK, taskDetailOut)
	}
	return taskID
}

func waitForLatestTaskIDSince(t *testing.T, ctx context.Context, pool *pgxpool.Pool, kind string, since time.Time, timeout time.Duration) string {
	t.Helper()
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		var taskID string
		err := pool.QueryRow(ctx, `
SELECT id::text
FROM gateway_tasks
WHERE kind = $1
  AND created_at >= $2
ORDER BY created_at DESC
LIMIT 1`, kind, since).Scan(&taskID)
		if err == nil && taskID != "" {
			return taskID
		}
		time.Sleep(10 * time.Millisecond)
	}
	t.Fatalf("latest %s task was not persisted within %s", kind, timeout)
	return ""
}

type taskWaitDetail struct {
	ID       string `json:"id"`
	Status   string `json:"status"`
	Attempts []struct {
		Status    string `json:"status"`
		ErrorCode string `json:"errorCode"`
	} `json:"attempts"`
}

type cacheAffinityPlatformFixture struct {
	ID          string `json:"id"`
	PlatformKey string `json:"platformKey"`
	Name        string `json:"name"`
}

type cacheAffinityTaskDetail struct {
	ID       string         `json:"id"`
	Status   string         `json:"status"`
	Usage    map[string]any `json:"usage"`
	Metrics  map[string]any `json:"metrics"`
	Attempts []struct {
		AttemptNo    int            `json:"attemptNo"`
		PlatformName string         `json:"platformName"`
		Status       string         `json:"status"`
		Metrics      map[string]any `json:"metrics"`
	} `json:"attempts"`
}

func createSimulationTextBaseModel(t *testing.T, baseURL string, adminToken string, model string) {
	t.Helper()
	createSimulationBaseModel(t, baseURL, adminToken, model, []string{"text_generate"})
}

func createSimulationBaseModel(t *testing.T, baseURL string, adminToken string, model string, modelTypes []string) {
	t.Helper()
	doJSON(t, baseURL, http.MethodPost, "/api/admin/catalog/base-models", adminToken, map[string]any{
		"providerKey":       "openai",
		"canonicalModelKey": "openai:" + model,
		"invocationName":    model,
		"providerModelName": model,
		"modelType":         modelTypes,
		"displayName":       model,
		"capabilities":      map[string]any{"originalTypes": modelTypes},
		"metadata":          map[string]any{"source": "cache-affinity-integration-test"},
	}, http.StatusCreated, nil)
}

func createSimulationTextPlatformModel(t *testing.T, baseURL string, adminToken string, platformKey string, platformName string, model string, priority int, rateLimitPolicy map[string]any) (cacheAffinityPlatformFixture, string) {
	t.Helper()
	var platform cacheAffinityPlatformFixture
	doJSON(t, baseURL, http.MethodPost, "/api/admin/platforms", adminToken, map[string]any{
		"provider":    "openai",
		"platformKey": platformKey,
		"name":        platformName,
		"baseUrl":     "https://api.openai.com/v1",
		"authType":    "bearer",
		"credentials": map[string]any{"mode": "simulation"},
		"priority":    priority,
	}, http.StatusCreated, &platform)
	if platform.ID == "" || platform.PlatformKey == "" {
		t.Fatalf("cache affinity platform was not created: %+v", platform)
	}
	payload := map[string]any{
		"canonicalModelKey": "openai:" + model,
		"modelName":         model,
		"providerModelName": model,
		"modelAlias":        model,
		"modelType":         []string{"text_generate"},
		"displayName":       platformName,
	}
	if len(rateLimitPolicy) > 0 {
		payload["rateLimitPolicy"] = rateLimitPolicy
	}
	var platformModel struct {
		ID string `json:"id"`
	}
	doJSON(t, baseURL, http.MethodPost, "/api/admin/platforms/"+platform.ID+"/models", adminToken, payload, http.StatusCreated, &platformModel)
	if platformModel.ID == "" {
		t.Fatalf("cache affinity platform model was not created: %+v", platformModel)
	}
	return platform, platformModel.ID
}

func runCacheAffinityChatTask(t *testing.T, ctx context.Context, pool *pgxpool.Pool, baseURL string, token string, model string, cacheAffinityKey string, marker string, simulationUsage map[string]any) cacheAffinityTaskDetail {
	t.Helper()
	return runCacheAffinityChatTaskWithBody(t, ctx, pool, baseURL, token, model, marker, map[string]any{
		"cacheAffinityKey": cacheAffinityKey,
		"simulationUsage":  simulationUsage,
		"messages":         []any{map[string]any{"role": "user", "content": "cache affinity route"}},
	})
}

func runCacheAffinityChatTaskWithBody(t *testing.T, ctx context.Context, pool *pgxpool.Pool, baseURL string, token string, model string, marker string, body map[string]any) cacheAffinityTaskDetail {
	t.Helper()
	var detail cacheAffinityTaskDetail
	requestBody := map[string]any{
		"model":                model,
		"runMode":              "simulation",
		"simulation":           true,
		"simulationDurationMs": 5,
	}
	for key, value := range body {
		requestBody[key] = value
	}
	doAPIV1ChatCompletionAndLoadTask(t, ctx, pool, baseURL, token, requestBody, marker, http.StatusOK, nil, &detail)
	return detail
}

type miniMaxModelCatalogItem struct {
	Alias     string                      `json:"alias"`
	ModelName string                      `json:"modelName"`
	ModelType []string                    `json:"modelType"`
	Sources   []miniMaxModelCatalogSource `json:"sources"`
}

type miniMaxModelCatalogSource struct {
	ModelAlias string   `json:"modelAlias"`
	ModelName  string   `json:"modelName"`
	ModelType  []string `json:"modelType"`
	Enabled    bool     `json:"enabled"`
}

func findModelCatalogItem(items []miniMaxModelCatalogItem, alias string) *miniMaxModelCatalogItem {
	for index := range items {
		if items[index].Alias == alias {
			return &items[index]
		}
	}
	return nil
}

func modelCatalogItemHasEnabledSource(sources []miniMaxModelCatalogSource, alias string, modelName string, modelType string) bool {
	for _, source := range sources {
		if source.Enabled && source.ModelAlias == alias && source.ModelName == modelName && stringSliceContains(source.ModelType, modelType) {
			return true
		}
	}
	return false
}

func waitForTaskStatus(t *testing.T, baseURL string, token string, taskID string, statuses []string, timeout time.Duration) taskWaitDetail {
	t.Helper()
	wanted := map[string]bool{}
	for _, status := range statuses {
		wanted[status] = true
	}
	var detail taskWaitDetail
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		doJSON(t, baseURL, http.MethodGet, "/api/v1/tasks/"+taskID, token, nil, http.StatusOK, &detail)
		if wanted[detail.Status] {
			return detail
		}
		time.Sleep(100 * time.Millisecond)
	}
	return detail
}

func waitForTaskIDByRequestField(t *testing.T, ctx context.Context, pool *pgxpool.Pool, key string, value string, timeout time.Duration) string {
	t.Helper()
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		var taskID string
		err := pool.QueryRow(ctx, `
SELECT id::text
FROM gateway_tasks
WHERE request->>$1 = $2
ORDER BY created_at DESC
LIMIT 1`, key, value).Scan(&taskID)
		if err == nil && taskID != "" {
			return taskID
		}
		time.Sleep(50 * time.Millisecond)
	}
	t.Fatalf("task with request %s=%s was not created within %s", key, value, timeout)
	return ""
}

func assertLoadAvoidanceSimulatedRetryChain(t *testing.T, ctx context.Context, testPool *pgxpool.Pool, baseURL string, adminToken string, runtimeToken string, suffixText string) {
	t.Helper()
	model := "load-avoidance-smoke-" + suffixText
	createSimulationTextBaseModel(t, baseURL, adminToken, model)
	type scenario struct {
		keySuffix string
		name      string
		failure   string
		priority  int
		full      bool
	}
	scenarios := []scenario{
		{keySuffix: "hard-stop", name: "Load Avoidance Hard Stop", failure: "fatal_failure", priority: 20},
		{keySuffix: "retryable", name: "Load Avoidance Retryable", failure: "retryable_failure", priority: 30},
		{keySuffix: "full-rate-limit", name: "Load Avoidance Full Rate Limit", failure: "rate_limit", priority: 1, full: true},
		{keySuffix: "full-overloaded", name: "Load Avoidance Full Overloaded", failure: "overloaded", priority: 2, full: true},
		{keySuffix: "full-fatal", name: "Load Avoidance Full Fatal", failure: "fatal_failure", priority: 3, full: true},
	}
	for _, item := range scenarios {
		var platform struct {
			ID          string `json:"id"`
			PlatformKey string `json:"platformKey"`
			Name        string `json:"name"`
		}
		credentials := map[string]any{"mode": "simulation"}
		if item.failure != "" {
			credentials["simulationFailure"] = item.failure
		}
		doJSON(t, baseURL, http.MethodPost, "/api/admin/platforms", adminToken, map[string]any{
			"provider":    "openai",
			"platformKey": "openai-load-" + item.keySuffix + "-" + suffixText,
			"name":        item.name,
			"baseUrl":     "https://api.openai.com/v1",
			"authType":    "bearer",
			"credentials": credentials,
			"priority":    item.priority,
		}, http.StatusCreated, &platform)
		if platform.ID == "" || platform.PlatformKey == "" {
			t.Fatalf("load avoidance platform was not created: %+v", platform)
		}

		var platformModel struct {
			ID string `json:"id"`
		}
		payload := map[string]any{
			"canonicalModelKey": "openai:" + model,
			"modelName":         model,
			"providerModelName": model,
			"modelAlias":        model,
			"modelType":         []string{"text_generate"},
			"displayName":       item.name,
			"retryPolicy":       map[string]any{"enabled": true, "maxAttempts": 1},
		}
		if item.full {
			payload["rateLimitPolicy"] = map[string]any{
				"rules": []map[string]any{
					{"metric": "rpm", "limit": 10, "windowSeconds": 60},
					{"metric": "tpm_total", "limit": 200, "windowSeconds": 60},
					{"metric": "concurrent", "limit": 1, "leaseTtlSeconds": 120},
				},
			}
		}
		doJSON(t, baseURL, http.MethodPost, "/api/admin/platforms/"+platform.ID+"/models", adminToken, payload, http.StatusCreated, &platformModel)
		if platformModel.ID == "" {
			t.Fatalf("load avoidance platform model was not created for %s: %+v", item.name, platformModel)
		}
		if item.full {
			seedQueuedConcurrencyLoad(t, ctx, testPool, platform.PlatformKey, model, platformModel.ID)
		}
	}

	var taskResponse struct {
		Task struct {
			ID        string `json:"id"`
			Status    string `json:"status"`
			ErrorCode string `json:"errorCode"`
		} `json:"task"`
	}
	doAPIV1ChatCompletionAndLoadTask(t, ctx, testPool, baseURL, runtimeToken, map[string]any{
		"model":                model,
		"runMode":              "simulation",
		"simulation":           true,
		"simulationDurationMs": 5,
		"messages":             []map[string]any{{"role": "user", "content": "load avoidance retry chain"}},
	}, "load-avoidance-"+suffixText, http.StatusBadRequest, nil, &taskResponse.Task)
	if taskResponse.Task.ID == "" || taskResponse.Task.Status != "failed" || taskResponse.Task.ErrorCode != "bad_request" {
		t.Fatalf("load avoidance task should only fail after avoided clients are retried, got %+v", taskResponse.Task)
	}

	var detail struct {
		ID        string `json:"id"`
		Status    string `json:"status"`
		ErrorCode string `json:"errorCode"`
		Attempts  []struct {
			AttemptNo    int            `json:"attemptNo"`
			PlatformName string         `json:"platformName"`
			Status       string         `json:"status"`
			Retryable    bool           `json:"retryable"`
			ErrorCode    string         `json:"errorCode"`
			Metrics      map[string]any `json:"metrics"`
		} `json:"attempts"`
		Metrics map[string]any `json:"metrics"`
	}
	doJSON(t, baseURL, http.MethodGet, "/api/v1/tasks/"+taskResponse.Task.ID, runtimeToken, nil, http.StatusOK, &detail)
	if detail.Status != "failed" || len(detail.Attempts) != 1 {
		t.Fatalf("non-retryable failure should stop before load-avoided fallbacks, got status=%s attempts=%+v", detail.Status, detail.Attempts)
	}
	attempt := detail.Attempts[0]
	if attempt.AttemptNo != 1 ||
		attempt.PlatformName != "Load Avoidance Hard Stop" ||
		attempt.Status != "failed" ||
		attempt.ErrorCode != "bad_request" ||
		attempt.Retryable {
		t.Fatalf("unexpected hard-stop attempt: %+v", attempt)
	}
	if boolFromTestMap(attempt.Metrics, "loadAvoided") {
		t.Fatalf("selected non-full candidate must not be marked load avoided: %+v", attempt.Metrics)
	}
	if ratio := floatFromTestAny(attempt.Metrics["loadRatio"]); ratio != 0 {
		t.Fatalf("selected non-full candidate should not carry load pressure, got ratio=%v metrics=%+v", ratio, attempt.Metrics)
	}
	if !attemptTraceHasReason(attempt.Metrics, "failover_action_rule") {
		t.Fatalf("hard-stop attempt should record the failure-policy decision, metrics=%+v", attempt.Metrics)
	}
	if _, ok := detail.Metrics["attempts"]; ok {
		t.Fatalf("task metrics should not duplicate the dedicated attempt chain: %+v", detail.Metrics)
	}
}

func seedQueuedConcurrencyLoad(t *testing.T, ctx context.Context, testPool *pgxpool.Pool, platformKey string, model string, platformModelID string) {
	t.Helper()
	queueKey := platformKey + ":text_generate:" + model
	if _, err := testPool.Exec(ctx, `
INSERT INTO gateway_tasks (
  kind, run_mode, user_id, model, requested_model, model_type,
  request, normalized_request, status, queue_key, priority, async_mode,
  next_run_at, result, billings
)
VALUES (
  'chat.completions', 'simulation', $1, $2, $2, 'text_generate',
  '{}'::jsonb, '{}'::jsonb, 'queued', $3, 999, true,
  now() + interval '1 hour', '{}'::jsonb, '[]'::jsonb
)`, "load-avoidance-seed-"+platformModelID, model, queueKey); err != nil {
		t.Fatalf("seed queued load for %s: %v", queueKey, err)
	}
}

func boolFromTestMap(values map[string]any, key string) bool {
	value, _ := values[key].(bool)
	return value
}

func intFromTestAny(value any) int {
	switch typed := value.(type) {
	case int:
		return typed
	case int64:
		return int(typed)
	case float64:
		return int(typed)
	case json.Number:
		out, _ := typed.Int64()
		return int(out)
	default:
		return 0
	}
}

func floatFromTestAny(value any) float64 {
	switch typed := value.(type) {
	case float64:
		return typed
	case float32:
		return float64(typed)
	case int:
		return float64(typed)
	case int64:
		return float64(typed)
	case json.Number:
		out, _ := typed.Float64()
		return out
	default:
		return 0
	}
}

func attemptTraceHasReason(metrics map[string]any, reason string) bool {
	trace, _ := metrics["trace"].([]any)
	for _, raw := range trace {
		item, _ := raw.(map[string]any)
		if item["reason"] == reason {
			return true
		}
	}
	return false
}

func waitForRateLimitWindowHead(t *testing.T, windowSeconds int) {
	t.Helper()
	if windowSeconds <= 0 {
		return
	}
	window := time.Duration(windowSeconds) * time.Second
	deadline := time.Now().Add(window + time.Second)
	for time.Now().Before(deadline) {
		elapsed := time.Duration(time.Now().UnixNano() % int64(window))
		if elapsed < 300*time.Millisecond {
			return
		}
		time.Sleep(20 * time.Millisecond)
	}
	t.Fatalf("timed out waiting for %ds rate limit window head", windowSeconds)
}

func stringSliceContains(values []string, target string) bool {
	for _, value := range values {
		if value == target {
			return true
		}
	}
	return false
}

func modelListContains(values []struct {
	ID        string `json:"id"`
	ModelName string `json:"modelName"`
}, target string) bool {
	for _, value := range values {
		if value.ID == target {
			return true
		}
	}
	return false
}

func taskListContains(values []struct {
	ID           string  `json:"id"`
	Status       string  `json:"status"`
	APIKeyName   string  `json:"apiKeyName"`
	ModelType    string  `json:"modelType"`
	FinalCharge  float64 `json:"finalChargeAmount"`
	ErrorCode    string  `json:"errorCode"`
	ErrorMessage string  `json:"errorMessage"`
}, target string) bool {
	for _, value := range values {
		if value.ID == target {
			return true
		}
	}
	return false
}

func walletTransactionListContains(values []struct {
	TransactionType string  `json:"transactionType"`
	Direction       string  `json:"direction"`
	ReferenceID     string  `json:"referenceId"`
	Currency        string  `json:"currency"`
	Amount          float64 `json:"amount"`
}, target string) bool {
	for _, value := range values {
		if value.ReferenceID == target && value.TransactionType == "task_billing" && value.Direction == "debit" && value.Currency == "resource" {
			return true
		}
	}
	return false
}

func walletTransactionListContainsWithMetadata(values []struct {
	TransactionType string         `json:"transactionType"`
	Direction       string         `json:"direction"`
	ReferenceID     string         `json:"referenceId"`
	Currency        string         `json:"currency"`
	Amount          float64        `json:"amount"`
	Metadata        map[string]any `json:"metadata"`
}, target string, model string, platformName string, apiKeyName string, finalChargeAmount float64) bool {
	for _, value := range values {
		usage := objectValue(value.Metadata["usage"])
		billingSummary := objectValue(value.Metadata["billingSummary"])
		finalCharge, hasFinalCharge := numberValue(value.Metadata["finalChargeAmount"])
		billingTotal, hasBillingTotal := numberValue(billingSummary["totalAmount"])
		if value.ReferenceID == target &&
			value.Metadata["model"] == model &&
			value.Metadata["modelType"] == "text_generate" &&
			value.Metadata["platformName"] == platformName &&
			value.Metadata["apiKeyName"] == apiKeyName &&
			usage["totalTokens"] != nil &&
			hasFinalCharge &&
			hasBillingTotal &&
			floatNear(finalCharge, finalChargeAmount) &&
			floatNear(billingTotal, finalChargeAmount) {
			return true
		}
	}
	return false
}

func floatNear(value float64, expected float64) bool {
	return math.Abs(value-expected) < 0.000001
}
