package runner

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"os"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"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/easyai/easyai-ai-gateway/apps/api/internal/workerload"
	"github.com/jackc/pgx/v5"
	"github.com/riverqueue/river"
)

type Service struct {
	cfg                    config.Config
	store                  *store.Store
	coordinationStore      *store.Store
	riverStore             *store.Store
	logger                 *slog.Logger
	clients                map[string]clients.Client
	scriptExecutor         *scriptengine.Executor
	httpClients            *httpClientCache
	riverMu                sync.RWMutex
	riverControlClient     *river.Client[pgx.Tx]
	riverExecutionClient   asyncExecutionClient
	riverDrainingClients   map[asyncExecutionClient]struct{}
	riverWorkerCapacity    int
	workerInstanceID       string
	workerLoad             *workerload.Controller
	workerLoadSampler      workerLoadSampler
	asyncCapacityLoader    func(context.Context, int) (store.AsyncWorkerCapacitySnapshot, error)
	admissionWakeMu        sync.Mutex
	admissionWake          chan struct{}
	asyncAdmissionWake     chan struct{}
	admissionTaskWaiters   map[string]*admissionTaskWaiter
	admissionListener      sync.Once
	asyncAdmissionRunner   sync.Once
	asyncClientFactory     func(int) (asyncExecutionClient, error)
	mediaResultSlots       chan struct{}
	imageNormalizeSlots    chan struct{}
	directOSS              *directOSSUploader
	taskCompletionOnce     sync.Once
	taskCompletionMu       sync.Mutex
	taskCompletionWaiters  map[string]map[chan struct{}]struct{}
	taskCompletionPollWake chan struct{}
	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 submissionConfirmationPendingError struct {
	AttemptID string
	Cause     error
}

func shouldEnterSubmissionConfirmationPending(simulated bool, submissionStatus string, err error) bool {
	return !simulated && submissionStatus == "submitting" && clients.ErrorCode(err) != "timeout"
}

func (e *submissionConfirmationPendingError) Error() string {
	return "upstream submission confirmation timed out"
}

func (e *submissionConfirmationPendingError) ErrorCode() string {
	return "upstream_timeout"
}

func (e *submissionConfirmationPendingError) 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 {
	return NewWithStores(cfg, db, db, db, logger, observers...)
}

func NewWithCoordinationStore(
	cfg config.Config,
	db *store.Store,
	coordinationDB *store.Store,
	logger *slog.Logger,
	observers ...billingMetricsObserver,
) *Service {
	return NewWithStores(cfg, db, coordinationDB, db, logger, observers...)
}

func NewWithStores(
	cfg config.Config,
	db *store.Store,
	coordinationDB *store.Store,
	riverDB *store.Store,
	logger *slog.Logger,
	observers ...billingMetricsObserver,
) *Service {
	if cfg.AsyncWorkerHardLimit == 0 {
		cfg.AsyncWorkerHardLimit = 2048
	}
	if strings.TrimSpace(cfg.ExecutionPoolID) == "" {
		cfg.ExecutionPoolID = "legacy-default"
	}
	if cfg.RouteProbeHotIntervalSeconds == 0 {
		cfg.RouteProbeHotIntervalSeconds = 15
	}
	if cfg.RouteProbeColdIntervalSeconds == 0 {
		cfg.RouteProbeColdIntervalSeconds = 60
	}
	if cfg.RouteProbeTimeoutMS == 0 {
		cfg.RouteProbeTimeoutMS = 3000
	}
	if cfg.AsyncWorkerInstanceHardLimit == 0 {
		cfg.AsyncWorkerInstanceHardLimit = 32
	}
	if cfg.AsyncWorkerRefreshIntervalSeconds == 0 {
		cfg.AsyncWorkerRefreshIntervalSeconds = 5
	}
	if strings.TrimSpace(cfg.AsyncWorkerLoadMode) == "" {
		cfg.AsyncWorkerLoadMode = workerload.ModeAdaptive
	}
	if cfg.MediaMaterializationConcurrency == 0 {
		cfg.MediaMaterializationConcurrency = 8
	}
	if cfg.MediaImageNormalizationConcurrency == 0 {
		cfg.MediaImageNormalizationConcurrency = 2
	}
	if cfg.TaskProgressCallbackTimeoutMS == 0 {
		cfg.TaskProgressCallbackTimeoutMS = 5000
	}
	if cfg.TaskProgressCallbackMaxAttempts == 0 {
		cfg.TaskProgressCallbackMaxAttempts = 10
	}
	if cfg.TaskRetentionDays == 0 {
		cfg.TaskRetentionDays = 30
	}
	if cfg.TaskAnalysisRetentionDays == 0 {
		cfg.TaskAnalysisRetentionDays = 7
	}
	if cfg.TaskCleanupIntervalSeconds == 0 {
		cfg.TaskCleanupIntervalSeconds = 300
	}
	if cfg.TaskCleanupBatchSize == 0 {
		cfg.TaskCleanupBatchSize = 1000
	}
	httpClients := newHTTPClientCache()
	scriptExecutor := &scriptengine.Executor{Logger: logger}
	openAIParameterCorrections := clients.NewParameterCorrectionCache()
	service := &Service{
		cfg:               cfg,
		store:             db,
		coordinationStore: coordinationDB,
		riverStore:        riverDB,
		logger:            logger,
		scriptExecutor:    scriptExecutor,
		clients: map[string]clients.Client{
			"openai":                clients.OpenAIClient{HTTPClient: httpClients.none, Corrections: openAIParameterCorrections},
			"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,
		workerInstanceID: asyncWorkerID(),
		workerLoad: workerload.New(workerload.Config{
			Mode: cfg.AsyncWorkerLoadMode, HardLimit: cfg.AsyncWorkerInstanceHardLimit,
			InitialActive: 4, InitialHeavy: 1, HealthySamples: 3,
		}),
		workerLoadSampler:      workerload.NewSystemSampler(),
		admissionWake:          make(chan struct{}, 4096),
		asyncAdmissionWake:     make(chan struct{}, 1),
		admissionTaskWaiters:   map[string]*admissionTaskWaiter{},
		mediaResultSlots:       make(chan struct{}, cfg.MediaMaterializationConcurrency),
		imageNormalizeSlots:    make(chan struct{}, cfg.MediaImageNormalizationConcurrency),
		directOSS:              newDirectOSSUploader(cfg),
		taskCompletionWaiters:  map[string]map[chan struct{}]struct{}{},
		taskCompletionPollWake: make(chan struct{}, 1),
	}
	if len(observers) > 0 {
		service.billingMetrics = observers[0]
	}
	if service.coordinationStore == nil {
		service.coordinationStore = db
	}
	return service
}

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

func (s *Service) observeTaskEventSkip(reason string) {
	observer, ok := s.billingMetrics.(interface {
		ObserveTaskEventSkip(string)
	})
	if ok {
		observer.ObserveTaskEventSkip(reason)
	}
}

func (s *Service) observeObjectStorage(event string, provider string, bytes int, duration time.Duration) {
	observer, ok := s.billingMetrics.(interface {
		ObserveObjectStorage(string, string, int64, time.Duration)
	})
	if ok {
		observer.ObserveObjectStorage(event, provider, int64(bytes), duration)
	}
}

func (s *Service) observeResultStorage(source string) {
	observer, ok := s.billingMetrics.(interface{ ObserveResultStorage(string) })
	if ok {
		observer.ObserveResultStorage(source)
	}
}

func (s *Service) observeResultURLSigning(outcome string) {
	observer, ok := s.billingMetrics.(interface{ ObserveResultURLSigning(string) })
	if ok {
		observer.ObserveResultURLSigning(outcome)
	}
}

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

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

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

func (s *Service) executeWithToken(ctx context.Context, task store.GatewayTask, user *auth.User, onDelta clients.StreamDelta, executionToken string) (Result, error) {
	wasRunning := task.Status == "running"
	preliminaryModelType := modelTypeFromKind(task.Kind, normalizeRequest(task.Kind, task.Request))
	distributedAdmission := distributedAdmissionModelType(preliminaryModelType)
	var claimed store.GatewayTask
	var err error
	if distributedAdmission && !wasRunning {
		claimed, err = s.store.ClaimTaskPreparationForOwner(
			ctx,
			task.ID,
			executionToken,
			taskExecutionLeaseTTL,
			s.workerInstanceID,
		)
	} else {
		claimed, err = s.store.ClaimTaskExecutionForOwner(
			ctx,
			task.ID,
			executionToken,
			taskExecutionLeaseTTL,
			s.workerInstanceID,
		)
	}
	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
	defer func() {
		if !requestCancelledBeforeSubmission(ctx) {
			return
		}
		cleanupCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
		defer cancel()
		if _, _, cleanupErr := s.store.CancelTaskBeforeUpstreamSubmission(
			cleanupCtx,
			task.ID,
			task.ExecutionToken,
			"client disconnected before upstream submission",
		); cleanupErr != nil && s.logger != nil {
			s.logger.Warn("cancel disconnected task before upstream submission failed", "taskID", task.ID, "error", cleanupErr)
		}
	}()
	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)
	distributedAdmission = distributedAdmissionModelType(modelType)
	if !distributedAdmission {
		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,
		QuotaScopePrefix:    acceptanceQuotaScopePrefix(task),
	})
	if err == nil {
		candidates, err = filterCandidatesByRequestedPlatform(candidates, body)
	}
	if err != nil {
		if task.Kind == "responses" && responseExecution.PreviousChain != nil {
			err = responseChainUnavailableError()
		}
		if task.AsyncMode && store.ModelCandidateRetryAfter(err) > 0 {
			s.observeCandidateRouting("cooldown_skipped")
			queued, delay, queueErr := s.requeueModelCoolingTask(ctx, task, err)
			if queueErr != nil {
				return Result{}, queueErr
			}
			return Result{Task: queued, Output: queued.Result}, &TaskQueuedError{Delay: delay}
		}
		if store.ModelCandidateRetryAfter(err) > 0 {
			s.observeCandidateRouting("cooldown_skipped")
			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
		}
		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, 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_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
		}
	}
	if task.RoutingPlatformModelID != "" {
		candidates = pinCandidatesToRoutingAssignment(candidates, task.RoutingPlatformID, task.RoutingPlatformModelID)
		if len(candidates) == 0 {
			err = &clients.ClientError{Code: "routing_candidate_unavailable", Message: "the routed upstream candidate is no longer available", Retryable: true}
			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
		}
	}
	for _, candidate := range candidates {
		if candidate.LoadAvoided {
			s.observeCandidateRouting("full_avoided")
		}
	}
	var asyncAdmission *store.TaskAdmission
	if distributedAdmission && task.AsyncMode {
		asyncAdmission, err = s.loadAsyncTaskAdmission(ctx, task)
		if err != nil {
			return Result{}, err
		}
		var pinned bool
		candidates, pinned = pinCandidatesToTaskAdmission(candidates, asyncAdmission)
		if pinned {
			s.observeTaskAdmission("candidate_pinned")
		}
	}
	pricingByCandidate := map[string]resolvedPricing{}
	preprocessingByCandidate := map[string]parameterPreprocessResult{}
	reservationBillings := []any(nil)
	reservationPricingSnapshot := map[string]any(nil)
	if isBillableRunMode(task.RunMode) {
		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,
			}
		}
	}
	var admittedLeases []store.ConcurrencyLease
	admittedPlatformModelID := ""
	if distributedAdmission && len(candidates) > 0 {
		admissionScopes, groupID, scopeErr := s.taskAdmissionScopes(ctx, task, user, candidates[0])
		if scopeErr != nil {
			return Result{}, scopeErr
		}
		hasConcurrentLimit := false
		for _, scope := range admissionScopes {
			if scope.ConcurrentLimit > 0 {
				hasConcurrentLimit = true
				break
			}
		}
		deferInitialAdmission := false
		reservations := acceptanceInfrastructureReservations(
			task,
			s.rateLimitReservations(ctx, user, candidates[0], body),
		)
		if err := s.store.CheckRateLimits(ctx, reservations); err != nil {
			if _, platformLimited := platformModelRateLimitError(err); platformLimited && len(candidates) > 1 {
				if task.AsyncMode {
					if deleteErr := s.store.DeleteTaskAdmission(context.WithoutCancel(ctx), task.ID); deleteErr != nil {
						return Result{}, deleteErr
					}
					asyncAdmission = nil
				}
				deferInitialAdmission = true
			} else if task.AsyncMode && errors.Is(err, store.ErrRateLimited) && store.RateLimitRetryable(err) {
				queued, delay, queueErr := s.requeueRateLimitedTask(ctx, task, err, candidates[0])
				if queueErr != nil {
					return Result{}, queueErr
				}
				return Result{Task: queued, Output: queued.Result}, &TaskQueuedError{Delay: delay}
			} else {
				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 hasConcurrentLimit && !deferInitialAdmission {
			plan := taskAdmissionPlan{
				Candidate: candidates[0],
				Body:      body,
				ModelType: modelType,
				GroupID:   groupID,
				Scopes:    admissionScopes,
				Eligible:  true,
			}
			var admissionResult store.TaskAdmissionResult
			var admissionErr error
			if task.AsyncMode {
				var alreadyAdmitted bool
				admissionResult, alreadyAdmitted, admissionErr = s.activeAsyncTaskAdmission(ctx, task, plan, asyncAdmission)
				if admissionErr == nil && !alreadyAdmitted {
					admissionResult, admissionErr = s.tryTaskAdmission(ctx, task, plan, "")
				}
				if admissionErr == nil && !admissionResult.Admitted {
					_ = s.store.ReleaseTaskPreparation(context.WithoutCancel(ctx), task.ID, task.ExecutionToken)
					return Result{Task: task, Output: task.Result}, &TaskQueuedError{Delay: time.Second}
				}
			} else {
				admissionResult, admissionErr = s.waitForSynchronousAdmission(ctx, task, plan)
			}
			if admissionErr != nil {
				if requestCancelledBeforeSubmission(ctx) {
					cancelled, changed, cancelErr := s.store.CancelTaskBeforeUpstreamSubmission(
						context.WithoutCancel(ctx),
						task.ID,
						task.ExecutionToken,
						"client disconnected before upstream submission",
					)
					if cancelErr != nil {
						return Result{}, cancelErr
					}
					if changed {
						return Result{Task: cancelled, Output: cancelled.Result}, admissionErr
					}
				}
				code := clients.ErrorCode(admissionErr)
				if errors.Is(admissionErr, context.Canceled) {
					code = "client_disconnected"
				}
				failed, finishErr := s.failTask(context.WithoutCancel(ctx), task.ID, task.ExecutionToken, code, admissionErr.Error(), task.RunMode == "simulation", admissionErr)
				if finishErr != nil && !errors.Is(finishErr, store.ErrTaskExecutionLeaseLost) {
					return Result{}, finishErr
				}
				return Result{Task: failed, Output: failed.Result}, admissionErr
			}
			admittedLeases = admissionResult.Leases
			if len(admittedLeases) == 0 {
				admittedLeases, err = s.store.ActiveTaskAdmissionLeases(ctx, task.ID)
				if err != nil {
					return Result{}, err
				}
			}
			admittedPlatformModelID = admissionResult.Admission.PlatformModelID
		}
		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 admitted and started", map[string]any{"modelType": modelType}, task.RunMode == "simulation"); err != nil {
				return Result{}, err
			}
		}
	}
	admissionFinalized := false
	if distributedAdmission {
		defer func() {
			if !admissionFinalized {
				_ = s.store.DeleteTaskAdmission(context.WithoutCancel(ctx), task.ID)
			}
		}()
	}
	firstCandidateBody := body
	normalizedModelType := modelType
	attemptNo := task.AttemptCount
	var firstPreprocessing parameterPreprocessingLog
	var walletReservations []store.WalletBillingReservation
	walletReservationFinalized := false
	retainWalletReservationForQueue := func() {
		// A queued task still owns its billing hold. Keeping the same active
		// reservation lets the next Worker claim reuse its idempotency key
		// instead of emitting another reserve/release pair.
		walletReservationFinalized = true
	}
	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
		}
		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)
	singleSourceProtection := runnerSingleSourceProtectionEnabled(runnerPolicy)
	var downstreamStarted atomic.Bool
	candidateDelta := onDelta
	if onDelta != nil {
		candidateDelta = func(event clients.StreamDeltaEvent) error {
			downstreamStarted.Store(true)
			return onDelta(event)
		}
	}
	var lastErr error
	var lastCandidate store.RuntimeModelCandidate
	var lastPreprocessing *parameterPreprocessingLog
	platformsVisited := 0
