package server

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"sync"
	"time"

	"github.com/pingcap/failpoint"
	"go.uber.org/zap"

	"github.com/mem9-ai/drive9/pkg/backend"
	"github.com/mem9-ai/drive9/pkg/datastore"
	"github.com/mem9-ai/drive9/pkg/embedding"
	"github.com/mem9-ai/drive9/pkg/logger"
	"github.com/mem9-ai/drive9/pkg/meta"
	"github.com/mem9-ai/drive9/pkg/metrics"
	"github.com/mem9-ai/drive9/pkg/semantic"
)

// semanticTenantRef identifies a schedulable tenant for the worker.
type semanticTenantRef struct {
	id     string
	tenant *meta.Tenant
}

// semanticTaskAction classifies the terminal action for a processed task.
type semanticTaskAction string

const (
	semanticTaskActionAck        semanticTaskAction = "ack"
	semanticTaskActionDiscard    semanticTaskAction = "discard"
	semanticTaskActionRetry      semanticTaskAction = "retry"
	semanticTaskActionDefer      semanticTaskAction = "defer"
	semanticTaskActionDeadLetter semanticTaskAction = "dead_letter"

	terminalMediaDiscardBatchSize = 1000
	embeddingBackfillPageSize     = 100
)

// semanticTaskOutcome is the result of dispatching one task.
type semanticTaskOutcome struct {
	action                 semanticTaskAction
	result                 string
	message                string
	discardFreshMediaPeers bool
}

// semanticTaskLeaseExecution manages lease renewal for a single in-flight task.
type semanticTaskLeaseExecution struct {
	ctx    context.Context
	cancel context.CancelFunc

	stopCh   chan struct{}
	doneCh   chan struct{}
	stopOnce sync.Once

	mu             sync.Mutex
	lost           bool
	renewAttempted bool
	lastLeaseUntil *time.Time
}

// semanticTaskLeaseOutcome is the result of stopping a lease execution.
type semanticTaskLeaseOutcome struct {
	lost           bool
	renewAttempted bool
	lastLeaseUntil *time.Time
}

// processTask claims, processes, and finalizes one semantic task. This is the
// core task processing pipeline moved from semantic_worker.go, preserving all
// lease execution, dispatch, ack/retry/dead-letter, and failpoint hooks.
func (m *tenantWorkerManager) processTask(ctx context.Context, target *tenantTarget, task *semantic.Task) {
	if task == nil || target == nil {
		return
	}
	start := time.Now()
	result := "ok"
	tenantID := target.tenantID
	tidbCloudOrgID := target.metricOrgID()
	defer func() {
		metrics.RecordTenantOperationWithOrg(tenantID, tidbCloudOrgID, "semantic_worker", string(task.TaskType), result, time.Since(start))
	}()
	leaseExec := m.startTaskLeaseExecution(ctx, target, task)
	var (
		leaseStopOnce sync.Once
		leaseOutcome  semanticTaskLeaseOutcome
	)
	stopLease := func() semanticTaskLeaseOutcome {
		leaseStopOnce.Do(func() {
			leaseOutcome = leaseExec.stop()
		})
		return leaseOutcome
	}
	defer stopLease()
	outcome := m.dispatchTask(leaseExec.leaseCtx(), target, task)
	// Stop renewal before ack/retry so only one lease owner can finalize outcome.
	leaseOutcome = stopLease()
	if leaseOutcome.lost {
		result = "lease_lost"
		return
	}
	result = outcome.result
	m.injectBeforeSemanticTaskFinalize(target.tenantID, target.store, task, outcome)
	// Re-check ownership after the handler returns so tests can deterministically
	// force a lease-loss window before any terminal state mutation happens.
	stillOwned, err := m.semanticTaskStillOwned(ctx, target, task, outcome.result)
	if err != nil {
		logger.Warn(ctx, "tenant_worker_finalize_ownership_check_failed",
			append([]zap.Field{
				zap.String("tenant_id", target.tenantID),
				zap.String("result", outcome.result),
			}, append(semanticTaskLogFields(task), zap.Error(err))...)...)
	} else if !stillOwned {
		return
	}
	var finalized bool
	switch outcome.action {
	case semanticTaskActionAck:
		finalized = m.ackTask(ctx, target.tenantID, target.store, task, outcome.message)
	case semanticTaskActionDiscard:
		finalized = m.discardTask(ctx, target, task, outcome.message)
	case semanticTaskActionRetry:
		m.retryTask(ctx, target, task, outcome.message)
	case semanticTaskActionDefer:
		m.deferTask(ctx, target, task, outcome.message)
	case semanticTaskActionDeadLetter:
		m.deadLetterTask(ctx, target, task, outcome.message)
	}
	if finalized && outcome.discardFreshMediaPeers {
		m.discardFreshMediaPeers(ctx, target, task, outcome.result)
	}
}

func (m *tenantWorkerManager) discardTask(ctx context.Context, target *tenantTarget, task *semantic.Task, reason string) bool {
	if err := target.store.DiscardSemanticTask(ctx, task.TaskID, task.Receipt); err != nil {
		logger.Warn(ctx, "tenant_worker_discard_failed",
			append([]zap.Field{zap.String("tenant_id", target.tenantID), zap.String("reason", reason)},
				append(semanticTaskLogFields(task), zap.Error(err))...)...)
		return false
	}
	logger.Info(ctx, "tenant_worker_discard_ok",
		append([]zap.Field{zap.String("tenant_id", target.tenantID), zap.String("reason", reason)}, semanticTaskLogFields(task)...)...)
	return true
}

func (m *tenantWorkerManager) deferTask(ctx context.Context, target *tenantTarget, task *semantic.Task, message string) {
	start := time.Now()
	// The only current defer outcome is image extraction global_disabled.
	// Requeue immediately because claimableTaskTypes excludes image extraction
	// while the switch is on; eligibility resumes as soon as it is lifted.
	retryAt := time.Now().UTC()
	if err := target.store.DeferSemanticTask(ctx, task.TaskID, task.Receipt, retryAt, message); err != nil {
		metrics.RecordTenantOperationWithOrg(target.tenantID, target.metricOrgID(), "semantic_worker", "defer", "error", time.Since(start))
		logger.Warn(ctx, "tenant_worker_defer_failed",
			append([]zap.Field{
				zap.String("tenant_id", target.tenantID),
				zap.String("result", "error"),
			}, append(semanticTaskLogFields(task), zap.Error(err))...)...)
		return
	}
	metrics.RecordTenantOperationWithOrg(target.tenantID, target.metricOrgID(), "semantic_worker", "defer", "scheduled", time.Since(start))
	logger.Info(ctx, "tenant_worker_defer_scheduled",
		append([]zap.Field{
			zap.String("tenant_id", target.tenantID),
			zap.String("result", "scheduled"),
			zap.String("message", message),
			zap.Time("retry_at", retryAt),
		}, semanticTaskLogFields(task)...)...)
}

