package runner

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"os"
	"regexp"
	"strconv"
	"strings"
	"time"

	"github.com/easyai/easyai-ai-gateway/apps/api/internal/auth"
	"github.com/easyai/easyai-ai-gateway/apps/api/internal/clients"
	"github.com/easyai/easyai-ai-gateway/apps/api/internal/config"
	scriptengine "github.com/easyai/easyai-ai-gateway/apps/api/internal/script"
	"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/riverqueue/river"
)

type Service struct {
	cfg            config.Config
	store          *store.Store
	logger         *slog.Logger
	clients        map[string]clients.Client
	scriptExecutor *scriptengine.Executor
	httpClients    *httpClientCache
	riverClient    *river.Client[pgx.Tx]
	billingMetrics billingMetricsObserver
}

type billingMetricsObserver interface {
	ObserveBillingEvent(string)
}

type Result struct {
	Task   store.GatewayTask
	Output map[string]any
	Wire   *clients.WireResponse
}

var ErrTaskQueued = errors.New("task queued")

type TaskQueuedError struct {
	Delay time.Duration
}

type upstreamSubmissionUnknownError struct {
	AttemptID string
	Cause     error
}

func (e *upstreamSubmissionUnknownError) Error() string {
	return "upstream submission result is unknown"
}

func (e *upstreamSubmissionUnknownError) ErrorCode() string {
	return "upstream_submission_unknown"
}

func (e *upstreamSubmissionUnknownError) Unwrap() error {
	return e.Cause
}

func (e *TaskQueuedError) Error() string {
	return ErrTaskQueued.Error()
}

func (e *TaskQueuedError) Is(target error) bool {
	return target == ErrTaskQueued
}

func New(cfg config.Config, db *store.Store, logger *slog.Logger, observers ...billingMetricsObserver) *Service {
	httpClients := newHTTPClientCache()
	scriptExecutor := &scriptengine.Executor{Logger: logger}
	service := &Service{
		cfg:            cfg,
		store:          db,
		logger:         logger,
		scriptExecutor: scriptExecutor,
		clients: map[string]clients.Client{
			"openai":                clients.OpenAIClient{HTTPClient: httpClients.none},
			"aliyun-bailian":        clients.AliyunBailianClient{HTTPClient: httpClients.none},
			"blackforest":           clients.BlackforestClient{HTTPClient: httpClients.none},
			"gemini":                clients.GeminiClient{HTTPClient: httpClients.none},
			"jimeng":                clients.JimengClient{HTTPClient: httpClients.none},
			"midjourney":            clients.MidjourneyClient{HTTPClient: httpClients.none},
			"minimax":               clients.MinimaxClient{HTTPClient: httpClients.none},
			"newapi":                clients.NewAPIClient{HTTPClient: httpClients.none},
			"suno":                  clients.SunoClient{HTTPClient: httpClients.none},
			"tencent-hunyuan-image": clients.HunyuanImageClient{HTTPClient: httpClients.none},
			"tencent-hunyuan-video": clients.HunyuanVideoClient{HTTPClient: httpClients.none},
			"vidu":                  clients.ViduClient{HTTPClient: httpClients.none},
			"volces":                clients.VolcesClient{HTTPClient: httpClients.none},
			"keling":                clients.KelingClient{HTTPClient: httpClients.none},
			"kling":                 clients.KelingClient{HTTPClient: httpClients.none},
			"vectorizer":            clients.VectorizerClient{HTTPClient: httpClients.none},
			"topaz":                 clients.TopazClient{HTTPClient: httpClients.none},
			"universal":             clients.UniversalClient{HTTPClient: httpClients.none, ScriptExecutor: scriptExecutor},
			"simulation":            clients.SimulationClient{},
		},
		httpClients: httpClients,
	}
	if len(observers) > 0 {
		service.billingMetrics = observers[0]
	}
	return service
}

func (s *Service) observeBillingEvent(event string) {
	if s.billingMetrics != nil {
		s.billingMetrics.ObserveBillingEvent(event)
	}
}

func (s *Service) Execute(ctx context.Context, task store.GatewayTask, user *auth.User) (Result, error) {
	return s.execute(ctx, task, user, nil)
}

func (s *Service) ExecuteStream(ctx context.Context, task store.GatewayTask, user *auth.User, onDelta clients.StreamDelta) (Result, error) {
	return s.execute(ctx, task, user, onDelta)
}

func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *auth.User, onDelta clients.StreamDelta) (Result, error) {
	return s.executeWithToken(ctx, task, user, onDelta, uuid.NewString())
}

