package store

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"time"
	"unicode/utf8"

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

type TaskListFilter struct {
	Query       string
	ModelType   string
	CreatedFrom *time.Time
	CreatedTo   *time.Time
	Page        int
	PageSize    int
}

type TaskListResult struct {
	Items    []GatewayTask
	Total    int
	Page     int
	PageSize int
}

type CompatTaskListFilter struct {
	Provider    string
	Version     string
	Statuses    []string
	CreatedFrom *time.Time
	CreatedTo   *time.Time
	Page        int
	PageSize    int
}

func (s *Store) ListTasks(ctx context.Context, user *auth.User, filter TaskListFilter) (TaskListResult, error) {
	page := filter.Page
	if page <= 0 {
		page = 1
	}
	pageSize := filter.PageSize
	if pageSize <= 0 {
		pageSize = 50
	}
	if pageSize > 100 {
		pageSize = 100
	}
	offset := (page - 1) * pageSize
	gatewayUserID := localGatewayUserID(user)
	apiKeyID := ""
	userID := ""
	if user != nil {
		apiKeyID = strings.TrimSpace(user.APIKeyID)
		userID = strings.TrimSpace(user.ID)
	}
	if gatewayUserID == "" && userID == "" {
		return TaskListResult{}, ErrLocalUserRequired
	}
	queryPattern := ""
	if query := strings.TrimSpace(filter.Query); query != "" {
		queryPattern = "%" + query + "%"
	}
	args := []any{
		gatewayUserID,
		userID,
		apiKeyID,
		queryPattern,
		strings.TrimSpace(filter.ModelType),
		nullableTaskListTime(filter.CreatedFrom),
		nullableTaskListTime(filter.CreatedTo),
	}
	whereSQL := `
WHERE (
    (
      NULLIF($1, '')::uuid IS NOT NULL
      AND gateway_user_id = NULLIF($1, '')::uuid
    )
    OR (
      NULLIF($1, '')::uuid IS NULL
      AND NULLIF($2, '') IS NOT NULL
      AND user_id = $2
    )
  )
  AND (
    NULLIF($3, '') IS NULL
    OR api_key_id = $3
  )
  AND (
    NULLIF($4, '') IS NULL
    OR id::text ILIKE $4
    OR COALESCE(request_id, '') ILIKE $4
    OR kind ILIKE $4
    OR model ILIKE $4
    OR COALESCE(requested_model, '') ILIKE $4
    OR COALESCE(resolved_model, '') ILIKE $4
    OR COALESCE(api_key_id, '') ILIKE $4
    OR COALESCE(api_key_name, '') ILIKE $4
    OR COALESCE(api_key_prefix, '') ILIKE $4
    OR status ILIKE $4
    OR COALESCE(model_type, '') ILIKE $4
  )
  AND (
    NULLIF($5, '') IS NULL
    OR model_type = $5
  )
  AND (
    $6::timestamptz IS NULL
    OR created_at >= $6::timestamptz
  )
  AND (
    $7::timestamptz IS NULL
    OR created_at <= $7::timestamptz
  )`
	var total int
	if err := s.pool.QueryRow(ctx, `SELECT count(*) FROM gateway_tasks `+whereSQL, args...).Scan(&total); err != nil {
		return TaskListResult{}, err
	}
	queryArgs := append(args, pageSize, offset)
	rows, err := s.pool.Query(ctx, `
SELECT `+gatewayTaskColumns+`
FROM gateway_tasks
`+whereSQL+`
ORDER BY created_at DESC
LIMIT $8 OFFSET $9`, queryArgs...)
	if err != nil {
		return TaskListResult{}, err
	}
	defer rows.Close()

	items := make([]GatewayTask, 0)
	for rows.Next() {
		task, err := scanGatewayTask(rows)
		if err != nil {
			return TaskListResult{}, err
		}
		items = append(items, task)
	}
	if err := rows.Err(); err != nil {
		return TaskListResult{}, err
	}
	items, err = s.attachTaskAttempts(ctx, items)
	if err != nil {
		return TaskListResult{}, err
	}
	return TaskListResult{
		Items:    items,
		Total:    total,
		Page:     page,
		PageSize: pageSize,
	}, nil
}

func (s *Store) GetCompatTask(ctx context.Context, user *auth.User, provider string, version string, identifier string) (GatewayTask, error) {
	gatewayUserID := localGatewayUserID(user)
	apiKeyID := ""
	userID := ""
	if user != nil {
		apiKeyID = strings.TrimSpace(user.APIKeyID)
		userID = strings.TrimSpace(user.ID)
	}
	if gatewayUserID == "" && userID == "" {
		return GatewayTask{}, ErrLocalUserRequired
	}
	task, err := scanGatewayTask(s.pool.QueryRow(ctx, `
SELECT `+gatewayTaskColumns+`
FROM gateway_tasks
WHERE (
    (NULLIF($1, '')::uuid IS NOT NULL AND gateway_user_id = NULLIF($1, '')::uuid)
    OR (NULLIF($1, '')::uuid IS NULL AND NULLIF($2, '') IS NOT NULL AND user_id = $2)
  )
  AND (NULLIF($3, '') IS NULL OR api_key_id = $3)
  AND request->>'_compat_provider' = $4
  AND request->>'_kling_compat_version' = $5
  AND (id::text = $6 OR external_task_id = $6 OR remote_task_id = $6 OR compatibility_public_id = $6)
ORDER BY created_at DESC
LIMIT 1`, gatewayUserID, userID, apiKeyID, strings.TrimSpace(provider), strings.TrimSpace(version), strings.TrimSpace(identifier)))
	if err != nil {
		return GatewayTask{}, err
	}
	attempts, err := s.ListTaskAttempts(ctx, task.ID)
	if err != nil {
		return GatewayTask{}, err
	}
	task.Attempts = attempts
	return task, nil
}

func (s *Store) ListCompatTasks(ctx context.Context, user *auth.User, filter CompatTaskListFilter) (TaskListResult, error) {
	page := filter.Page
	if page <= 0 {
		page = 1
	}
	pageSize := filter.PageSize
	if pageSize <= 0 {
		pageSize = 30
	}
	if pageSize > 500 {
		pageSize = 500
	}
	offset := (page - 1) * pageSize
	gatewayUserID := localGatewayUserID(user)
	apiKeyID := ""
	userID := ""
	if user != nil {
		apiKeyID = strings.TrimSpace(user.APIKeyID)
		userID = strings.TrimSpace(user.ID)
	}
	if gatewayUserID == "" && userID == "" {
		return TaskListResult{}, ErrLocalUserRequired
	}
	statuses := filter.Statuses
	if len(statuses) == 0 {
		statuses = nil
	}
	args := []any{
		gatewayUserID,
		userID,
		apiKeyID,
		strings.TrimSpace(filter.Provider),
		strings.TrimSpace(filter.Version),
		nullableTaskListTime(filter.CreatedFrom),
		nullableTaskListTime(filter.CreatedTo),
		statuses,
	}
	whereSQL := `
WHERE (
    (NULLIF($1, '')::uuid IS NOT NULL AND gateway_user_id = NULLIF($1, '')::uuid)
    OR (NULLIF($1, '')::uuid IS NULL AND NULLIF($2, '') IS NOT NULL AND user_id = $2)
  )
  AND (NULLIF($3, '') IS NULL OR api_key_id = $3)
  AND request->>'_compat_provider' = $4
  AND request->>'_kling_compat_version' = $5
  AND ($6::timestamptz IS NULL OR created_at >= $6::timestamptz)
  AND ($7::timestamptz IS NULL OR created_at <= $7::timestamptz)
  AND ($8::text[] IS NULL OR status = ANY($8::text[]))`
	var total int
	if err := s.pool.QueryRow(ctx, `SELECT count(*) FROM gateway_tasks `+whereSQL, args...).Scan(&total); err != nil {
		return TaskListResult{}, err
	}
	queryArgs := append(args, pageSize, offset)
	rows, err := s.pool.Query(ctx, `
SELECT `+gatewayTaskColumns+`
FROM gateway_tasks
`+whereSQL+`
ORDER BY created_at DESC
LIMIT $9 OFFSET $10`, queryArgs...)
	if err != nil {
		return TaskListResult{}, err
	}
	defer rows.Close()
	items := make([]GatewayTask, 0)
	for rows.Next() {
		task, err := scanGatewayTask(rows)
		if err != nil {
			return TaskListResult{}, err
		}
		items = append(items, task)
	}
	if err := rows.Err(); err != nil {
		return TaskListResult{}, err
	}
	items, err = s.attachTaskAttempts(ctx, items)
	if err != nil {
		return TaskListResult{}, err
	}
	return TaskListResult{Items: items, Total: total, Page: page, PageSize: pageSize}, nil
}

func nullableTaskListTime(value *time.Time) any {
	if value == nil {
		return nil
	}
	return *value
}

func (s *Store) ClaimTaskExecution(ctx context.Context, taskID string, executionToken string, leaseTTL time.Duration) (GatewayTask, error) {
	return s.ClaimTaskExecutionForOwner(ctx, taskID, executionToken, leaseTTL, "")
}

func (s *Store) ClaimTaskExecutionForOwner(
	ctx context.Context,
	taskID string,
	executionToken string,
	leaseTTL time.Duration,
	ownerInstanceID string,
) (GatewayTask, error) {
	if leaseTTL <= 0 {
		leaseTTL = 5 * time.Minute
	}
	var task GatewayTask
	manualReview := false
	err := s.beginTransaction(ctx, func(tx pgx.Tx) error {
		var queuedReady bool
		var runningExpired bool
		var production bool
		var hasGatewayUser bool
		if err := tx.QueryRow(ctx, `
SELECT status = 'queued' AND next_run_at <= now(),
       status = 'running' AND (execution_lease_expires_at IS NULL OR execution_lease_expires_at <= now()),
       run_mode IN ('production', 'acceptance', 'acceptance_canary'),
       gateway_user_id IS NOT NULL
FROM gateway_tasks
WHERE id = $1::uuid
FOR UPDATE`, taskID).Scan(&queuedReady, &runningExpired, &production, &hasGatewayUser); err != nil {
			return err
		}
		if !queuedReady && !runningExpired {
			return ErrTaskExecutionLeaseUnavailable
		}
		if (queuedReady || runningExpired) && production {
			submissionAmbiguous, err := taskExecutionRequiresManualReviewTx(ctx, tx, taskID)
			if err != nil {
				return err
			}
			if submissionAmbiguous {
				if err := markTaskExecutionManualReviewTx(ctx, tx, taskID, hasGatewayUser); err != nil {
					return err
				}
				manualReview = true
				return nil
			}
		}
		var err error
		task, err = scanGatewayTask(tx.QueryRow(ctx, `
UPDATE gateway_tasks
SET status = 'running',
    execution_token = $2::uuid,
    execution_lease_expires_at = now() + ($3::int * interval '1 second'),
    locked_by = NULLIF($4::text, ''),
    locked_at = now(),
    heartbeat_at = now(),
    updated_at = now()
WHERE id = $1::uuid
  AND (
    (status = 'queued' AND next_run_at <= now())
    OR (status = 'running' AND (execution_lease_expires_at IS NULL OR execution_lease_expires_at <= now()))
  )
RETURNING `+gatewayTaskColumns, taskID, executionToken, int(leaseTTL/time.Second), strings.TrimSpace(ownerInstanceID)))
		return err
	})
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return GatewayTask{}, ErrTaskExecutionLeaseUnavailable
		}
		return GatewayTask{}, err
	}
	if manualReview {
		return GatewayTask{}, ErrTaskExecutionManualReview
	}
	return task, nil
}

func taskExecutionRequiresManualReviewTx(ctx context.Context, tx pgx.Tx, taskID string) (bool, error) {
	var submissionAmbiguous bool
	err := tx.QueryRow(ctx, `
SELECT COALESCE(task.remote_task_id, '') = ''
       AND COALESCE((
  SELECT (
    (status = 'running' AND upstream_submission_status IN ('submitting', 'response_received'))
    OR (status = 'failed' AND upstream_submission_status = 'submitting')
    OR (
      status = 'failed'
      AND upstream_submission_status = 'response_received'
      AND error_code = 'execution_lease_expired'
    )
  )
  FROM gateway_task_attempts
  WHERE task_id = $1::uuid
  ORDER BY attempt_no DESC, started_at DESC
  LIMIT 1
), false)
FROM gateway_tasks task
WHERE task.id = $1::uuid`, taskID).Scan(&submissionAmbiguous)
	return submissionAmbiguous, err
}