func (m *tenantWorkerManager) ackTask(ctx context.Context, tenantID string, store *datastore.Store, task *semantic.Task, reason string) bool {
	if err := store.AckSemanticTask(ctx, task.TaskID, task.Receipt); err != nil {
		logger.Warn(ctx, "tenant_worker_ack_failed",
			append([]zap.Field{
				zap.String("tenant_id", tenantID),
				zap.String("reason", reason),
				zap.String("result", "error"),
			}, append(semanticTaskLogFields(task), zap.Error(err))...)...)
		return false
	}
	logger.Info(ctx, "tenant_worker_ack_ok",
		append([]zap.Field{
			zap.String("tenant_id", tenantID),
			zap.String("reason", reason),
			zap.String("result", reason),
		}, semanticTaskLogFields(task)...)...)
	return true
}

func (m *tenantWorkerManager) retryTask(ctx context.Context, target *tenantTarget, task *semantic.Task, message string) {
	start := time.Now()
	tenantID := target.tenantID
	store := target.store
	retryAt := time.Now().UTC().Add(m.retryDelay(task.AttemptCount))
	willDeadLetter := task.AttemptCount >= task.MaxAttempts
	if err := store.RetrySemanticTask(ctx, task.TaskID, task.Receipt, retryAt, message); err != nil {
		metrics.RecordTenantOperationWithOrg(tenantID, target.metricOrgID(), "semantic_worker", "retry", "error", time.Since(start))
		logger.Warn(ctx, "tenant_worker_retry_failed",
			append([]zap.Field{
				zap.String("tenant_id", tenantID),
				zap.String("result", "error"),
			}, append(semanticTaskLogFields(task), zap.Error(err))...)...)
		return
	}
	result := "scheduled"
	logMessage := "tenant_worker_retry_scheduled"
	if willDeadLetter {
		result = "dead_lettered"
		logMessage = "tenant_worker_dead_lettered"
	}
	metrics.RecordTenantOperationWithOrg(tenantID, target.metricOrgID(), "semantic_worker", "retry", result, time.Since(start))
	fields := append([]zap.Field{
		zap.String("tenant_id", tenantID),
		zap.String("result", result),
		zap.String("message", message),
	}, semanticTaskLogFields(task)...)
	if !willDeadLetter {
		fields = append(fields, zap.Time("retry_at", retryAt))
	}
	logger.Warn(ctx, logMessage, fields...)
}

func (m *tenantWorkerManager) deadLetterTask(ctx context.Context, target *tenantTarget, task *semantic.Task, message string) {
	start := time.Now()
	tenantID := target.tenantID
	store := target.store
	if err := store.DeadLetterSemanticTask(ctx, task.TaskID, task.Receipt, message); err != nil {
		metrics.RecordTenantOperationWithOrg(tenantID, target.metricOrgID(), "semantic_worker", "dead_letter", "error", time.Since(start))
		logger.Warn(ctx, "tenant_worker_dead_letter_failed",
			append([]zap.Field{
				zap.String("tenant_id", tenantID),
				zap.String("result", "error"),
			}, append(semanticTaskLogFields(task), zap.Error(err))...)...)
		return
	}
	metrics.RecordTenantOperationWithOrg(tenantID, target.metricOrgID(), "semantic_worker", "dead_letter", "dead_lettered", time.Since(start))
	logger.Warn(ctx, "tenant_worker_dead_lettered",
		append([]zap.Field{
			zap.String("tenant_id", tenantID),
			zap.String("result", "dead_lettered"),
			zap.String("message", message),
		}, semanticTaskLogFields(task)...)...)
}

func (m *tenantWorkerManager) retryDelay(attemptCount int) time.Duration {
	if attemptCount < 1 {
		attemptCount = 1
	}
	delay := m.opts.RetryBaseDelay
	for i := 1; i < attemptCount; i++ {
		delay *= 2
		if delay >= m.opts.RetryMaxDelay {
			return m.opts.RetryMaxDelay
		}
	}
	if delay > m.opts.RetryMaxDelay {
		return m.opts.RetryMaxDelay
	}
	return delay
}

func (m *tenantWorkerManager) dispatchTask(ctx context.Context, target *tenantTarget, task *semantic.Task) semanticTaskOutcome {
	switch task.TaskType {
	case semantic.TaskTypeEmbed:
		return m.processEmbedTask(ctx, target, task)
	case semantic.TaskTypeEmbedBackfill:
		return m.processEmbeddingBackfillTask(ctx, target, task)
	case semantic.TaskTypeImgExtractText:
		return m.processImgExtractTask(ctx, target, task)
	case semantic.TaskTypeAudioExtractText:
		return m.processAudioExtractTask(ctx, target, task)
	case semantic.TaskTypeVideoExtractVisual:
		return m.processVideoExtractTask(ctx, target, task)
	default:
		message := fmt.Sprintf("unsupported task type %q", task.TaskType)
		return semanticTaskOutcome{action: semanticTaskActionRetry, result: "unsupported", message: message}
	}
}

func (m *tenantWorkerManager) startTaskLeaseExecution(ctx context.Context, target *tenantTarget, task *semantic.Task) *semanticTaskLeaseExecution {
	leaseCtx, cancel := context.WithCancel(ctx)
	exec := &semanticTaskLeaseExecution{
		ctx:    leaseCtx,
		cancel: cancel,
		stopCh: make(chan struct{}),
		doneCh: make(chan struct{}),
	}
	if task != nil && task.LeaseUntil != nil {
		leaseUntil := task.LeaseUntil.UTC()
		exec.lastLeaseUntil = &leaseUntil
	}
	go exec.run(m, target, task)
	return exec
}

func (m *tenantWorkerManager) leaseRenewInterval() time.Duration {
	interval := m.opts.LeaseDuration / 2
	if interval > 0 {
		return interval
	}
	if m.opts.LeaseDuration > 0 {
		return m.opts.LeaseDuration
	}
	return time.Second
}

func (e *semanticTaskLeaseExecution) leaseCtx() context.Context {
	if e == nil || e.ctx == nil {
		return context.Background()
	}
	return e.ctx
}

func (e *semanticTaskLeaseExecution) stop() semanticTaskLeaseOutcome {
	if e == nil {
		return semanticTaskLeaseOutcome{}
	}
	e.stopOnce.Do(func() {
		close(e.stopCh)
		e.cancel()
		<-e.doneCh
	})

	e.mu.Lock()
	defer e.mu.Unlock()
	outcome := semanticTaskLeaseOutcome{
		lost:           e.lost,
		renewAttempted: e.renewAttempted,
	}
	if e.lastLeaseUntil != nil {
		leaseUntil := e.lastLeaseUntil.UTC()
		outcome.lastLeaseUntil = &leaseUntil
	}
	return outcome
}