func (s *Service) executeWithToken(ctx context.Context, task store.GatewayTask, user *auth.User, onDelta clients.StreamDelta, executionToken string) (Result, error) {
	wasRunning := task.Status == "running"
	claimed, err := s.store.ClaimTaskExecution(ctx, task.ID, executionToken, taskExecutionLeaseTTL)
	if err != nil {
		return Result{}, err
	}
	task = claimed
	executionCtx, stopExecution := context.WithCancel(ctx)
	defer stopExecution()
	go s.renewTaskExecutionLease(executionCtx, stopExecution, task.ID, task.ExecutionToken)
	ctx = executionCtx
	executeStartedAt := time.Now()
	restoredRequest, err := s.restoreTaskRequestReferences(ctx, task)
	if err != nil {
		return Result{}, err
	}
	body := normalizeRequest(task.Kind, restoredRequest)
	responseExecution := responseExecutionContext{}
	modelType := modelTypeFromKind(task.Kind, body)
	if err := s.store.MarkTaskRunning(ctx, task.ID, task.ExecutionToken, modelType, s.slimTaskRequestSnapshot(task, body)); err != nil {
		return Result{}, err
	}
	if !wasRunning {
		if err := s.emit(ctx, task.ID, "task.running", "running", "starting", 0.12, "task pulled from queue and started", map[string]any{"modelType": modelType}, task.RunMode == "simulation"); err != nil {
			return Result{}, err
		}
	}
	if err := validateRequest(task.Kind, body); err != nil {
		validationErr := parameterPreprocessClientError(err)
		s.recordFailedAttempt(ctx, failedAttemptRecord{
			Task:      task,
			Body:      body,
			AttemptNo: task.AttemptCount + 1,
			Code:      "bad_request",
			Cause:     validationErr,
			Simulated: task.RunMode == "simulation",
			Scope:     "request_validation",
			Reason:    "request_validation_failed",
			ModelType: modelType,
		})
		failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, "bad_request", validationErr.Error(), task.RunMode == "simulation", validationErr)
		if finishErr != nil {
			return Result{}, finishErr
		}
		return Result{Task: failed, Output: failed.Result}, validationErr
	}
	var clonedVoice clonedVoiceBinding
	body, clonedVoice, err = s.resolveClonedVoiceBinding(ctx, user, task.Kind, body)
	if err != nil {
		s.recordFailedAttempt(ctx, failedAttemptRecord{
			Task:      task,
			Body:      body,
			AttemptNo: task.AttemptCount + 1,
			Code:      clients.ErrorCode(err),
			Cause:     err,
			Simulated: task.RunMode == "simulation",
			Scope:     "cloned_voice_binding",
			Reason:    "cloned_voice_binding_failed",
			ModelType: modelType,
		})
		failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, clients.ErrorCode(err), err.Error(), task.RunMode == "simulation", err)
		if finishErr != nil {
			return Result{}, finishErr
		}
		return Result{Task: failed, Output: failed.Result}, err
	}
	if clonedVoice.Found {
		if err := s.store.MarkTaskRunning(ctx, task.ID, task.ExecutionToken, modelType, s.slimTaskRequestSnapshot(task, body)); err != nil {
			return Result{}, err
		}
	}
	if task.Kind == "responses" {
		responseExecution, err = s.prepareResponseExecution(ctx, task, body)
		if err != nil {
			code, message := responseExecutionFailure(err)
			s.recordFailedAttempt(ctx, failedAttemptRecord{Task: task, Body: body, AttemptNo: task.AttemptCount + 1, Code: code, Cause: err, Simulated: task.RunMode == "simulation", Scope: "response_chain", Reason: code, ModelType: modelType})
			failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, code, message, task.RunMode == "simulation", err)
			if finishErr != nil {
				return Result{}, finishErr
			}
			return Result{Task: failed, Output: failed.Result}, err
		}
	}
	runnerPolicy, err := s.store.GetActiveRunnerPolicy(ctx)
	if err != nil {
		return Result{}, err
	}
	cacheAffinityKeys := buildCacheAffinityKeys(task.Kind, modelType, body)
	candidates, err := s.store.ListModelCandidates(ctx, task.Model, modelType, user, store.ListModelCandidatesOptions{
		CacheAffinityKey:    cacheAffinityKeys.Primary,
		CacheAffinityKeys:   cacheAffinityKeys.Lookup,
		CacheAffinityPolicy: runnerPolicy.CacheAffinityPolicy,
	})
	if err != nil {
		if task.Kind == "responses" && responseExecution.PreviousChain != nil {
			err = responseChainUnavailableError()
		}
		s.recordFailedAttempt(ctx, failedAttemptRecord{
			Task:      task,
			Body:      body,
			AttemptNo: task.AttemptCount + 1,
			Code:      store.ModelCandidateErrorCode(err),
			Cause:     err,
			Simulated: task.RunMode == "simulation",
			Scope:     "candidate_selection",
			Reason:    "candidate_selection_failed",
			ModelType: modelType,
		})
		failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, store.ModelCandidateErrorCode(err), err.Error(), task.RunMode == "simulation", err)
		if finishErr != nil {
			return Result{}, finishErr
		}
		return Result{Task: failed, Output: failed.Result}, err
	}
	candidates, err = filterCandidatesByClonedVoiceBinding(candidates, clonedVoice)
	if err != nil {
		s.recordFailedAttempt(ctx, failedAttemptRecord{
			Task:      task,
			Body:      body,
			AttemptNo: task.AttemptCount + 1,
			Code:      store.ModelCandidateErrorCode(err),
			Cause:     err,
			Simulated: task.RunMode == "simulation",
			Scope:     "cloned_voice_binding",
			Reason:    store.ModelCandidateErrorCode(err),
			ModelType: modelType,
		})
		failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, store.ModelCandidateErrorCode(err), err.Error(), task.RunMode == "simulation", err)
		if finishErr != nil {
			return Result{}, finishErr
		}
		return Result{Task: failed, Output: failed.Result}, err
	}
	var candidateFilterSummary map[string]any
	candidates, candidateFilterSummary, err = filterRuntimeCandidatesByRequest(task.Kind, task.Model, modelType, body, candidates)
	if err != nil {
		candidateFilterMetrics := candidateCapabilityFilterMetrics(candidateFilterSummary)
		s.recordFailedAttempt(ctx, failedAttemptRecord{
			Task:         task,
			Body:         body,
			AttemptNo:    task.AttemptCount + 1,
			Code:         store.ModelCandidateErrorCode(err),
			Cause:        err,
			Simulated:    task.RunMode == "simulation",
			Scope:        "candidate_request_filter",
			Reason:       store.ModelCandidateErrorCode(err),
			ExtraMetrics: []map[string]any{candidateFilterMetrics},
			ModelType:    modelType,
		})
		failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, store.ModelCandidateErrorCode(err), err.Error(), task.RunMode == "simulation", err, candidateFilterMetrics)
		if finishErr != nil {
			return Result{}, finishErr
		}
		return Result{Task: failed, Output: failed.Result}, err
	}
	var outputTokenFilterSummary map[string]any
	candidates, outputTokenFilterSummary, err = filterRuntimeCandidatesByOutputTokens(task.Kind, task.Model, modelType, body, candidates)
	candidateFilterSummary = mergeCandidateFilterSummaries(candidateFilterSummary, outputTokenFilterSummary)
	if err != nil {
		candidateFilterMetrics := candidateCapabilityFilterMetrics(candidateFilterSummary)
		s.recordFailedAttempt(ctx, failedAttemptRecord{
			Task: task, Body: body, AttemptNo: task.AttemptCount + 1, Code: store.ModelCandidateErrorCode(err), Cause: err,
			Simulated: task.RunMode == "simulation", Scope: "candidate_output_token_filter", Reason: store.ModelCandidateErrorCode(err),
			ExtraMetrics: []map[string]any{candidateFilterMetrics}, ModelType: modelType,
		})
		failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, store.ModelCandidateErrorCode(err), err.Error(), task.RunMode == "simulation", err, candidateFilterMetrics)
		if finishErr != nil {
			return Result{}, finishErr
		}
		return Result{Task: failed, Output: failed.Result}, err
	}
	if task.Kind == "responses" {
		candidates, err = prepareResponseCandidates(candidates, responseExecution)
		if err != nil {
			code, message := responseExecutionFailure(err)
			s.recordFailedAttempt(ctx, failedAttemptRecord{Task: task, Body: body, AttemptNo: task.AttemptCount + 1, Code: code, Cause: err, Simulated: task.RunMode == "simulation", Scope: "response_protocol", Reason: code, ModelType: modelType})
			failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, code, message, task.RunMode == "simulation", err)
			if finishErr != nil {
				return Result{}, finishErr
			}
			return Result{Task: failed, Output: failed.Result}, err
		}
	}
	pricingByCandidate := map[string]resolvedPricing{}
	preprocessingByCandidate := map[string]parameterPreprocessResult{}
	reservationBillings := []any(nil)
	reservationPricingSnapshot := map[string]any(nil)
	if task.RunMode == "production" {
		billingMode := normalizedBillingEngineMode(s.cfg.BillingEngineMode)
		if billingMode == "hold" {
			holdErr := &clients.ClientError{Code: "billing_hold", Message: "production billing is temporarily on hold", Retryable: true}
			failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, "billing_hold", holdErr.Error(), false, holdErr)
			if finishErr != nil {
				return Result{}, finishErr
			}
			return Result{Task: failed, Output: failed.Result}, holdErr
		}
		estimates := make([]candidateEstimate, 0, len(candidates))
		var pricingErr error
		var preprocessingErr error
		var preprocessingCandidate store.RuntimeModelCandidate
		for _, candidate := range candidates {
			preprocessing := s.preprocessRequestWithScripts(ctx, task.Kind, body, candidate)
			preprocessingByCandidate[pricingCandidateKey(candidate)] = preprocessing
			if preprocessing.Err != nil {
				preprocessingErr = parameterPreprocessClientError(preprocessing.Err)
				preprocessingCandidate = candidate
				break
			}
			estimate, estimateErr := s.estimateCandidateV2(ctx, user, task.Kind, preprocessing.Body, candidate)
			if estimateErr != nil {
				pricingErr = estimateErr
				if billingMode == "enforce" {
					break
				}
				continue
			}
			estimates = append(estimates, estimate)
			pricingByCandidate[pricingCandidateKey(candidate)] = estimate.Pricing
		}
		if preprocessingErr != nil {
			preprocessing := preprocessingByCandidate[pricingCandidateKey(preprocessingCandidate)]
			s.recordFailedAttempt(ctx, failedAttemptRecord{
				Task: task, Body: preprocessing.Body, Candidate: &preprocessingCandidate,
				AttemptNo: task.AttemptCount + 1, Code: clients.ErrorCode(preprocessingErr), Cause: preprocessingErr,
				Simulated: false, Scope: "parameter_preprocessing", Reason: "parameter_preprocessing_failed",
				ExtraMetrics: []map[string]any{parameterPreprocessingMetrics(preprocessing.Log)}, Preprocessing: &preprocessing.Log,
				ModelType: preprocessingCandidate.ModelType,
			})
			failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, clients.ErrorCode(preprocessingErr), preprocessingErr.Error(), false, preprocessingErr, parameterPreprocessingMetrics(preprocessing.Log))
			if finishErr != nil {
				return Result{}, finishErr
			}
			return Result{Task: failed, Output: failed.Result}, preprocessingErr
		}
		if billingMode == "observe" {
			legacyItems, legacyAmount := s.maximumLegacyCandidateEstimate(ctx, user, task.Kind, body, candidates, preprocessingByCandidate)
			reservationBillings = legacyItems
			candidateSnapshots := make([]any, 0, len(estimates))
			for _, estimate := range estimates {
				candidateSnapshots = append(candidateSnapshots, estimate.Snapshot)
			}
			observedAmount := ""
			if maximumEstimate, estimateErr := maximumCandidateEstimate(estimates); estimateErr == nil {
				observedAmount = maximumEstimate.Amount.String()
			}
			reservationPricingSnapshot = map[string]any{
				"pricingVersion": "legacy-observe", "observedPricingVersion": pricingVersionV2,
				"requestFingerprint": pricingRequestFingerprint(task.Kind, task.Model, body),
				"reservationAmount":  legacyAmount.String(), "observedReservationAmount": observedAmount,
				"candidateCount": len(candidates), "observedCandidates": candidateSnapshots,
			}
			s.logger.Info("billing observe comparison", "taskID", task.ID, "legacyAmount", legacyAmount.String(), "v2Amount", observedAmount, "pricedCandidates", len(estimates), "candidateCount", len(candidates))
		} else if pricingErr != nil || len(estimates) != len(candidates) {
			if pricingErr == nil {
				pricingErr = &PricingUnavailableError{Reason: "not every candidate has effective pricing"}
			}
			s.observeBillingEvent("pricing_unavailable")
			failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, "pricing_unavailable", pricingErr.Error(), false, pricingErr)
			if finishErr != nil {
				return Result{}, finishErr
			}
			return Result{Task: failed, Output: failed.Result}, pricingErr
		} else {
			maximumEstimate, estimateErr := maximumCandidateEstimate(estimates)
			if estimateErr != nil {
				failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, "pricing_unavailable", estimateErr.Error(), false, estimateErr)
				if finishErr != nil {
					return Result{}, finishErr
				}
				return Result{Task: failed, Output: failed.Result}, estimateErr
			}
			pricedCandidates := make([]store.RuntimeModelCandidate, 0, len(estimates))
			candidateSnapshots := make([]any, 0, len(estimates))
			for _, candidate := range candidates {
				if _, ok := pricingByCandidate[pricingCandidateKey(candidate)]; ok {
					pricedCandidates = append(pricedCandidates, candidate)
				}
			}
			for _, estimate := range estimates {
				candidateSnapshots = append(candidateSnapshots, estimate.Snapshot)
			}
			candidates = pricedCandidates
			reservationBillings = maximumEstimate.Items
			reservationPricingSnapshot = map[string]any{
				"pricingVersion": pricingVersionV2, "requestFingerprint": pricingRequestFingerprint(task.Kind, task.Model, body),
				"reservationAmount": maximumEstimate.Amount.String(), "candidateCount": len(estimates), "candidates": candidateSnapshots,
			}
		}
	}
	firstCandidateBody := body
	normalizedModelType := modelType
	attemptNo := task.AttemptCount
	var firstPreprocessing parameterPreprocessingLog
	var walletReservations []store.WalletBillingReservation
	walletReservationFinalized := false
	defer func() {
		if !walletReservationFinalized && len(walletReservations) > 0 {
			_ = s.store.ReleaseTaskBillingReservations(context.WithoutCancel(ctx), walletReservations, "task_not_settled")
		}
	}()
	if len(candidates) > 0 {
		preprocessing, ok := preprocessingByCandidate[pricingCandidateKey(candidates[0])]
		if !ok {
			preprocessing = s.preprocessRequestWithScripts(ctx, task.Kind, body, candidates[0])
		}
		firstCandidateBody = preprocessing.Body
		firstPreprocessing = preprocessing.Log
		normalizedModelType = candidates[0].ModelType
		if preprocessing.Err != nil {
			clientErr := parameterPreprocessClientError(preprocessing.Err)
			attemptNo = s.recordFailedAttempt(ctx, failedAttemptRecord{
				Task:          task,
				Body:          firstCandidateBody,
				Candidate:     &candidates[0],
				AttemptNo:     attemptNo + 1,
				Code:          clients.ErrorCode(clientErr),
				Cause:         clientErr,
				Simulated:     task.RunMode == "simulation",
				Scope:         "parameter_preprocessing",
				Reason:        "parameter_preprocessing_failed",
				ExtraMetrics:  []map[string]any{parameterPreprocessingMetrics(firstPreprocessing)},
				Preprocessing: &firstPreprocessing,
				ModelType:     normalizedModelType,
			})
			failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, clients.ErrorCode(clientErr), clientErr.Error(), task.RunMode == "simulation", clientErr, parameterPreprocessingMetrics(firstPreprocessing))
			if finishErr != nil {
				return Result{}, finishErr
			}
			return Result{Task: failed, Output: failed.Result}, clientErr
		}
		if err := s.store.MarkTaskRunning(ctx, task.ID, task.ExecutionToken, candidates[0].ModelType, s.slimTaskRequestSnapshot(task, firstCandidateBody)); err != nil {
			return Result{}, err
		}
		var reserveErr error
		walletReservations, reserveErr = s.store.ReserveTaskBilling(ctx, task, user, reservationBillings, reservationPricingSnapshot)
		if reserveErr != nil {
			if errors.Is(reserveErr, store.ErrInsufficientWalletBalance) {
				attemptNo = s.recordFailedAttempt(ctx, failedAttemptRecord{
					Task:          task,
					Body:          firstCandidateBody,
					Candidate:     &candidates[0],
					AttemptNo:     attemptNo + 1,
					Code:          "insufficient_balance",
					Cause:         reserveErr,
					Simulated:     task.RunMode == "simulation",
					Scope:         "wallet_balance",
					Reason:        "wallet_balance_check_failed",
					ExtraMetrics:  []map[string]any{parameterPreprocessingMetrics(firstPreprocessing)},
					Preprocessing: &firstPreprocessing,
					ModelType:     normalizedModelType,
				})
				failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, "insufficient_balance", reserveErr.Error(), task.RunMode == "simulation", reserveErr, parameterPreprocessingMetrics(firstPreprocessing))
				if finishErr != nil {
					return Result{}, finishErr
				}
				return Result{Task: failed, Output: failed.Result}, reserveErr
			}
			return Result{}, reserveErr
		}
	}
	if err := s.emit(ctx, task.ID, "task.progress", "running", "normalizing", 0.15, "request normalized", map[string]any{"modelType": normalizedModelType}, task.RunMode == "simulation"); err != nil {
		return Result{}, err
	}

	maxPlatforms := maxPlatformsForCandidates(candidates, runnerPolicy)
	maxFailoverDuration := maxFailoverDurationForCandidates(candidates, runnerPolicy)
	singleSourceProtected := singleSourceProtectionActive(candidates, maxPlatforms, runnerPolicy)
	var lastErr error
	var lastCandidate store.RuntimeModelCandidate
	var lastPreprocessing *parameterPreprocessingLog