candidatesLoop:
	for index, candidate := range candidates {
		if platformsVisited >= maxPlatforms {
			break
		}
		available, unavailableReason, availabilityErr := s.store.RuntimeCandidateAvailability(ctx, candidate.PlatformID, candidate.PlatformModelID)
		if availabilityErr != nil {
			return Result{}, availabilityErr
		}
		if !available {
			s.observeCandidateRouting(unavailableReason)
			continue
		}
		if candidate.SelectionReason == "normal_rotation" || candidate.SelectionReason == "" {
			s.observeCandidateRouting("normal_rotation")
		}
		if distributedAdmission && candidate.PlatformModelID != admittedPlatformModelID {
			admissionResult, candidateLimited, admissionErr := s.ensureCandidateAdmission(ctx, task, user, body, candidate)
			if admissionErr != nil {
				lastErr = admissionErr
				lastCandidate = candidate
				if errors.Is(admissionErr, store.ErrQueueTimeout) {
					break
				}
				if _, platformLimited := platformModelRateLimitError(admissionErr); platformLimited {
					s.observeCandidateRouting("quota_race_rotated")
					continue
				}
				if task.AsyncMode && errors.Is(admissionErr, store.ErrRateLimited) && store.RateLimitRetryable(admissionErr) {
					queued, delay, queueErr := s.requeueRateLimitedTask(ctx, task, admissionErr, candidate)
					if queueErr != nil {
						return Result{}, queueErr
					}
					retainWalletReservationForQueue()
					return Result{Task: queued, Output: queued.Result}, &TaskQueuedError{Delay: delay}
				}
				return Result{}, admissionErr
			}
			if candidateLimited {
				admittedPlatformModelID = admissionResult.Admission.PlatformModelID
				admittedLeases = admissionResult.Leases
				if len(admittedLeases) == 0 {
					admittedLeases, err = s.store.ActiveTaskAdmissionLeases(ctx, task.ID)
					if err != nil {
						return Result{}, err
					}
				}
			} else {
				admittedPlatformModelID = ""
				admittedLeases = nil
			}
		}
		platformsVisited++
		lastCandidate = candidate
		clientAttempts := clientAttemptsForCandidate(candidate)
		var candidateErr error
		var candidateDecision failureDecision
		var candidateDecisionAttempt int
		var candidateClientAttempt int
		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, candidateDelta, responseExecution, runnerPolicy.CacheAffinityPolicy, cacheAffinityKeys.Record, admittedPlatformModelID, admittedLeases)
			if err != nil && requestCancelledBeforeSubmission(ctx) {
				cancelled, changed, cancelErr := s.store.CancelTaskBeforeUpstreamSubmission(
					context.WithoutCancel(ctx),
					task.ID,
					task.ExecutionToken,
					"client disconnected before upstream submission",
				)
				if cancelErr != nil {
					return Result{}, cancelErr
				}
				if changed {
					return Result{Task: cancelled, Output: cancelled.Result}, err
				}
			}
			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
				response.Result = canonicalTaskResult(response.Result)
				var billings []any
				finalAmount := fixedAmount(0)
				finalAmountText := ""
				pricingSnapshot := map[string]any{"pricingVersion": pricingVersionV2, "source": "simulation"}
				if isBillableRunMode(task.RunMode) {
					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")
						output, hydrateErr := s.taskResultForSynchronousResponse(ctx, task, response.Result)
						if hydrateErr != nil {
							return Result{Task: review}, hydrateErr
						}
						return Result{Task: review, Output: output, 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)
				callbackURL, callbackErr := s.taskCallbackURL(ctx, task.ID)
				if callbackErr != nil {
					return Result{}, callbackErr
				}
				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,
					CompletionCallbackURL: callbackURL,
					Simulated:             isSimulation(task, candidate),
					FinalizeAdmission:     distributedAdmission,
				})
				if finishErr != nil {
					if errors.Is(finishErr, store.ErrTaskResultBinaryNotMaterialized) {
						s.logger.Error("task result rejected by binary persistence guard",
							"taskID", task.ID,
							"error_category", "result_binary_not_materialized",
						)
						failureCtx := context.WithoutCancel(ctx)
						_ = s.store.FinishTaskAttempt(failureCtx, store.FinishTaskAttemptInput{
							AttemptID:          response.AttemptID,
							Status:             "failed",
							Retryable:          false,
							RequestID:          response.RequestID,
							ResponseStartedAt:  response.ResponseStartedAt,
							ResponseFinishedAt: response.ResponseFinishedAt,
							ResponseDurationMS: response.ResponseDurationMS,
							ErrorCode:          clients.ErrorCode(finishErr),
							ErrorMessage:       finishErr.Error(),
						})
						failed, failErr := s.failTask(
							failureCtx,
							task.ID,
							task.ExecutionToken,
							clients.ErrorCode(finishErr),
							finishErr.Error(),
							isSimulation(task, candidate),
							finishErr,
						)
						if failErr != nil {
							return Result{}, failErr
						}
						walletReservationFinalized = true
						return Result{Task: failed, Output: failed.Result}, finishErr
					}
					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
				admissionFinalized = distributedAdmission
				if task.AsyncMode {
					// Async callers hydrate the canonical stored result through
					// the task detail endpoint. The River Worker does not consume
					// Result.Output, so downloading and rebuilding Base64 here
					// only retains the execution slot and media memory a second
					// time after the task is already durably complete.
					return Result{Task: finished, Output: response.Result}, nil
				}
				output, hydrateErr := s.taskResultForSynchronousResponse(ctx, task, response.Result)
				if hydrateErr != nil {
					return Result{Task: finished}, hydrateErr
				}
				return Result{Task: finished, Output: output, Wire: response.Wire}, nil
			}
			var confirmationPending *submissionConfirmationPendingError
			if errors.As(err, &confirmationPending) {
				_ = s.store.SetTaskSubmissionState(context.WithoutCancel(ctx), task.ID, task.ExecutionToken, "submission_confirmation_pending")
				review, reviewErr := s.store.FinishTaskManualReview(context.WithoutCancel(ctx), store.FinishTaskManualReviewInput{
					TaskID: task.ID, ExecutionToken: task.ExecutionToken, AttemptID: confirmationPending.AttemptID, TaskStatus: "failed",
					Code: "upstream_timeout", Message: confirmationPending.Error(),
					PricingSnapshot:    candidatePricing.Snapshot,
					RequestFingerprint: pricingRequestFingerprint(task.Kind, task.Model, candidateBody),
				})
				if reviewErr != nil {
					return Result{}, reviewErr
				}
				walletReservationFinalized = true
				s.logger.Warn("upstream submission confirmation pending", "taskID", task.ID, "attemptID", confirmationPending.AttemptID, "error_category", "upstream_timeout")
				return Result{Task: review, Output: review.Result}, confirmationPending
			}
			if isLocalRateLimitError(err) {
				lastErr = err
				candidateErr = err
				candidateDecision = resolveCandidateFailure(resolveCandidateFailureInput{
					RunnerPolicy:      runnerPolicy,
					Candidate:         candidate,
					Err:               err,
					ClientAttempt:     clientAttempt,
					MaxClientAttempts: clientAttempts,
					HasNextCandidate:  platformsVisited < maxPlatforms && index+1 < len(candidates),
					FailoverExpired:   failoverTimeBudgetExceeded(executeStartedAt, maxFailoverDuration),
					Async:             task.AsyncMode,
					DownstreamStarted: downstreamStarted.Load(),
					TaskKind:          task.Kind,
				})
				if candidateDecision.Route == "requeue" {
					if _, platformLimited := platformModelRateLimitError(err); platformLimited {
						s.observeCandidateRouting("all_full_queued")
					}
					queued, delay, queueErr := s.requeueRateLimitedTask(ctx, task, err, candidate)
					if queueErr != nil {
						return Result{}, queueErr
					}
					retainWalletReservationForQueue()
					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),
						{"selectionReason": candidateDecision.Reason},
					},
					ModelType: candidate.ModelType,
				})
				candidateDecisionAttempt = attemptNo
				candidateClientAttempt = clientAttempt
				s.recordAttemptTrace(ctx, task.ID, attemptNo, failureDecisionTraceEntry(candidateDecision, candidate, clientAttempt, clientAttempts, false, ""))
				if candidateDecision.Route == "next" {
					s.observeCandidateRouting("quota_race_rotated")
					if admittedPlatformModelID == candidate.PlatformModelID {
						if deleteErr := s.store.DeleteTaskAdmission(context.WithoutCancel(ctx), task.ID); deleteErr != nil {
							return Result{}, deleteErr
						}
						admittedPlatformModelID = ""
						admittedLeases = nil
					}
					break
				}
				break candidatesLoop
			}
			attemptNo = nextAttemptNo
			lastErr = err
			candidateErr = err
			candidateDecision = resolveCandidateFailure(resolveCandidateFailureInput{
				RunnerPolicy:      runnerPolicy,
				Candidate:         candidate,
				Err:               err,
				ClientAttempt:     clientAttempt,
				MaxClientAttempts: clientAttempts,
				HasNextCandidate:  platformsVisited < maxPlatforms && index+1 < len(candidates),
				FailoverExpired:   failoverTimeBudgetExceeded(executeStartedAt, maxFailoverDuration),
				Async:             task.AsyncMode,
				DownstreamStarted: downstreamStarted.Load(),
				TaskKind:          task.Kind,
			})
			candidateDecisionAttempt = attemptNo
			candidateClientAttempt = clientAttempt
			if candidateDecision.Route != "retry_same" {
				break
			}
			s.recordAttemptTrace(ctx, task.ID, attemptNo, failureDecisionTraceEntry(candidateDecision, candidate, clientAttempt, clientAttempts, false, ""))
			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":        candidateDecision.Reason,
				"scope":         "same_client",
			}, candidateDecision.Match, candidateDecision.Info), isSimulation(task, candidate)); err != nil {
				return Result{}, err
			}
		}
		if candidateErr == nil {
			break
		}
		effectResult, effectErr := s.applyFailureDecision(
			context.WithoutCancel(ctx),
			task.ID,
			task.Model,
			candidate,
			candidateDecision,
			singleSourceProtection,
			isSimulation(task, candidate),
		)
		if effectErr != nil {
			s.logger.Warn("apply candidate failure effect failed", "taskID", task.ID, "effect", candidateDecision.Effect, "error", effectErr)
			effectResult.GuardReason = "effect_error"
		}
		s.recordAttemptTrace(ctx, task.ID, candidateDecisionAttempt, failureDecisionTraceEntry(
			candidateDecision,
			candidate,
			candidateClientAttempt,
			clientAttempts,
			effectResult.Applied,
			effectResult.GuardReason,
		))
		if candidateDecision.Route != "next" {
			if candidateDecision.Reason == "failover_time_budget_exceeded" {
				elapsedSeconds := int(time.Since(executeStartedAt).Seconds())
				if err := s.emit(ctx, task.ID, "task.failover.stopped", "running", "retry", 0.55, "failover time budget exceeded", map[string]any{
					"elapsedSeconds":     elapsedSeconds,
					"maxDurationSeconds": int(maxFailoverDuration.Seconds()),
					"scope":              "next_platform",
					"statusCode":         clients.ErrorResponseMetadata(candidateErr).StatusCode,
				}, isSimulation(task, candidate)); err != nil {
					return Result{}, err
				}
			}
			break
		}
		if err := s.emit(ctx, task.ID, "task.retrying", "running", "retry", 0.55, "retrying next client", addPolicyTracePayload(map[string]any{
			"attempt": attemptNo,
			"route":   candidateDecision.Route,
			"effect":  candidateDecision.Effect,
			"error":   candidateErr.Error(),
			"reason":  candidateDecision.Reason,
			"scope":   "next_platform",
		}, candidateDecision.Match, candidateDecision.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
		}
		retainWalletReservationForQueue()
		return Result{Task: queued, Output: queued.Result}, &TaskQueuedError{Delay: 0}
	}
	if task.AsyncMode && errors.Is(lastErr, store.ErrRateLimited) && store.RateLimitRetryable(lastErr) {
		if _, platformLimited := platformModelRateLimitError(lastErr); platformLimited {
			s.observeCandidateRouting("all_full_queued")
		}
		queued, delay, queueErr := s.requeueRateLimitedTask(ctx, task, lastErr, lastCandidate)
		if queueErr != nil {
			return Result{}, queueErr
		}
		retainWalletReservationForQueue()
		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 (s *Service) taskResultForSynchronousResponse(ctx context.Context, task store.GatewayTask, result map[string]any) (map[string]any, error) {
	if taskRequestsInlineMediaResult(task.Request) {
		if rawBytes := SynchronousInlineResultBytes(result); rawBytes > MaxSynchronousInlineResponseBytes {
			return nil, &clients.ClientError{
				Code:       "response_format_too_large",
				Message:    "synchronous Base64 response exceeds the 20 MiB limit",
				StatusCode: http.StatusRequestEntityTooLarge,
				Retryable:  false,
				Details: map[string]any{
					"task_id":         task.ID,
					"query_url":       "/api/v1/ai/result/" + task.ID,
					"limit_bytes":     MaxSynchronousInlineResponseBytes,
					"actual_bytes":    rawBytes,
					"response_format": "url",
				},
			}
		}
		return s.HydrateTaskResult(ctx, task.ID, result)
	}
	return s.ProjectTaskResultURLs(ctx, task.ID, result)
}

func taskRequestsInlineMediaResult(request map[string]any) bool {
	value, _ := request["response_format"].(string)
	return strings.EqualFold(strings.TrimSpace(value), "b64_json")
}

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,
	cacheAffinityPolicy map[string]any,
	cacheAffinityRecordKeys []string,
	admittedPlatformModelID string,
	admittedLeases []store.ConcurrencyLease,
) (clients.Response, error) {
	var err error
	if task.RunMode == "acceptance" {
		candidate, err = s.acceptanceCandidate(ctx, task, candidate)
		if err != nil {
			return clients.Response{}, err
		}
	} else {
		candidate, err = candidateWithEnvironmentCredentials(candidate)
		if err != nil {
			return clients.Response{}, err
		}
	}
	simulated := isSimulation(task, candidate)
	baseAttemptMetrics := mergeMetrics(attemptMetrics(candidate, attemptNo, simulated), parameterPreprocessingMetrics(preprocessing))
	reservations := acceptanceInfrastructureReservations(task, s.rateLimitReservations(ctx, user, candidate, body))
	if admittedPlatformModelID == candidate.PlatformModelID && len(admittedLeases) > 0 {
		filtered := make([]store.RateLimitReservation, 0, len(reservations))
		for _, reservation := range reservations {
			if reservation.Metric != "concurrent" {
				filtered = append(filtered, reservation)
			}
		}
		reservations = filtered
	}
	limitResult, err := s.store.ReserveRateLimits(ctx, task.ID, "", reservations)
	if err != nil {
		var limitErr *store.RateLimitExceededError
		if errors.As(err, &limitErr) {
			s.observeProviderQuotaWait(limitErr.Metric)
		}
		retryable := store.RateLimitRetryable(err)
		clientErr := &clients.ClientError{Code: clients.ErrorCode(err), Message: err.Error(), Retryable: retryable}
		return clients.Response{}, &localRateLimitError{clientErr: clientErr, cause: err, retryAfter: localRateLimitRetryAfter(err)}
	}
	attemptOwnedLeases := append([]store.ConcurrencyLease(nil), limitResult.Leases...)
	if admittedPlatformModelID == candidate.PlatformModelID {
		limitResult.Leases, attemptOwnedLeases = bindAttemptConcurrencyLeases(limitResult.Leases, admittedLeases)
	}
	rateReservationsFinalized := false
	defer func() {
		if !rateReservationsFinalized {
			_ = s.store.ReleaseRateLimitReservations(context.WithoutCancel(ctx), limitResult.Reservations, "attempt_failed")
		}
	}()
	defer func() {
		_ = s.store.ReleaseConcurrencyLeases(context.WithoutCancel(ctx), attemptOwnedLeases)
	}()

	attemptID, err := s.store.CreateTaskAttempt(ctx, store.CreateTaskAttemptInput{
		TaskID:             task.ID,
		ExecutionToken:     task.ExecutionToken,
		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, task.Kind)
	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
	}
	mediaSlotPreacquired := mediaTaskNeedsPreUpstreamMaterializationSlot(candidate.ModelType)
	releaseMediaSlot := func() {}
	if mediaSlotPreacquired {
		releaseMediaSlot, err = s.acquireMediaMaterializationSlot(ctx)
		if err != nil {
			return clients.Response{}, err
		}
		defer releaseMediaSlot()
	}
	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
	}
	if strings.TrimSpace(task.RemoteTaskID) == "" {
		providerBody, err = s.normalizeVideoInputImages(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"
		markUpstreamSubmissionStarted(ctx)
		if err := s.store.SetAttemptUpstreamSubmissionStatusForExecution(
			ctx,
			attemptID,
			task.ID,
			task.ExecutionToken,
			submissionStatus,
		); err != nil {
			return clients.Response{}, fmt.Errorf("restore upstream submission status: %w", err)
		}
		if err := enterWorkerWaiting(ctx); err != nil {
			return clients.Response{}, err
		}
	}
	setSubmissionStatus := func(status string) error {
		if submissionStatus == "response_received" && status != "response_received" {
			return nil
		}
		if submissionStatus == status {
			return nil
		}
		if err := s.store.SetAttemptUpstreamSubmissionStatusForExecution(
			context.WithoutCancel(ctx),
			attemptID,
			task.ID,
			task.ExecutionToken,
			status,
		); err != nil {
			return err
		}
		submissionStatus = status
		return nil
	}
	var submissionWire *clients.WireResponse
	var routeObserver *passiveRouteObserver
	if task.RouteProfileKey != "" {
		routeObserver = &passiveRouteObserver{}
		requestHTTPClient = routeObserver.wrap(requestHTTPClient)
	}
	runCtx, stopLeaseRenewal := s.startConcurrencyLeaseRenewal(ctx, task.ID, limitResult.Leases)
	response, err := client.Run(runCtx, clients.Request{
		Kind:                   task.Kind,
		ModelType:              candidate.ModelType,
		Model:                  task.Model,
		Body:                   providerBody,
		OriginalBody:           preprocessing.Input,
		Candidate:              candidate,
		HTTPClient:             requestHTTPClient,
		RemoteTaskID:           task.RemoteTaskID,
		RemoteTaskPayload:      task.RemoteTaskPayload,
		UpstreamIdempotencyKey: task.ID,
		OnRemoteTaskSubmitted: func(remoteTaskID string, payload map[string]any) error {
			if strings.TrimSpace(remoteTaskID) == "" {
				return nil
			}
			checkpoint := minimalRemoteTaskCheckpoint(candidate.Provider, candidate.SpecType, payload)
			if err := s.store.SetTaskRemoteTask(context.WithoutCancel(ctx), task.ID, task.ExecutionToken, attemptID, remoteTaskID, checkpoint); err != nil {
				return err
			}
			task.RemoteTaskID = remoteTaskID
			task.RemoteTaskPayload = checkpoint
			if err := s.persistCompatibilitySubmission(context.WithoutCancel(ctx), task, candidate, remoteTaskID, checkpoint, submissionWire); err != nil {
				return err
			}
			if err := setSubmissionStatus("response_received"); err != nil {
				return err
			}
			if err := s.store.SetTaskSubmissionState(context.WithoutCancel(ctx), task.ID, task.ExecutionToken, "submitted"); err != nil {
				return err
			}
			return enterWorkerWaiting(ctx)
		},
		OnRemoteTaskPolled: func(remoteTaskID string, payload map[string]any) error {
			if strings.TrimSpace(remoteTaskID) == "" {
				return nil
			}
			checkpoint := minimalRemoteTaskCheckpoint(candidate.Provider, candidate.SpecType, payload)
			if err := s.store.SetTaskRemoteTask(context.WithoutCancel(ctx), task.ID, task.ExecutionToken, attemptID, remoteTaskID, checkpoint); err != nil {
				return err
			}
			task.RemoteTaskID = remoteTaskID
			task.RemoteTaskPayload = checkpoint
			if err := setSubmissionStatus("response_received"); err != nil {
				return err
			}
			return enterWorkerWaiting(ctx)
		},
		OnUpstreamSubmissionStarted: func() error {
			if err := setSubmissionStatus("submitting"); err != nil {
				return err
			}
			if err := s.store.SetTaskSubmissionState(context.WithoutCancel(ctx), task.ID, task.ExecutionToken, "submitting"); err != nil {
				return err
			}
			markUpstreamSubmissionStarted(ctx)
			return enterWorkerWaiting(ctx)
		},
		OnUpstreamResponseReceived: func() error {
			submissionStatus = "response_received"
			if err := s.store.SetTaskSubmissionState(context.WithoutCancel(ctx), task.ID, task.ExecutionToken, "submitted"); err != nil {
				return err
			}
			return enterWorkerFinalizing(ctx)
		},
		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,
	})
	s.recordPassiveRouteObservation(task, routeObserver)
	if phaseErr := enterWorkerFinalizing(runCtx); err == nil && phaseErr != nil {
		err = phaseErr
	}
	if leaseErr := stopLeaseRenewal(); leaseErr != nil {
		err = &clients.ClientError{
			Code:      "concurrency_lease_lost",
			Message:   leaseErr.Error(),
			Retryable: true,
		}
	}
	callFinishedAt := time.Now()
	if err == nil {
		submissionStatus = "response_received"
	}
	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 && clients.ErrorCode(err) == "gemini_image_output_missing" && s.logger != nil {
		meta := clients.ErrorResponseMetadata(err)
		s.logger.Warn("gemini image response contains no extractable resource",
			"taskID", task.ID,
			"attempt", attemptNo,
			"platformID", candidate.PlatformID,
			"platformModelID", candidate.PlatformModelID,
			"clientID", candidate.ClientID,
			"requestID", meta.RequestID,
			"error_category", clients.ErrorCode(err),
			"reason", err.Error(),
			"details", clients.ErrorDetails(err),
		)
	}
	if err != nil {
		if clients.ErrorResponseMetadata(err).StatusCode > 0 && submissionStatus != "response_received" {
			submissionStatus = "response_received"
		}
		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,
			UpstreamSubmissionStatus: submissionStatus,
			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 shouldEnterSubmissionConfirmationPending(simulated, submissionStatus, err) {
			return clients.Response{}, &submissionConfirmationPendingError{AttemptID: attemptID, Cause: err}
		}
		return clients.Response{}, err
	}
	if task.AsyncMode {
		// Async callers consume the canonical task result through the task API,
		// never the original provider wire response. Release the raw JSON before
		// decoding/uploading inline media so a large Base64 response is not held
		// twice during the most memory-intensive phase.
		response.Wire = nil
		submissionWire = nil
	}
	if !mediaSlotPreacquired {
		releaseMediaSlot, err = s.acquireMediaResultSlot(ctx, response.Result)
		if err != nil {
			return clients.Response{}, err
		}
		defer releaseMediaSlot()
	}
	uploadedResult, err := s.uploadGeneratedAssets(ctx, task.ID, task.Kind, task.AcceptanceRunID, 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),
			UpstreamSubmissionStatus: submissionStatus,
			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 localBinaryResultHasPlaceholders(response.Result) {
		// A provider wire response can still contain its original Base64 body.
		// Force protocol responses to be rebuilt from the verified standard
		// result instead of bypassing local result materialization.
		response.Wire = nil
	}
	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
}