func (e *semanticTaskLeaseExecution) run(m *tenantWorkerManager, target *tenantTarget, task *semantic.Task) {
	defer close(e.doneCh)
	if m == nil || target == nil || target.store == nil || task == nil {
		return
	}

	ticker := time.NewTicker(m.leaseRenewInterval())
	defer ticker.Stop()

	for {
		select {
		case <-e.stopCh:
			e.logStopped(m, target, task)
			return
		case <-e.ctx.Done():
			select {
			case <-e.stopCh:
				e.logStopped(m, target, task)
			default:
			}
			return
		case <-ticker.C:
			e.markRenewAttempted()
			failpoint.InjectCall("semanticWorkerBeforeRenew", target.tenantID, target.store, task)
			renewStart := time.Now()
			leaseUntil, err := target.store.RenewSemanticTask(e.ctx, task.TaskID, task.Receipt, m.opts.LeaseDuration)
			if err != nil {
				if errors.Is(err, semantic.ErrTaskLeaseMismatch) || errors.Is(err, semantic.ErrTaskNotFound) {
					metrics.RecordTenantOperationWithOrg(target.tenantID, target.metricOrgID(), "semantic_worker", "renew", "error", time.Since(renewStart))
					e.markLeaseLost()
					failpoint.InjectCall("semanticWorkerOnLeaseLost", target.tenantID, target.store, task, err)
					metrics.RecordTenantOperationWithOrg(target.tenantID, target.metricOrgID(), "semantic_worker", "lease_lost", "ok", time.Since(renewStart))
					logger.Warn(e.ctx, "tenant_worker_lease_lost",
						append([]zap.Field{
							zap.String("tenant_id", target.tenantID),
							zap.String("result", "lease_lost"),
						}, append(semanticTaskLogFields(task), zap.Error(err))...)...)
					e.cancel()
					e.logStopped(m, target, task)
					return
				}
				if e.shouldStop() {
					e.logStopped(m, target, task)
					return
				}
				metrics.RecordTenantOperationWithOrg(target.tenantID, target.metricOrgID(), "semantic_worker", "renew", "error", time.Since(renewStart))
				logger.Warn(e.ctx, "tenant_worker_lease_renew_failed",
					append([]zap.Field{
						zap.String("tenant_id", target.tenantID),
						zap.String("result", "error"),
					}, append(semanticTaskLogFields(task), zap.Error(err))...)...)
				continue
			}
			e.recordRenewedLease(leaseUntil)
			failpoint.InjectCall("semanticWorkerAfterRenew", target.tenantID, target.store, task, leaseUntil)
			metrics.RecordTenantOperationWithOrg(target.tenantID, target.metricOrgID(), "semantic_worker", "renew", "ok", time.Since(renewStart))
		}
	}
}

func (e *semanticTaskLeaseExecution) shouldStop() bool {
	if e == nil {
		return true
	}
	select {
	case <-e.stopCh:
		return true
	default:
		return false
	}
}

func (e *semanticTaskLeaseExecution) recordRenewedLease(leaseUntil time.Time) {
	if e == nil {
		return
	}
	e.mu.Lock()
	defer e.mu.Unlock()
	e.renewAttempted = true
	t := leaseUntil.UTC()
	e.lastLeaseUntil = &t
}

func (e *semanticTaskLeaseExecution) markRenewAttempted() {
	if e == nil {
		return
	}
	e.mu.Lock()
	defer e.mu.Unlock()
	e.renewAttempted = true
}

func (e *semanticTaskLeaseExecution) markLeaseLost() {
	if e == nil {
		return
	}
	e.mu.Lock()
	defer e.mu.Unlock()
	e.lost = true
	e.renewAttempted = true
}

func (e *semanticTaskLeaseExecution) logStopped(m *tenantWorkerManager, target *tenantTarget, task *semantic.Task) {
	if e == nil || m == nil || target == nil || task == nil {
		return
	}
	e.mu.Lock()
	renewAttempted := e.renewAttempted
	lost := e.lost
	var leaseUntil *time.Time
	if e.lastLeaseUntil != nil {
		t := e.lastLeaseUntil.UTC()
		leaseUntil = &t
	}
	e.mu.Unlock()
	if !renewAttempted {
		return
	}

	result := "owned"
	if lost {
		result = "lease_lost"
	}
	fields := append([]zap.Field{
		zap.String("tenant_id", target.tenantID),
		zap.String("result", result),
	}, semanticTaskLogFields(task)...)
	if leaseUntil != nil {
		fields = append(fields, zap.Time("latest_lease_until", leaseUntil.UTC()))
	}
	logger.Info(e.ctx, "tenant_worker_lease_renew_stopped", fields...)
}