candidatesLoop:
	for index, candidate := range candidates {
		if index >= maxPlatforms {
			break
		}
		lastCandidate = candidate
		clientAttempts := clientAttemptsForCandidate(candidate)
		var candidateErr error
		for clientAttempt := 1; clientAttempt <= clientAttempts; clientAttempt++ {
			nextAttemptNo := attemptNo + 1
			preprocessing, ok := preprocessingByCandidate[pricingCandidateKey(candidate)]
			if !ok {
				preprocessing = s.preprocessRequestWithScripts(ctx, task.Kind, body, candidate)
			}
			preprocessingLog := preprocessing.Log
			lastPreprocessing = &preprocessingLog
			if preprocessing.Err != nil {
				lastErr = parameterPreprocessClientError(preprocessing.Err)
				attemptNo = s.recordFailedAttempt(ctx, failedAttemptRecord{
					Task:          task,
					Body:          preprocessing.Body,
					Candidate:     &candidate,
					AttemptNo:     nextAttemptNo,
					Code:          clients.ErrorCode(lastErr),
					Cause:         lastErr,
					Simulated:     isSimulation(task, candidate),
					Scope:         "parameter_preprocessing",
					Reason:        "parameter_preprocessing_failed",
					ExtraMetrics:  []map[string]any{parameterPreprocessingMetrics(preprocessingLog)},
					Preprocessing: &preprocessingLog,
					ModelType:     candidate.ModelType,
				})
				break candidatesLoop
			}
			candidateBody := preprocessing.Body
			candidatePricing := pricingByCandidate[pricingCandidateKey(candidate)]
			response, err := s.runCandidate(ctx, task, user, candidateBody, preprocessing.Log, candidate, candidatePricing, nextAttemptNo, onDelta, responseExecution, singleSourceProtected, runnerPolicy.CacheAffinityPolicy, cacheAffinityKeys.Record)
			if err != nil && isVolcesRemoteTaskCancellation(candidate, err) {
				cancelled, changed, cancelErr := s.store.CancelSubmittedTask(ctx, task.ID, task.ExecutionToken, "任务已由火山引擎取消")
				if cancelErr != nil {
					return Result{}, cancelErr
				}
				if changed {
					// CancelSubmittedTask atomically transfers any reservation to the release Outbox.
					walletReservationFinalized = true
					if emitErr := s.emit(ctx, task.ID, "task.cancelled", "cancelled", "cancelled", 1, "任务已由火山引擎取消", map[string]any{"taskId": task.ID, "reason": "upstream_cancelled"}, isSimulation(task, candidate)); emitErr != nil {
						return Result{}, emitErr
					}
					return Result{Task: cancelled, Output: cancelled.Result}, nil
				}
			}
			if err == nil {
				attemptNo = nextAttemptNo
				var billings []any
				finalAmount := fixedAmount(0)
				finalAmountText := ""
				pricingSnapshot := map[string]any{"pricingVersion": pricingVersionV2, "source": "simulation"}
				if task.RunMode == "production" {
					billingMode := normalizedBillingEngineMode(s.cfg.BillingEngineMode)
					var billingErr error
					if billingMode == "observe" {
						billings = s.billings(ctx, user, task.Kind, candidateBody, candidate, response, false)
						finalAmount = billingItemsFixedTotal(billings)
						pricingSnapshot = map[string]any{
							"pricingVersion": "legacy-observe", "observedPricingVersion": pricingVersionV2,
							"observedPricing": candidatePricing.Snapshot,
						}
					} else {
						billings, finalAmount, _, billingErr = s.billingsWithResolvedPricingV2(ctx, user, task.Kind, candidateBody, candidate, response, false, candidatePricing)
						pricingSnapshot = candidatePricing.Snapshot
					}
					if billingErr != nil {
						review, reviewErr := s.store.FinishTaskManualReview(context.WithoutCancel(ctx), store.FinishTaskManualReviewInput{
							TaskID: task.ID, ExecutionToken: task.ExecutionToken, AttemptID: response.AttemptID, TaskStatus: "succeeded",
							Code: "billing_calculation_failed", Message: "billing calculation requires manual review",
							Result: response.Result, RequestID: response.RequestID,
							PricingSnapshot:    candidatePricing.Snapshot,
							RequestFingerprint: pricingRequestFingerprint(task.Kind, task.Model, candidateBody),
							ResponseStartedAt:  response.ResponseStartedAt, ResponseFinishedAt: response.ResponseFinishedAt,
							ResponseDurationMS: response.ResponseDurationMS,
						})
						if reviewErr != nil {
							return Result{}, reviewErr
						}
						walletReservationFinalized = true
						s.logger.Warn("task succeeded but billing requires manual review", "taskID", task.ID, "error_category", "billing_calculation_failed")
						return Result{Task: review, Output: response.Result, Wire: response.Wire}, nil
					}
					finalAmountText = finalAmount.String()
				} else {
					billings = s.billings(ctx, user, task.Kind, candidateBody, candidate, response, true)
				}
				record := buildSuccessRecord(task, user, candidateBody, candidate, response, billings, isSimulation(task, candidate))
				record.Metrics = mergeMetrics(record.Metrics, candidateCapabilityFilterMetrics(candidateFilterSummary))
				record.Metrics = mergeMetrics(record.Metrics, parameterPreprocessingMetrics(preprocessing.Log))
				record.Metrics = s.withAttemptHistory(ctx, task.ID, record.Metrics)
				finished, finishErr := s.store.FinishTaskSuccess(ctx, store.FinishTaskSuccessInput{
					TaskID:                task.ID,
					ExecutionToken:        task.ExecutionToken,
					AttemptID:             response.AttemptID,
					Result:                response.Result,
					Billings:              billings,
					RequestID:             record.RequestID,
					ResolvedModel:         record.ResolvedModel,
					Usage:                 record.Usage,
					Metrics:               record.Metrics,
					BillingSummary:        record.BillingSummary,
					FinalChargeAmount:     record.FinalChargeAmount,
					FinalChargeAmountText: finalAmountText,
					BillingCurrency:       stringFromAny(record.BillingSummary["currency"]),
					PricingSnapshot:       pricingSnapshot,
					RequestFingerprint:    pricingRequestFingerprint(task.Kind, task.Model, candidateBody),
					ResponseStartedAt:     record.ResponseStartedAt,
					ResponseFinishedAt:    record.ResponseFinishedAt,
					ResponseDurationMS:    record.ResponseDurationMS,
				})
				if finishErr != nil {
					if errors.Is(finishErr, store.ErrTaskExecutionLeaseLost) {
						latest, latestErr := s.store.GetTask(ctx, task.ID)
						if latestErr == nil && latest.Status == "cancelled" {
							walletReservationFinalized = true
							return Result{Task: latest, Output: latest.Result}, nil
						}
					}
					return Result{}, finishErr
				}
				walletReservationFinalized = true
				if finished.BillingStatus == "pending" {
					if err := s.emit(ctx, task.ID, "task.billing.pending", "succeeded", "billing", 0.98, "task billing queued", map[string]any{
						"amount": finished.FinalChargeAmount, "currency": finished.BillingCurrency,
					}, isSimulation(task, candidate)); err != nil {
						return Result{}, err
					}
				}
				if err := s.emit(ctx, task.ID, "task.completed", "succeeded", "completed", 1, "task completed", map[string]any{
					"result":         response.Result,
					"billings":       billings,
					"usage":          record.Usage,
					"metrics":        record.Metrics,
					"billingSummary": record.BillingSummary,
					"requestId":      record.RequestID,
				}, isSimulation(task, candidate)); err != nil {
					return Result{}, err
				}
				return Result{Task: finished, Output: response.Result, Wire: response.Wire}, nil
			}
			var submissionUnknown *upstreamSubmissionUnknownError
			if errors.As(err, &submissionUnknown) {
				review, reviewErr := s.store.FinishTaskManualReview(context.WithoutCancel(ctx), store.FinishTaskManualReviewInput{
					TaskID: task.ID, ExecutionToken: task.ExecutionToken, AttemptID: submissionUnknown.AttemptID, TaskStatus: "failed",
					Code: "upstream_submission_unknown", Message: submissionUnknown.Error(),
					PricingSnapshot:    candidatePricing.Snapshot,
					RequestFingerprint: pricingRequestFingerprint(task.Kind, task.Model, candidateBody),
				})
				if reviewErr != nil {
					return Result{}, reviewErr
				}
				walletReservationFinalized = true
				s.logger.Warn("upstream submission requires manual review", "taskID", task.ID, "attemptID", submissionUnknown.AttemptID, "error_category", "upstream_submission_unknown")
				return Result{Task: review, Output: review.Result}, submissionUnknown
			}
			if isLocalRateLimitError(err) {
				lastErr = err
				candidateErr = err
				if task.AsyncMode && store.RateLimitRetryable(err) {
					queued, delay, queueErr := s.requeueRateLimitedTask(ctx, task, err, candidate)
					if queueErr != nil {
						return Result{}, queueErr
					}
					return Result{Task: queued, Output: queued.Result}, &TaskQueuedError{Delay: delay}
				}
				attemptNo = s.recordFailedAttempt(ctx, failedAttemptRecord{
					Task:         task,
					Body:         candidateBody,
					Candidate:    &candidate,
					AttemptNo:    nextAttemptNo,
					Code:         clients.ErrorCode(err),
					Cause:        err,
					Simulated:    isSimulation(task, candidate),
					Scope:        "rate_limit",
					Reason:       "local_rate_limit_blocked",
					ExtraMetrics: []map[string]any{parameterPreprocessingMetrics(preprocessing.Log)},
					ModelType:    candidate.ModelType,
				})
				break candidatesLoop
			}
			attemptNo = nextAttemptNo
			lastErr = err
			candidateErr = err
			retryDecision := retryDecisionForCandidate(candidate, err)
			retryAction := "retry"
			if !retryDecision.Retry {
				retryAction = "stop"
			}
			if clientAttempt >= clientAttempts {
				retryDecision.Retry = false
				retryDecision.Reason = "same_client_max_attempts"
				retryDecision.Match = policyRuleMatch{
					Source: "model_runtime_policy_sets.retry_policy",
					Policy: "retryPolicy",
					Rule:   "maxAttempts",
					Value:  strconv.Itoa(clientAttempts),
				}
				retryDecision.Info = failureInfoFromError(err)
				retryAction = "stop"
			}
			if failoverTimeBudgetExceeded(executeStartedAt, maxFailoverDuration) {
				retryDecision.Retry = false
				retryDecision.Reason = "failover_time_budget_exceeded"
				retryDecision.Match = policyRuleMatch{
					Source: "gateway_runner_policies.failover_policy",
					Policy: "failoverPolicy",
					Rule:   "maxDurationSeconds",
					Value:  strconv.Itoa(int(maxFailoverDuration.Seconds())),
				}
				retryDecision.Info = failureInfoFromError(err)
				retryAction = "stop"
			}
			s.recordAttemptTrace(ctx, task.ID, attemptNo, retryTraceEntry(candidate, retryDecision, retryAction, clientAttempt, clientAttempts))
			if !retryDecision.Retry {
				break
			}
			if err := s.emit(ctx, task.ID, "task.retrying", "running", "retry", 0.45, "retrying same client", addPolicyTracePayload(map[string]any{
				"attempt":       attemptNo,
				"clientAttempt": clientAttempt,
				"clientId":      candidate.ClientID,
				"error":         err.Error(),
				"reason":        retryDecision.Reason,
				"scope":         "same_client",
			}, retryDecision.Match, retryDecision.Info), isSimulation(task, candidate)); err != nil {
				return Result{}, err
			}
		}
		if candidateErr == nil || index+1 >= len(candidates) || index+1 >= maxPlatforms {
			if candidateErr != nil {
				s.applyPriorityDemotePolicy(ctx, task.ID, attemptNo, runnerPolicy, candidate, task.Model, candidateErr, isSimulation(task, candidate))
				decision := failoverDecisionForCandidate(runnerPolicy, candidate, candidateErr)
				if decision.Retry {
					decision.Retry = false
					decision.Action = "stop"
					decision.Reason = "no_next_platform"
					decision.Match = policyRuleMatch{Source: "runner_candidates", Policy: "candidateSelection", Rule: "candidateCount", Value: strconv.Itoa(len(candidates))}
					if index+1 >= maxPlatforms {
						decision.Reason = "max_platforms_reached"
						decision.Match = policyRuleMatch{Source: "gateway_runner_policies.failover_policy", Policy: "failoverPolicy", Rule: "maxPlatforms", Value: strconv.Itoa(maxPlatforms)}
					}
				}
				s.recordAttemptTrace(ctx, task.ID, attemptNo, failoverTraceEntry(decision, candidate))
			}
			break
		}
		s.applyPriorityDemotePolicy(ctx, task.ID, attemptNo, runnerPolicy, candidate, task.Model, candidateErr, isSimulation(task, candidate))
		if failoverTimeBudgetExceeded(executeStartedAt, maxFailoverDuration) {
			elapsedSeconds := int(time.Since(executeStartedAt).Seconds())
			maxDurationSeconds := int(maxFailoverDuration.Seconds())
			s.recordAttemptTrace(ctx, task.ID, attemptNo, failoverTimeBudgetTraceEntry(elapsedSeconds, maxDurationSeconds, failureInfoFromError(candidateErr), candidate))
			if err := s.emit(ctx, task.ID, "task.failover.stopped", "running", "retry", 0.55, "failover time budget exceeded", map[string]any{
				"elapsedSeconds":     elapsedSeconds,
				"maxDurationSeconds": maxDurationSeconds,
				"scope":              "next_platform",
				"statusCode":         clients.ErrorResponseMetadata(candidateErr).StatusCode,
			}, isSimulation(task, candidate)); err != nil {
				return Result{}, err
			}
			break
		}
		decision := failoverDecisionForCandidate(runnerPolicy, candidate, candidateErr)
		if !decision.Retry && hasLoadAvoidanceFallback(candidates, index, maxPlatforms) {
			decision = loadAvoidanceFallbackDecision(candidateErr)
		}
		s.recordAttemptTrace(ctx, task.ID, attemptNo, failoverTraceEntry(decision, candidate))
		if !decision.Retry {
			break
		}
		s.applyFailoverAction(ctx, task.ID, candidate, task.Model, decision, isSimulation(task, candidate), singleSourceProtected)
		if err := s.emit(ctx, task.ID, "task.retrying", "running", "retry", 0.55, "retrying next client", addPolicyTracePayload(map[string]any{
			"attempt": attemptNo,
			"action":  decision.Action,
			"error":   candidateErr.Error(),
			"reason":  decision.Reason,
			"scope":   "next_platform",
		}, decision.Match, decision.Info), isSimulation(task, candidate)); err != nil {
			return Result{}, err
		}
	}
	code := clients.ErrorCode(lastErr)
	message := "task failed"
	if lastErr != nil {
		message = lastErr.Error()
	}
	if task.AsyncMode && ctx.Err() != nil {
		queued, queueErr := s.requeueInterruptedAsyncTask(context.WithoutCancel(ctx), task)
		if queueErr != nil {
			return Result{}, queueErr
		}
		return Result{Task: queued, Output: queued.Result}, &TaskQueuedError{Delay: 0}
	}
	if task.AsyncMode && errors.Is(lastErr, store.ErrRateLimited) && store.RateLimitRetryable(lastErr) {
		queued, delay, queueErr := s.requeueRateLimitedTask(ctx, task, lastErr, lastCandidate)
		if queueErr != nil {
			return Result{}, queueErr
		}
		return Result{Task: queued, Output: queued.Result}, &TaskQueuedError{Delay: delay}
	}
	extraMetrics := []map[string]any{}
	if lastPreprocessing != nil {
		extraMetrics = append(extraMetrics, parameterPreprocessingMetrics(*lastPreprocessing))
	}
	failed, err := s.failTask(ctx, task.ID, task.ExecutionToken, code, message, task.RunMode == "simulation", lastErr, extraMetrics...)
	if err != nil {
		return Result{}, err
	}
	// FinishTaskFailure atomically transfers ownership of any reservation to the release Outbox.
	walletReservationFinalized = true
	return Result{Task: failed, Output: failed.Result}, lastErr
}