func markTaskExecutionManualReviewTx(
	ctx context.Context,
	tx pgx.Tx,
	taskID string,
	hasGatewayUser bool,
) error {
	if _, err := tx.Exec(ctx, `
UPDATE gateway_task_attempts
SET status = 'failed',
    retryable = false,
    error_code = 'upstream_timeout',
    error_message = 'upstream submission confirmation timed out',
    finished_at = COALESCE(finished_at, now()),
    upstream_submission_updated_at = now()
WHERE task_id = $1::uuid
  AND status = 'running'
  AND upstream_submission_status IN ('submitting', 'response_received')`, taskID); err != nil {
		return err
	}
	if _, err := tx.Exec(ctx, `
UPDATE gateway_tasks
SET status = 'failed',
    billing_status = CASE WHEN $2 THEN 'manual_review' ELSE 'not_required' END,
    billing_updated_at = now(),
    error = NULL,
    error_code = 'upstream_timeout',
    error_message = 'upstream submission confirmation timed out',
    locked_by = NULL,
    locked_at = NULL,
    heartbeat_at = NULL,
    execution_token = NULL,
    execution_lease_expires_at = NULL,
    remote_task_payload = '{}'::jsonb,
	    submission_state = 'submission_confirmation_pending',
    finished_at = now(),
    updated_at = now()
WHERE id = $1::uuid`, taskID, hasGatewayUser); err != nil {
		return err
	}
	if !hasGatewayUser {
		return nil
	}
	payloadJSON, _ := json.Marshal(map[string]any{
		"taskId": taskID, "classification": "upstream_timeout",
	})
	if _, err := tx.Exec(ctx, `
INSERT INTO settlement_outbox (
  task_id, event_type, action, amount, currency, pricing_snapshot, payload,
  status, next_attempt_at, manual_review_reason
)
SELECT id, 'task.billing.review', 'release', reservation_amount, billing_currency,
       pricing_snapshot, $2::jsonb, 'manual_review', now(), 'upstream_timeout'
FROM gateway_tasks
WHERE id = $1::uuid
ON CONFLICT (task_id, event_type) DO NOTHING`, taskID, string(payloadJSON)); err != nil {
		return err
	}
	return nil
}

func (s *Store) ClaimTaskPreparation(ctx context.Context, taskID string, executionToken string, leaseTTL time.Duration) (GatewayTask, error) {
	return s.ClaimTaskPreparationForOwner(ctx, taskID, executionToken, leaseTTL, "")
}

func (s *Store) ClaimTaskPreparationForOwner(
	ctx context.Context,
	taskID string,
	executionToken string,
	leaseTTL time.Duration,
	ownerInstanceID string,
) (GatewayTask, error) {
	if leaseTTL <= 0 {
		leaseTTL = 5 * time.Minute
	}
	var task GatewayTask
	manualReview := false
	err := s.beginTransaction(ctx, func(tx pgx.Tx) error {
		var queuedReady bool
		var production bool
		var hasGatewayUser bool
		if err := tx.QueryRow(ctx, `
SELECT status = 'queued'
         AND next_run_at <= now()
         AND (
           execution_token IS NULL
           OR execution_lease_expires_at IS NULL
           OR execution_lease_expires_at <= now()
         ),
       run_mode IN ('production', 'acceptance', 'acceptance_canary'),
       gateway_user_id IS NOT NULL
FROM gateway_tasks
WHERE id = $1::uuid
FOR UPDATE`, taskID).Scan(&queuedReady, &production, &hasGatewayUser); err != nil {
			return err
		}
		if !queuedReady {
			return ErrTaskExecutionLeaseUnavailable
		}
		if production {
			submissionAmbiguous, err := taskExecutionRequiresManualReviewTx(ctx, tx, taskID)
			if err != nil {
				return err
			}
			if submissionAmbiguous {
				if err := markTaskExecutionManualReviewTx(ctx, tx, taskID, hasGatewayUser); err != nil {
					return err
				}
				manualReview = true
				return nil
			}
		}
		var err error
		task, err = scanGatewayTask(tx.QueryRow(ctx, `
UPDATE gateway_tasks
SET execution_token = $2::uuid,
    execution_lease_expires_at = now() + ($3::int * interval '1 second'),
    locked_by = NULLIF($4::text, ''),
    locked_at = now(),
    heartbeat_at = now(),
    updated_at = now()
WHERE id = $1::uuid
  AND status = 'queued'
  AND next_run_at <= now()
  AND (
    execution_token IS NULL
    OR execution_lease_expires_at IS NULL
    OR execution_lease_expires_at <= now()
  )
RETURNING `+gatewayTaskColumns, taskID, executionToken, int(leaseTTL/time.Second), strings.TrimSpace(ownerInstanceID)))
		return err
	})
	if errors.Is(err, pgx.ErrNoRows) {
		return GatewayTask{}, ErrTaskExecutionLeaseUnavailable
	}
	if err != nil {
		return GatewayTask{}, err
	}
	if manualReview {
		return GatewayTask{}, ErrTaskExecutionManualReview
	}
	return task, err
}

func (s *Store) ReleaseTaskPreparation(ctx context.Context, taskID string, executionToken string) error {
	tag, err := s.pool.Exec(ctx, `
UPDATE gateway_tasks
SET execution_token = NULL,
    execution_lease_expires_at = NULL,
    locked_at = NULL,
    heartbeat_at = NULL,
    updated_at = now()
WHERE id = $1::uuid
  AND status = 'queued'
  AND execution_token = $2::uuid`, taskID, executionToken)
	if err != nil {
		return err
	}
	if tag.RowsAffected() != 1 {
		return ErrTaskExecutionLeaseLost
	}
	return nil
}

func (s *Store) FailQueuedTask(ctx context.Context, taskID string, code string, message string) (GatewayTask, error) {
	return s.FailQueuedTaskWithCallback(ctx, taskID, code, message, "", false)
}