func (m *tenantWorkerManager) processEmbedTask(ctx context.Context, target *tenantTarget, task *semantic.Task) semanticTaskOutcome {
	if target == nil || target.backend == nil || target.store == nil {
		return semanticTaskOutcome{action: semanticTaskActionRetry, result: "handler_missing", message: "embed target not configured"}
	}
	runtime, ready := target.backend.AppEmbeddingRuntime()
	if !ready {
		return semanticTaskOutcome{action: semanticTaskActionAck, result: "obsolete", message: "app embedding disabled"}
	}
	var payload semantic.EmbedTaskPayload
	if len(task.PayloadJSON) > 0 {
		if err := json.Unmarshal(task.PayloadJSON, &payload); err != nil {
			return semanticTaskOutcome{action: semanticTaskActionRetry, result: "invalid_payload", message: "decode embed task payload"}
		}
	}
	currentPayload := semantic.EmbedTaskPayload{
		Generation: runtime.Generation, ConfigRevision: runtime.ConfigRevision,
		TenantManaged: runtime.TenantManaged,
	}
	configOrder := semantic.CompareEmbedConfig(currentPayload, payload)
	if configOrder == semantic.EmbedConfigStale {
		return semanticTaskOutcome{action: semanticTaskActionDefer, result: "superseded", message: "embedding configuration is newer than backend runtime"}
	}
	if configOrder == semantic.EmbedConfigNewer {
		payloadJSON, err := json.Marshal(semantic.EmbedTaskPayload{
			Generation: runtime.Generation, ConfigRevision: runtime.ConfigRevision,
			TenantManaged: runtime.TenantManaged,
		})
		if err != nil {
			return semanticTaskOutcome{action: semanticTaskActionRetry, result: "payload_refresh_error", message: "encode current embed task payload"}
		}
		if err := target.store.UpdateSemanticTaskPayload(ctx, task.TaskID, task.Receipt, payloadJSON); err != nil {
			return semanticTaskOutcome{action: semanticTaskActionRetry, result: "payload_refresh_error", message: fmt.Sprintf("refresh embed task payload: %v", err)}
		}
		return semanticTaskOutcome{action: semanticTaskActionDefer, result: "superseded", message: "embedding configuration changed"}
	}
	store := target.store
	file, err := store.GetFile(ctx, task.ResourceID)
	if err != nil {
		if errors.Is(err, datastore.ErrNotFound) {
			return semanticTaskOutcome{action: semanticTaskActionAck, result: "obsolete", message: "file_not_found"}
		}
		return semanticTaskOutcome{action: semanticTaskActionRetry, result: "get_file_error", message: fmt.Sprintf("get file: %v", err)}
	}
	if file.Status != datastore.StatusConfirmed || file.Revision != task.ResourceVersion {
		return semanticTaskOutcome{action: semanticTaskActionAck, result: "obsolete", message: "stale"}
	}
	if strings.TrimSpace(file.ContentText) == "" && strings.TrimSpace(file.Description) == "" {
		return semanticTaskOutcome{action: semanticTaskActionAck, result: "obsolete", message: "empty_content_and_description"}
	}

	var contentUpdated, descUpdated bool
	if strings.TrimSpace(file.ContentText) != "" {
		vec, err := runtime.Client.EmbedText(ctx, file.ContentText)
		if err != nil {
			return embeddingProviderFailureOutcome(ctx, target, task, "content", "embed_error", err)
		}
		if len(vec) != embedding.Dimensions {
			return semanticTaskOutcome{action: semanticTaskActionRetry, result: "embed_dimensions", message: fmt.Sprintf("embed text returned %d dimensions", len(vec))}
		}
		contentUpdated, err = store.UpdateFileEmbeddingForTask(ctx, task.ResourceID, task.ResourceVersion, runtime.Generation, task.TaskID, task.Receipt, file.ContentText, vec)
		if err != nil {
			return semanticTaskOutcome{action: semanticTaskActionRetry, result: "writeback_error", message: fmt.Sprintf("write embedding: %v", err)}
		}
	}

	descChanged := false
	if strings.TrimSpace(file.Description) != "" {
		vecDesc, err := runtime.Client.EmbedText(ctx, file.Description)
		if err != nil {
			return embeddingProviderFailureOutcome(ctx, target, task, "description", "embed_desc_error", err)
		}
		if len(vecDesc) != embedding.Dimensions {
			return semanticTaskOutcome{action: semanticTaskActionRetry, result: "embed_desc_dimensions", message: fmt.Sprintf("embed description returned %d dimensions", len(vecDesc))}
		}
		descUpdated, err = store.UpdateFileDescriptionEmbeddingForTask(ctx, task.ResourceID, task.ResourceVersion, runtime.Generation, task.TaskID, task.Receipt, file.Description, vecDesc)
		if err != nil {
			return semanticTaskOutcome{action: semanticTaskActionRetry, result: "writeback_desc_error", message: fmt.Sprintf("write description embedding: %v", err)}
		}
		// setmeta changes the description without bumping the revision, so the
		// revision check above cannot catch a description that changed while
		// this task was in flight; the writeback text comparison does. Retry to
		// converge on the new description. When the miss came from lease loss
		// instead, the retry is a no-op and the new owner proceeds with fresh
		// text.
		descChanged = !descUpdated
	}

	if descChanged {
		return semanticTaskOutcome{action: semanticTaskActionRetry, result: "description_changed", message: "description changed during embedding"}
	}
	if !contentUpdated && !descUpdated {
		return semanticTaskOutcome{action: semanticTaskActionAck, result: "obsolete", message: "conditional_write_miss"}
	}
	return semanticTaskOutcome{action: semanticTaskActionAck, result: "ok", message: "written"}
}

func embeddingProviderFailureOutcome(
	ctx context.Context,
	target *tenantTarget,
	task *semantic.Task,
	source string,
	result string,
	err error,
) semanticTaskOutcome {
	category := "request_failed"
	providerStatus := 0
	if providerCategory, status, _, ok := embedding.ProviderFailure(err); ok {
		category = string(providerCategory)
		providerStatus = status
	} else if errors.Is(err, context.DeadlineExceeded) {
		category = "timeout"
	}
	metrics.RecordTenantOperationWithOrg(
		target.tenantID,
		target.metricOrgID(),
		"embedding",
		"provider_request",
		category,
		0,
	)
	logger.Warn(ctx, "tenant_worker_embedding_provider_failed",
		append([]zap.Field{
			zap.String("tenant_id", target.tenantID),
			zap.String("embedding_source", source),
			zap.String("failure_category", category),
			zap.Int("provider_status", providerStatus),
		}, semanticTaskLogFields(task)...)...)
	return semanticTaskOutcome{
		action:  semanticTaskActionRetry,
		result:  result,
		message: "embedding_provider_" + category,
	}
}

func (m *tenantWorkerManager) processEmbeddingBackfillTask(ctx context.Context, target *tenantTarget, task *semantic.Task) semanticTaskOutcome {
	if target == nil || target.backend == nil || target.store == nil {
		return semanticTaskOutcome{action: semanticTaskActionRetry, result: "handler_missing", message: "embedding backfill target not configured"}
	}
	runtime, ready := target.backend.AppEmbeddingRuntime()
	if !ready {
		return semanticTaskOutcome{action: semanticTaskActionAck, result: "obsolete", message: "app embedding disabled"}
	}
	var payload semantic.EmbedBackfillTaskPayload
	if err := json.Unmarshal(task.PayloadJSON, &payload); err != nil {
		return semanticTaskOutcome{action: semanticTaskActionRetry, result: "invalid_payload", message: "decode embedding backfill payload"}
	}
	if payload.Generation == 0 || payload.ConfigRevision == 0 ||
		payload.Generation != runtime.Generation || payload.ConfigRevision != runtime.ConfigRevision {
		return semanticTaskOutcome{action: semanticTaskActionAck, result: "obsolete", message: "embedding config changed"}
	}
	candidates, err := target.store.ListEmbeddingBackfillCandidates(ctx, payload.AfterInodeID, runtime.Generation, embeddingBackfillPageSize)
	if err != nil {
		return semanticTaskOutcome{action: semanticTaskActionRetry, result: "scan_error", message: fmt.Sprintf("scan embedding backfill: %v", err)}
	}
	if len(candidates) == 0 {
		return semanticTaskOutcome{action: semanticTaskActionAck, result: "ok", message: "embedding backfill complete"}
	}
	if _, err := target.backend.EnsureEmbeddingBackfillCandidates(ctx, candidates); err != nil {
		return semanticTaskOutcome{action: semanticTaskActionRetry, result: "enqueue_error", message: fmt.Sprintf("enqueue embedding backfill page: %v", err)}
	}
	payload.AfterInodeID = candidates[len(candidates)-1].InodeID
	encoded, err := json.Marshal(payload)
	if err != nil {
		return semanticTaskOutcome{action: semanticTaskActionRetry, result: "checkpoint_error", message: "encode embedding backfill checkpoint"}
	}
	if err := target.store.UpdateSemanticTaskPayload(ctx, task.TaskID, task.Receipt, encoded); err != nil {
		return semanticTaskOutcome{action: semanticTaskActionRetry, result: "checkpoint_error", message: fmt.Sprintf("checkpoint embedding backfill: %v", err)}
	}
	return semanticTaskOutcome{action: semanticTaskActionDefer, result: "page", message: "embedding backfill page queued"}
}