func pricingCandidateKey(candidate store.RuntimeModelCandidate) string {
	return firstNonEmptyString(candidate.PlatformModelID, candidate.PlatformID+":"+candidate.ModelName)
}

func normalizedBillingEngineMode(value string) string {
	switch strings.ToLower(strings.TrimSpace(value)) {
	case "enforce", "hold":
		return strings.ToLower(strings.TrimSpace(value))
	default:
		return "observe"
	}
}

func (s *Service) maximumLegacyCandidateEstimate(ctx context.Context, user *auth.User, kind string, body map[string]any, candidates []store.RuntimeModelCandidate, preprocessingByCandidate map[string]parameterPreprocessResult) ([]any, fixedAmount) {
	var maximumItems []any
	maximumAmount := fixedAmount(0)
	for index, candidate := range candidates {
		candidateBody := preprocessRequest(kind, cloneMap(body), candidate)
		if preprocessing, ok := preprocessingByCandidate[pricingCandidateKey(candidate)]; ok && preprocessing.Err == nil {
			candidateBody = preprocessing.Body
		}
		items := s.estimatedBillings(ctx, user, kind, candidateBody, candidate)
		amount := billingItemsFixedTotal(items)
		if index == 0 || amount > maximumAmount {
			maximumItems = items
			maximumAmount = amount
		}
	}
	return maximumItems, maximumAmount
}

func billingItemsFixedTotal(items []any) fixedAmount {
	total := fixedAmount(0)
	for _, raw := range items {
		line, _ := raw.(map[string]any)
		if line == nil {
			continue
		}
		amount, err := fixedAmountFromAny(line["amount"])
		if err == nil && amount > 0 {
			next, addErr := addFixedAmounts(total, amount)
			if addErr != nil {
				return maxFixedAmount
			}
			total = next
		}
	}
	return total
}