// FailQueuedTaskWithCallback terminally fails a task before upstream
// submission and atomically releases every durable resource owned by its
// admission. The task advisory lock makes competing dispatchers idempotent.
func (s *Store) FailQueuedTaskWithCallback(
	ctx context.Context,
	taskID string,
	code string,
	message string,
	callbackURL string,
	simulated bool,
) (GatewayTask, error) {
	code = strings.TrimSpace(code)
	message = truncateUTF8Bytes(message, 2048)
	publicErrorJSON := encodePublicErrorSnapshot(code, message, 0, true, "", taskID)
	var task GatewayTask
	err := s.beginTransaction(ctx, func(tx pgx.Tx) error {
		if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1, 0))`, "task-admission:"+taskID); err != nil {
			return err
		}
		var err error
		task, err = scanGatewayTask(tx.QueryRow(ctx, `
UPDATE gateway_tasks
SET status = 'failed',
    error = NULL,
    error_code = NULLIF($2, ''),
    error_message = NULLIF($3, ''),
    public_error = $4::jsonb,
    billing_status = CASE
      WHEN run_mode NOT IN ('production', 'acceptance', 'acceptance_canary') OR gateway_user_id IS NULL THEN 'not_required'
	  WHEN reservation_amount > 0 THEN 'pending'
	  ELSE 'released'
    END,
    billing_updated_at = now(),
    locked_by = NULL,
    locked_at = NULL,
    heartbeat_at = NULL,
    execution_token = NULL,
    execution_lease_expires_at = NULL,
    finished_at = now(),
    updated_at = now()
WHERE id = $1::uuid
  AND status = 'queued'
  AND COALESCE(remote_task_id, '') = ''
RETURNING `+gatewayTaskColumns, taskID, code, message, string(publicErrorJSON)))
		if errors.Is(err, pgx.ErrNoRows) {
			return ErrTaskExecutionFinished
		}
		if err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `
UPDATE gateway_concurrency_leases
SET released_at = statement_timestamp()
WHERE task_id = $1::uuid
  AND released_at IS NULL`, taskID); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `DELETE FROM gateway_task_admissions WHERE task_id = $1::uuid`, taskID); err != nil {
			return err
		}
		if _, err := releaseRuntimeRateReservationsTx(ctx, tx, []string{taskID}); err != nil {
			return err
		}
		payloadJSON, _ := json.Marshal(map[string]any{"taskId": taskID, "reason": code})
		if _, err := tx.Exec(ctx, `
INSERT INTO settlement_outbox (
  task_id, event_type, action, amount, currency, pricing_snapshot, payload, status, next_attempt_at
)
SELECT id, 'task.billing.release', 'release', reservation_amount, billing_currency,
       pricing_snapshot, $2::jsonb, 'pending', now()
FROM gateway_tasks
WHERE id = $1::uuid
  AND run_mode IN ('production', 'acceptance', 'acceptance_canary')
  AND gateway_user_id IS NOT NULL
  AND reservation_amount > 0
ON CONFLICT (task_id, event_type) DO NOTHING`, taskID, string(payloadJSON)); err != nil {
			return err
		}
		var eventID string
		var eventSeq int64
		if err := tx.QueryRow(ctx, `
WITH next_seq AS (
  SELECT COALESCE(MAX(seq), 0) + 1 AS seq
  FROM gateway_task_events
  WHERE task_id = $1::uuid
)
INSERT INTO gateway_task_events (
  task_id, seq, event_type, status, phase, progress, message, payload, simulated
)
SELECT $1::uuid, next_seq.seq, 'task.failed', 'failed', 'admission', 0,
       NULLIF($2, ''), '{}'::jsonb, $3
FROM next_seq
RETURNING id::text, seq`, taskID, message, simulated).Scan(&eventID, &eventSeq); err != nil {
			return err
		}
		callbackURL = strings.TrimSpace(callbackURL)
		if callbackURL != "" {
			if _, err := tx.Exec(ctx, `
INSERT INTO gateway_task_callback_outbox (task_id, event_id, seq, callback_url, payload)
VALUES ($1::uuid, $2::uuid, $3, $4, '{}'::jsonb)
ON CONFLICT (task_id, seq, callback_url) DO NOTHING`, taskID, eventID, eventSeq, callbackURL); err != nil {
				return err
			}
		}
		return nil
	})
	if err != nil {
		return GatewayTask{}, err
	}
	s.notifyTaskAdmissionBestEffort(ctx, "*")
	return task, nil
}

func (s *Store) RenewTaskExecutionLease(ctx context.Context, taskID string, executionToken string, leaseTTL time.Duration) error {
	if leaseTTL <= 0 {
		leaseTTL = 5 * time.Minute
	}
	tag, err := s.pool.Exec(ctx, `
UPDATE gateway_tasks
SET execution_lease_expires_at = now() + ($3::int * interval '1 second'),
    heartbeat_at = now(),
    updated_at = now()
WHERE id = $1::uuid
  AND status IN ('queued', 'running')
  AND execution_token = $2::uuid`, taskID, executionToken, int(leaseTTL/time.Second))
	if err != nil {
		return err
	}
	if tag.RowsAffected() != 1 {
		var stillRunning bool
		if err := s.pool.QueryRow(ctx, `
SELECT COALESCE((SELECT status IN ('queued', 'running') FROM gateway_tasks WHERE id = $1::uuid), false)`, taskID).Scan(&stillRunning); err != nil {
			return err
		}
		if !stillRunning {
			return ErrTaskExecutionFinished
		}
		return ErrTaskExecutionLeaseLost
	}
	return nil
}

func (s *Store) MarkTaskRunning(ctx context.Context, taskID string, executionToken string, modelType string, _ map[string]any) error {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer rollbackTransaction(tx)
	// The execution claim is the durable ownership boundary. This derived
	// status update is always followed by either the durable upstream
	// submission barrier or a terminal task transition, so it must be locally
	// visible immediately but does not need its own synchronous-replica wait.
	if _, err := tx.Exec(ctx, `SET LOCAL synchronous_commit = off`); err != nil {
		return err
	}
	tag, err := tx.Exec(ctx, `
UPDATE gateway_tasks
SET status = 'running',
	    model_type = NULLIF($3::text, ''),
	    normalized_request = '{}'::jsonb,
	    locked_at = now(),
	    heartbeat_at = now(),
	    updated_at = now()
WHERE id = $1::uuid
  AND status IN ('queued', 'running')
  AND execution_token = $2::uuid`, taskID, executionToken, modelType)
	if err != nil {
		return err
	}
	if tag.RowsAffected() != 1 {
		return ErrTaskExecutionLeaseLost
	}
	return tx.Commit(ctx)
}

func (s *Store) ClaimAsyncQueuedTask(ctx context.Context, workerID string) (GatewayTask, error) {
	return scanGatewayTask(s.pool.QueryRow(ctx, `
WITH picked AS (
  SELECT id AS task_id
  FROM gateway_tasks
  WHERE async_mode = true
    AND status = 'queued'
    AND next_run_at <= now()
  ORDER BY priority ASC, created_at ASC
  LIMIT 1
  FOR UPDATE SKIP LOCKED
)
UPDATE gateway_tasks t
SET status = 'running',
    locked_by = NULLIF($1::text, ''),
    locked_at = now(),
    heartbeat_at = now(),
    updated_at = now()
FROM picked
WHERE t.id = picked.task_id
RETURNING `+gatewayTaskColumns, workerID))
}

func (s *Store) RequeueTask(ctx context.Context, taskID string, executionToken string, delay time.Duration, queueKey string) (GatewayTask, error) {
	if delay < time.Second {
		delay = time.Second
	}
	if delay > 10*time.Minute {
		delay = 10 * time.Minute
	}
	nextRunAt := time.Now().Add(delay)
	return s.RequeueTaskUntil(ctx, taskID, executionToken, nextRunAt, queueKey)
}

func (s *Store) RequeueTaskUntil(ctx context.Context, taskID string, executionToken string, nextRunAt time.Time, queueKey string) (GatewayTask, error) {
	if nextRunAt.Before(time.Now().Add(time.Second)) {
		nextRunAt = time.Now().Add(time.Second)
	}
	return scanGatewayTask(s.pool.QueryRow(ctx, `
UPDATE gateway_tasks
SET status = 'queued',
    locked_by = NULL,
    locked_at = NULL,
    heartbeat_at = NULL,
    execution_token = NULL,
    execution_lease_expires_at = NULL,
    next_run_at = $3::timestamptz,
    queue_key = COALESCE(NULLIF($4::text, ''), queue_key),
    error = NULL,
    error_code = NULL,
    error_message = NULL,
	public_error = NULL,
    updated_at = now()
WHERE id = $1::uuid
  AND status IN ('queued', 'running')
  AND execution_token = $2::uuid
RETURNING `+gatewayTaskColumns, taskID, executionToken, nextRunAt, strings.TrimSpace(queueKey)))
}

func (s *Store) RequeueTaskBeforeUpstreamSubmission(
	ctx context.Context,
	taskID string,
	executionToken string,
	delay time.Duration,
) (GatewayTask, bool, error) {
	if delay < time.Second {
		delay = time.Second
	}
	if delay > 10*time.Minute {
		delay = 10 * time.Minute
	}
	nextRunAt := time.Now().Add(delay)
	var task GatewayTask
	changed := false
	err := s.beginTransaction(ctx, func(tx pgx.Tx) error {
		var err error
		task, err = scanGatewayTask(tx.QueryRow(ctx, `
UPDATE gateway_tasks task
SET status = 'queued',
    locked_by = NULL,
    locked_at = NULL,
    heartbeat_at = NULL,
    execution_token = NULL,
    execution_lease_expires_at = NULL,
    next_run_at = $3::timestamptz,
    error = NULL,
    error_code = NULL,
    error_message = NULL,
	public_error = NULL,
    updated_at = now()
WHERE task.id = $1::uuid
  AND task.status IN ('queued', 'running')
  AND task.execution_token = $2::uuid
  AND COALESCE(task.remote_task_id, '') = ''
  AND NOT EXISTS (
    SELECT 1
    FROM gateway_task_attempts attempt
    WHERE attempt.task_id = task.id
      AND COALESCE(attempt.upstream_submission_status, 'not_submitted') <> 'not_submitted'
  )
RETURNING `+gatewayTaskColumns, taskID, executionToken, nextRunAt))
		if errors.Is(err, pgx.ErrNoRows) {
			return nil
		}
		if err != nil {
			return err
		}
		changed = true
		if _, err := tx.Exec(ctx, `
DELETE FROM gateway_task_param_preprocessing_logs log
USING gateway_task_attempts attempt
WHERE log.attempt_id = attempt.id
  AND attempt.task_id = $1::uuid
  AND COALESCE(attempt.upstream_submission_status, 'not_submitted') = 'not_submitted'`, taskID); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `
DELETE FROM gateway_task_attempts
WHERE task_id = $1::uuid
  AND COALESCE(upstream_submission_status, 'not_submitted') = 'not_submitted'`, taskID); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `
UPDATE gateway_concurrency_leases
SET released_at = statement_timestamp()
WHERE task_id = $1::uuid
  AND released_at IS NULL`, taskID); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `DELETE FROM gateway_task_admissions WHERE task_id = $1::uuid`, taskID); err != nil {
			return err
		}
		return nil
	})
	if err != nil {
		return GatewayTask{}, false, err
	}
	if changed {
		s.notifyTaskAdmissionBestEffort(ctx, "*")
	}
	return task, changed, nil
}

// ResolveInterruptedTaskExecution atomically preserves the no-duplicate-submit
// invariant when a Worker disappears or its execution context is cancelled.
// A known remote task is safe to requeue for polling takeover. An ambiguous
// submission without a remote ID is moved to manual review instead of replayed.
func (s *Store) ResolveInterruptedTaskExecution(
	ctx context.Context,
	taskID string,
	executionToken string,
) (GatewayTask, error) {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return GatewayTask{}, err
	}
	defer rollbackTransaction(tx)
	var remoteTaskID string
	var hasGatewayUser bool
	if err := tx.QueryRow(ctx, `
SELECT COALESCE(remote_task_id, ''), gateway_user_id IS NOT NULL
FROM gateway_tasks
WHERE id = $1::uuid
  AND status = 'running'
  AND execution_token = $2::uuid
FOR UPDATE`, taskID, executionToken).Scan(&remoteTaskID, &hasGatewayUser); err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return GatewayTask{}, ErrTaskExecutionLeaseLost
		}
		return GatewayTask{}, err
	}
	ambiguous, err := taskExecutionRequiresManualReviewTx(ctx, tx, taskID)
	if err != nil {
		return GatewayTask{}, err
	}
	if remoteTaskID == "" && ambiguous {
		if err := markTaskExecutionManualReviewTx(ctx, tx, taskID, hasGatewayUser); err != nil {
			return GatewayTask{}, err
		}
		if err := tx.Commit(ctx); err != nil {
			return GatewayTask{}, err
		}
		return GatewayTask{}, ErrTaskExecutionManualReview
	}
	nextRunAt := time.Now().Add(time.Second)
	queued, err := scanGatewayTask(tx.QueryRow(ctx, `
UPDATE gateway_tasks
SET status = 'queued',
    locked_by = NULL,
    locked_at = NULL,
    heartbeat_at = NULL,
    execution_token = NULL,
    execution_lease_expires_at = NULL,
    next_run_at = $3,
    error = NULL,
    error_code = NULL,
    error_message = NULL,
    public_error = NULL,
    updated_at = now()
WHERE id = $1::uuid
  AND status = 'running'
  AND execution_token = $2::uuid
RETURNING `+gatewayTaskColumns, taskID, executionToken, nextRunAt))
	if err != nil {
		return GatewayTask{}, err
	}
	if err := tx.Commit(ctx); err != nil {
		return GatewayTask{}, err
	}
	return queued, nil
}

func (s *Store) SetTaskRiverJobID(ctx context.Context, taskID string, riverJobID int64) error {
	if riverJobID <= 0 {
		return nil
	}
	_, err := s.pool.Exec(ctx, `
UPDATE gateway_tasks
SET river_job_id = $2,
    updated_at = now()
WHERE id = $1::uuid`, taskID, riverJobID)
	return err
}

func (s *Store) SetTaskRemoteTask(ctx context.Context, taskID string, executionToken string, attemptID string, remoteTaskID string, payload map[string]any) error {
	payloadJSON, _ := json.Marshal(sanitizeJSONForStorage(emptyObjectIfNil(payload)))
	return s.beginTransaction(ctx, func(tx pgx.Tx) error {
		tag, err := tx.Exec(ctx, `
UPDATE gateway_tasks
SET remote_task_id = NULLIF($3::text, ''),
    remote_task_payload = $4::jsonb,
    updated_at = now()
WHERE id = $1::uuid
  AND status = 'running'
  AND execution_token = $2::uuid`,
			taskID,
			executionToken,
			remoteTaskID,
			string(payloadJSON),
		)
		if err != nil {
			return err
		}
		if tag.RowsAffected() != 1 {
			return ErrTaskExecutionLeaseLost
		}
		if strings.TrimSpace(attemptID) == "" {
			return nil
		}
		_, err = tx.Exec(ctx, `
UPDATE gateway_task_attempts
SET remote_task_id = NULLIF($2::text, '')
WHERE id = $1::uuid
  AND status = 'running'`,
			attemptID,
			remoteTaskID,
		)
		return err
	})
}

func validateAttemptUpstreamSubmissionStatus(status string) error {
	switch status {
	case "not_submitted", "submitting", "response_received":
	default:
		return fmt.Errorf("invalid upstream submission status %q", status)
	}
	return nil
}

func (s *Store) SetAttemptUpstreamSubmissionStatus(ctx context.Context, attemptID string, status string) error {
	if err := validateAttemptUpstreamSubmissionStatus(status); err != nil {
		return err
	}
	_, err := s.pool.Exec(ctx, `
UPDATE gateway_task_attempts
SET upstream_submission_status = $2,
    upstream_submission_updated_at = now()
WHERE id = $1::uuid`, attemptID, status)
	return err
}

func (s *Store) SetAttemptUpstreamSubmissionStatusForExecution(
	ctx context.Context,
	attemptID string,
	taskID string,
	executionToken string,
	status string,
) error {
	if err := validateAttemptUpstreamSubmissionStatus(status); err != nil {
		return err
	}
	tag, err := s.pool.Exec(ctx, `
UPDATE gateway_task_attempts attempt
SET upstream_submission_status = $4,
    upstream_submission_updated_at = now()
FROM gateway_tasks task
WHERE attempt.id = $1::uuid
  AND attempt.task_id = task.id
  AND attempt.task_id = $2::uuid
  AND attempt.status = 'running'
  AND task.status = 'running'
  AND task.execution_token = $3::uuid
  AND task.execution_lease_expires_at > now()`,
		attemptID,
		taskID,
		executionToken,
		status,
	)
	if err != nil {
		return err
	}
	if tag.RowsAffected() != 1 {
		return ErrTaskExecutionLeaseLost
	}
	return nil
}

func (s *Store) CancelQueuedTask(ctx context.Context, taskID string, message string) (GatewayTask, bool, error) {
	return s.cancelQueuedTask(ctx, taskID, message, "", false)
}

func (s *Store) CancelQueuedTaskWithCallback(
	ctx context.Context,
	taskID string,
	message string,
	callbackURL string,
) (GatewayTask, bool, error) {
	return s.cancelQueuedTask(ctx, taskID, message, callbackURL, true)
}

func (s *Store) cancelQueuedTask(
	ctx context.Context,
	taskID string,
	message string,
	callbackURL string,
	emitTerminalEvent bool,
) (GatewayTask, bool, error) {
	message = strings.TrimSpace(message)
	if message == "" {
		message = "任务已取消"
	}
	message = truncateUTF8Bytes(message, 2048)
	var task GatewayTask
	changed := false
	err := s.beginTransaction(ctx, func(tx pgx.Tx) error {
		if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1, 0))`, "task-admission:"+taskID); err != nil {
			return err
		}
		var err error
		task, err = scanGatewayTask(tx.QueryRow(ctx, `
UPDATE gateway_tasks
SET status = 'cancelled',
    error = NULL,
    error_code = 'task_cancelled',
    error_message = NULLIF($2::text, ''),
    billing_status = CASE
      WHEN run_mode NOT IN ('production', 'acceptance', 'acceptance_canary') OR gateway_user_id IS NULL THEN 'not_required'
      WHEN reservation_amount > 0 THEN 'pending'
      ELSE 'released'
    END,
    billing_updated_at = now(),
    remote_task_payload = '{}'::jsonb,
    locked_by = NULL,
    locked_at = NULL,
    heartbeat_at = NULL,
    execution_token = NULL,
    execution_lease_expires_at = NULL,
    finished_at = now(),
    updated_at = now()
WHERE id = $1::uuid
  AND status = 'queued'
  AND COALESCE(remote_task_id, '') = ''
RETURNING `+gatewayTaskColumns, taskID, message))
		if errors.Is(err, pgx.ErrNoRows) {
			return nil
		}
		if err != nil {
			return err
		}
		changed = true
		if _, err := tx.Exec(ctx, `
UPDATE gateway_concurrency_leases
SET released_at = statement_timestamp()
WHERE task_id = $1::uuid
  AND released_at IS NULL`, taskID); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `DELETE FROM gateway_task_admissions WHERE task_id = $1::uuid`, taskID); err != nil {
			return err
		}
		if _, err := releaseRuntimeRateReservationsTx(ctx, tx, []string{taskID}); err != nil {
			return err
		}
		payloadJSON, _ := json.Marshal(map[string]any{"taskId": taskID, "reason": "queued_cancelled"})
		if _, err = tx.Exec(ctx, `
INSERT INTO settlement_outbox (
  task_id, event_type, action, amount, currency, pricing_snapshot, payload, status, next_attempt_at
)
SELECT id, 'task.billing.release', 'release', reservation_amount, billing_currency,
       pricing_snapshot, $2::jsonb, 'pending', now()
FROM gateway_tasks
WHERE id = $1::uuid
  AND run_mode IN ('production', 'acceptance', 'acceptance_canary')
  AND gateway_user_id IS NOT NULL
  AND reservation_amount > 0
ON CONFLICT (task_id, event_type) DO NOTHING`, taskID, string(payloadJSON)); err != nil {
			return err
		}
		if !emitTerminalEvent {
			return nil
		}
		var eventID string
		var eventSeq int64
		if err := tx.QueryRow(ctx, `
WITH next_seq AS (
  SELECT COALESCE(MAX(seq), 0) + 1 AS seq
  FROM gateway_task_events
  WHERE task_id = $1::uuid
)
INSERT INTO gateway_task_events (
  task_id, seq, event_type, status, phase, progress, message, payload, simulated
)
SELECT $1::uuid, next_seq.seq, 'task.cancelled', 'cancelled', 'admission', 0,
       NULLIF($2, ''), '{}'::jsonb, $3
FROM next_seq
RETURNING id::text, seq`, taskID, message, task.RunMode == "simulation").Scan(&eventID, &eventSeq); err != nil {
			return err
		}
		callbackURL = strings.TrimSpace(callbackURL)
		if callbackURL == "" {
			return nil
		}
		_, err = tx.Exec(ctx, `
INSERT INTO gateway_task_callback_outbox (task_id, event_id, seq, callback_url, payload)
VALUES ($1::uuid, $2::uuid, $3, $4, '{}'::jsonb)
ON CONFLICT (task_id, seq, callback_url) DO NOTHING`, taskID, eventID, eventSeq, callbackURL)
		return err
	})
	if err != nil {
		return GatewayTask{}, false, err
	}
	if changed {
		s.notifyTaskAdmissionBestEffort(ctx, "*")
	}
	return task, changed, nil
}