// processImgExtractTask resolves the tenant's effective image extract runtime
// at execution time (never at enqueue time) and runs the backend-owned
// extraction with it. Kill switch semantics: a claimed task that has not yet
// called the provider is deferred with global_disabled without consuming an
// attempt (claim-time filtering in claimableTaskTypes stops new claims; file
// writes keep enqueueing; nothing is acked because of the switch). The switch
// is checked at task entry and again via a BeforeExtract guard immediately
// before the provider call, so a flip during resolve/file-load still prevents
// the call. Resolution outcomes: tenant-disabled and not-configured tasks are
// discarded as terminal, with fresh due peers collapsed in bounded batches;
// resolution failures are fail-closed retries.
func (m *tenantWorkerManager) processImgExtractTask(ctx context.Context, target *tenantTarget, task *semantic.Task) semanticTaskOutcome {
	if m.imageExtractGloballyDisabled() {
		return semanticTaskOutcome{action: semanticTaskActionDefer, result: "global_disabled", message: "image extract globally disabled"}
	}
	if m.extractResolver == nil {
		// No resolver wired (single-tenant local server): use the
		// process-default backend runtime.
		return m.processImgExtractTaskWithDefaultRuntime(ctx, target, task)
	}
	res, err := m.extractResolver.ResolveImage(ctx, target.tenantID)
	if err != nil {
		return semanticTaskOutcome{action: semanticTaskActionRetry, result: "resolve_error", message: err.Error()}
	}
	switch res.state {
	case extractResolveDisabled:
		return terminalMediaConfigOutcome("disabled", "tenant disabled image extract", false)
	case extractResolveNotConfigured:
		return terminalMediaConfigOutcome("not_configured", "no default extractor configured", false)
	case extractResolveReady:
		// Continue below.
	default:
		return semanticTaskOutcome{action: semanticTaskActionRetry, result: "resolve_state_invalid", message: "image extract resolver returned an invalid state"}
	}
	// Re-check the kill switch immediately before the provider call: the
	// switch may have flipped while the config resolved and the image loaded.
	res.runtime.BeforeExtract = m.imageExtractKillSwitchGuard()
	result, err := target.backend.ProcessImageExtractTaskWithRuntime(ctx, imageExtractTaskSpecFromSemanticTask(task), res.runtime)
	if err != nil {
		if errors.Is(err, errImageExtractGloballyDisabled) {
			return semanticTaskOutcome{action: semanticTaskActionDefer, result: "global_disabled", message: "image extract globally disabled"}
		}
		if outcome, ok := imageExtractProviderFailureOutcome(ctx, target, task, result, err); ok {
			return outcome
		}
		return semanticTaskOutcome{action: semanticTaskActionRetry, result: string(result), message: err.Error()}
	}
	if result == backend.ImageExtractResultBudgetExhausted {
		return semanticTaskOutcome{action: semanticTaskActionAck, result: string(result), message: "monthly_llm_cost_budget_exhausted"}
	}
	return semanticTaskOutcome{action: semanticTaskActionAck, result: string(result), message: string(result)}
}

// errImageExtractGloballyDisabled is returned by the pre-provider-call guard
// when the cluster image-extract kill switch flipped on after the task-entry
// check; the task is deferred instead of calling the provider.
var errImageExtractGloballyDisabled = errors.New("image extract globally disabled")

// imageExtractKillSwitchGuard returns the BeforeExtract guard wired to the
// cluster kill switch.
func (m *tenantWorkerManager) imageExtractKillSwitchGuard() func() error {
	return func() error {
		if m.imageExtractGloballyDisabled() {
			return errImageExtractGloballyDisabled
		}
		return nil
	}
}

func (m *tenantWorkerManager) processImgExtractTaskWithDefaultRuntime(ctx context.Context, target *tenantTarget, task *semantic.Task) semanticTaskOutcome {
	result, err := target.backend.ProcessImageExtractTaskWithGuard(ctx, imageExtractTaskSpecFromSemanticTask(task), m.imageExtractKillSwitchGuard())
	if err != nil {
		if errors.Is(err, errImageExtractGloballyDisabled) {
			return semanticTaskOutcome{action: semanticTaskActionDefer, result: "global_disabled", message: "image extract globally disabled"}
		}
		if outcome, ok := imageExtractProviderFailureOutcome(ctx, target, task, result, err); ok {
			return outcome
		}
		return semanticTaskOutcome{action: semanticTaskActionRetry, result: string(result), message: err.Error()}
	}
	if result == backend.ImageExtractResultBudgetExhausted {
		return semanticTaskOutcome{action: semanticTaskActionAck, result: string(result), message: "monthly_llm_cost_budget_exhausted"}
	}
	return semanticTaskOutcome{action: semanticTaskActionAck, result: string(result), message: string(result)}
}

func imageExtractProviderFailureOutcome(
	ctx context.Context,
	target *tenantTarget,
	task *semantic.Task,
	result backend.ImageExtractResult,
	err error,
) (semanticTaskOutcome, bool) {
	category, providerStatus, ok := backend.ImageProviderFailure(err)
	if !ok {
		return semanticTaskOutcome{}, false
	}
	metrics.RecordTenantOperationWithOrg(
		target.tenantID,
		target.metricOrgID(),
		"image_extract",
		"provider_request",
		string(category),
		0,
	)
	logger.Warn(ctx, "tenant_worker_image_extract_provider_failed",
		append([]zap.Field{
			zap.String("tenant_id", target.tenantID),
			zap.String("failure_category", string(category)),
			zap.Int("provider_status", providerStatus),
		}, semanticTaskLogFields(task)...)...)
	// Provider rejections stay retryable: request_rejected also covers
	// configuration-level 400s (e.g. an unsupported or renamed model), which
	// must not be silently acked — after the tenant fixes the config, only
	// retried/dead-lettered tasks remain visible for recovery. Inputs that
	// are not real images never reach the provider: the backend sniffs the
	// bytes and returns not_image before this point.
	return semanticTaskOutcome{
		action:  semanticTaskActionRetry,
		result:  string(result),
		message: "image_extract_provider_" + string(category),
	}, true
}