func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user *auth.User, body map[string]any, preprocessing parameterPreprocessingLog, candidate store.RuntimeModelCandidate, pricing resolvedPricing, attemptNo int, onDelta clients.StreamDelta, responseExecution responseExecutionContext, singleSourceProtected bool, cacheAffinityPolicy map[string]any, cacheAffinityRecordKeys []string) (clients.Response, error) {
	var err error
	candidate, err = candidateWithEnvironmentCredentials(candidate)
	if err != nil {
		return clients.Response{}, err
	}
	simulated := isSimulation(task, candidate)
	baseAttemptMetrics := mergeMetrics(attemptMetrics(candidate, attemptNo, simulated), parameterPreprocessingMetrics(preprocessing))
	reservations := s.rateLimitReservations(ctx, user, candidate, body)
	limitResult, err := s.store.ReserveRateLimits(ctx, task.ID, "", reservations)
	if err != nil {
		retryable := store.RateLimitRetryable(err)
		clientErr := &clients.ClientError{Code: "rate_limit", Message: err.Error(), Retryable: retryable}
		return clients.Response{}, &localRateLimitError{clientErr: clientErr, cause: err, retryAfter: localRateLimitRetryAfter(err)}
	}
	rateReservationsFinalized := false
	defer func() {
		if !rateReservationsFinalized {
			_ = s.store.ReleaseRateLimitReservations(context.WithoutCancel(ctx), limitResult.Reservations, "attempt_failed")
		}
	}()
	defer s.store.ReleaseConcurrencyLeases(context.WithoutCancel(ctx), limitResult.LeaseIDs)

	attemptID, err := s.store.CreateTaskAttempt(ctx, store.CreateTaskAttemptInput{
		TaskID:             task.ID,
		AttemptNo:          attemptNo,
		PlatformID:         candidate.PlatformID,
		PlatformModelID:    candidate.PlatformModelID,
		ClientID:           candidate.ClientID,
		QueueKey:           candidate.QueueKey,
		Status:             "running",
		Simulated:          simulated,
		RequestSnapshot:    s.slimTaskRequestSnapshot(task, body),
		Metrics:            baseAttemptMetrics,
		PricingSnapshot:    pricing.Snapshot,
		RequestFingerprint: pricingRequestFingerprint(task.Kind, task.Model, body),
	})
	if err != nil {
		return clients.Response{}, fmt.Errorf("create task attempt: %w", err)
	}
	if err := s.recordTaskParameterPreprocessing(ctx, task.ID, attemptID, attemptNo, candidate, s.slimParameterPreprocessingLog(task, preprocessing)); err != nil {
		clientErr := &clients.ClientError{Code: "runtime_error", Message: err.Error(), Retryable: false}
		_ = s.store.FinishTaskAttempt(ctx, store.FinishTaskAttemptInput{
			AttemptID:    attemptID,
			Status:       "failed",
			Retryable:    false,
			Metrics:      mergeMetrics(baseAttemptMetrics, map[string]any{"error": err.Error(), "retryable": false, "trace": []any{failureTraceEntry(clientErr, false)}}),
			ErrorCode:    clients.ErrorCode(clientErr),
			ErrorMessage: err.Error(),
		})
		return clients.Response{}, fmt.Errorf("record parameter preprocessing: %w", err)
	}
	if err := s.store.AttachRateLimitResultToAttempt(ctx, attemptID, limitResult); err != nil {
		clientErr := &clients.ClientError{Code: "runtime_error", Message: err.Error(), Retryable: false}
		_ = s.store.FinishTaskAttempt(ctx, store.FinishTaskAttemptInput{
			AttemptID:    attemptID,
			Status:       "failed",
			Retryable:    false,
			Metrics:      mergeMetrics(baseAttemptMetrics, map[string]any{"error": err.Error(), "retryable": false, "trace": []any{failureTraceEntry(clientErr, false)}}),
			ErrorCode:    clients.ErrorCode(clientErr),
			ErrorMessage: err.Error(),
		})
		return clients.Response{}, fmt.Errorf("attach rate limit reservations to attempt: %w", err)
	}
	if err := s.emit(ctx, task.ID, "task.attempt.started", "running", "submitting", 0.25, "client attempt started", map[string]any{"attempt": attemptNo, "clientId": candidate.ClientID}, simulated); err != nil {
		return clients.Response{}, fmt.Errorf("emit attempt started: %w", err)
	}

	if err := s.store.RecordClientAssignment(ctx, candidate); err != nil {
		return clients.Response{}, fmt.Errorf("record client assignment: %w", err)
	}
	defer s.store.RecordClientRelease(context.WithoutCancel(ctx), candidate.ClientID, "")

	requestHTTPClient, err := s.httpClientForCandidate(candidate, simulated)
	if err != nil {
		_ = s.store.FinishTaskAttempt(ctx, store.FinishTaskAttemptInput{
			AttemptID:    attemptID,
			Status:       "failed",
			Retryable:    false,
			Metrics:      mergeMetrics(baseAttemptMetrics, map[string]any{"error": err.Error(), "retryable": false, "trace": []any{failureTraceEntry(err, false)}}),
			ErrorCode:    clients.ErrorCode(err),
			ErrorMessage: err.Error(),
		})
		return clients.Response{}, fmt.Errorf("prepare http client: %w", err)
	}
	client := s.clientFor(candidate, simulated)
	providerBody, err := s.compilePortraitAssetReferences(ctx, user, task.Kind, body, candidate)
	if err != nil {
		_ = s.store.FinishTaskAttempt(ctx, store.FinishTaskAttemptInput{
			AttemptID:    attemptID,
			Status:       "failed",
			Retryable:    false,
			Metrics:      mergeMetrics(baseAttemptMetrics, map[string]any{"error": err.Error(), "retryable": false, "trace": []any{failureTraceEntry(err, false)}}),
			ErrorCode:    clients.ErrorCode(err),
			ErrorMessage: err.Error(),
		})
		return clients.Response{}, err
	}
	providerBody, err = s.preparePrivateProviderRequest(ctx, user, task.Kind, providerBody)
	if err != nil {
		_ = s.store.FinishTaskAttempt(ctx, store.FinishTaskAttemptInput{
			AttemptID: attemptID, Status: "failed", Retryable: false,
			Metrics:   mergeMetrics(baseAttemptMetrics, map[string]any{"error": err.Error(), "retryable": false}),
			ErrorCode: clients.ErrorCode(err), ErrorMessage: err.Error(),
		})
		return clients.Response{}, err
	}
	providerBody, err = s.hydrateProviderRequestAssets(ctx, providerBody, candidate)
	if err != nil {
		_ = s.store.FinishTaskAttempt(ctx, store.FinishTaskAttemptInput{
			AttemptID:    attemptID,
			Status:       "failed",
			Retryable:    false,
			Metrics:      mergeMetrics(baseAttemptMetrics, map[string]any{"error": err.Error(), "retryable": false, "trace": []any{failureTraceEntry(err, false)}}),
			ErrorCode:    clients.ErrorCode(err),
			ErrorMessage: err.Error(),
		})
		return clients.Response{}, err
	}
	callStartedAt := time.Now()
	publicResponseID := ""
	publicPreviousResponseID := ""
	if task.Kind == "responses" && candidate.ResponseProtocol == clients.ProtocolOpenAIChatCompletions {
		publicResponseID = responseExecution.PublicResponseID
		publicPreviousResponseID = responseExecution.PublicPreviousResponseID
	}
	submissionStatus := "not_submitted"
	if strings.TrimSpace(task.RemoteTaskID) != "" {
		submissionStatus = "response_received"
		if err := s.store.SetAttemptUpstreamSubmissionStatus(ctx, attemptID, submissionStatus); err != nil {
			return clients.Response{}, fmt.Errorf("restore upstream submission status: %w", err)
		}
	}
	setSubmissionStatus := func(status string) error {
		if submissionStatus == "response_received" && status != "response_received" {
			return nil
		}
		if submissionStatus == status {
			return nil
		}
		if err := s.store.SetAttemptUpstreamSubmissionStatus(context.WithoutCancel(ctx), attemptID, status); err != nil {
			return err
		}
		submissionStatus = status
		return nil
	}
	var submissionWire *clients.WireResponse
	response, err := client.Run(ctx, clients.Request{
		Kind:              task.Kind,
		ModelType:         candidate.ModelType,
		Model:             task.Model,
		Body:              providerBody,
		Candidate:         candidate,
		HTTPClient:        requestHTTPClient,
		RemoteTaskID:      task.RemoteTaskID,
		RemoteTaskPayload: task.RemoteTaskPayload,
		OnRemoteTaskSubmitted: func(remoteTaskID string, payload map[string]any) error {
			if strings.TrimSpace(remoteTaskID) == "" {
				return nil
			}
			if err := s.store.SetTaskRemoteTask(context.WithoutCancel(ctx), task.ID, task.ExecutionToken, attemptID, remoteTaskID, payload); err != nil {
				return err
			}
			task.RemoteTaskID = remoteTaskID
			task.RemoteTaskPayload = payload
			if err := s.persistCompatibilitySubmission(context.WithoutCancel(ctx), task, candidate, remoteTaskID, payload, submissionWire); err != nil {
				return err
			}
			return setSubmissionStatus("response_received")
		},
		OnRemoteTaskPolled: func(remoteTaskID string, payload map[string]any) error {
			if strings.TrimSpace(remoteTaskID) == "" {
				return nil
			}
			if err := s.store.SetTaskRemoteTask(context.WithoutCancel(ctx), task.ID, task.ExecutionToken, attemptID, remoteTaskID, payload); err != nil {
				return err
			}
			task.RemoteTaskID = remoteTaskID
			task.RemoteTaskPayload = payload
			return setSubmissionStatus("response_received")
		},
		OnUpstreamSubmissionStarted: func() error {
			return setSubmissionStatus("submitting")
		},
		OnUpstreamResponseReceived: func() error {
			return setSubmissionStatus("response_received")
		},
		OnUpstreamWireResponse: func(wire *clients.WireResponse) error {
			submissionWire = wire
			return s.persistCompatibilitySubmission(context.WithoutCancel(ctx), task, candidate, "", wire.Body, wire)
		},
		Stream:                     boolFromMap(providerBody, "stream"),
		StreamDelta:                onDelta,
		UpstreamProtocol:           candidate.ResponseProtocol,
		PublicResponseID:           publicResponseID,
		PublicPreviousResponseID:   publicPreviousResponseID,
		UpstreamPreviousResponseID: responseExecution.UpstreamPreviousResponseID,
		PreviousResponseTurns:      responseExecution.PreviousTurns,
	})
	callFinishedAt := time.Now()
	if err == nil {
		if markErr := setSubmissionStatus("response_received"); markErr != nil {
			return clients.Response{}, &upstreamSubmissionUnknownError{AttemptID: attemptID, Cause: markErr}
		}
	}
	if response.ResponseStartedAt.IsZero() {
		response.ResponseStartedAt = callStartedAt
	}
	if response.ResponseFinishedAt.IsZero() {
		response.ResponseFinishedAt = callFinishedAt
	}
	if response.ResponseDurationMS == 0 {
		response.ResponseDurationMS = response.ResponseFinishedAt.Sub(response.ResponseStartedAt).Milliseconds()
		if response.ResponseDurationMS < 0 {
			response.ResponseDurationMS = 0
		}
	}
	if err != nil {
		if clients.ErrorResponseMetadata(err).StatusCode > 0 && submissionStatus != "response_received" {
			if markErr := setSubmissionStatus("response_received"); markErr != nil {
				return clients.Response{}, &upstreamSubmissionUnknownError{AttemptID: attemptID, Cause: markErr}
			}
		}
		retryable := clients.IsRetryable(err)
		requestID, metrics, responseStartedAt, responseFinishedAt, responseDurationMS := failureMetrics(err, simulated)
		if responseStartedAt.IsZero() {
			responseStartedAt = callStartedAt
		}
		if responseFinishedAt.IsZero() {
			responseFinishedAt = callFinishedAt
		}
		if responseDurationMS == 0 {
			responseDurationMS = responseFinishedAt.Sub(responseStartedAt).Milliseconds()
			if responseDurationMS < 0 {
				responseDurationMS = 0
			}
		}
		metrics = mergeMetrics(baseAttemptMetrics, metrics)
		_ = s.store.FinishTaskAttempt(ctx, store.FinishTaskAttemptInput{
			AttemptID:          attemptID,
			Status:             "failed",
			Retryable:          retryable,
			RequestID:          requestID,
			Metrics:            metrics,
			ResponseStartedAt:  responseStartedAt,
			ResponseFinishedAt: responseFinishedAt,
			ResponseDurationMS: responseDurationMS,
			ErrorCode:          clients.ErrorCode(err),
			ErrorMessage:       err.Error(),
		})
		_ = s.emit(ctx, task.ID, "task.attempt.failed", "running", "attempt_failed", 0.45, err.Error(), map[string]any{"attempt": attemptNo, "retryable": retryable, "requestId": requestID, "statusCode": clients.ErrorResponseMetadata(err).StatusCode, "metrics": metrics}, simulated)
		if !simulated && submissionStatus == "submitting" {
			return clients.Response{}, &upstreamSubmissionUnknownError{AttemptID: attemptID, Cause: err}
		}
		s.applyCandidateFailurePolicies(ctx, task.ID, candidate, err, simulated, singleSourceProtected)
		return clients.Response{}, err
	}
	uploadedResult, err := s.uploadGeneratedAssets(ctx, task.ID, task.Kind, response.Result)
	if err != nil {
		metrics := mergeMetrics(taskMetrics(task, user, body, candidate, response, simulated), parameterPreprocessingMetrics(preprocessing), map[string]any{
			"error":     err.Error(),
			"retryable": clients.IsRetryable(err),
			"trace":     []any{failureTraceEntry(err, clients.IsRetryable(err))},
		})
		_ = s.store.FinishTaskAttempt(ctx, store.FinishTaskAttemptInput{
			AttemptID:          attemptID,
			Status:             "failed",
			Retryable:          clients.IsRetryable(err),
			RequestID:          response.RequestID,
			Usage:              usageToMap(response.Usage),
			Metrics:            metrics,
			ResponseSnapshot:   response.Result,
			ResponseStartedAt:  response.ResponseStartedAt,
			ResponseFinishedAt: response.ResponseFinishedAt,
			ResponseDurationMS: response.ResponseDurationMS,
			ErrorCode:          clients.ErrorCode(err),
			ErrorMessage:       err.Error(),
		})
		return clients.Response{}, err
	}
	response.Result = uploadedResult
	if task.Kind == "responses" {
		response.UpstreamProtocol = candidate.ResponseProtocol
		response.ParentResponseID = responseExecution.PublicPreviousResponseID
		response.ResponseChainDepth = responseExecution.ChainDepth
		if candidate.ResponseProtocol == clients.ProtocolOpenAIChatCompletions {
			response.Result["id"] = responseExecution.PublicResponseID
			response.PublicResponseID = responseExecution.PublicResponseID
		} else {
			response.PublicResponseID = strings.TrimSpace(stringFromAny(response.Result["id"]))
		}
		if err := s.persistResponseChain(ctx, task, body, candidate, responseExecution, response); err != nil {
			return clients.Response{}, fmt.Errorf("persist response chain: %w", err)
		}
	}
	if task.Kind == "voice.clone" {
		voice, err := s.persistVoiceCloneResult(ctx, task, user, candidate, attemptID, body, response.Result)
		if err != nil {
			metrics := mergeMetrics(taskMetrics(task, user, body, candidate, response, simulated), parameterPreprocessingMetrics(preprocessing), map[string]any{
				"error":     err.Error(),
				"retryable": false,
				"trace":     []any{failureTraceEntry(err, false)},
			})
			_ = s.store.FinishTaskAttempt(ctx, store.FinishTaskAttemptInput{
				AttemptID:          attemptID,
				Status:             "failed",
				Retryable:          false,
				RequestID:          response.RequestID,
				Usage:              usageToMap(response.Usage),
				Metrics:            metrics,
				ResponseSnapshot:   response.Result,
				ResponseStartedAt:  response.ResponseStartedAt,
				ResponseFinishedAt: response.ResponseFinishedAt,
				ResponseDurationMS: response.ResponseDurationMS,
				ErrorCode:          "cloned_voice_persist_failed",
				ErrorMessage:       err.Error(),
			})
			return clients.Response{}, err
		}
		response.Result["cloned_voice"] = voice
		response.Result["clonedVoice"] = voice
	}
	if task.Kind == "speech.generations" {
		s.touchClonedVoiceUsage(ctx, user, body, candidate)
	}
	response.Result = s.enrichGeneratedVideoMetadata(ctx, task.Kind, response.Result)
	for _, progress := range response.Progress {
		if err := s.emit(ctx, task.ID, "task.progress", "running", progress.Phase, progress.Progress, progress.Message, progress.Payload, simulated); err != nil {
			return clients.Response{}, fmt.Errorf("emit task progress: %w", err)
		}
	}
	if err := s.store.CommitRateLimitReservations(ctx, limitResult.Reservations, tokenUsageAmounts(response.Usage)); err != nil {
		return clients.Response{}, fmt.Errorf("commit rate limit reservations: %w", err)
	}
	rateReservationsFinalized = true
	response.AttemptID = attemptID
	if err := s.store.RecordCacheAffinityObservation(context.WithoutCancel(ctx), store.CacheAffinityObservationInput{
		CacheAffinityKey:       candidate.CacheAffinity.Key,
		CacheAffinityKeys:      cacheAffinityRecordKeys,
		CacheAffinityPolicy:    cacheAffinityPolicy,
		PlatformID:             candidate.PlatformID,
		PlatformModelID:        candidate.PlatformModelID,
		ClientID:               candidate.ClientID,
		ModelType:              candidate.ModelType,
		InputTokens:            response.Usage.InputTokens,
		CachedInputTokens:      response.Usage.CachedInputTokens,
		CachedInputTokensKnown: response.Usage.CachedInputTokensKnown,
	}); err != nil {
		s.logger.Warn("record cache affinity observation failed", "error", err, "clientId", candidate.ClientID)
	}
	return response, nil
}