func (s *Store) CancelTaskBeforeUpstreamSubmission(
	ctx context.Context,
	taskID string,
	executionToken string,
	message string,
) (GatewayTask, bool, error) {
	message = strings.TrimSpace(message)
	if message == "" {
		message = "client disconnected before upstream submission"
	}
	message = truncateUTF8Bytes(message, 2048)
	var task GatewayTask
	changed := false
	err := s.beginTransaction(ctx, func(tx pgx.Tx) error {
		if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1, 0))`, "task-admission:"+taskID); err != nil {
			return err
		}
		var err error
		task, err = scanGatewayTask(tx.QueryRow(ctx, `
UPDATE gateway_tasks task
SET status = 'cancelled',
    error = NULL,
    error_code = 'client_disconnected',
    error_message = NULLIF($3::text, ''),
    billing_status = CASE
      WHEN run_mode NOT IN ('production', 'acceptance', 'acceptance_canary') OR gateway_user_id IS NULL THEN 'not_required'
      ELSE 'released'
    END,
    billing_updated_at = now(),
    locked_by = NULL,
    locked_at = NULL,
    heartbeat_at = NULL,
    execution_token = NULL,
    execution_lease_expires_at = NULL,
    finished_at = now(),
    updated_at = now()
WHERE id = $1::uuid
  AND status IN ('queued', 'running')
  AND execution_token = NULLIF($2, '')::uuid
  AND COALESCE(remote_task_id, '') = ''
  AND NOT EXISTS (
    SELECT 1
    FROM gateway_task_attempts attempt
    WHERE attempt.task_id = task.id
      AND attempt.upstream_submission_status <> 'not_submitted'
  )
RETURNING `+gatewayTaskColumns, taskID, strings.TrimSpace(executionToken), message))
		if errors.Is(err, pgx.ErrNoRows) {
			return nil
		}
		if err != nil {
			return err
		}
		changed = true
		if _, err := tx.Exec(ctx, `
DELETE FROM gateway_task_param_preprocessing_logs log
USING gateway_task_attempts attempt
WHERE log.attempt_id = attempt.id
  AND attempt.task_id = $1::uuid
  AND attempt.upstream_submission_status = 'not_submitted'`, taskID); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `
DELETE FROM gateway_task_attempts
WHERE task_id = $1::uuid
  AND upstream_submission_status = 'not_submitted'`, taskID); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `
UPDATE gateway_concurrency_leases
SET released_at = statement_timestamp()
WHERE task_id = $1::uuid
  AND released_at IS NULL`, taskID); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `DELETE FROM gateway_task_admissions WHERE task_id = $1::uuid`, taskID); err != nil {
			return err
		}
		return nil
	})
	if err != nil {
		return GatewayTask{}, false, err
	}
	if changed {
		s.notifyTaskAdmissionBestEffort(ctx, "*")
	}
	return task, changed, nil
}

// CancelSubmittedTask records a confirmed upstream cancellation. Callers must
// first complete the provider-side DELETE so local status never claims a remote
// task was cancelled when the upstream request was not accepted.
func (s *Store) CancelSubmittedTask(ctx context.Context, taskID string, executionToken string, message string) (GatewayTask, bool, error) {
	message = strings.TrimSpace(message)
	if message == "" {
		message = "任务已由上游取消"
	}
	message = truncateUTF8Bytes(message, 2048)
	var task GatewayTask
	changed := false
	err := s.beginTransaction(ctx, func(tx pgx.Tx) error {
		var err error
		task, err = scanGatewayTask(tx.QueryRow(ctx, `
UPDATE gateway_tasks
SET status = 'cancelled',
    error = NULL,
    error_code = 'task_cancelled',
    error_message = NULLIF($2, ''),
    billing_status = CASE
      WHEN run_mode NOT IN ('production', 'acceptance', 'acceptance_canary') OR gateway_user_id IS NULL THEN 'not_required'
      WHEN reservation_amount > 0 THEN 'pending'
      ELSE 'released'
    END,
    billing_updated_at = now(),
    locked_by = NULL,
    locked_at = NULL,
    heartbeat_at = NULL,
    execution_token = NULL,
    execution_lease_expires_at = NULL,
    remote_task_payload = '{}'::jsonb,
    finished_at = now(),
    updated_at = now()
WHERE id = $1::uuid
  AND NULLIF(remote_task_id, '') IS NOT NULL
  AND (
    (status = 'running' AND execution_token = NULLIF($3, '')::uuid)
    OR status = 'queued'
  )
RETURNING `+gatewayTaskColumns, taskID, message, strings.TrimSpace(executionToken)))
		if IsNotFound(err) {
			return nil
		}
		if err != nil {
			return err
		}
		changed = true
		payloadJSON, _ := json.Marshal(map[string]any{"taskId": taskID, "reason": "upstream_cancelled"})
		_, err = tx.Exec(ctx, `
INSERT INTO settlement_outbox (
  task_id, event_type, action, amount, currency, pricing_snapshot, payload, status, next_attempt_at
)
SELECT id, 'task.billing.release', 'release', reservation_amount, billing_currency,
       pricing_snapshot, $2::jsonb, 'pending', now()
FROM gateway_tasks
WHERE id = $1::uuid
  AND run_mode IN ('production', 'acceptance', 'acceptance_canary')
  AND gateway_user_id IS NOT NULL
  AND reservation_amount > 0
ON CONFLICT (task_id, event_type) DO NOTHING`, taskID, string(payloadJSON))
		return err
	})
	if err != nil {
		return GatewayTask{}, false, err
	}
	return task, changed, nil
}

func (s *Store) ListRecoverableAsyncTasks(ctx context.Context, limit int) ([]AsyncTaskQueueItem, error) {
	if limit <= 0 {
		limit = 500
	}
	// API handlers create the durable task before they finish media restoration
	// and register distributed admission. Give that normal submission path
	// ownership of fresh tasks so the generic crash recovery scan cannot create
	// a River job inside the create-to-admission window.
	const minimumRecoveryAge = 5 * time.Minute
	rows, err := s.pool.Query(ctx, `
SELECT task.id::text, task.priority, task.next_run_at
FROM gateway_tasks task
LEFT JOIN river_job job ON job.id = task.river_job_id
WHERE task.async_mode = true
  AND task.created_at <= now() - $2::interval
  AND (
    task.status = 'queued'
    OR (
      task.status = 'running'
      AND (task.execution_lease_expires_at IS NULL OR task.execution_lease_expires_at <= now())
    )
  )
  AND (
    task.river_job_id IS NULL
    OR job.id IS NULL
    OR job.state NOT IN ('available', 'pending', 'retryable', 'running', 'scheduled')
  )
  AND NOT EXISTS (
    SELECT 1
    FROM gateway_task_admissions admission
    WHERE admission.task_id = task.id
      AND admission.status = 'waiting'
  )
ORDER BY task.priority ASC, task.created_at ASC
LIMIT $1`, limit, minimumRecoveryAge.String())
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	items := make([]AsyncTaskQueueItem, 0)
	for rows.Next() {
		var item AsyncTaskQueueItem
		if err := rows.Scan(&item.ID, &item.Priority, &item.NextRunAt); err != nil {
			return nil, err
		}
		items = append(items, item)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

func (s *Store) CreateTaskAttempt(ctx context.Context, input CreateTaskAttemptInput) (string, error) {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return "", err
	}
	defer rollbackTransaction(tx)
	// OnUpstreamSubmissionStarted synchronously persists `submitting` before
	// any provider request leaves the process. PostgreSQL WAL ordering makes
	// that later barrier cover this attempt row as well, avoiding a redundant
	// cross-region wait without weakening duplicate-submission protection.
	if _, err := tx.Exec(ctx, `SET LOCAL synchronous_commit = off`); err != nil {
		return "", err
	}

	if strings.TrimSpace(input.ExecutionToken) != "" {
		var ownsExecution bool
		if err := tx.QueryRow(ctx, `
SELECT status = 'running'
   AND execution_token = $2::uuid
   AND execution_lease_expires_at > now()
FROM gateway_tasks
WHERE id = $1::uuid
FOR UPDATE`,
			input.TaskID,
			input.ExecutionToken,
		).Scan(&ownsExecution); err != nil {
			return "", err
		}
		if !ownsExecution {
			return "", ErrTaskExecutionLeaseLost
		}
	}
	var attemptID string
	err = tx.QueryRow(ctx, `
INSERT INTO gateway_task_attempts (
  task_id, attempt_no, platform_id, platform_model_id, client_id, queue_key,
  status, simulated, request_snapshot, metrics, pricing_snapshot, request_fingerprint,
  upstream_submission_status, upstream_submission_updated_at
)
VALUES (
  $1::uuid, $2::int, NULLIF($3::text, '')::uuid, NULLIF($4::text, '')::uuid, NULLIF($5::text, ''), $6,
  $7, $8, '{}'::jsonb, '{}'::jsonb, '{}'::jsonb, NULL,
  'not_submitted', now()
)
RETURNING id::text`,
		input.TaskID,
		input.AttemptNo,
		input.PlatformID,
		input.PlatformModelID,
		input.ClientID,
		input.QueueKey,
		firstNonEmpty(input.Status, "running"),
		input.Simulated,
	).Scan(&attemptID)
	if err != nil {
		return "", err
	}
	if _, err := tx.Exec(ctx, `
UPDATE gateway_tasks
SET attempt_count = GREATEST(attempt_count, $2::int), updated_at = now()
WHERE id = $1::uuid`, input.TaskID, input.AttemptNo); err != nil {
		return "", err
	}
	return attemptID, tx.Commit(ctx)
}

func (s *Store) CreateTaskParamPreprocessingLog(ctx context.Context, input CreateTaskParamPreprocessingLogInput) (string, error) {
	if !input.Changed {
		return "", nil
	}
	changesJSON, _ := json.Marshal(sanitizeJSONForStorage(input.Changes))
	if input.Changes == nil {
		changesJSON = []byte("[]")
	}
	if len(changesJSON) > 1024 {
		changesJSON = []byte("[]")
	}
	var attemptNo any
	if input.AttemptNo > 0 {
		attemptNo = input.AttemptNo
	}
	var id string
	err := s.pool.QueryRow(ctx, `
INSERT INTO gateway_task_param_preprocessing_logs (
  task_id, attempt_id, attempt_no, model_type, platform_id, platform_model_id, client_id,
  changed, change_count, actual_input, converted_output, changes, model_snapshot
)
VALUES (
  $1::uuid, NULLIF($2::text, '')::uuid, $3::int, NULLIF($4::text, ''),
  NULLIF($5::text, '')::uuid, NULLIF($6::text, '')::uuid, NULLIF($7::text, ''),
  true, $8::int, '{}'::jsonb, '{}'::jsonb, $9::jsonb, '{}'::jsonb
)
RETURNING id::text`,
		input.TaskID,
		input.AttemptID,
		attemptNo,
		input.ModelType,
		input.PlatformID,
		input.PlatformModelID,
		input.ClientID,
		input.ChangeCount,
		string(changesJSON),
	).Scan(&id)
	return id, err
}

func (s *Store) attachTaskAttempts(ctx context.Context, items []GatewayTask) ([]GatewayTask, error) {
	if len(items) == 0 {
		return items, nil
	}
	taskIDs := make([]string, 0, len(items))
	for _, item := range items {
		taskIDs = append(taskIDs, item.ID)
	}
	attemptsByTaskID, err := s.listTaskAttemptsByTaskIDs(ctx, taskIDs)
	if err != nil {
		return nil, err
	}
	for index := range items {
		items[index].Attempts = attemptsByTaskID[items[index].ID]
	}
	return items, nil
}

func (s *Store) ListTaskAttempts(ctx context.Context, taskID string) ([]TaskAttempt, error) {
	attemptsByTaskID, err := s.listTaskAttemptsByTaskIDs(ctx, []string{taskID})
	if err != nil {
		return nil, err
	}
	return attemptsByTaskID[taskID], nil
}

func (s *Store) ListTaskParamPreprocessingLogs(ctx context.Context, taskID string) ([]TaskParamPreprocessingLog, error) {
	rows, err := s.pool.Query(ctx, `
SELECT id::text, task_id::text, COALESCE(attempt_id::text, ''), COALESCE(attempt_no, 0),
       COALESCE(model_type, ''), COALESCE(platform_id::text, ''), COALESCE(platform_model_id::text, ''),
       COALESCE(client_id, ''), changed, change_count,
       COALESCE(actual_input, '{}'::jsonb), COALESCE(converted_output, '{}'::jsonb),
       COALESCE(changes, '[]'::jsonb), COALESCE(model_snapshot, '{}'::jsonb), created_at
FROM gateway_task_param_preprocessing_logs
WHERE task_id = $1::uuid
ORDER BY COALESCE(attempt_no, 0), created_at`, taskID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	items := make([]TaskParamPreprocessingLog, 0)
	for rows.Next() {
		item, err := scanTaskParamPreprocessingLog(rows)
		if err != nil {
			return nil, err
		}
		items = append(items, item)
	}
	return items, rows.Err()
}

func (s *Store) AppendTaskAttemptTrace(ctx context.Context, taskID string, attemptNo int, entry map[string]any) error {
	if strings.TrimSpace(taskID) == "" || attemptNo <= 0 || len(entry) == 0 {
		return nil
	}
	encoded, err := json.Marshal(entry)
	if err != nil {
		return err
	}
	result, err := s.pool.Exec(ctx, `
UPDATE gateway_task_attempts
SET metrics = jsonb_set(
      COALESCE(metrics, '{}'::jsonb),
      '{trace}',
      COALESCE(metrics->'trace', '[]'::jsonb) || jsonb_build_array($3::jsonb),
      true
    )
WHERE task_id = $1::uuid
  AND attempt_no = $2::int`, taskID, attemptNo, encoded)
	if err != nil {
		return err
	}
	if result.RowsAffected() == 0 {
		return pgx.ErrNoRows
	}
	return nil
}

func (s *Store) listTaskAttemptsByTaskIDs(ctx context.Context, taskIDs []string) (map[string][]TaskAttempt, error) {
	itemsByTaskID := map[string][]TaskAttempt{}
	if len(taskIDs) == 0 {
		return itemsByTaskID, nil
	}
	rows, err := s.pool.Query(ctx, `
SELECT a.id::text, a.task_id::text, a.attempt_no,
       COALESCE(a.platform_id::text, ''), COALESCE(p.name, ''), COALESCE(p.provider, ''),
       COALESCE(a.platform_model_id::text, ''), COALESCE(pm.model_name, ''),
       COALESCE(NULLIF(pm.provider_model_name, ''), pm.model_name, ''),
       COALESCE(pm.model_alias, ''),
       COALESCE(a.client_id, ''), a.queue_key, a.status, a.retryable, a.simulated,
       COALESCE(a.request_id, ''), COALESCE(a.status_code, 0),
       COALESCE(a.usage, '{}'::jsonb), COALESCE(a.metrics, '{}'::jsonb),
       a.request_snapshot, COALESCE(a.response_snapshot, '{}'::jsonb),
       COALESCE(a.response_started_at::text, ''), COALESCE(a.response_finished_at::text, ''),
       COALESCE(a.response_duration_ms, 0), COALESCE(a.error_code, ''), COALESCE(a.error_message, ''),
       COALESCE(a.public_error, '{}'::jsonb),
       COALESCE(a.pricing_snapshot, '{}'::jsonb), COALESCE(a.request_fingerprint, ''),
       a.upstream_submission_status, COALESCE(a.upstream_submission_updated_at::text, ''),
       a.started_at, COALESCE(a.finished_at::text, '')
FROM gateway_task_attempts a
LEFT JOIN integration_platforms p ON p.id = a.platform_id
LEFT JOIN platform_models pm ON pm.id = a.platform_model_id
WHERE a.task_id::text = ANY($1)
ORDER BY a.task_id, a.attempt_no`, taskIDs)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	for rows.Next() {
		item, err := scanTaskAttempt(rows)
		if err != nil {
			return nil, err
		}
		itemsByTaskID[item.TaskID] = append(itemsByTaskID[item.TaskID], item)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return itemsByTaskID, nil
}

func scanTaskAttempt(scanner taskScanner) (TaskAttempt, error) {
	var item TaskAttempt
	var usageBytes []byte
	var metricsBytes []byte
	var requestBytes []byte
	var responseBytes []byte
	var pricingSnapshotBytes []byte
	var publicErrorBytes []byte
	if err := scanner.Scan(
		&item.ID,
		&item.TaskID,
		&item.AttemptNo,
		&item.PlatformID,
		&item.PlatformName,
		&item.Provider,
		&item.PlatformModelID,
		&item.ModelName,
		&item.ProviderModelName,
		&item.ModelAlias,
		&item.ClientID,
		&item.QueueKey,
		&item.Status,
		&item.Retryable,
		&item.Simulated,
		&item.RequestID,
		&item.StatusCode,
		&usageBytes,
		&metricsBytes,
		&requestBytes,
		&responseBytes,
		&item.ResponseStartedAt,
		&item.ResponseFinishedAt,
		&item.ResponseDurationMS,
		&item.ErrorCode,
		&item.ErrorMessage,
		&publicErrorBytes,
		&pricingSnapshotBytes,
		&item.RequestFingerprint,
		&item.UpstreamSubmissionStatus,
		&item.UpstreamSubmissionUpdatedAt,
		&item.StartedAt,
		&item.FinishedAt,
	); err != nil {
		return TaskAttempt{}, err
	}
	item.Usage = decodeObject(usageBytes)
	item.Metrics = decodeObject(metricsBytes)
	item.RequestSnapshot = decodeObject(requestBytes)
	item.ResponseSnapshot = decodeObject(responseBytes)
	item.PricingSnapshot = decodeObject(pricingSnapshotBytes)
	if len(publicErrorBytes) > 0 {
		var snapshot publicerror.Error
		if json.Unmarshal(publicErrorBytes, &snapshot) == nil && snapshot.Code != "" {
			item.PublicError = &snapshot
		}
	}
	enrichTaskAttemptFromMetrics(&item)
	return item, nil
}

func scanTaskParamPreprocessingLog(scanner taskScanner) (TaskParamPreprocessingLog, error) {
	var item TaskParamPreprocessingLog
	var actualInputBytes []byte
	var convertedOutputBytes []byte
	var changesBytes []byte
	var modelSnapshotBytes []byte
	if err := scanner.Scan(
		&item.ID,
		&item.TaskID,
		&item.AttemptID,
		&item.AttemptNo,
		&item.ModelType,
		&item.PlatformID,
		&item.PlatformModelID,
		&item.ClientID,
		&item.Changed,
		&item.ChangeCount,
		&actualInputBytes,
		&convertedOutputBytes,
		&changesBytes,
		&modelSnapshotBytes,
		&item.CreatedAt,
	); err != nil {
		return TaskParamPreprocessingLog{}, err
	}
	item.ActualInput = decodeObject(actualInputBytes)
	item.ConvertedOutput = decodeObject(convertedOutputBytes)
	item.Changes = decodeArray(changesBytes)
	item.ModelSnapshot = decodeObject(modelSnapshotBytes)
	return item, nil
}

func enrichTaskAttemptFromMetrics(item *TaskAttempt) {
	if item == nil || len(item.Metrics) == 0 {
		return
	}
	item.PlatformID = firstNonEmpty(item.PlatformID, taskAttemptMetricString(item.Metrics, "platformId"))
	item.PlatformName = firstNonEmpty(item.PlatformName, taskAttemptMetricString(item.Metrics, "platformName"))
	item.Provider = firstNonEmpty(item.Provider, taskAttemptMetricString(item.Metrics, "provider"))
	item.PlatformModelID = firstNonEmpty(item.PlatformModelID, taskAttemptMetricString(item.Metrics, "platformModelId"))
	item.ModelName = firstNonEmpty(item.ModelName, taskAttemptMetricString(item.Metrics, "resolvedModel"), taskAttemptMetricString(item.Metrics, "modelName"))
	item.ProviderModelName = firstNonEmpty(item.ProviderModelName, taskAttemptMetricString(item.Metrics, "providerModel"))
	item.ModelAlias = firstNonEmpty(item.ModelAlias, taskAttemptMetricString(item.Metrics, "modelAlias"))
	item.ModelType = firstNonEmpty(item.ModelType, taskAttemptMetricString(item.Metrics, "modelType"))
	item.ClientID = firstNonEmpty(item.ClientID, taskAttemptMetricString(item.Metrics, "clientId"))
	if item.StatusCode == 0 {
		item.StatusCode = taskAttemptMetricInt(item.Metrics, "statusCode")
	}
}

func taskAttemptMetricString(metrics map[string]any, key string) string {
	value, _ := metrics[key].(string)
	return strings.TrimSpace(value)
}

func taskAttemptMetricInt(metrics map[string]any, key string) int {
	switch value := metrics[key].(type) {
	case int:
		return value
	case int64:
		return int(value)
	case float64:
		return int(value)
	case json.Number:
		next, _ := value.Int64()
		return int(next)
	case string:
		next, _ := strconv.Atoi(strings.TrimSpace(value))
		return next
	default:
		return 0
	}
}

func (s *Store) FinishTaskAttempt(ctx context.Context, input FinishTaskAttemptInput) error {
	if input.UpstreamSubmissionStatus != "" {
		if err := validateAttemptUpstreamSubmissionStatus(input.UpstreamSubmissionStatus); err != nil {
			return err
		}
	}
	statusCode := input.StatusCode
	if statusCode == 0 {
		statusCode = taskAttemptMetricInt(input.Metrics, "statusCode")
	}
	usageJSON, _ := json.Marshal(sanitizeJSONForStorage(minimalTaskAttemptUsage(input.Usage)))
	metricsJSON, _ := json.Marshal(sanitizeJSONForStorage(minimalTaskAttemptMetrics(input.Metrics)))
	publicErrorJSON := encodePublicErrorSnapshot(input.ErrorCode, input.ErrorMessage, statusCode, input.Retryable, input.RequestID, "")
	_, err := s.pool.Exec(ctx, `
UPDATE gateway_task_attempts
SET status = $2::text,
    retryable = $3,
    request_id = NULLIF($4::text, ''),
    status_code = NULLIF($5::int, 0),
    usage = $11::jsonb,
    metrics = $12::jsonb,
    response_snapshot = '{}'::jsonb,
    pricing_snapshot = '{}'::jsonb,
    request_fingerprint = NULL,
    response_started_at = $6::timestamptz,
    response_finished_at = $7::timestamptz,
    response_duration_ms = $8,
    error_code = NULLIF($9::text, ''),
    error_message = NULLIF(left($10::text, 2048), ''),
	public_error = CASE WHEN $2::text = 'failed' THEN $14::jsonb ELSE NULL END,
    upstream_submission_status = COALESCE(NULLIF($13::text, ''), upstream_submission_status),
    upstream_submission_updated_at = CASE
      WHEN NULLIF($13::text, '') IS NULL THEN upstream_submission_updated_at
      ELSE now()
    END,
    finished_at = now()
WHERE id = $1::uuid`,
		input.AttemptID,
		input.Status,
		input.Retryable,
		input.RequestID,
		statusCode,
		nullableTime(input.ResponseStartedAt),
		nullableTime(input.ResponseFinishedAt),
		input.ResponseDurationMS,
		input.ErrorCode,
		truncateUTF8Bytes(input.ErrorMessage, 2048),
		usageJSON,
		metricsJSON,
		input.UpstreamSubmissionStatus,
		string(publicErrorJSON),
	)
	return err
}

func minimalTaskAttemptUsage(usage map[string]any) map[string]any {
	return whitelistedTaskAttemptMap(usage, []string{
		"inputTokens", "promptTokens", "input_tokens", "prompt_tokens",
		"cachedInputTokens", "cachedPromptTokens", "cached_input_tokens", "cached_tokens",
		"cachedInputTokensKnown",
		"outputTokens", "completionTokens", "output_tokens", "completion_tokens",
		"totalTokens", "total_tokens",
	})
}

func minimalTaskAttemptMetrics(metrics map[string]any) map[string]any {
	return whitelistedTaskAttemptMap(metrics, []string{
		"platformPriority", "currentPriority", "loadRatio", "loadAvoided",
		"loadMetrics", "fullReasons", "selectionReason",
		"cacheAffinityKey", "cacheAdjustedPriority", "cacheAffinitySamples",
		"cacheAffinityScore", "cacheAffinityConfidence", "cacheAffinityHitRatio",
		"cacheAffinityEMAHitRatio", "cacheAffinityLastHitRatio",
		"cacheAffinityCachedInputTokens", "cacheAffinityBoost",
		"cacheAffinityApplied", "cacheAffinityMatched",
		"cacheAffinityCandidateCount", "cacheAffinityMatchedPrefixDepth",
		"cacheAffinityOverrideReason",
	})
}

func whitelistedTaskAttemptMap(input map[string]any, keys []string) map[string]any {
	out := map[string]any{}
	for _, key := range keys {
		if value, ok := input[key]; ok {
			out[key] = value
		}
	}
	return out
}

func (s *Store) FinishTaskSuccess(ctx context.Context, input FinishTaskSuccessInput) (GatewayTask, error) {
	resultReport := sanitizeJSONForStorageWithReport(minimalTaskResult(input.Result))
	if resultReport.BinaryCount > 0 {
		return GatewayTask{}, &taskPayloadBinaryError{
			target: ErrTaskResultBinaryNotMaterialized,
			code:   "result_binary_not_materialized",
			count:  resultReport.BinaryCount,
		}
	}
	resultJSON, _ := json.Marshal(resultReport.Value)
	billingsJSON, _ := json.Marshal(sanitizeJSONForStorage(input.Billings))
	usageJSON, _ := json.Marshal(sanitizeJSONForStorage(emptyObjectIfNil(input.Usage)))
	metricsJSON, _ := json.Marshal(sanitizeJSONForStorage(emptyObjectIfNil(input.Metrics)))
	attemptUsageJSON, _ := json.Marshal(sanitizeJSONForStorage(minimalTaskAttemptUsage(input.Usage)))
	attemptMetricsJSON, _ := json.Marshal(sanitizeJSONForStorage(minimalTaskAttemptMetrics(input.Metrics)))
	billingSummaryJSON, _ := json.Marshal(sanitizeJSONForStorage(emptyObjectIfNil(input.BillingSummary)))
	pricingSnapshotJSON, _ := json.Marshal(sanitizeJSONForStorage(emptyObjectIfNil(input.PricingSnapshot)))
	finalChargeAmount := strings.TrimSpace(input.FinalChargeAmountText)
	if finalChargeAmount == "" {
		finalChargeAmount = strconv.FormatFloat(input.FinalChargeAmount, 'f', 9, 64)
	}
	currency := normalizeWalletCurrency(input.BillingCurrency)
	finalizedAdmission := false
	err := s.beginTransaction(ctx, func(tx pgx.Tx) error {
		if strings.TrimSpace(input.AttemptID) != "" {
			if _, err := tx.Exec(ctx, `
UPDATE gateway_task_attempts
SET status = 'succeeded',
    retryable = false,
    request_id = NULLIF($2, ''),
    status_code = NULL,
    usage = $6::jsonb,
    metrics = $7::jsonb,
    response_snapshot = '{}'::jsonb,
    pricing_snapshot = '{}'::jsonb,
    request_fingerprint = NULL,
    upstream_submission_status = 'response_received',
    upstream_submission_updated_at = now(),
    response_started_at = $3::timestamptz,
    response_finished_at = $4::timestamptz,
    response_duration_ms = $5,
    error_code = NULL,
    error_message = NULL,
	public_error = NULL,
    finished_at = now()
WHERE id = $1::uuid`,
				input.AttemptID, input.RequestID,
				nullableTime(input.ResponseStartedAt), nullableTime(input.ResponseFinishedAt), input.ResponseDurationMS,
				attemptUsageJSON, attemptMetricsJSON,
			); err != nil {
				return err
			}
		}
		var billingStatus string
		err := tx.QueryRow(ctx, `
UPDATE gateway_tasks
SET status = 'succeeded',
    result = $2::jsonb,
    billings = $3::jsonb,
    request_id = NULLIF($4, ''),
    resolved_model = NULLIF($5, ''),
    usage = $6::jsonb,
    metrics = $7::jsonb,
    billing_summary = $8::jsonb,
    final_charge_amount = $9::numeric,
    billing_version = 'effective-pricing-v2',
    billing_status = CASE
      WHEN run_mode IN ('production', 'acceptance', 'acceptance_canary') AND gateway_user_id IS NOT NULL THEN 'pending'
      ELSE 'not_required'
    END,
    billing_currency = $10,
    pricing_snapshot = $11::jsonb,
    request_fingerprint = NULLIF($12, ''),
    billing_updated_at = now(),
    response_started_at = $13::timestamptz,
    response_finished_at = $14::timestamptz,
    response_duration_ms = $15,
    error = NULL,
    error_code = NULL,
    error_message = NULL,
	public_error = NULL,
    locked_by = NULL,
    locked_at = NULL,
    heartbeat_at = NULL,
    execution_token = NULL,
    execution_lease_expires_at = NULL,
    remote_task_payload = '{}'::jsonb,
    finished_at = now(),
    updated_at = now()
WHERE id = $1::uuid
  AND status = 'running'
  AND execution_token = $16::uuid
RETURNING billing_status`,
			input.TaskID,
			string(resultJSON),
			string(billingsJSON),
			input.RequestID,
			input.ResolvedModel,
			string(usageJSON),
			string(metricsJSON),
			string(billingSummaryJSON),
			finalChargeAmount,
			currency,
			string(pricingSnapshotJSON),
			input.RequestFingerprint,
			nullableTime(input.ResponseStartedAt),
			nullableTime(input.ResponseFinishedAt),
			input.ResponseDurationMS,
			input.ExecutionToken,
		).Scan(&billingStatus)
		if errors.Is(err, pgx.ErrNoRows) {
			return ErrTaskExecutionLeaseLost
		}
		if err != nil {
			return err
		}
		payloadJSON, _ := json.Marshal(map[string]any{"taskId": input.TaskID, "pricingVersion": "effective-pricing-v2"})
		_, err = tx.Exec(ctx, `
INSERT INTO settlement_outbox (
  task_id, event_type, action, amount, currency, pricing_snapshot, payload, status, next_attempt_at
)
SELECT id, 'task.billing.settle', 'settle', $2::numeric, $3, $4::jsonb, $5::jsonb, 'pending', now()
FROM gateway_tasks
WHERE id = $1::uuid
  AND run_mode IN ('production', 'acceptance', 'acceptance_canary')
  AND gateway_user_id IS NOT NULL
ON CONFLICT (task_id, event_type) DO NOTHING`,
			input.TaskID, finalChargeAmount, currency, string(pricingSnapshotJSON), string(payloadJSON))
		if err != nil {
			return err
		}
		if input.FinalizeAdmission {
			if _, err := tx.Exec(ctx, `
UPDATE gateway_concurrency_leases
SET released_at = statement_timestamp()
WHERE task_id = $1::uuid
  AND released_at IS NULL`, input.TaskID); err != nil {
				return err
			}
			if _, err := tx.Exec(ctx, `
DELETE FROM gateway_task_admissions
WHERE task_id = $1::uuid`, input.TaskID); err != nil {
				return err
			}
			finalizedAdmission = true
		}
		var eventID string
		var eventSeq int64
		if err := tx.QueryRow(ctx, `
WITH next_seq AS (
  SELECT COALESCE(MAX(seq), 0) + 1 AS seq
  FROM gateway_task_events
  WHERE task_id = $1::uuid
)
INSERT INTO gateway_task_events (
  task_id, seq, event_type, status, phase, progress, message, payload, simulated
)
SELECT $1::uuid, next_seq.seq, 'task.completed', 'succeeded', NULL, 0, NULL, '{}'::jsonb, $2
FROM next_seq
RETURNING id::text, seq`,
			input.TaskID,
			input.Simulated,
		).Scan(&eventID, &eventSeq); err != nil {
			return err
		}
		callbackURL := strings.TrimSpace(input.CompletionCallbackURL)
		if callbackURL != "" {
			if _, err := tx.Exec(ctx, `
INSERT INTO gateway_task_callback_outbox (task_id, event_id, seq, callback_url, payload)
VALUES ($1::uuid, $2::uuid, $3, $4, '{}'::jsonb)
ON CONFLICT (task_id, seq, callback_url) DO NOTHING`,
				input.TaskID,
				eventID,
				eventSeq,
				callbackURL,
			); err != nil {
				return err
			}
		}
		_ = billingStatus
		return nil
	})
	if err != nil {
		return GatewayTask{}, err
	}
	if finalizedAdmission {
		s.notifyTaskAdmissionBestEffort(ctx, "*")
	}
	return s.GetTask(ctx, input.TaskID)
}

func (s *Store) FinishTaskManualReview(ctx context.Context, input FinishTaskManualReviewInput) (GatewayTask, error) {
	status := strings.TrimSpace(input.TaskStatus)
	if status != "succeeded" && status != "failed" {
		return GatewayTask{}, fmt.Errorf("manual review task status must be succeeded or failed")
	}
	message := truncateUTF8Bytes(input.Message, 2048)
	result := input.Result
	if status == "failed" {
		result = nil
	}
	resultReport := sanitizeJSONForStorageWithReport(minimalTaskResult(result))
	if resultReport.BinaryCount > 0 {
		return GatewayTask{}, &taskPayloadBinaryError{
			target: ErrTaskResultBinaryNotMaterialized,
			code:   "result_binary_not_materialized",
			count:  resultReport.BinaryCount,
		}
	}
	resultJSON, _ := json.Marshal(resultReport.Value)
	pricingSnapshotJSON, _ := json.Marshal(sanitizeJSONForStorage(emptyObjectIfNil(input.PricingSnapshot)))
	err := s.beginTransaction(ctx, func(tx pgx.Tx) error {
		if strings.TrimSpace(input.AttemptID) != "" {
			attemptStatus := "failed"
			if status == "succeeded" {
				attemptStatus = "succeeded"
			}
			if _, err := tx.Exec(ctx, `
UPDATE gateway_task_attempts
SET status = $2,
    request_id = NULLIF($3, ''),
    error_code = CASE WHEN $2 = 'failed' THEN NULLIF($4, '') ELSE NULL END,
    error_message = CASE WHEN $2 = 'failed' THEN NULLIF($5, '') ELSE NULL END,
    upstream_submission_status = CASE WHEN $2 = 'succeeded' THEN 'response_received' ELSE upstream_submission_status END,
    upstream_submission_updated_at = now(),
    response_started_at = $6::timestamptz,
    response_finished_at = $7::timestamptz,
    response_duration_ms = $8,
    finished_at = now()
WHERE id = $1::uuid`, input.AttemptID, attemptStatus, input.RequestID, input.Code, message,
				nullableTime(input.ResponseStartedAt), nullableTime(input.ResponseFinishedAt), input.ResponseDurationMS); err != nil {
				return err
			}
		}
		tag, err := tx.Exec(ctx, `
UPDATE gateway_tasks
SET status = $2,
    result = $3::jsonb,
    request_id = NULLIF($4, ''),
    error = NULL,
    error_code = NULLIF($5, ''),
    error_message = NULLIF($6, ''),
    billing_status = 'manual_review',
    pricing_snapshot = CASE WHEN $7::jsonb = '{}'::jsonb THEN pricing_snapshot ELSE $7::jsonb END,
    request_fingerprint = COALESCE(NULLIF($8, ''), request_fingerprint),
    billing_updated_at = now(),
    response_started_at = $9::timestamptz,
    response_finished_at = $10::timestamptz,
    response_duration_ms = $11,
    locked_by = NULL,
    locked_at = NULL,
    heartbeat_at = NULL,
    execution_token = NULL,
    execution_lease_expires_at = NULL,
    remote_task_payload = '{}'::jsonb,
    finished_at = now(),
    updated_at = now()
WHERE id = $1::uuid
  AND status = 'running'
  AND execution_token = $12::uuid`, input.TaskID, status, string(resultJSON), input.RequestID, input.Code, message,
			string(pricingSnapshotJSON), input.RequestFingerprint, nullableTime(input.ResponseStartedAt),
			nullableTime(input.ResponseFinishedAt), input.ResponseDurationMS, input.ExecutionToken)
		if err != nil {
			return err
		}
		if tag.RowsAffected() != 1 {
			return ErrTaskExecutionLeaseLost
		}
		payloadJSON, _ := json.Marshal(map[string]any{
			"taskId": input.TaskID, "classification": input.Code,
		})
		_, err = tx.Exec(ctx, `
INSERT INTO settlement_outbox (
  task_id, event_type, action, amount, currency, pricing_snapshot, payload,
  status, next_attempt_at, manual_review_reason
)
SELECT id, 'task.billing.review', 'release', reservation_amount, billing_currency,
       pricing_snapshot, $2::jsonb, 'manual_review', now(), $3
FROM gateway_tasks
WHERE id = $1::uuid
  AND run_mode IN ('production', 'acceptance', 'acceptance_canary')
  AND gateway_user_id IS NOT NULL
ON CONFLICT (task_id, event_type) DO NOTHING`, input.TaskID, string(payloadJSON), input.Code)
		return err
	})
	if err != nil {
		return GatewayTask{}, err
	}
	return s.GetTask(ctx, input.TaskID)
}

func (s *Store) SettleTaskBilling(ctx context.Context, task GatewayTask) error {
	if task.FinalChargeAmount <= 0 || strings.TrimSpace(task.GatewayUserID) == "" {
		return nil
	}
	currency := strings.TrimSpace(taskBillingString(task.BillingSummary["currency"]))
	if currency == "" || currency == "mixed" {
		currency = "resource"
	}
	modelIdentity := taskBillingModelIdentity(task)
	metadataMap := map[string]any{
		"taskId":         task.ID,
		"kind":           task.Kind,
		"model":          modelIdentity.Model,
		"requestedModel": modelIdentity.RequestedModel,
		"resolvedModel":  modelIdentity.ResolvedModel,
		"modelName":      modelIdentity.ModelName,
		"modelAlias":     modelIdentity.ModelAlias,
		"providerModel":  modelIdentity.ProviderModelName,
		"billings":       task.Billings,
		"billingSummary": task.BillingSummary,
	}
	metadata, _ := json.Marshal(sanitizeJSONForStorage(metadataMap))
	return s.beginTransaction(ctx, func(tx pgx.Tx) error {
		if _, err := tx.Exec(ctx, `
INSERT INTO gateway_wallet_accounts (
  gateway_tenant_id, gateway_user_id, tenant_id, tenant_key, user_id, currency
)
VALUES (NULLIF($1, '')::uuid, $2::uuid, NULLIF($3, ''), NULLIF($4, ''), NULLIF($5, ''), $6)
ON CONFLICT (gateway_user_id, currency) DO NOTHING`,
			task.GatewayTenantID, task.GatewayUserID, task.TenantID, task.TenantKey, task.UserID, currency); err != nil {
			return err
		}
		if err := ensureWalletAccountAuditGuard(ctx, tx, task.GatewayUserID, currency); err != nil {
			return err
		}
		var exists bool
		var accountID string
		var balanceBefore float64
		var frozenBefore float64
		var gatewayTenantID string
		if err := tx.QueryRow(ctx, `
SELECT id::text, balance::float8, frozen_balance::float8, COALESCE(gateway_tenant_id::text, '')
FROM gateway_wallet_accounts
WHERE gateway_user_id = $1::uuid
  AND currency = $2
FOR UPDATE`, task.GatewayUserID, currency).Scan(&accountID, &balanceBefore, &frozenBefore, &gatewayTenantID); err != nil {
			return err
		}
		if err := tx.QueryRow(ctx, `
SELECT EXISTS (
  SELECT 1
  FROM gateway_wallet_transactions
  WHERE account_id = $1::uuid
    AND idempotency_key = $2
)`, accountID, billingIdempotencyKey(task.ID)).Scan(&exists); err != nil {
			return err
		}
		if exists {
			return nil
		}

		amount := roundMoney(task.FinalChargeAmount)
		reservationKey, reservedAmount, err := activeWalletReservation(ctx, tx, accountID, task.ID)
		if err != nil {
			return err
		}
		reservedAmount = roundMoney(reservedAmount)
		spendableForTask := roundMoney(balanceBefore - frozenBefore + reservedAmount)
		if spendableForTask+0.000001 < amount {
			return fmt.Errorf("%w: required %.6f %s, available %.6f", ErrInsufficientWalletBalance, amount, currency, spendableForTask)
		}

		balanceAfter := roundMoney(balanceBefore - amount)
		frozenAfter := roundMoney(frozenBefore - reservedAmount)
		if frozenAfter < 0 {
			frozenAfter = 0
		}
		if _, err := tx.Exec(ctx, `
UPDATE gateway_wallet_accounts
SET balance = $2,
    total_spent = total_spent + $3,
    frozen_balance = $4,
    updated_at = now()
WHERE id = $1::uuid`, accountID, balanceAfter, amount, frozenAfter); err != nil {
			return err
		}
		if reservedAmount > 0 {
			releaseMetadata, _ := json.Marshal(map[string]any{
				"taskId":       task.ID,
				"reason":       "task_billing_settled",
				"reserved":     reservedAmount,
				"frozenBefore": roundMoney(frozenBefore),
				"frozenAfter":  frozenAfter,
			})
			if _, err := tx.Exec(ctx, `
INSERT INTO gateway_wallet_transactions (
  account_id, gateway_tenant_id, gateway_user_id, direction, transaction_type,
  amount, balance_before, balance_after, idempotency_key, reference_type, reference_id, metadata
)
VALUES (
  $1::uuid, NULLIF($2, '')::uuid, $3::uuid, 'credit', 'release',
  $4, $5, $6, $7, 'gateway_task', $8, $9::jsonb
)
ON CONFLICT (account_id, idempotency_key) WHERE idempotency_key IS NOT NULL DO NOTHING`,
				accountID, firstNonEmpty(gatewayTenantID, task.GatewayTenantID), task.GatewayUserID, reservedAmount, roundMoney(balanceBefore), roundMoney(balanceBefore), billingReservationReleaseIdempotencyKey(reservationKey), task.ID, string(releaseMetadata)); err != nil {
				return err
			}
		}
		billingMetadata := mergeObjects(metadataMap, map[string]any{
			"reservedAmount": reservedAmount,
			"frozenBefore":   roundMoney(frozenBefore),
			"frozenAfter":    frozenAfter,
		})
		metadata, _ = json.Marshal(sanitizeJSONForStorage(billingMetadata))
		if _, err := tx.Exec(ctx, `
INSERT INTO gateway_wallet_transactions (
  account_id, gateway_tenant_id, gateway_user_id, direction, transaction_type,
  amount, balance_before, balance_after, idempotency_key, reference_type, reference_id, metadata
)
VALUES (
  $1::uuid, NULLIF($2, '')::uuid, $3::uuid, 'debit', 'task_billing',
  $4, $5, $6, $7, 'gateway_task', $8, $9::jsonb
)`,
			accountID, firstNonEmpty(gatewayTenantID, task.GatewayTenantID), task.GatewayUserID, amount, roundMoney(balanceBefore), balanceAfter, billingIdempotencyKey(task.ID), task.ID, string(metadata)); err != nil {
			return err
		}
		return nil
	})
}

func billingIdempotencyKey(taskID string) string {
	return "task:" + taskID + ":billing"
}

func roundMoney(value float64) float64 {
	if value < 0 {
		return -roundMoney(-value)
	}
	return float64(int64(value*1000000+0.5)) / 1000000
}

func taskBillingString(value any) string {
	if text, ok := value.(string); ok {
		return strings.TrimSpace(text)
	}
	return ""
}

func (s *Store) FinishTaskFailure(ctx context.Context, input FinishTaskFailureInput) (GatewayTask, error) {
	metricsJSON, _ := json.Marshal(sanitizeJSONForStorage(emptyObjectIfNil(input.Metrics)))
	resultJSON, _ := json.Marshal(minimalTaskResult(nil))
	message := truncateUTF8Bytes(input.Message, 2048)
	publicErrorJSON := encodePublicErrorSnapshot(input.Code, message, input.StatusCode, true, input.RequestID, input.TaskID)
	finalizedAdmission := false
	err := s.beginTransaction(ctx, func(tx pgx.Tx) error {
		tag, err := tx.Exec(ctx, `
	UPDATE gateway_tasks
	SET status = 'failed',
	    error = NULL,
	    error_code = NULLIF($3::text, ''),
	    error_message = NULLIF($2::text, ''),
	    public_error = $11::jsonb,
	    request_id = NULLIF($4::text, ''),
	    metrics = $5::jsonb,
	    response_started_at = $6::timestamptz,
	    response_finished_at = $7::timestamptz,
	    response_duration_ms = $8,
	    result = $9::jsonb,
	    billing_status = CASE
	      WHEN run_mode NOT IN ('production', 'acceptance', 'acceptance_canary') OR gateway_user_id IS NULL THEN 'not_required'
	      WHEN reservation_amount > 0 THEN 'pending'
	      ELSE 'released'
	    END,
	    billing_updated_at = now(),
	    locked_by = NULL,
	    locked_at = NULL,
	    heartbeat_at = NULL,
	    execution_token = NULL,
	    execution_lease_expires_at = NULL,
	    remote_task_payload = '{}'::jsonb,
	    finished_at = now(),
	    updated_at = now()
WHERE id = $1::uuid
  AND status IN ('queued', 'running')
  AND execution_token = $10::uuid`,
			input.TaskID,
			message,
			input.Code,
			input.RequestID,
			string(metricsJSON),
			nullableTime(input.ResponseStartedAt),
			nullableTime(input.ResponseFinishedAt),
			input.ResponseDurationMS,
			string(resultJSON),
			input.ExecutionToken,
			string(publicErrorJSON),
		)
		if err != nil {
			return err
		}
		if tag.RowsAffected() != 1 {
			return ErrTaskExecutionLeaseLost
		}
		payloadJSON, _ := json.Marshal(map[string]any{"taskId": input.TaskID, "reason": input.Code})
		_, err = tx.Exec(ctx, `
INSERT INTO settlement_outbox (
  task_id, event_type, action, amount, currency, pricing_snapshot, payload, status, next_attempt_at
)
SELECT id, 'task.billing.release', 'release', reservation_amount, billing_currency,
       pricing_snapshot, $2::jsonb, 'pending', now()
FROM gateway_tasks
WHERE id = $1::uuid
  AND run_mode IN ('production', 'acceptance', 'acceptance_canary')
  AND gateway_user_id IS NOT NULL
  AND reservation_amount > 0
ON CONFLICT (task_id, event_type) DO NOTHING`, input.TaskID, string(payloadJSON))
		if err != nil {
			return err
		}
		if input.FinalizeAdmission {
			if _, err := tx.Exec(ctx, `
UPDATE gateway_concurrency_leases
SET released_at = statement_timestamp()
WHERE task_id = $1::uuid
  AND released_at IS NULL`, input.TaskID); err != nil {
				return err
			}
			if _, err := tx.Exec(ctx, `
DELETE FROM gateway_task_admissions
WHERE task_id = $1::uuid`, input.TaskID); err != nil {
				return err
			}
			finalizedAdmission = true
		}
		var eventID string
		var eventSeq int64
		if err := tx.QueryRow(ctx, `
WITH next_seq AS (
  SELECT COALESCE(MAX(seq), 0) + 1 AS seq
  FROM gateway_task_events
  WHERE task_id = $1::uuid
)
INSERT INTO gateway_task_events (
  task_id, seq, event_type, status, phase, progress, message, payload, simulated
)
SELECT $1::uuid, next_seq.seq, 'task.failed', 'failed', NULL, 0, NULL, '{}'::jsonb, $2
FROM next_seq
RETURNING id::text, seq`,
			input.TaskID,
			input.Simulated,
		).Scan(&eventID, &eventSeq); err != nil {
			return err
		}
		callbackURL := strings.TrimSpace(input.CompletionCallbackURL)
		if callbackURL != "" {
			if _, err := tx.Exec(ctx, `
INSERT INTO gateway_task_callback_outbox (task_id, event_id, seq, callback_url, payload)
VALUES ($1::uuid, $2::uuid, $3, $4, '{}'::jsonb)
ON CONFLICT (task_id, seq, callback_url) DO NOTHING`,
				input.TaskID,
				eventID,
				eventSeq,
				callbackURL,
			); err != nil {
				return err
			}
		}
		return nil
	})
	if err != nil {
		return GatewayTask{}, err
	}
	if finalizedAdmission {
		s.notifyTaskAdmissionBestEffort(ctx, "*")
	}
	return s.GetTask(ctx, input.TaskID)
}

func nullableTime(value time.Time) any {
	if value.IsZero() {
		return nil
	}
	return value
}

func truncateUTF8Bytes(value string, maxBytes int) string {
	if maxBytes <= 0 || len(value) <= maxBytes {
		return value
	}
	end := maxBytes
	for end > 0 && !utf8.ValidString(value[:end]) {
		end--
	}
	return value[:end]
}

func minimalTaskResult(input map[string]any) map[string]any {
	result := make(map[string]any, len(input))
	for key, value := range input {
		switch key {
		case "raw",
			"raw_data",
			"rawData",
			"provider_response",
			"providerResponse",
			"submit",
			"file_retrieve",
			"fileRetrieve",
			"upstream_task_id",
			"remote_task_id":
			continue
		}
		result[key] = value
	}
	return result
}

var allowedTaskEventTypes = map[string]struct{}{
	"task.accepted":                    {},
	"task.running":                     {},
	"task.queued":                      {},
	"task.completed":                   {},
	"task.failed":                      {},
	"task.cancelled":                   {},
	"task.attempt.failed":              {},
	"task.billing.settled":             {},
	"task.billing.released":            {},
	"task.billing.review":              {},
	"task.policy.auto_disabled":        {},
	"task.policy.degraded":             {},
	"task.policy.failover_disabled":    {},
	"task.policy.failover_cooled_down": {},
	"task.policy.priority_demoted":     {},
}

func taskEventAllowed(eventType string) bool {
	_, ok := allowedTaskEventTypes[strings.TrimSpace(eventType)]
	return ok
}

func taskEventTerminal(eventType string) bool {
	switch strings.TrimSpace(eventType) {
	case "task.completed", "task.failed", "task.cancelled",
		"task.billing.settled", "task.billing.released", "task.billing.review":
		return true
	default:
		return false
	}
}

func taskEventPlatformID(payload map[string]any) string {
	platformID, _ := payload["platformId"].(string)
	return strings.TrimSpace(platformID)
}

func (s *Store) AddTaskEvent(ctx context.Context, taskID string, eventType string, status string, phase string, progress float64, message string, payload map[string]any, simulated bool) (TaskEvent, error) {
	return s.addTaskEvent(ctx, taskID, eventType, status, phase, progress, message, payload, simulated, "")
}

// AddTaskEventWithCallback persists the event and its callback outbox item in
// one transaction. This closes the crash window between event creation and
// callback registration and avoids a second synchronous-replication commit for
// every emitted task event.
func (s *Store) AddTaskEventWithCallback(
	ctx context.Context,
	taskID string,
	eventType string,
	status string,
	phase string,
	progress float64,
	message string,
	payload map[string]any,
	simulated bool,
	callbackURL string,
) (TaskEvent, error) {
	return s.addTaskEvent(ctx, taskID, eventType, status, phase, progress, message, payload, simulated, callbackURL)
}

func (s *Store) addTaskEvent(
	ctx context.Context,
	taskID string,
	eventType string,
	status string,
	phase string,
	progress float64,
	message string,
	payload map[string]any,
	simulated bool,
	callbackURL string,
) (TaskEvent, error) {
	eventType = strings.TrimSpace(eventType)
	if !taskEventAllowed(eventType) {
		return TaskEvent{SkippedReason: "unknown_type"}, nil
	}
	platformID := taskEventPlatformID(payload)
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return TaskEvent{}, err
	}
	defer rollbackTransaction(tx)
	if !taskEventTerminal(eventType) {
		// Progress/history events are recoverable hints. A later synchronous
		// upstream or terminal task transition is their durability barrier, so
		// waiting for the cross-region replica on every event only amplifies
		// queue latency under media bursts.
		if _, err := tx.Exec(ctx, `SET LOCAL synchronous_commit = off`); err != nil {
			return TaskEvent{}, err
		}
	}
	if _, err := tx.Exec(ctx, `SELECT 1 FROM gateway_tasks WHERE id = $1::uuid FOR UPDATE`, taskID); err != nil {
		return TaskEvent{}, err
	}
	var previousType string
	var previousStatus string
	var previousPlatformID string
	var previousSimulated bool
	err = tx.QueryRow(ctx, `
SELECT event_type, COALESCE(status, ''), COALESCE(platform_id::text, ''), simulated
FROM gateway_task_events
WHERE task_id = $1::uuid
ORDER BY seq DESC
LIMIT 1`, taskID).Scan(&previousType, &previousStatus, &previousPlatformID, &previousSimulated)
	if err != nil && !errors.Is(err, pgx.ErrNoRows) {
		return TaskEvent{}, err
	}
	if err == nil &&
		previousType == eventType &&
		previousStatus == strings.TrimSpace(status) &&
		previousPlatformID == platformID &&
		previousSimulated == simulated {
		if err := tx.Commit(ctx); err != nil {
			return TaskEvent{}, err
		}
		return TaskEvent{SkippedReason: "duplicate"}, nil
	}
	if !taskEventTerminal(eventType) {
		var nonTerminalCount int
		if err := tx.QueryRow(ctx, `
SELECT count(*)
FROM gateway_task_events
WHERE task_id = $1::uuid
  AND event_type NOT IN (
    'task.completed', 'task.failed', 'task.cancelled',
    'task.billing.settled', 'task.billing.released', 'task.billing.review'
  )`, taskID).Scan(&nonTerminalCount); err != nil {
			return TaskEvent{}, err
		}
		if nonTerminalCount >= 16 {
			if err := tx.Commit(ctx); err != nil {
				return TaskEvent{}, err
			}
			return TaskEvent{SkippedReason: "budget_exceeded"}, nil
		}
	}
	var event TaskEvent
	var payloadBytes []byte
	err = tx.QueryRow(ctx, `
WITH next_seq AS (
  SELECT COALESCE(MAX(seq), 0) + 1 AS seq
  FROM gateway_task_events
  WHERE task_id = $1::uuid
)
INSERT INTO gateway_task_events (task_id, seq, event_type, status, phase, progress, message, payload, simulated, platform_id)
SELECT $1::uuid, next_seq.seq, $2::text, NULLIF($3::text, ''), NULL, 0, NULL, '{}'::jsonb, $4,
       NULLIF($5::text, '')::uuid
FROM next_seq
RETURNING id::text, task_id::text, seq, event_type, COALESCE(status, ''), COALESCE(phase, ''),
          COALESCE(progress, 0)::float8, COALESCE(message, ''), payload, simulated, created_at,
          COALESCE(platform_id::text, '')`,
		taskID,
		eventType,
		status,
		simulated,
		platformID,
	).Scan(
		&event.ID,
		&event.TaskID,
		&event.Seq,
		&event.EventType,
		&event.Status,
		&event.Phase,
		&event.Progress,
		&event.Message,
		&payloadBytes,
		&event.Simulated,
		&event.CreatedAt,
		&event.PlatformID,
	)
	if err != nil {
		return TaskEvent{}, err
	}
	event.Payload = decodeObject(payloadBytes)
	callbackURL = strings.TrimSpace(callbackURL)
	if callbackURL != "" {
		if _, err := tx.Exec(ctx, `
INSERT INTO gateway_task_callback_outbox (task_id, event_id, seq, callback_url, payload)
VALUES ($1::uuid, $2::uuid, $3, $4, '{}'::jsonb)
ON CONFLICT (task_id, seq, callback_url) DO NOTHING`,
			event.TaskID,
			event.ID,
			event.Seq,
			callbackURL,
		); err != nil {
			return TaskEvent{}, err
		}
	}
	if err := tx.Commit(ctx); err != nil {
		return TaskEvent{}, err
	}
	return event, nil
}

func (s *Store) QueueTaskCallback(ctx context.Context, event TaskEvent, callbackURL string) error {
	if callbackURL == "" || event.ID == "" {
		return nil
	}
	_, err := s.pool.Exec(ctx, `
INSERT INTO gateway_task_callback_outbox (task_id, event_id, seq, callback_url, payload)
VALUES ($1::uuid, $2::uuid, $3, $4, '{}'::jsonb)
ON CONFLICT (task_id, seq, callback_url) DO NOTHING`,
		event.TaskID,
		event.ID,
		event.Seq,
		callbackURL,
	)
	return err
}

func (s *Store) RecordClientAssignment(ctx context.Context, candidate RuntimeModelCandidate) error {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer rollbackTransaction(tx)
	// This counter only influences candidate load balancing and is rebuilt from
	// active attempts during runtime recovery. Keeping a synchronous-replication
	// commit open while updating one row per upstream client serializes every
	// assignment for a hot model across regions.
	if _, err := tx.Exec(ctx, `SET LOCAL synchronous_commit = off`); err != nil {
		return err
	}
	if _, err := tx.Exec(ctx, `
INSERT INTO runtime_client_states (
  client_id, platform_id, provider, method_name, queue_key, running_count, last_assigned_at
)
VALUES ($1, $2::uuid, $3, $4, $5, 1, now())
ON CONFLICT (client_id) DO UPDATE
SET running_count = runtime_client_states.running_count + 1,
    last_assigned_at = now(),
    updated_at = now()`,
		candidate.ClientID,
		candidate.PlatformID,
		candidate.Provider,
		candidate.ModelType,
		candidate.QueueKey,
	); err != nil {
		return err
	}
	return tx.Commit(ctx)
}

func (s *Store) RecordClientRelease(ctx context.Context, clientID string, lastError string) error {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer rollbackTransaction(tx)
	// runtime_client_states is derived scheduler state; see
	// RecordClientAssignment for why it must not hold its hot row lock through
	// a cross-region synchronous replication wait.
	if _, err := tx.Exec(ctx, `SET LOCAL synchronous_commit = off`); err != nil {
		return err
	}
	if _, err := tx.Exec(ctx, `
UPDATE runtime_client_states
SET running_count = GREATEST(running_count - 1, 0),
    last_error = NULLIF($2::text, ''),
    updated_at = now()
WHERE client_id = $1`, clientID, lastError); err != nil {
		return err
	}
	return tx.Commit(ctx)
}