func (m *tenantWorkerManager) processAudioExtractTask(ctx context.Context, target *tenantTarget, task *semantic.Task) semanticTaskOutcome {
	videoOwnsAudio, routeErr := m.videoOwnsFallbackAudio(ctx, target, task)
	if routeErr != nil {
		return semanticTaskOutcome{action: semanticTaskActionRetry, result: "resolve_error", message: "video fallback runtime resolution failed"}
	}
	if videoOwnsAudio {
		return semanticTaskOutcome{action: semanticTaskActionDiscard, result: "handled_by_video", message: "video runtime handles the audio modality"}
	}
	b := target.backend
	var result backend.AudioExtractResult
	var err error
	if m.extractResolver != nil {
		res, resolveErr := m.extractResolver.ResolveAudio(ctx, target.tenantID)
		if resolveErr != nil {
			return semanticTaskOutcome{action: semanticTaskActionRetry, result: "resolve_error", message: "audio extract runtime resolution failed"}
		}
		switch res.state {
		case extractResolveDisabled:
			return terminalMediaConfigOutcome("disabled", "tenant disabled audio extract", false)
		case extractResolveNotConfigured:
			if runtime, ok := b.DefaultAudioExtractRuntime(); ok {
				runtime = prepareAudioExtractRuntime(ctx, target, task, runtime)
				result, err = b.ProcessAudioExtractTaskWithRuntime(ctx, audioExtractTaskSpecFromSemanticTask(task), runtime)
				break
			}
			return terminalMediaConfigOutcome("not_configured", "no audio extractor configured", false)
		case extractResolveReady:
			res.runtime = prepareAudioExtractRuntime(ctx, target, task, res.runtime)
			result, err = b.ProcessAudioExtractTaskWithRuntime(ctx, audioExtractTaskSpecFromSemanticTask(task), res.runtime)
		default:
			return semanticTaskOutcome{action: semanticTaskActionRetry, result: "resolve_state_invalid", message: "invalid audio extract runtime state"}
		}
	} else {
		if runtime, ok := b.DefaultAudioExtractRuntime(); ok {
			runtime = prepareAudioExtractRuntime(ctx, target, task, runtime)
			result, err = b.ProcessAudioExtractTaskWithRuntime(ctx, audioExtractTaskSpecFromSemanticTask(task), runtime)
		} else {
			result, err = b.ProcessAudioExtractTask(ctx, audioExtractTaskSpecFromSemanticTask(task))
		}
	}
	if err != nil {
		if category, providerStatus, ok := backend.ImageProviderFailure(err); ok {
			recordMediaProviderFailure(ctx, target, task, "audio_extract", category, providerStatus)
		}
		if backend.IsNonRetryableAudioExtractError(err) {
			return semanticTaskOutcome{action: semanticTaskActionDeadLetter, result: string(result), message: "audio_extract_provider_non_retryable"}
		}
		return semanticTaskOutcome{action: semanticTaskActionRetry, result: string(result), message: "audio_extract_failed"}
	}
	if result == backend.AudioExtractResultBudgetExhausted {
		return semanticTaskOutcome{action: semanticTaskActionAck, result: string(result), message: "monthly_llm_cost_budget_exhausted"}
	}
	return semanticTaskOutcome{action: semanticTaskActionAck, result: string(result), message: string(result)}
}

func (m *tenantWorkerManager) videoOwnsFallbackAudio(ctx context.Context, target *tenantTarget, task *semantic.Task) (bool, error) {
	if task == nil {
		return false, nil
	}
	var payload semantic.AudioExtractTaskPayload
	if json.Unmarshal(task.PayloadJSON, &payload) != nil || !payload.VideoFallback {
		return false, nil
	}
	if m.extractResolver != nil {
		res, err := m.extractResolver.ResolveVideo(ctx, target.tenantID)
		if err != nil {
			return false, err
		}
		switch res.state {
		case extractResolveReady:
			return res.runtime.AudioRuntime != nil && res.runtime.AudioRuntime.Extractor != nil, nil
		case extractResolveDisabled:
			return false, nil
		case extractResolveNotConfigured:
			if target != nil && target.backend != nil {
				runtime, ok := target.backend.DefaultVideoExtractRuntime()
				return ok && runtime.AudioRuntime != nil && runtime.AudioRuntime.Extractor != nil, nil
			}
			return false, nil
		default:
			return false, fmt.Errorf("invalid video extract runtime state %d", res.state)
		}
	}
	if target != nil && target.backend != nil {
		runtime, ok := target.backend.DefaultVideoExtractRuntime()
		return ok && runtime.AudioRuntime != nil && runtime.AudioRuntime.Extractor != nil, nil
	}
	return false, nil
}

func (m *tenantWorkerManager) processVideoExtractTask(ctx context.Context, target *tenantTarget, task *semantic.Task) semanticTaskOutcome {
	b := target.backend
	var result backend.VideoExtractResult
	var err error
	if m.extractResolver != nil {
		res, resolveErr := m.extractResolver.ResolveVideo(ctx, target.tenantID)
		if resolveErr != nil {
			return semanticTaskOutcome{action: semanticTaskActionRetry, result: "resolve_error", message: "video extract runtime resolution failed"}
		}
		switch res.state {
		case extractResolveDisabled:
			return terminalMediaConfigOutcome("disabled", "tenant disabled video extract", videoTaskHasProgress(task))
		case extractResolveNotConfigured:
			if runtime, ok := b.DefaultVideoExtractRuntime(); ok {
				runtime = prepareVideoExtractRuntime(ctx, target, task, runtime)
				result, err = b.ProcessVideoExtractTaskWithRuntime(ctx, videoExtractTaskSpecFromSemanticTask(task), runtime)
				break
			}
			return terminalMediaConfigOutcome("not_configured", "no video extractor configured", videoTaskHasProgress(task))
		case extractResolveReady:
			res.runtime = prepareVideoExtractRuntime(ctx, target, task, res.runtime)
			result, err = b.ProcessVideoExtractTaskWithRuntime(ctx, videoExtractTaskSpecFromSemanticTask(task), res.runtime)
		default:
			return semanticTaskOutcome{action: semanticTaskActionRetry, result: "resolve_state_invalid", message: "invalid video extract runtime state"}
		}
	} else {
		if runtime, ok := b.DefaultVideoExtractRuntime(); ok {
			runtime = prepareVideoExtractRuntime(ctx, target, task, runtime)
			result, err = b.ProcessVideoExtractTaskWithRuntime(ctx, videoExtractTaskSpecFromSemanticTask(task), runtime)
		} else {
			result, err = b.ProcessVideoExtractTask(ctx, videoExtractTaskSpecFromSemanticTask(task))
		}
	}
	if err != nil {
		if category, providerStatus, ok := backend.ImageProviderFailure(err); ok {
			recordMediaProviderFailure(ctx, target, task, "video_extract", category, providerStatus)
			if category == backend.ImageProviderFailureAuthRejected || category == backend.ImageProviderFailureNotFound || category == backend.ImageProviderFailureRequestRejected {
				return semanticTaskOutcome{action: semanticTaskActionDeadLetter, result: string(result), message: "video_extract_provider_" + string(category)}
			}
			return semanticTaskOutcome{action: semanticTaskActionRetry, result: string(result), message: "video_extract_provider_" + string(category)}
		}
		return semanticTaskOutcome{action: semanticTaskActionRetry, result: string(result), message: "video_extract_failed"}
	}
	if result == backend.VideoExtractResultBudgetExhausted {
		return semanticTaskOutcome{action: semanticTaskActionAck, result: string(result), message: "monthly_llm_cost_budget_exhausted"}
	}
	return semanticTaskOutcome{action: semanticTaskActionAck, result: string(result), message: string(result)}
}