func (s *Service) persistCompatibilitySubmission(ctx context.Context, task store.GatewayTask, candidate store.RuntimeModelCandidate, remoteTaskID string, payload map[string]any, wire *clients.WireResponse) error {
	targetProtocol := strings.TrimSpace(stringFromMap(task.Request, "_gateway_target_protocol"))
	if targetProtocol == "" {
		return nil
	}
	sourceProtocol := compatibilitySourceProtocol(task.Kind, candidate)
	if wire != nil && strings.TrimSpace(wire.Protocol) != "" {
		sourceProtocol = strings.TrimSpace(wire.Protocol)
	}
	publicID := task.ID
	if sourceProtocol == targetProtocol && strings.TrimSpace(remoteTaskID) != "" {
		publicID = strings.TrimSpace(remoteTaskID)
	}
	submitBody := payload
	if nested, ok := payload["submit"].(map[string]any); ok && len(nested) > 0 {
		submitBody = nested
	}
	httpStatus := http.StatusOK
	headers := map[string]any{}
	if wire != nil {
		httpStatus = wire.StatusCode
		if len(wire.Body) > 0 {
			submitBody = wire.Body
		}
		for name, values := range wire.Headers {
			headers[name] = append([]string(nil), values...)
		}
	}
	return s.store.SetTaskCompatibilitySubmission(ctx, task.ID, store.CompatibilitySubmission{
		TargetProtocol: targetProtocol,
		PublicID:       publicID,
		SourceProtocol: sourceProtocol,
		HTTPStatus:     httpStatus,
		Headers:        headers,
		Body:           submitBody,
	})
}

func compatibilitySourceProtocol(kind string, candidate store.RuntimeModelCandidate) string {
	if protocol := strings.TrimSpace(candidate.ResponseProtocol); protocol != "" {
		return protocol
	}
	switch strings.ToLower(strings.TrimSpace(candidate.Provider)) {
	case "volces":
		if kind == "videos.generations" {
			return clients.ProtocolVolcesContents
		}
	case "keling", "kling":
		if kind == "videos.generations" {
			return clients.ProtocolKlingV1Omni
		}
	case "gemini":
		return clients.ProtocolGeminiGenerateContent
	case "openai":
		switch kind {
		case "chat.completions":
			return clients.ProtocolOpenAIChatCompletions
		case "responses":
			return clients.ProtocolOpenAIResponses
		case "embeddings":
			return clients.ProtocolOpenAIEmbeddings
		case "images.generations", "images.edits":
			return clients.ProtocolOpenAIImages
		}
	}
	return strings.ToLower(strings.TrimSpace(candidate.Provider))
}

func (s *Service) recordTaskParameterPreprocessing(ctx context.Context, taskID string, attemptID string, attemptNo int, candidate store.RuntimeModelCandidate, log parameterPreprocessingLog) error {
	if skipTaskParameterPreprocessingLog(log.ModelType) && !log.Changed {
		return nil
	}
	_, err := s.store.CreateTaskParamPreprocessingLog(ctx, store.CreateTaskParamPreprocessingLogInput{
		TaskID:          taskID,
		AttemptID:       attemptID,
		AttemptNo:       attemptNo,
		ModelType:       log.ModelType,
		PlatformID:      candidate.PlatformID,
		PlatformModelID: candidate.PlatformModelID,
		ClientID:        candidate.ClientID,
		Changed:         log.Changed,
		ChangeCount:     len(log.Changes),
		ActualInput:     log.Input,
		ConvertedOutput: log.Output,
		Changes:         log.Changes,
		ModelSnapshot:   log.Model,
	})
	return err
}

func skipTaskParameterPreprocessingLog(modelType string) bool {
	switch strings.TrimSpace(modelType) {
	case "text_generate", "text_embedding", "text_rerank", "chat", "responses", "text":
		return true
	default:
		return false
	}
}

func (s *Service) clientFor(candidate store.RuntimeModelCandidate, simulated bool) clients.Client {
	if simulated {
		return s.clients["simulation"]
	}
	key := strings.ToLower(strings.TrimSpace(candidate.SpecType))
	if key == "" {
		key = strings.ToLower(strings.TrimSpace(candidate.Provider))
	}
	provider := strings.ToLower(strings.TrimSpace(candidate.Provider))
	baseURL := strings.ToLower(strings.TrimSpace(candidate.BaseURL))
	if key == "google-gemini" || provider == "gemini" || provider == "google-gemini" || provider == "gemini-openai" || strings.Contains(baseURL, "generativelanguage.googleapis.com") {
		key = "gemini"
	}
	if client, ok := s.clients[key]; ok {
		return client
	}
	return s.clients["openai"]
}

func (s *Service) failTask(ctx context.Context, taskID string, executionToken string, code string, message string, simulated bool, cause error, extraMetrics ...map[string]any) (store.GatewayTask, error) {
	requestID, metrics, responseStartedAt, responseFinishedAt, responseDurationMS := failureMetrics(cause, simulated)
	if len(extraMetrics) > 0 {
		values := append([]map[string]any{metrics}, extraMetrics...)
		metrics = mergeMetrics(values...)
	}
	metrics = s.withAttemptHistory(ctx, taskID, metrics)
	failed, err := s.store.FinishTaskFailure(ctx, store.FinishTaskFailureInput{
		TaskID:             taskID,
		ExecutionToken:     executionToken,
		Code:               code,
		Message:            message,
		Result:             buildFailureResult(code, message, requestID, cause),
		RequestID:          requestID,
		Metrics:            metrics,
		ResponseStartedAt:  responseStartedAt,
		ResponseFinishedAt: responseFinishedAt,
		ResponseDurationMS: responseDurationMS,
	})
	if err != nil {
		return store.GatewayTask{}, err
	}
	if failed.Status == "cancelled" {
		return failed, nil
	}
	if eventErr := s.emit(ctx, taskID, "task.failed", "failed", "failed", 1, message, map[string]any{"code": code, "requestId": requestID, "metrics": metrics}, simulated); eventErr != nil {
		return store.GatewayTask{}, eventErr
	}
	return failed, nil
}

func isVolcesRemoteTaskCancellation(candidate store.RuntimeModelCandidate, err error) bool {
	return isVolcesCancellationCandidate(candidate) && strings.EqualFold(clients.ErrorCode(err), "volces_task_cancelled")
}

type failedAttemptRecord struct {
	Task          store.GatewayTask
	Body          map[string]any
	Candidate     *store.RuntimeModelCandidate
	AttemptNo     int
	Code          string
	Cause         error
	Simulated     bool
	Scope         string
	Reason        string
	ExtraMetrics  []map[string]any
	Preprocessing *parameterPreprocessingLog
	ModelType     string
}