// bindAttemptConcurrencyLeases combines task-scoped admission leases with
// leases acquired by the current attempt. Only the latter are returned for
// attempt cleanup: admission owns its leases across retry_same and releases
// them when the task changes binding or reaches a terminal state.
func bindAttemptConcurrencyLeases(attemptLeases []store.ConcurrencyLease, admissionLeases []store.ConcurrencyLease) ([]store.ConcurrencyLease, []store.ConcurrencyLease) {
	attemptOwned := append([]store.ConcurrencyLease(nil), attemptLeases...)
	active := append([]store.ConcurrencyLease(nil), attemptLeases...)
	active = append(active, admissionLeases...)
	return active, attemptOwned
}

func (s *Service) observeProviderQuotaWait(metric string) {
	observer, ok := s.billingMetrics.(interface {
		ObserveProviderQuotaWait(string)
	})
	if ok {
		observer.ObserveProviderQuotaWait(metric)
	}
}

func (s *Service) observeCandidateRouting(reason string) {
	observer, ok := s.billingMetrics.(interface {
		ObserveCandidateRouting(string)
	})
	if ok {
		observer.ObserveCandidateRouting(reason)
	}
}

func minimalRemoteTaskCheckpoint(provider string, specType string, payload map[string]any) map[string]any {
	const maxBytes = 8192
	provider = strings.ToLower(strings.TrimSpace(provider))
	specType = strings.ToLower(strings.TrimSpace(specType))
	allowedByProvider := map[string][]string{
		"keling":     {"endpoint", "pollEndpoint", "mode", "taskType", "cleanupElementIds"},
		"kling":      {"endpoint", "pollEndpoint", "mode", "taskType", "cleanupElementIds"},
		"topaz":      {"phase", "targetResolution"},
		"vectorizer": {"imageToken", "receipt"},
	}
	allowedKeys := allowedByProvider[provider]
	if len(allowedKeys) == 0 {
		allowedKeys = allowedByProvider[specType]
	}
	checkpoint := map[string]any{}
	for _, key := range allowedKeys {
		value, ok := payload[key]
		if !ok || value == nil {
			continue
		}
		if text, ok := value.(string); ok {
			value = truncateText(text, 4096)
		}
		next := cloneMap(checkpoint)
		next[key] = value
		encoded, err := json.Marshal(next)
		if err != nil || len(encoded) > maxBytes {
			continue
		}
		checkpoint = next
	}
	return checkpoint
}