// videoTaskHasProgress reports whether a durable video task has checkpointed
// any visual or audio work. AttemptCount alone is not evidence of progress:
// resolver failures can consume attempts before a provider is ever called.
func videoTaskHasProgress(task *semantic.Task) bool {
	if task == nil || len(task.PayloadJSON) == 0 {
		return false
	}
	var payload semantic.VideoExtractTaskPayload
	if err := json.Unmarshal(task.PayloadJSON, &payload); err != nil {
		return false
	}
	return payload.VisualDone || payload.VisualCompletedSegments > 0 || payload.AudioDone || payload.AudioCompletedSegments > 0 ||
		strings.TrimSpace(payload.VisualText) != "" || strings.TrimSpace(payload.AudioText) != ""
}

// terminalMediaConfigOutcome marks fresh due peers for cleanup after the
// claimed task has been finalized through its lease fence.
func terminalMediaConfigOutcome(
	result string,
	message string,
	preserveCurrent bool,
) semanticTaskOutcome {
	action := semanticTaskActionDiscard
	if preserveCurrent {
		action = semanticTaskActionAck
	}
	return semanticTaskOutcome{
		action: action, result: result, message: message,
		discardFreshMediaPeers: true,
	}
}

func (m *tenantWorkerManager) discardFreshMediaPeers(ctx context.Context, target *tenantTarget, task *semantic.Task, result string) {
	if target == nil || target.store == nil || task == nil {
		return
	}
	switch task.TaskType {
	case semantic.TaskTypeImgExtractText, semantic.TaskTypeAudioExtractText, semantic.TaskTypeVideoExtractVisual:
	default:
		return
	}
	deleted, err := target.store.DiscardFreshClaimableSemanticTasksByType(
		ctx, task.TaskType, time.Now().UTC(), terminalMediaDiscardBatchSize,
	)
	if err != nil {
		logger.Warn(ctx, "tenant_worker_media_terminal_batch_discard_failed",
			append([]zap.Field{
				zap.String("tenant_id", target.tenantID),
				zap.String("result", result),
			}, append(semanticTaskLogFields(task), zap.Error(err))...)...)
		return
	}
	if deleted > 0 {
		logger.Info(ctx, "tenant_worker_media_terminal_batch_discarded",
			append([]zap.Field{
				zap.String("tenant_id", target.tenantID),
				zap.String("result", result),
				zap.Int64("deleted", deleted),
			}, semanticTaskLogFields(task)...)...)
	}
}

func prepareAudioExtractRuntime(ctx context.Context, target *tenantTarget, task *semantic.Task, runtime backend.AudioExtractRuntime) backend.AudioExtractRuntime {
	var payload semantic.AudioExtractTaskPayload
	if task != nil {
		_ = json.Unmarshal(task.PayloadJSON, &payload)
		runtime.FinalAttempt = task.MaxAttempts > 0 && task.AttemptCount >= task.MaxAttempts
	}
	runtime.CheckpointState = backend.AudioExtractCheckpoint{CompletedSegments: payload.CompletedSegments, Text: payload.Text, Tags: payload.Tags, Structured: payload.Structured}
	runtime.Checkpoint = func(next backend.AudioExtractCheckpoint) error {
		payload.CompletedSegments = next.CompletedSegments
		payload.Text = next.Text
		payload.Tags = next.Tags
		payload.Structured = next.Structured
		encoded, err := marshalSemanticCheckpointPayload(payload)
		if err != nil {
			return err
		}
		if target == nil || target.store == nil || task == nil {
			return errors.New("audio checkpoint task store is not configured")
		}
		return target.store.UpdateSemanticTaskPayload(ctx, task.TaskID, task.Receipt, encoded)
	}
	return runtime
}

func prepareVideoExtractRuntime(ctx context.Context, target *tenantTarget, task *semantic.Task, runtime backend.VideoExtractRuntime) backend.VideoExtractRuntime {
	var payload semantic.VideoExtractTaskPayload
	if task != nil {
		_ = json.Unmarshal(task.PayloadJSON, &payload)
		runtime.FinalAttempt = task.MaxAttempts > 0 && task.AttemptCount >= task.MaxAttempts
		if target != nil && target.backend != nil {
			beforeExtract := runtime.BeforeExtract
			runtime.BeforeExtract = func() {
				if beforeExtract != nil {
					beforeExtract()
				}
				target.backend.RecordVideoLLMCall(ctx)
			}
		}
	}
	runtime.CheckpointState = backend.VideoExtractCheckpoint{
		VisualDone: payload.VisualDone, VisualText: payload.VisualText,
		VisualCompletedSegments: payload.VisualCompletedSegments,
		VisualSegmentCount:      payload.VisualSegmentCount, VisualSegmentsText: payload.VisualSegmentsText,
		AudioDone: payload.AudioDone, AudioPresent: payload.AudioPresent,
		AudioCompletedSegments: payload.AudioCompletedSegments, AudioText: payload.AudioText,
		VisualTags: payload.VisualTags, AudioTags: payload.AudioTags,
	}
	runtime.Checkpoint = func(next backend.VideoExtractCheckpoint) error {
		payload.VisualDone = next.VisualDone
		payload.VisualText = next.VisualText
		payload.VisualCompletedSegments = next.VisualCompletedSegments
		payload.VisualSegmentCount = next.VisualSegmentCount
		payload.VisualSegmentsText = next.VisualSegmentsText
		payload.AudioDone = next.AudioDone
		payload.AudioPresent = next.AudioPresent
		payload.AudioCompletedSegments = next.AudioCompletedSegments
		payload.AudioText = next.AudioText
		payload.VisualTags = next.VisualTags
		payload.AudioTags = next.AudioTags
		encoded, err := marshalSemanticCheckpointPayload(payload)
		if err != nil {
			return err
		}
		if target == nil || target.store == nil || task == nil {
			return errors.New("video checkpoint task store is not configured")
		}
		return target.store.UpdateSemanticTaskPayload(ctx, task.TaskID, task.Receipt, encoded)
	}
	return runtime
}