func (s *Service) recordFailedAttempt(ctx context.Context, input failedAttemptRecord) int {
	attemptNo := input.AttemptNo
	if attemptNo <= 0 {
		attemptNo = input.Task.AttemptCount + 1
	}
	code := firstNonEmptyString(input.Code, clients.ErrorCode(input.Cause))
	message := ""
	if input.Cause != nil {
		message = input.Cause.Error()
	}
	retryable := clients.IsRetryable(input.Cause)
	requestID, failure, responseStartedAt, responseFinishedAt, responseDurationMS := failureMetrics(input.Cause, input.Simulated)
	scope := firstNonEmptyString(input.Scope, "pre_provider")
	reason := firstNonEmptyString(input.Reason, "pre_provider_failed")
	trace := failureTraceEntryWithReason(input.Cause, retryable, scope, reason)
	statusCode := clients.ErrorResponseMetadata(input.Cause).StatusCode
	category := failureCategory(strings.ToLower(strings.TrimSpace(code)), statusCode, message)
	if code != "" {
		failure["errorCode"] = code
		trace["errorCode"] = code
	}
	if category != "" {
		failure["errorCategory"] = category
		trace["category"] = category
	}
	failure["failureScope"] = scope
	failure["failureReason"] = reason
	failure["trace"] = []any{trace}

	baseMetrics := map[string]any{
		"attempt":        attemptNo,
		"kind":           input.Task.Kind,
		"runMode":        input.Task.RunMode,
		"requestedModel": input.Task.Model,
		"simulated":      input.Simulated,
	}
	if input.ModelType != "" {
		baseMetrics["modelType"] = input.ModelType
	}
	var platformID, platformModelID, clientID, queueKey string
	if input.Candidate != nil {
		baseMetrics = attemptMetrics(*input.Candidate, attemptNo, input.Simulated)
		baseMetrics["kind"] = input.Task.Kind
		baseMetrics["runMode"] = input.Task.RunMode
		baseMetrics["requestedModel"] = input.Task.Model
		platformID = input.Candidate.PlatformID
		platformModelID = input.Candidate.PlatformModelID
		clientID = input.Candidate.ClientID
		queueKey = input.Candidate.QueueKey
	}
	metrics := mergeMetrics(append([]map[string]any{baseMetrics, failure}, input.ExtraMetrics...)...)
	attemptID, err := s.store.CreateTaskAttempt(ctx, store.CreateTaskAttemptInput{
		TaskID:          input.Task.ID,
		AttemptNo:       attemptNo,
		PlatformID:      platformID,
		PlatformModelID: platformModelID,
		ClientID:        clientID,
		QueueKey:        queueKey,
		Status:          "running",
		Simulated:       input.Simulated,
		RequestSnapshot: s.slimTaskRequestSnapshot(input.Task, input.Body),
		Metrics:         metrics,
	})
	if err != nil {
		s.logger.Warn("record failed task attempt failed", "taskID", input.Task.ID, "attempt", attemptNo, "error", err)
		return attemptNo
	}
	if input.Preprocessing != nil && input.Candidate != nil {
		preprocessing := s.slimParameterPreprocessingLog(input.Task, *input.Preprocessing)
		if err := s.recordTaskParameterPreprocessing(ctx, input.Task.ID, attemptID, attemptNo, *input.Candidate, preprocessing); err != nil {
			s.logger.Warn("record failed attempt parameter preprocessing failed", "taskID", input.Task.ID, "attempt", attemptNo, "error", err)
		}
	}
	if err := s.store.FinishTaskAttempt(ctx, store.FinishTaskAttemptInput{
		AttemptID:          attemptID,
		Status:             "failed",
		Retryable:          retryable,
		RequestID:          requestID,
		Metrics:            metrics,
		ResponseStartedAt:  responseStartedAt,
		ResponseFinishedAt: responseFinishedAt,
		ResponseDurationMS: responseDurationMS,
		ErrorCode:          code,
		ErrorMessage:       message,
	}); err != nil {
		s.logger.Warn("finish failed task attempt failed", "taskID", input.Task.ID, "attempt", attemptNo, "error", err)
	}
	return attemptNo
}

func (s *Service) requeueRateLimitedTask(ctx context.Context, task store.GatewayTask, cause error, candidate store.RuntimeModelCandidate) (store.GatewayTask, time.Duration, error) {
	delay := localRateLimitRetryAfter(cause)
	if delay <= 0 {
		delay = 5 * time.Second
	}
	queued, err := s.store.RequeueTask(ctx, task.ID, task.ExecutionToken, delay, candidate.QueueKey)
	if err != nil {
		return store.GatewayTask{}, 0, err
	}
	payload := map[string]any{
		"code":         "rate_limit",
		"message":      cause.Error(),
		"retryAfterMs": delay.Milliseconds(),
	}
	if eventErr := s.emit(ctx, task.ID, "task.queued", "queued", "rate_limited", 0.2, "task queued by local rate limit", payload, task.RunMode == "simulation"); eventErr != nil {
		return store.GatewayTask{}, 0, eventErr
	}
	return queued, delay, nil
}

func (s *Service) requeueInterruptedAsyncTask(ctx context.Context, task store.GatewayTask) (store.GatewayTask, error) {
	queued, err := s.store.RequeueTask(ctx, task.ID, task.ExecutionToken, 0, "")
	if err != nil {
		return store.GatewayTask{}, err
	}
	payload := map[string]any{"code": "worker_interrupted"}
	if task.RemoteTaskID != "" {
		payload["remoteTaskId"] = task.RemoteTaskID
	}
	if eventErr := s.emit(ctx, task.ID, "task.queued", "queued", "worker_interrupted", 0.2, "async task queued after worker interruption", payload, task.RunMode == "simulation"); eventErr != nil {
		return store.GatewayTask{}, eventErr
	}
	return queued, nil
}

func (s *Service) withAttemptHistory(ctx context.Context, taskID string, metrics map[string]any) map[string]any {
	attempts, err := s.store.ListTaskAttempts(ctx, taskID)
	if err != nil {
		s.logger.Warn("list task attempts for metrics failed", "taskID", taskID, "error", err)
		return metrics
	}
	if len(attempts) == 0 {
		return metrics
	}
	metrics = mergeMetrics(metrics)
	metrics["attemptCount"] = len(attempts)
	metrics["attempts"] = summarizeAttempts(attempts)
	return metrics
}

func (s *Service) emit(ctx context.Context, taskID string, eventType string, status string, phase string, progress float64, message string, payload map[string]any, simulated bool) error {
	event, err := s.store.AddTaskEvent(ctx, taskID, eventType, status, phase, progress, message, payload, simulated)
	if err != nil {
		return err
	}
	if s.cfg.TaskProgressCallbackEnabled {
		return s.store.QueueTaskCallback(ctx, event, s.cfg.TaskProgressCallbackURL)
	}
	return nil
}

func modelTypeFromKind(kind string, body map[string]any) string {
	if requested := requestedModelTypeFromBody(body); requested != "" {
		return requested
	}
	switch kind {
	case "chat.completions", "responses":
		return "text_generate"
	case "embeddings":
		return "text_embedding"
	case "reranks":
		return "text_rerank"
	case "images.generations", "images.edits":
		if kind == "images.edits" {
			return "image_edit"
		}
		return "image_generate"
	case "images.vectorize":
		return "image_vectorize"
	case "videos.generations":
		if videoRequestHasVideoOrAudioReference(body) {
			return "omni_video"
		}
		if videoRequestHasReferenceImage(body) {
			return "image_to_video"
		}
		return "video_generate"
	case "videos.upscales":
		return "video_enhance"
	case "song.generations", "music.generations":
		return "audio_generate"
	case "speech.generations":
		return "text_to_speech"
	case "voice.clone":
		return "voice_clone"
	default:
		return "task"
	}
}

func requestedModelTypeFromBody(body map[string]any) string {
	for _, key := range []string{"modelType", "model_type", "capability", "capabilityType"} {
		value := canonicalModelType(strings.TrimSpace(stringFromMap(body, key)))
		if isKnownModelType(value) {
			return value
		}
	}
	return ""
}

func canonicalModelType(value string) string {
	normalized := strings.ReplaceAll(strings.ToLower(strings.TrimSpace(value)), "-", "_")
	switch normalized {
	case "embedding":
		return "text_embedding"
	case "rerank", "reranks":
		return "text_rerank"
	case "audio", "music", "music_generate", "song", "songs":
		return "audio_generate"
	case "speech", "tts":
		return "text_to_speech"
	case "voice", "voice_clone", "voiceclone", "voice.cloning":
		return "voice_clone"
	case "vectorize", "image_vectorizer", "image_vectorize":
		return "image_vectorize"
	case "video_upscale", "video_enhance", "upscale":
		return "video_enhance"
	default:
		return normalized
	}
}

func isKnownModelType(value string) bool {
	switch value {
	case "text_generate", "text_embedding", "text_rerank", "image_generate", "image_edit", "image_vectorize", "video_generate", "video_enhance", "image_to_video", "text_to_video", "video_edit", "video_reference", "video_first_last_frame", "omni_video", "omni", "audio_generate", "text_to_speech", "voice_clone":
		return true
	default:
		return false
	}
}

func videoRequestHasReferenceImage(body map[string]any) bool {
	if body == nil {
		return false
	}
	for _, key := range []string{
		"image", "images", "image_url", "imageUrl", "image_urls", "imageUrls",
		"reference_image", "referenceImage", "first_frame", "firstFrame", "last_frame", "lastFrame",
	} {
		if hasAnyString(body, key) {
			return true
		}
	}
	for _, item := range contentItems(body["content"]) {
		if isImageContent(item) {
			return true
		}
	}
	return false
}

func videoRequestHasVideoOrAudioReference(body map[string]any) bool {
	if body == nil {
		return false
	}
	for _, key := range []string{"video", "video_url", "videoUrl", "reference_video", "referenceVideo", "audio_url", "audioUrl", "reference_audio", "referenceAudio"} {
		if hasAnyString(body, key) {
			return true
		}
	}
	for _, item := range contentItems(body["content"]) {
		if isVideoContent(item) || isAudioContent(item) {
			return true
		}
	}
	return false
}

func isTextGenerationKind(kind string) bool {
	return kind == "chat.completions" || kind == "responses"
}

func isTextInputOnlyKind(kind string) bool {
	return kind == "embeddings" || kind == "reranks"
}

func isTextBillingKind(kind string) bool {
	return isTextGenerationKind(kind) || isTextInputOnlyKind(kind)
}

func isSimulation(task store.GatewayTask, candidate store.RuntimeModelCandidate) bool {
	if task.RunMode == "simulation" {
		return true
	}
	return stringFromMap(candidate.Credentials, "mode") == "simulation" || boolFromMap(candidate.PlatformConfig, "testMode")
}

func retryEnabled(candidate store.RuntimeModelCandidate) bool {
	policy := effectiveRetryPolicy(candidate)
	if enabled, ok := policy["enabled"].(bool); ok {
		return enabled
	}
	return true
}

func clientAttemptsForCandidate(candidate store.RuntimeModelCandidate) int {
	if !retryEnabled(candidate) {
		return 1
	}
	if value := intFromPolicy(effectiveRetryPolicy(candidate), "maxAttempts"); value > 0 {
		return value
	}
	return 1
}