func canonicalTaskResult(result map[string]any) map[string]any {
	canonical := cloneMap(result)
	for _, key := range []string{
		"raw",
		"raw_data",
		"rawData",
		"provider_response",
		"providerResponse",
		"submit",
		"file_retrieve",
		"fileRetrieve",
		"upstream_task_id",
		"remote_task_id",
	} {
		delete(canonical, key)
	}
	return canonical
}

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)
	}
	return s.store.SetTaskCompatibilitySubmission(ctx, task.ID, store.CompatibilitySubmission{
		TargetProtocol: targetProtocol,
		SourceProtocol: sourceProtocol,
	})
}

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":
		if kind == "videos.generations" && strings.Contains(strings.ToLower(strings.TrimSpace(firstNonEmptyString(candidate.ProviderModelName, candidate.ModelName))), "veo") {
			return clients.ProtocolGeminiVeo
		}
		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 !log.Changed {
		return nil
	}
	changes := minimalParameterPreprocessingChanges(log.Changes)
	_, 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:         true,
		ChangeCount:     len(log.Changes),
		Changes:         changes,
	})
	return err
}

func minimalParameterPreprocessingChanges(changes []parameterPreprocessChange) []any {
	const maxBytes = 1024
	out := make([]any, 0, len(changes))
	for _, change := range changes {
		item := map[string]any{
			"path":   truncateText(change.Path, 256),
			"action": truncateText(change.Action, 64),
		}
		next := append(append([]any(nil), out...), item)
		encoded, err := json.Marshal(next)
		if err != nil || len(encoded) > maxBytes {
			break
		}
		out = next
	}
	return out
}

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 truncateText(value string, maxRunes int) string {
	value = strings.TrimSpace(value)
	runes := []rune(value)
	if len(runes) <= maxRunes {
		return value
	}
	return string(runes[:maxRunes])
}

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)
	callbackURL, callbackErr := s.taskCallbackURL(ctx, taskID)
	if callbackErr != nil {
		return store.GatewayTask{}, callbackErr
	}
	failed, err := s.store.FinishTaskFailure(ctx, store.FinishTaskFailureInput{
		TaskID:                taskID,
		ExecutionToken:        executionToken,
		Code:                  code,
		Message:               message,
		StatusCode:            clients.ErrorResponseMetadata(cause).StatusCode,
		Result:                buildFailureResult(code, message, requestID, cause),
		RequestID:             requestID,
		Metrics:               metrics,
		ResponseStartedAt:     responseStartedAt,
		ResponseFinishedAt:    responseFinishedAt,
		ResponseDurationMS:    responseDurationMS,
		CompletionCallbackURL: callbackURL,
		Simulated:             simulated,
		FinalizeAdmission:     true,
	})
	if err != nil {
		return store.GatewayTask{}, err
	}
	if failed.Status == "cancelled" {
		return failed, nil
	}
	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,
		ExecutionToken:  input.Task.ExecutionToken,
		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
	}
	delay += taskRetryJitter(task.ID)
	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) requeueModelCoolingTask(ctx context.Context, task store.GatewayTask, cause error) (store.GatewayTask, time.Duration, error) {
	delay := store.ModelCandidateRetryAfter(cause)
	recoveryAt := store.ModelCandidateRecoveryAt(cause)
	if recoveryAt.IsZero() {
		recoveryAt = time.Now().Add(delay)
	}
	if delay <= 0 {
		delay = time.Until(recoveryAt)
	}
	if delay < time.Second {
		delay = time.Second
		recoveryAt = time.Now().Add(delay)
	}
	queued, err := s.store.RequeueTaskUntil(ctx, task.ID, task.ExecutionToken, recoveryAt, "")
	if err != nil {
		return store.GatewayTask{}, 0, err
	}
	payload := map[string]any{
		"code":              store.ModelCandidateErrorCode(cause),
		"retryAfterSeconds": int((delay + time.Second - 1) / time.Second),
		"recoveryAt":        recoveryAt.UTC().Format(time.RFC3339),
	}
	if eventErr := s.emit(ctx, task.ID, "task.queued", "queued", "model_cooling_down", 0.2, "task queued until a model candidate recovers", payload, task.RunMode == "simulation"); eventErr != nil {
		return store.GatewayTask{}, 0, eventErr
	}
	return queued, delay, nil
}