func marshalSemanticCheckpointPayload(payload any) ([]byte, error) {
	encoded, err := json.Marshal(payload)
	if err != nil {
		return nil, err
	}
	if len(encoded) > datastore.MaxSemanticTaskPayloadBytes {
		return nil, datastore.ErrSemanticTaskPayloadTooLarge
	}
	return encoded, nil
}

func recordMediaProviderFailure(ctx context.Context, target *tenantTarget, task *semantic.Task, operation string, category backend.ImageProviderFailureCategory, providerStatus int) {
	metrics.RecordTenantOperationWithOrg(target.tenantID, target.metricOrgID(), operation, "provider_request", string(category), 0)
	logger.Warn(ctx, "tenant_worker_media_extract_provider_failed",
		append([]zap.Field{
			zap.String("tenant_id", target.tenantID),
			zap.String("media_operation", operation),
			zap.String("failure_category", string(category)),
			zap.Int("provider_status", providerStatus),
		}, semanticTaskLogFields(task)...)...)
}

func (m *tenantWorkerManager) semanticTaskStillOwned(ctx context.Context, target *tenantTarget, task *semantic.Task, result string) (bool, error) {
	tenantID := target.tenantID
	store := target.store
	if store == nil || task == nil {
		return false, fmt.Errorf("check semantic task ownership: nil store or task")
	}
	now := semanticWorkerLeaseNow()
	var count int
	err := store.DB().QueryRowContext(ctx, `SELECT COUNT(*) FROM semantic_tasks
		WHERE task_id = ? AND status = ? AND receipt = ? AND lease_until IS NOT NULL AND lease_until > ?`,
		task.TaskID, semantic.TaskProcessing, task.Receipt, now).Scan(&count)
	if err != nil {
		return false, fmt.Errorf("check semantic task ownership: %w", err)
	}
	if count > 0 {
		return true, nil
	}
	metrics.RecordTenantOperationWithOrg(tenantID, target.metricOrgID(), "semantic_worker", "lease_lost", "ok", 0)
	logger.Warn(ctx, "tenant_worker_finalize_skipped_lease_lost",
		append([]zap.Field{
			zap.String("tenant_id", tenantID),
			zap.String("result", result),
		}, semanticTaskLogFields(task)...)...)
	return false, nil
}

func semanticWorkerLeaseNow() time.Time {
	now := time.Now().UTC()
	failpoint.Inject("semanticWorkerLeaseNow", func(val failpoint.Value) {
		switch injected := val.(type) {
		case string:
			parsed, err := time.Parse(time.RFC3339Nano, injected)
			if err == nil {
				now = parsed.UTC()
			}
		}
	})
	return now
}

func semanticTaskLogFields(task *semantic.Task) []zap.Field {
	if task == nil {
		return nil
	}
	fields := []zap.Field{
		zap.String("task_id", task.TaskID),
		zap.String("task_type", string(task.TaskType)),
		zap.String("resource_id", task.ResourceID),
		zap.Int64("resource_version", task.ResourceVersion),
		zap.Bool("has_receipt", task.Receipt != ""),
		zap.Int("attempt_count", task.AttemptCount),
	}
	if task.LeaseUntil != nil {
		fields = append(fields, zap.Time("lease_until", task.LeaseUntil.UTC()))
	}
	return fields
}

func imageExtractTaskSpecFromSemanticTask(task *semantic.Task) backend.ImageExtractTaskSpec {
	if task == nil {
		return backend.ImageExtractTaskSpec{}
	}
	spec := backend.ImageExtractTaskSpec{FileID: task.ResourceID, Revision: task.ResourceVersion}
	if len(task.PayloadJSON) == 0 {
		return spec
	}
	var payload semantic.ImgExtractTaskPayload
	if err := json.Unmarshal(task.PayloadJSON, &payload); err == nil {
		spec.Path = payload.Path
		spec.ContentType = payload.ContentType
	}
	return spec
}

func audioExtractTaskSpecFromSemanticTask(task *semantic.Task) backend.AudioExtractTaskSpec {
	if task == nil {
		return backend.AudioExtractTaskSpec{}
	}
	spec := backend.AudioExtractTaskSpec{FileID: task.ResourceID, Revision: task.ResourceVersion}
	if len(task.PayloadJSON) == 0 {
		return spec
	}
	var payload semantic.AudioExtractTaskPayload
	if err := json.Unmarshal(task.PayloadJSON, &payload); err == nil {
		spec.Path = payload.Path
		spec.ContentType = payload.ContentType
	}
	return spec
}

func videoExtractTaskSpecFromSemanticTask(task *semantic.Task) backend.VideoExtractTaskSpec {
	if task == nil {
		return backend.VideoExtractTaskSpec{}
	}
	spec := backend.VideoExtractTaskSpec{FileID: task.ResourceID, Revision: task.ResourceVersion}
	if len(task.PayloadJSON) == 0 {
		return spec
	}
	var payload semantic.VideoExtractTaskPayload
	if err := json.Unmarshal(task.PayloadJSON, &payload); err == nil {
		spec.Path = payload.Path
		spec.ContentType = payload.ContentType
	}
	return spec
}

func hasAnyTaskTypes(types []semantic.TaskType) bool {
	return len(types) > 0
}

// unionTaskTypes merges two task-type slices, deduplicating by value.
func unionTaskTypes(a, b []semantic.TaskType) []semantic.TaskType {
	if len(a) == 0 {
		return b
	}
	if len(b) == 0 {
		return a
	}
	seen := make(map[semantic.TaskType]struct{}, len(a)+len(b))
	out := make([]semantic.TaskType, 0, len(a)+len(b))
	for _, t := range a {
		if _, ok := seen[t]; ok {
			continue
		}
		seen[t] = struct{}{}
		out = append(out, t)
	}
	for _, t := range b {
		if _, ok := seen[t]; ok {
			continue
		}
		seen[t] = struct{}{}
		out = append(out, t)
	}
	return out
}

// tenantWorkerLogTaskTypesFromTypes stringifies task-type slices for zap.
func tenantWorkerLogTaskTypesFromTypes(types []semantic.TaskType) []string {
	if len(types) == 0 {
		return nil
	}
	out := make([]string, len(types))
	for i, t := range types {
		out[i] = string(t)
	}
	return out
}