func maxPlatformsForCandidates(candidates []store.RuntimeModelCandidate, runnerPolicy store.RunnerPolicy) int {
	if len(candidates) == 0 {
		return 0
	}
	maxPlatforms := len(candidates)
	if value := intFromPolicy(runnerPolicy.FailoverPolicy, "maxPlatforms"); value > 0 {
		if value < maxPlatforms {
			maxPlatforms = value
		}
	} else if maxPlatforms > 99 {
		maxPlatforms = 99
	}
	for _, candidate := range candidates {
		if value := intFromPolicy(effectiveFailoverPolicy(runnerPolicy.FailoverPolicy, candidate.RuntimePolicyOverride), "maxPlatforms"); value > 0 && value < maxPlatforms {
			maxPlatforms = value
		}
	}
	if maxPlatforms <= 0 {
		return 1
	}
	return maxPlatforms
}

func singleSourceProtectionActive(candidates []store.RuntimeModelCandidate, maxPlatforms int, runnerPolicy store.RunnerPolicy) bool {
	if !runnerSingleSourceProtectionEnabled(runnerPolicy) {
		return false
	}
	return runtimeCandidateSourceCount(candidates, maxPlatforms) <= 1
}

func runnerSingleSourceProtectionEnabled(runnerPolicy store.RunnerPolicy) bool {
	if enabled, ok := runnerPolicy.SingleSourcePolicy["enabled"].(bool); ok {
		return enabled
	}
	return true
}

func runtimeCandidateSourceCount(candidates []store.RuntimeModelCandidate, maxPlatforms int) int {
	limit := len(candidates)
	if maxPlatforms > 0 && maxPlatforms < limit {
		limit = maxPlatforms
	}
	sources := make(map[string]bool, limit)
	for index := 0; index < limit; index++ {
		key := strings.TrimSpace(candidates[index].PlatformID)
		if key == "" {
			key = strings.TrimSpace(candidates[index].PlatformModelID)
		}
		if key == "" {
			key = strings.TrimSpace(candidates[index].ClientID)
		}
		if key == "" {
			key = strconv.Itoa(index)
		}
		sources[key] = true
	}
	return len(sources)
}

func maxFailoverDurationForCandidates(candidates []store.RuntimeModelCandidate, runnerPolicy store.RunnerPolicy) time.Duration {
	seconds := intFromPolicy(runnerPolicy.FailoverPolicy, "maxDurationSeconds")
	if seconds <= 0 {
		seconds = 600
	}
	for _, candidate := range candidates {
		if value := intFromPolicy(effectiveFailoverPolicy(runnerPolicy.FailoverPolicy, candidate.RuntimePolicyOverride), "maxDurationSeconds"); value > 0 && value < seconds {
			seconds = value
		}
	}
	return time.Duration(seconds) * time.Second
}

func failoverTimeBudgetExceeded(start time.Time, maxDuration time.Duration) bool {
	return maxDuration > 0 && time.Since(start) >= maxDuration
}

func hasLoadAvoidanceFallback(candidates []store.RuntimeModelCandidate, index int, maxPlatforms int) bool {
	if index < 0 || index >= len(candidates) || candidates[index].LoadAvoided {
		return false
	}
	limit := len(candidates)
	if maxPlatforms > 0 && maxPlatforms < limit {
		limit = maxPlatforms
	}
	for next := index + 1; next < limit; next++ {
		if candidates[next].LoadAvoided {
			return true
		}
	}
	return false
}

func loadAvoidanceFallbackDecision(err error) failoverDecision {
	return failoverDecision{
		Retry:  true,
		Action: "next",
		Reason: "load_avoidance_fallback",
		Match: policyRuleMatch{
			Source: "runtime_client_load",
			Policy: "loadAvoidance",
			Rule:   "fallback",
			Value:  "loadRatio>=1",
		},
		Info: failureInfoFromError(err),
	}
}

func normalizeRequest(kind string, body map[string]any) map[string]any {
	out := cloneMap(body)
	return out
}

func validateRequest(kind string, body map[string]any) error {
	switch kind {
	case "chat.completions":
		if body["messages"] == nil {
			return errors.New("messages is required")
		}
		if err := clients.ValidateOpenAIReasoningEffort(body["reasoning_effort"]); err != nil {
			return err
		}
		for _, key := range []string{"max_tokens", "max_completion_tokens"} {
			if value, explicit := nonNullParameter(body, key); explicit {
				if _, ok := positiveInteger(value); !ok {
					return &clients.ClientError{Code: "invalid_parameter", Message: key + " must be a positive integer", Param: key, StatusCode: 400, Retryable: false}
				}
			}
		}
	case "responses":
		if body["input"] == nil && body["messages"] == nil {
			return errors.New("input or messages is required")
		}
		if value, explicit := nonNullParameter(body, "max_output_tokens"); explicit {
			if _, ok := positiveInteger(value); !ok {
				return &clients.ClientError{Code: "invalid_parameter", Message: "max_output_tokens must be a positive integer", Param: "max_output_tokens", StatusCode: 400, Retryable: false}
			}
		}
	case "embeddings":
		if body["input"] == nil {
			return errors.New("input is required")
		}
	case "reranks":
		if body["query"] == nil {
			return errors.New("query is required")
		}
		if !hasRerankDocuments(body["documents"]) {
			return errors.New("documents is required")
		}
	case "images.generations", "images.edits":
		if strings.TrimSpace(stringFromMap(body, "prompt")) == "" {
			return errors.New("prompt is required")
		}
	case "images.vectorize":
		if vectorizerSourceURL(body) == "" && vectorizerSourceTaskID(body) == "" {
			return &clients.ClientError{Code: "invalid_parameter", Message: "source.url or source.vectorizerTaskId is required", Param: "source", StatusCode: 400}
		}
	case "videos.upscales":
		if firstNonEmptyString(stringFromMap(body, "video_url"), stringFromMap(body, "videoUrl")) == "" {
			return &clients.ClientError{Code: "invalid_parameter", Message: "video_url is required", Param: "video_url", StatusCode: 400}
		}
	case "song.generations", "music.generations":
		if strings.TrimSpace(stringFromMap(body, "prompt")) == "" {
			return errors.New("prompt is required")
		}
	case "speech.generations":
		if strings.TrimSpace(stringFromMap(body, "text")) == "" && strings.TrimSpace(stringFromMap(body, "text_file_id")) == "" {
			return errors.New("text or text_file_id is required")
		}
		if strings.TrimSpace(stringFromMap(body, "voice_id")) == "" {
			return errors.New("voice_id is required")
		}
	case "voice.clone":
		if err := validateVoiceCloneRequest(body); err != nil {
			return err
		}
	}
	return nil
}

var credentialEnvironmentName = regexp.MustCompile(`^[A-Z_][A-Z0-9_]*$`)

func candidateWithEnvironmentCredentials(candidate store.RuntimeModelCandidate) (store.RuntimeModelCandidate, error) {
	references, _ := candidate.PlatformConfig["credentialEnv"].(map[string]any)
	if len(references) == 0 {
		references, _ = candidate.PlatformConfig["credential_env"].(map[string]any)
	}
	if len(references) == 0 {
		return candidate, nil
	}
	credentials := cloneMap(candidate.Credentials)
	for credentialName, rawEnvironmentName := range references {
		environmentName := strings.TrimSpace(fmt.Sprint(rawEnvironmentName))
		if !credentialEnvironmentName.MatchString(environmentName) {
			return candidate, &clients.ClientError{Code: "missing_credentials", Message: "platform credential environment reference is invalid", Retryable: false}
		}
		value, ok := os.LookupEnv(environmentName)
		if !ok || strings.TrimSpace(value) == "" {
			return candidate, &clients.ClientError{Code: "missing_credentials", Message: "platform credential environment variable is not configured", Retryable: false}
		}
		credentials[credentialName] = value
	}
	candidate.Credentials = credentials
	return candidate, nil
}

func (s *Service) preparePrivateProviderRequest(ctx context.Context, user *auth.User, kind string, body map[string]any) (map[string]any, error) {
	if kind != "images.vectorize" {
		return body, nil
	}
	taskID := vectorizerSourceTaskID(body)
	if taskID == "" {
		return body, nil
	}
	sourceTask, err := s.store.GetTask(ctx, taskID)
	if err != nil || !taskAccessibleToUser(sourceTask, user) || sourceTask.Kind != "images.vectorize" {
		return nil, &clients.ClientError{Code: "not_found", Message: "source vectorizer task not found", StatusCode: 404, Retryable: false}
	}
	imageToken := strings.TrimSpace(fmt.Sprint(sourceTask.RemoteTaskPayload["imageToken"]))
	if imageToken == "" {
		return nil, &clients.ClientError{Code: "invalid_parameter", Message: "source vectorizer task cannot be reused", Param: "source.vectorizerTaskId", StatusCode: 400, Retryable: false}
	}
	privateBody := cloneMap(body)
	privateBody["_vectorizer_image_token"] = imageToken
	if receipt := strings.TrimSpace(fmt.Sprint(sourceTask.RemoteTaskPayload["receipt"])); receipt != "" {
		privateBody["_vectorizer_receipt"] = receipt
	}
	return privateBody, nil
}

func vectorizerSourceURL(body map[string]any) string {
	if source, ok := body["source"].(map[string]any); ok {
		return strings.TrimSpace(firstNonEmptyString(stringFromMap(source, "url"), stringFromMap(source, "image_url"), stringFromMap(source, "imageUrl")))
	}
	return strings.TrimSpace(firstNonEmptyString(stringFromMap(body, "image_url"), stringFromMap(body, "imageUrl")))
}

func vectorizerSourceTaskID(body map[string]any) string {
	for _, key := range []string{"source", "sourceContext"} {
		if source, ok := body[key].(map[string]any); ok {
			if taskID := strings.TrimSpace(firstNonEmptyString(stringFromMap(source, "vectorizerTaskId"), stringFromMap(source, "vectorizer_task_id"), stringFromMap(source, "taskId"), stringFromMap(source, "task_id"))); taskID != "" {
				return taskID
			}
		}
	}
	return strings.TrimSpace(firstNonEmptyString(stringFromMap(body, "vectorizerTaskId"), stringFromMap(body, "vectorizer_task_id")))
}

func hasRerankDocuments(value any) bool {
	switch typed := value.(type) {
	case []any:
		return len(typed) > 0
	case []string:
		return len(typed) > 0
	default:
		return false
	}
}

func parameterPreprocessClientError(err error) *clients.ClientError {
	if err == nil {
		return nil
	}
	code := "invalid_parameter"
	var coded interface{ ErrorCode() string }
	if errors.As(err, &coded) && strings.TrimSpace(coded.ErrorCode()) != "" {
		code = coded.ErrorCode()
	}
	return &clients.ClientError{
		Code:       code,
		Message:    err.Error(),
		StatusCode: 400,
		Retryable:  false,
	}
}