func taskRetryJitter(taskID string) time.Duration {
	var sum uint32
	for _, value := range []byte(taskID) {
		sum = sum*33 + uint32(value)
	}
	return time.Duration(sum%251) * time.Millisecond
}

func (s *Service) startConcurrencyLeaseRenewal(ctx context.Context, taskID string, leases []store.ConcurrencyLease) (context.Context, func() error) {
	if len(leases) == 0 {
		return ctx, func() error { return nil }
	}
	interval := 30 * time.Second
	for _, lease := range leases {
		ttl := lease.TTL
		if ttl <= 0 {
			ttl = 120 * time.Second
		}
		candidate := ttl / 3
		if candidate < time.Second {
			candidate = time.Second
		}
		if candidate < interval {
			interval = candidate
		}
	}
	runCtx, cancelRun := context.WithCancel(ctx)
	renewCtx, cancelRenew := context.WithCancel(ctx)
	done := make(chan error, 1)
	minimumTTL := 120 * time.Second
	for _, lease := range leases {
		if lease.TTL > 0 && lease.TTL < minimumTTL {
			minimumTTL = lease.TTL
		}
	}
	go func() {
		timer := time.NewTimer(interval)
		defer timer.Stop()
		leaseDeadline := time.Now().Add(minimumTTL)
		for {
			select {
			case <-renewCtx.Done():
				done <- nil
				return
			case <-timer.C:
				if err := s.coordinationStore.RenewConcurrencyLeases(renewCtx, leases); err != nil {
					if renewCtx.Err() != nil {
						done <- nil
						return
					}
					if concurrencyLeaseRenewalIsFatal(err, time.Now(), leaseDeadline) {
						outcome := "failure"
						if errors.Is(err, store.ErrConcurrencyLeaseLost) {
							outcome = "lost"
						}
						s.observeConcurrencyLeaseRenewal(outcome)
						s.logger.Error("concurrency lease renewal failed; cancelling upstream execution",
							"taskID", taskID, "leaseCount", len(leases), "outcome", outcome, "error", err)
						done <- err
						cancelRun()
						return
					}
					s.observeConcurrencyLeaseRenewal("failure")
					s.logger.Warn("concurrency lease renewal failed; retrying before lease expiry",
						"taskID", taskID, "leaseCount", len(leases), "outcome", "transient", "error", err)
					timer.Reset(leaseRenewalRetryInterval)
					continue
				}
				s.observeConcurrencyLeaseRenewal("success")
				leaseDeadline = time.Now().Add(minimumTTL)
				timer.Reset(interval)
			}
		}
	}()
	return runCtx, func() error {
		cancelRenew()
		renewalErr := <-done
		cancelRun()
		return renewalErr
	}
}

func concurrencyLeaseRenewalIsFatal(err error, now time.Time, leaseDeadline time.Time) bool {
	return errors.Is(err, store.ErrConcurrencyLeaseLost) || !now.Before(leaseDeadline)
}

func (s *Service) observeConcurrencyLeaseRenewal(outcome string) {
	observer, ok := s.billingMetrics.(interface {
		ObserveConcurrencyLeaseRenewal(string)
	})
	if ok {
		observer.ObserveConcurrencyLeaseRenewal(outcome)
	}
}

func (s *Service) requeueInterruptedAsyncTask(ctx context.Context, task store.GatewayTask) (store.GatewayTask, error) {
	queued, err := s.store.ResolveInterruptedTaskExecution(ctx, task.ID, task.ExecutionToken)
	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 {
	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 {
	callbackURL, callbackErr := s.taskCallbackURL(ctx, taskID)
	if callbackErr != nil {
		return callbackErr
	}
	event, err := s.store.AddTaskEventWithCallback(
		ctx,
		taskID,
		eventType,
		status,
		phase,
		progress,
		message,
		payload,
		simulated,
		callbackURL,
	)
	if err != nil {
		return err
	}
	if event.ID == "" {
		s.observeTaskEventSkip(event.SkippedReason)
	}
	return nil
}

func (s *Service) taskCallbackURL(ctx context.Context, taskID string) (string, error) {
	acceptanceURL, acceptance, err := s.store.AcceptanceCallbackURL(ctx, taskID)
	if err != nil {
		return "", err
	}
	if acceptance && acceptanceURL != "" {
		return acceptanceURL, nil
	}
	if s.cfg.TaskProgressCallbackEnabled && s.cfg.TaskProgressCallbackURL != "" {
		return s.cfg.TaskProgressCallbackURL, nil
	}
	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_analysis", "image_vectorize",
		"video_generate", "video_enhance", "image_to_video", "text_to_video", "video_edit", "video_reference", "video_first_last_frame", "video_understanding",
		"omni_video", "omni", "audio_generate", "audio_understanding", "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 isBillableRunMode(runMode string) bool {
	return runMode == "production" || runMode == "acceptance" || runMode == "acceptance_canary"
}

func (s *Service) acceptanceCandidate(
	ctx context.Context,
	task store.GatewayTask,
	candidate store.RuntimeModelCandidate,
) (store.RuntimeModelCandidate, error) {
	if strings.TrimSpace(task.AcceptanceRunID) == "" {
		return candidate, &clients.ClientError{
			Code:      "acceptance_run_missing",
			Message:   "acceptance task is missing its run identifier",
			Retryable: false,
		}
	}
	baseURL, credential, err := s.store.AcceptanceCandidateOverride(ctx, task.AcceptanceRunID)
	if err != nil {
		return candidate, &clients.ClientError{
			Code:      "acceptance_run_inactive",
			Message:   err.Error(),
			Retryable: false,
		}
	}
	candidate.BaseURL = baseURL
	candidate.Credentials = map[string]any{"apiKey": credential}
	platformConfig := cloneMap(candidate.PlatformConfig)
	delete(platformConfig, "credentialEnv")
	delete(platformConfig, "credential_env")
	delete(platformConfig, "proxyMode")
	delete(platformConfig, "httpProxy")
	platformConfig["networkProxy"] = map[string]any{"mode": "none"}
	platformConfig["acceptanceRunId"] = task.AcceptanceRunID
	candidate.PlatformConfig = platformConfig
	return candidate, nil
}

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 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 reasoning, ok := body["reasoning"].(map[string]any); ok {
			if err := clients.ValidateOpenAIReasoningEffort(reasoning["effort"]); err != nil {
				return err
			}
		}
		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,
	}
}
