package server

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"github.com/c4pt0r/agfs/agfs-server/pkg/filesystem"
	"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/meta"
	"github.com/mem9-ai/drive9/pkg/metrics"
	"github.com/mem9-ai/drive9/pkg/s3client"
	"github.com/mem9-ai/drive9/pkg/semantic"
	"github.com/mem9-ai/drive9/pkg/testtidb"
)

func TestProcessEmbedTaskUsesTenantRuntimeAndGeneration(t *testing.T) {
	initServerTenantSchema(t, testDSN)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	testtidb.ResetDB(t, store.DB())
	t.Cleanup(func() { _ = store.Close() })
	s3c, err := s3client.NewLocal(mustTempDir(t), "http://localhost:9091/s3")
	if err != nil {
		t.Fatal(err)
	}
	vec := make([]float32, embedding.Dimensions)
	vec[0] = 0.25
	b, err := backend.NewWithS3ModeAndOptions(store, s3c, true, backend.Options{
		TenantID:                   "tenant-app-embed",
		AppSemanticTasksEnabled:    true,
		AppEmbeddingGeneration:     7,
		AppEmbeddingConfigRevision: 4,
		QueryEmbedding:             backend.QueryEmbeddingOptions{Client: staticSemanticEmbedder{vec: vec}},
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := b.Write("/document.txt", []byte("generation aware document"), 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	task, found, err := store.ClaimSemanticTask(context.Background(), time.Now().Add(time.Second), time.Minute, semantic.TaskTypeEmbed)
	if err != nil || !found {
		t.Fatalf("claim embed task found=%v err=%v", found, err)
	}
	m := newTenantWorkerManager(b, nil, nil, nil, TenantWorkerOptions{}, 0)
	target := &tenantTarget{tenantID: "tenant-app-embed", backend: b, store: store}
	outcome := m.processEmbedTask(context.Background(), target, task)
	if outcome.action != semanticTaskActionAck || outcome.result != "ok" {
		t.Fatalf("outcome=%+v, want ack/ok", outcome)
	}
	var generation uint64
	if err := store.DB().QueryRow(`SELECT embedding_generation FROM semantic WHERE inode_id = ?`, task.ResourceID).Scan(&generation); err != nil {
		t.Fatal(err)
	}
	if generation != 7 {
		t.Fatalf("embedding_generation=%d, want 7", generation)
	}
}

func TestProcessEmbedTaskDoesNotRestoreEmbeddingAfterContentCleared(t *testing.T) {
	initServerTenantSchema(t, testDSN)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	testtidb.ResetDB(t, store.DB())
	t.Cleanup(func() { _ = store.Close() })
	ctx := context.Background()
	now := time.Now().UTC()
	const fileID = "content-cleared-during-embed"
	if err := store.InsertFile(ctx, &datastore.File{
		FileID: fileID, StorageType: datastore.StorageDB9, StorageRef: "/blobs/" + fileID,
		Revision: 1, Status: datastore.StatusConfirmed, ContentText: "old media transcript",
		CreatedAt: now, ConfirmedAt: &now,
	}); err != nil {
		t.Fatal(err)
	}
	if updated, err := store.UpdateFileEmbeddingForGeneration(ctx, fileID, 1, 7, make([]float32, embedding.Dimensions)); err != nil || !updated {
		t.Fatalf("seed embedding updated=%v err=%v", updated, err)
	}
	payloadJSON, err := json.Marshal(semantic.EmbedTaskPayload{Generation: 7, ConfigRevision: 4})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := store.EnqueueSemanticTask(ctx, &semantic.Task{
		TaskID: "content-cleared-embed-task", TaskType: semantic.TaskTypeEmbed,
		ResourceID: fileID, ResourceVersion: 1, Status: semantic.TaskQueued, PayloadJSON: payloadJSON,
	}); err != nil {
		t.Fatal(err)
	}
	task, found, err := store.ClaimSemanticTask(ctx, now.Add(time.Second), time.Minute, semantic.TaskTypeEmbed)
	if err != nil || !found {
		t.Fatalf("claim embed task found=%v err=%v", found, err)
	}

	started := make(chan string, 1)
	release := make(chan struct{})
	vec := make([]float32, embedding.Dimensions)
	vec[0] = 0.5
	b, err := backend.NewWithOptions(store, backend.Options{
		TenantID: "tenant-content-cleared", AppSemanticTasksEnabled: true,
		AppEmbeddingGeneration: 7, AppEmbeddingConfigRevision: 4,
		QueryEmbedding: backend.QueryEmbeddingOptions{Client: blockingSemanticEmbedder{
			started: started, release: release, vec: vec,
		}},
	})
	if err != nil {
		t.Fatal(err)
	}
	target := &tenantTarget{tenantID: "tenant-content-cleared", backend: b, store: store}
	outcomes := make(chan semanticTaskOutcome, 1)
	go func() {
		outcomes <- newTenantWorkerManager(b, nil, nil, nil, TenantWorkerOptions{}, 0).
			processEmbedTask(ctx, target, task)
	}()

	select {
	case got := <-started:
		if got != "old media transcript" {
			t.Fatalf("embedded text=%q, want old transcript", got)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("embedding provider was not called")
	}
	if err := store.InTx(ctx, func(tx *sql.Tx) error {
		updated, err := store.UpdateFileSearchTextTx(tx, fileID, 1, "")
		if err != nil {
			return err
		}
		if !updated {
			return errors.New("empty media writeback missed current revision")
		}
		return store.ClearFileEmbeddingStateTx(tx, fileID)
	}); err != nil {
		t.Fatal(err)
	}
	close(release)

	select {
	case outcome := <-outcomes:
		if outcome.action != semanticTaskActionAck || outcome.result != "obsolete" || outcome.message != "conditional_write_miss" {
			t.Fatalf("outcome=%+v, want ack/obsolete conditional_write_miss", outcome)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("embed task did not finish")
	}
	file, err := store.GetFile(ctx, fileID)
	if err != nil {
		t.Fatal(err)
	}
	if file.ContentText != "" || file.EmbeddingRevision != nil {
		t.Fatalf("file content=%q embedding_revision=%v, want empty content and no embedding", file.ContentText, file.EmbeddingRevision)
	}
}

func TestProcessEmbedTaskProviderFailureUsesBoundedSignal(t *testing.T) {
	const providerSecret = "embedding-provider-secret-detail"
	provider := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		w.WriteHeader(http.StatusUnauthorized)
		_, _ = w.Write([]byte(`{"error":{"message":"` + providerSecret + `"}}`))
	}))
	defer provider.Close()
	client, err := embedding.NewOpenAIClient(embedding.OpenAIClientConfig{
		BaseURL: provider.URL, APIKey: "tenant-provider-key", Model: "embed-model", Client: provider.Client(),
	})
	if err != nil {
		t.Fatal(err)
	}

	initServerTenantSchema(t, testDSN)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	testtidb.ResetDB(t, store.DB())
	t.Cleanup(func() { _ = store.Close() })
	now := time.Now().UTC()
	if err := store.InsertFile(context.Background(), &datastore.File{
		FileID: "provider-failure-file", StorageType: datastore.StorageDB9,
		StorageRef: "/blobs/provider-failure-file", Revision: 2,
		Status: datastore.StatusConfirmed, ContentText: "must be embedded",
		CreatedAt: now, ConfirmedAt: &now,
	}); err != nil {
		t.Fatal(err)
	}
	payloadJSON, err := json.Marshal(semantic.EmbedTaskPayload{Generation: 7, ConfigRevision: 4})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := store.EnqueueSemanticTask(context.Background(), &semantic.Task{
		TaskID: "provider-failure-task", TaskType: semantic.TaskTypeEmbed,
		ResourceID: "provider-failure-file", ResourceVersion: 2,
		Status: semantic.TaskQueued, PayloadJSON: payloadJSON,
	}); err != nil {
		t.Fatal(err)
	}
	task, found, err := store.ClaimSemanticTask(context.Background(), time.Now().Add(time.Second), time.Minute, semantic.TaskTypeEmbed)
	if err != nil || !found {
		t.Fatalf("claim embed task found=%v err=%v", found, err)
	}
	b, err := backend.NewWithOptions(store, backend.Options{
		TenantID: "tenant-provider-failure", AppSemanticTasksEnabled: true,
		AppEmbeddingGeneration: 7, AppEmbeddingConfigRevision: 4,
		QueryEmbedding: backend.QueryEmbeddingOptions{Client: client},
	})
	if err != nil {
		t.Fatal(err)
	}
	target := &tenantTarget{tenantID: "tenant-provider-failure", tidbCloudOrgID: "org-provider-failure", backend: b, store: store}
	m := newTenantWorkerManager(b, nil, nil, nil, TenantWorkerOptions{}, 0)
	recorded := withObservabilityLogger(t, zap.WarnLevel)
	outcome := m.processEmbedTask(context.Background(), target, task)
	if outcome.action != semanticTaskActionRetry || outcome.result != "embed_error" || outcome.message != "embedding_provider_auth_rejected" {
		t.Fatalf("outcome=%+v, want retry/embed_error with bounded provider code", outcome)
	}
	entries := recorded.FilterMessage("tenant_worker_embedding_provider_failed").All()
	if len(entries) != 1 {
		t.Fatalf("provider failure logs=%d, want 1: %v", len(entries), recorded.All())
	}
	fields := entries[0].ContextMap()
	if fields["failure_category"] != "auth_rejected" || fmt.Sprint(fields["provider_status"]) != "401" {
		t.Fatalf("provider failure fields=%v", fields)
	}
	if strings.Contains(fmt.Sprint(entries), providerSecret) {
		t.Fatalf("provider failure log contains provider response detail: %v", entries)
	}
	recorder := httptest.NewRecorder()
	metrics.WritePrometheus(recorder)
	metric := `drive9_service_operations_total{component="embedding",operation="provider_request",result="auth_rejected"}`
	if !strings.Contains(recorder.Body.String(), metric) || strings.Contains(recorder.Body.String(), providerSecret) {
		t.Fatalf("embedding provider metric missing or unsafe: %s", recorder.Body.String())
	}
}

func TestProcessEmbedTaskUnknownProviderFailurePersistsBoundedSignal(t *testing.T) {
	const providerDetail = "dial tcp provider.internal.example:443: connection refused"
	initServerTenantSchema(t, testDSN)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	testtidb.ResetDB(t, store.DB())
	t.Cleanup(func() { _ = store.Close() })
	now := time.Now().UTC()
	if err := store.InsertFile(context.Background(), &datastore.File{
		FileID: "unknown-provider-failure-file", StorageType: datastore.StorageDB9,
		StorageRef: "/blobs/unknown-provider-failure-file", Revision: 2,
		Status: datastore.StatusConfirmed, ContentText: "must be embedded",
		CreatedAt: now, ConfirmedAt: &now,
	}); err != nil {
		t.Fatal(err)
	}
	payloadJSON, err := json.Marshal(semantic.EmbedTaskPayload{Generation: 7, ConfigRevision: 4})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := store.EnqueueSemanticTask(context.Background(), &semantic.Task{
		TaskID: "unknown-provider-failure-task", TaskType: semantic.TaskTypeEmbed,
		ResourceID: "unknown-provider-failure-file", ResourceVersion: 2,
		Status: semantic.TaskQueued, PayloadJSON: payloadJSON,
	}); err != nil {
		t.Fatal(err)
	}
	task, found, err := store.ClaimSemanticTask(context.Background(), time.Now().Add(time.Second), time.Minute, semantic.TaskTypeEmbed)
	if err != nil || !found {
		t.Fatalf("claim embed task found=%v err=%v", found, err)
	}
	b, err := backend.NewWithOptions(store, backend.Options{
		TenantID: "tenant-unknown-provider-failure", AppSemanticTasksEnabled: true,
		AppEmbeddingGeneration: 7, AppEmbeddingConfigRevision: 4,
		QueryEmbedding: backend.QueryEmbeddingOptions{Client: staticSemanticEmbedder{err: errors.New(providerDetail)}},
	})
	if err != nil {
		t.Fatal(err)
	}
	target := &tenantTarget{tenantID: "tenant-unknown-provider-failure", backend: b, store: store}
	m := newTenantWorkerManager(b, nil, nil, nil, TenantWorkerOptions{}, 0)
	recorded := withObservabilityLogger(t, zap.WarnLevel)
	m.processTask(context.Background(), target, task)

	var status, lastError string
	if err := store.DB().QueryRow(
		"SELECT status, last_error FROM semantic_tasks WHERE task_id = ?", task.TaskID,
	).Scan(&status, &lastError); err != nil {
		t.Fatal(err)
	}
	if status != string(semantic.TaskQueued) || lastError != "embedding_provider_request_failed" {
		t.Fatalf("persisted task = status:%q last_error:%q, want queued/bounded provider code", status, lastError)
	}
	if strings.Contains(lastError, providerDetail) || strings.Contains(fmt.Sprint(recorded.All()), providerDetail) {
		t.Fatalf("provider detail escaped into durable state or logs: last_error=%q logs=%v", lastError, recorded.All())
	}
	entries := recorded.FilterMessage("tenant_worker_embedding_provider_failed").All()
	if len(entries) != 1 || entries[0].ContextMap()["failure_category"] != "request_failed" {
		t.Fatalf("bounded provider failure logs=%v, want one request_failed entry", entries)
	}
}

func TestProcessEmbedTaskRefreshesStaleGenerationInsteadOfDroppingWork(t *testing.T) {
	initServerTenantSchema(t, testDSN)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	for _, tc := range []struct {
		name        string
		payloadJSON []byte
	}{
		{name: "stale generation", payloadJSON: []byte(`{"generation":6}`)},
		{name: "stale config revision", payloadJSON: []byte(`{"generation":7,"config_revision":3}`)},
		{name: "legacy empty payload"},
	} {
		t.Run(tc.name, func(t *testing.T) {
			testtidb.ResetDB(t, store.DB())
			now := time.Now().UTC()
			if err := store.InsertFile(context.Background(), &datastore.File{
				FileID: "stale-generation-file", StorageType: datastore.StorageDB9,
				StorageRef: "/blobs/stale-generation-file", Revision: 2,
				Status: datastore.StatusConfirmed, ContentText: "must still be embedded",
				CreatedAt: now, ConfirmedAt: &now,
			}); err != nil {
				t.Fatal(err)
			}
			if _, err := store.EnqueueSemanticTask(context.Background(), &semantic.Task{
				TaskID: "stale-generation-task", TaskType: semantic.TaskTypeEmbed,
				ResourceID: "stale-generation-file", ResourceVersion: 2,
				Status: semantic.TaskQueued, PayloadJSON: tc.payloadJSON,
			}); err != nil {
				t.Fatal(err)
			}
			task, found, err := store.ClaimSemanticTask(context.Background(), time.Now().Add(time.Second), time.Minute, semantic.TaskTypeEmbed)
			if err != nil || !found {
				t.Fatalf("claim stale embed task found=%v err=%v", found, err)
			}
			b, err := backend.NewWithOptions(store, backend.Options{
				TenantID: "tenant-stale-generation", AppSemanticTasksEnabled: true,
				AppEmbeddingGeneration: 7, AppEmbeddingConfigRevision: 4,
				QueryEmbedding: backend.QueryEmbeddingOptions{Client: staticSemanticEmbedder{vec: make([]float32, embedding.Dimensions)}},
			})
			if err != nil {
				t.Fatal(err)
			}
			target := &tenantTarget{tenantID: "tenant-stale-generation", backend: b, store: store}
			m := newTenantWorkerManager(b, nil, nil, nil, TenantWorkerOptions{}, 0)
			outcome := m.processEmbedTask(context.Background(), target, task)
			if outcome.action != semanticTaskActionDefer || outcome.result != "superseded" {
				t.Fatalf("outcome=%+v, want defer/superseded", outcome)
			}
			var payloadJSON []byte
			if err := store.DB().QueryRow(`SELECT payload_json FROM semantic_tasks WHERE task_id = ?`, task.TaskID).Scan(&payloadJSON); err != nil {
				t.Fatal(err)
			}
			var payload semantic.EmbedTaskPayload
			if err := json.Unmarshal(payloadJSON, &payload); err != nil {
				t.Fatal(err)
			}
			if payload.Generation != 7 || payload.ConfigRevision != 4 {
				t.Fatalf("refreshed payload=%+v, want generation 7 config revision 4", payload)
			}
		})
	}
}

func TestProcessEmbedTaskDoesNotRewriteNewerTenantConfigFromStaleBackend(t *testing.T) {
	initServerTenantSchema(t, testDSN)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	testtidb.ResetDB(t, store.DB())
	t.Cleanup(func() { _ = store.Close() })
	now := time.Now().UTC()
	if err := store.InsertFile(context.Background(), &datastore.File{
		FileID: "newer-config-file", StorageType: datastore.StorageDB9,
		StorageRef: "/blobs/newer-config-file", Revision: 2,
		Status: datastore.StatusConfirmed, ContentText: "must use the current provider",
		CreatedAt: now, ConfirmedAt: &now,
	}); err != nil {
		t.Fatal(err)
	}
	payloadJSON, err := json.Marshal(semantic.EmbedTaskPayload{
		Generation: 2, ConfigRevision: 4, TenantManaged: true,
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := store.EnqueueSemanticTask(context.Background(), &semantic.Task{
		TaskID: "newer-config-task", TaskType: semantic.TaskTypeEmbed,
		ResourceID: "newer-config-file", ResourceVersion: 2,
		Status: semantic.TaskQueued, PayloadJSON: payloadJSON,
	}); err != nil {
		t.Fatal(err)
	}
	task, found, err := store.ClaimSemanticTask(context.Background(), now.Add(time.Second), time.Minute, semantic.TaskTypeEmbed)
	if err != nil || !found {
		t.Fatalf("claim found=%v err=%v", found, err)
	}
	b, err := backend.NewWithOptions(store, backend.Options{
		TenantID: "tenant-stale-runtime", AppSemanticTasksEnabled: true,
		AppEmbeddingGeneration: 1, AppEmbeddingConfigRevision: 3, AppEmbeddingTenantManaged: true,
		QueryEmbedding: backend.QueryEmbeddingOptions{Client: staticSemanticEmbedder{err: errors.New("stale provider must not be called")}},
	})
	if err != nil {
		t.Fatal(err)
	}
	target := &tenantTarget{tenantID: "tenant-stale-runtime", backend: b, store: store}
	outcome := newTenantWorkerManager(b, nil, nil, nil, TenantWorkerOptions{}, 0).
		processEmbedTask(context.Background(), target, task)
	if outcome.action != semanticTaskActionDefer || outcome.result != "superseded" {
		t.Fatalf("outcome=%+v, want defer/superseded", outcome)
	}

	var storedJSON []byte
	if err := store.DB().QueryRow(`SELECT payload_json FROM semantic_tasks WHERE task_id = ?`, task.TaskID).Scan(&storedJSON); err != nil {
		t.Fatal(err)
	}
	var stored semantic.EmbedTaskPayload
	if err := json.Unmarshal(storedJSON, &stored); err != nil {
		t.Fatal(err)
	}
	if stored.Generation != 2 || stored.ConfigRevision != 4 || !stored.TenantManaged {
		t.Fatalf("stored payload=%+v, want newer tenant config unchanged", stored)
	}
}

func TestProcessEmbeddingBackfillTaskQueuesGenerationAwareEmbedWork(t *testing.T) {
	initServerTenantSchema(t, testDSN)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	testtidb.ResetDB(t, store.DB())
	t.Cleanup(func() { _ = store.Close() })
	now := time.Now().UTC()
	if err := store.InsertFile(context.Background(), &datastore.File{
		FileID: "backfill-file", StorageType: datastore.StorageDB9, StorageRef: "/blobs/backfill-file",
		Revision: 2, Status: datastore.StatusConfirmed, ContentText: "existing semantic text",
		CreatedAt: now, ConfirmedAt: &now,
	}); err != nil {
		t.Fatal(err)
	}
	s3c, err := s3client.NewLocal(mustTempDir(t), "http://localhost:9091/s3")
	if err != nil {
		t.Fatal(err)
	}
	b, err := backend.NewWithS3ModeAndOptions(store, s3c, true, backend.Options{
		TenantID: "tenant-backfill", AppSemanticTasksEnabled: true,
		AppEmbeddingGeneration: 7, AppEmbeddingConfigRevision: 4,
		QueryEmbedding: backend.QueryEmbeddingOptions{Client: staticSemanticEmbedder{vec: make([]float32, embedding.Dimensions)}},
	})
	if err != nil {
		t.Fatal(err)
	}
	if created, err := b.EnqueueEmbeddingBackfillTask(context.Background()); err != nil || !created {
		t.Fatalf("enqueue backfill created=%v err=%v", created, err)
	}
	task, found, err := store.ClaimSemanticTask(context.Background(), time.Now().Add(time.Second), time.Minute, semantic.TaskTypeEmbedBackfill)
	if err != nil || !found {
		t.Fatalf("claim backfill found=%v err=%v", found, err)
	}
	target := &tenantTarget{tenantID: "tenant-backfill", backend: b, store: store}
	m := newTenantWorkerManager(b, nil, nil, nil, TenantWorkerOptions{}, 0)
	outcome := m.processEmbeddingBackfillTask(context.Background(), target, task)
	if outcome.action != semanticTaskActionDefer || outcome.result != "page" {
		t.Fatalf("outcome=%+v, want defer/page", outcome)
	}
	var payloadJSON []byte
	if err := store.DB().QueryRow(`SELECT payload_json FROM semantic_tasks
		WHERE task_type = ? AND resource_id = ? AND resource_version = ?`, semantic.TaskTypeEmbed, "backfill-file", 2).Scan(&payloadJSON); err != nil {
		t.Fatal(err)
	}
	var payload semantic.EmbedTaskPayload
	if err := json.Unmarshal(payloadJSON, &payload); err != nil {
		t.Fatal(err)
	}
	if payload.Generation != 7 || payload.ConfigRevision != 4 {
		t.Fatalf("child payload=%+v, want generation 7 config revision 4", payload)
	}
}

// spyImageExtractor counts ExtractImageText calls so tests can prove the
// provider was (not) invoked.
type spyImageExtractor struct {
	calls int
	text  string
	usage backend.ImageExtractUsage
	err   error
}

// fakeImagePNG is fake image content prefixed with the PNG magic bytes so it
// passes the raster-image sniff in the extraction path.
var fakeImagePNG = append([]byte{0x89, 'P', 'N', 'G', '\r', '\n', 0x1A, '\n'}, "fake-png"...)

func (s *spyImageExtractor) ExtractImageText(context.Context, backend.ImageExtractRequest) (string, backend.ImageExtractUsage, error) {
	s.calls++
	return s.text, s.usage, s.err
}

func disabledImageRow(tenantID string) meta.TenantExtractConfig {
	row := enabledImageRow(tenantID)
	row.Enabled = false
	return row
}

func imgExtractTestTask() *semantic.Task {
	return &semantic.Task{TaskID: "task-img-1", TaskType: semantic.TaskTypeImgExtractText}
}

func newImageExtractFailureWorker(
	t *testing.T,
	extractor backend.ImageTextExtractor,
	processDefault bool,
	content []byte,
) (*tenantWorkerManager, *tenantTarget, *semantic.Task) {
	t.Helper()
	initServerTenantSchema(t, testDSN)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	testtidb.ResetDB(t, store.DB())
	t.Cleanup(func() { _ = store.Close() })

	s3c, err := s3client.NewLocal(mustTempDir(t), "http://localhost:9091/s3")
	if err != nil {
		t.Fatal(err)
	}
	imageOpts := backend.AsyncImageExtractOptions{Capability: true}
	if processDefault {
		imageOpts = backend.AsyncImageExtractOptions{Enabled: true, Extractor: extractor}
	}
	b, err := backend.NewWithS3ModeAndOptions(store, s3c, true, backend.Options{
		AsyncImageExtract: imageOpts,
		TenantID:          "tenant-provider-failure",
		TiDBCloudOrgID:    "org-provider-failure",
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := b.Write("/img/provider-failure.png", content, 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	nf, err := b.Store().Stat(context.Background(), "/img/provider-failure.png")
	if err != nil || nf.File == nil {
		t.Fatalf("stat provider failure image: %v", err)
	}
	task := &semantic.Task{
		TaskID:          "task-provider-failure",
		TaskType:        semantic.TaskTypeImgExtractText,
		ResourceID:      nf.File.FileID,
		ResourceVersion: nf.File.Revision,
		AttemptCount:    2,
	}
	target := &tenantTarget{
		tenantID:       "tenant-provider-failure",
		tidbCloudOrgID: "org-provider-failure",
		backend:        b,
		store:          b.Store(),
	}
	m := newTenantWorkerManager(b, nil, nil, nil, TenantWorkerOptions{}, 0)
	if m == nil {
		t.Fatal("expected viable worker manager")
	}
	if !processDefault {
		reader := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{
			target.tenantID: enabledImageRow(target.tenantID),
		}}
		resolver := newExtractConfigResolver(reader, &fakeExtractDecryptor{plaintext: "tenant-provider-key"}, nil, time.Minute)
		t.Cleanup(resolver.Close)
		resolver.newOpenAIExtractor = func(backend.OpenAIImageTextExtractorConfig) (backend.ImageTextExtractor, error) {
			return extractor, nil
		}
		m.setExtractRuntimeDeps(resolver, nil)
	}
	return m, target, task
}

func TestProcessImgExtractTaskProviderFailure(t *testing.T) {
	const providerSecret = "provider-response-secret-must-not-escape"
	providerErr := imageProviderHTTPFailure(t, http.StatusUnauthorized, providerSecret)

	for _, processDefault := range []bool{false, true} {
		name := "resolver runtime"
		if processDefault {
			name = "process default runtime"
		}
		t.Run(name, func(t *testing.T) {
			recorded := withObservabilityLogger(t, zap.WarnLevel)
			extractor := &spyImageExtractor{err: providerErr}
			m, target, task := newImageExtractFailureWorker(t, extractor, processDefault, fakeImagePNG)

			outcome := m.processImgExtractTask(context.Background(), target, task)
			if outcome.action != semanticTaskActionRetry || outcome.result != string(backend.ImageExtractResultExtractError) ||
				outcome.message != "image_extract_provider_auth_rejected" {
				t.Fatalf("outcome = %+v, want retry/extract_error with fixed provider code", outcome)
			}
			if extractor.calls != 1 {
				t.Fatalf("extractor calls = %d, want 1", extractor.calls)
			}

			entries := recorded.FilterMessage("tenant_worker_image_extract_provider_failed").All()
			if len(entries) != 1 {
				t.Fatalf("provider failure logs = %d, want 1: %v", len(entries), recorded.All())
			}
			fields := entries[0].ContextMap()
			if fields["tenant_id"] != target.tenantID || fields["task_id"] != task.TaskID ||
				fields["failure_category"] != "auth_rejected" || fmt.Sprint(fields["provider_status"]) != "401" {
				t.Fatalf("provider failure fields = %v", fields)
			}
			logText := fmt.Sprint(entries)
			for _, forbidden := range []string{providerSecret, "tenant-provider-key", "extractor.example.com", "vision-model-1"} {
				if strings.Contains(logText, forbidden) {
					t.Fatalf("provider failure log contains sensitive detail %q: %s", forbidden, logText)
				}
			}

			recorder := httptest.NewRecorder()
			metrics.WritePrometheus(recorder)
			metric := `drive9_service_operations_total{component="image_extract",operation="provider_request",result="auth_rejected"}`
			if !strings.Contains(recorder.Body.String(), metric) || strings.Contains(recorder.Body.String(), providerSecret) {
				t.Fatalf("provider request metric missing or unsafe: %s", recorder.Body.String())
			}
		})
	}
}

func TestProcessImgExtractTaskProviderFailurePersistsFixedLastError(t *testing.T) {
	const providerSecret = "provider-response-secret-must-not-persist"
	extractor := &spyImageExtractor{err: imageProviderHTTPFailure(t, http.StatusUnauthorized, providerSecret)}
	m, target, _ := newImageExtractFailureWorker(t, extractor, false, fakeImagePNG)

	claimed, found, err := target.store.ClaimSemanticTask(
		context.Background(), time.Now().UTC().Add(time.Second), time.Minute, semantic.TaskTypeImgExtractText,
	)
	if err != nil {
		t.Fatal(err)
	}
	if !found {
		t.Fatal("expected queued image extraction task")
	}
	m.processTask(context.Background(), target, claimed)

	var status, lastError string
	if err := target.store.DB().QueryRow(
		"SELECT status, last_error FROM semantic_tasks WHERE task_id = ?", claimed.TaskID,
	).Scan(&status, &lastError); err != nil {
		t.Fatal(err)
	}
	if status != string(semantic.TaskQueued) || lastError != "image_extract_provider_auth_rejected" {
		t.Fatalf("persisted task = status:%q last_error:%q, want queued/fixed provider code", status, lastError)
	}
	if strings.Contains(lastError, providerSecret) {
		t.Fatalf("persisted last_error contains provider response detail: %q", lastError)
	}
}

func TestProcessImgExtractTaskGenericFailureKeepsMessage(t *testing.T) {
	extractor := &spyImageExtractor{err: errors.New("generic extractor failure")}
	m, target, task := newImageExtractFailureWorker(t, extractor, false, fakeImagePNG)

	outcome := m.processImgExtractTask(context.Background(), target, task)
	if outcome.action != semanticTaskActionRetry || outcome.result != string(backend.ImageExtractResultExtractError) ||
		!strings.Contains(outcome.message, "generic extractor failure") {
		t.Fatalf("outcome = %+v, want unchanged generic retry message", outcome)
	}
}

// A provider 400 (request_rejected) stays retryable: the category also covers
// configuration-level failures (e.g. an unsupported or renamed model), which
// must not be silently acked. Same contract as the 401 case in
// TestProcessImgExtractTaskProviderFailure, pinned for 400 explicitly.
func TestProcessImgExtractTaskProviderRequestRejectedRetries(t *testing.T) {
	for _, processDefault := range []bool{false, true} {
		name := "resolver runtime"
		if processDefault {
			name = "process default runtime"
		}
		t.Run(name, func(t *testing.T) {
			extractor := &spyImageExtractor{err: imageProviderHTTPFailure(t, http.StatusBadRequest, "invalid request")}
			m, target, task := newImageExtractFailureWorker(t, extractor, processDefault, fakeImagePNG)

			outcome := m.processImgExtractTask(context.Background(), target, task)
			if outcome.action != semanticTaskActionRetry || outcome.result != string(backend.ImageExtractResultExtractError) ||
				outcome.message != "image_extract_provider_request_rejected" {
				t.Fatalf("outcome = %+v, want retry/extract_error with fixed provider code", outcome)
			}
			if extractor.calls != 1 {
				t.Fatalf("extractor calls = %d, want 1", extractor.calls)
			}
		})
	}
}

// Files declared as images whose bytes match no raster image format (e.g.
// macOS AppleDouble sidecars synced as .jpg) are acked as not_image without
// calling the provider; the provider would reject them with a permanent 400
// anyway.
func TestProcessImgExtractTaskNonImageBytesSkipped(t *testing.T) {
	// AppleDouble magic: 0x00051607 — not a raster image.
	junk := append([]byte{0x00, 0x05, 0x16, 0x07}, []byte("appledouble-junk")...)
	for _, processDefault := range []bool{false, true} {
		name := "resolver runtime"
		if processDefault {
			name = "process default runtime"
		}
		t.Run(name, func(t *testing.T) {
			extractor := &spyImageExtractor{text: "must not be called"}
			m, target, task := newImageExtractFailureWorker(t, extractor, processDefault, junk)

			outcome := m.processImgExtractTask(context.Background(), target, task)
			if outcome.action != semanticTaskActionAck || outcome.result != string(backend.ImageExtractResultNotImage) {
				t.Fatalf("outcome = %+v, want ack/not_image", outcome)
			}
			if extractor.calls != 0 {
				t.Fatalf("extractor calls = %d, want 0 for non-image bytes", extractor.calls)
			}
		})
	}
}

func TestProcessImgExtractTaskNonImageBytesSkippedPersistsSuccess(t *testing.T) {
	junk := append([]byte{0x00, 0x05, 0x16, 0x07}, []byte("appledouble-junk")...)
	extractor := &spyImageExtractor{text: "must not be called"}
	m, target, _ := newImageExtractFailureWorker(t, extractor, false, junk)

	claimed, found, err := target.store.ClaimSemanticTask(
		context.Background(), time.Now().UTC().Add(time.Second), time.Minute, semantic.TaskTypeImgExtractText,
	)
	if err != nil {
		t.Fatal(err)
	}
	if !found {
		t.Fatal("expected queued image extraction task")
	}
	m.processTask(context.Background(), target, claimed)

	var status, lastError string
	if err := target.store.DB().QueryRow(
		"SELECT status, COALESCE(last_error, '') FROM semantic_tasks WHERE task_id = ?", claimed.TaskID,
	).Scan(&status, &lastError); err != nil {
		t.Fatal(err)
	}
	if status != string(semantic.TaskSucceeded) {
		t.Fatalf("persisted task = status:%q last_error:%q, want succeeded", status, lastError)
	}
}

// (a) A disabled tenant row discards the task with result "disabled"; the backend
// is never touched.
func TestProcessImgExtractTaskTenantDisabled(t *testing.T) {
	reader := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{
		"t1": disabledImageRow("t1"),
	}}
	r := newExtractConfigResolver(reader, &fakeExtractDecryptor{}, defaultTestRuntime(), time.Minute)
	t.Cleanup(r.Close)
	m := &tenantWorkerManager{extractResolver: r}

	outcome := m.processImgExtractTask(context.Background(), &tenantTarget{tenantID: "t1"}, imgExtractTestTask())
	if outcome.action != semanticTaskActionDiscard || outcome.result != "disabled" {
		t.Fatalf("outcome = %+v, want discard/disabled", outcome)
	}
}

// (b) An absent row with no process default discards with "not_configured".
func TestProcessImgExtractTaskNotConfigured(t *testing.T) {
	reader := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{}}
	r := newExtractConfigResolver(reader, &fakeExtractDecryptor{}, nil, time.Minute)
	t.Cleanup(r.Close)
	m := &tenantWorkerManager{extractResolver: r}

	outcome := m.processImgExtractTask(context.Background(), &tenantTarget{tenantID: "t1"}, imgExtractTestTask())
	if outcome.action != semanticTaskActionDiscard || outcome.result != "not_configured" {
		t.Fatalf("outcome = %+v, want discard/not_configured", outcome)
	}
}

func TestTerminalMediaConfigDiscardsClaimableBatch(t *testing.T) {
	mediaTypes := []struct {
		name      string
		mediaType string
		taskType  semantic.TaskType
	}{
		{
			name: "image", mediaType: meta.ExtractMediaTypeImage, taskType: semantic.TaskTypeImgExtractText,
		},
		{
			name: "audio", mediaType: meta.ExtractMediaTypeAudio, taskType: semantic.TaskTypeAudioExtractText,
		},
		{
			name: "video", mediaType: meta.ExtractMediaTypeVideo, taskType: semantic.TaskTypeVideoExtractVisual,
		},
	}
	for _, tc := range mediaTypes {
		for _, state := range []string{"not_configured", "disabled"} {
			t.Run(tc.name+"/"+state, func(t *testing.T) {
				initServerTenantSchema(t, testDSN)
				store, err := datastore.Open(testDSN)
				if err != nil {
					t.Fatal(err)
				}
				testtidb.ResetDB(t, store.DB())
				t.Cleanup(func() { _ = store.Close() })
				ctx := context.Background()
				now := time.Now().UTC()
				for i := range 3 {
					task := &semantic.Task{
						TaskID: fmt.Sprintf("%s-terminal-%d", tc.name, i), TaskType: tc.taskType,
						ResourceID: fmt.Sprintf("%s-file-%d", tc.name, i), ResourceVersion: 1,
						Status: semantic.TaskQueued, MaxAttempts: 5, AvailableAt: now,
						CreatedAt: now.Add(time.Duration(i) * time.Millisecond), UpdatedAt: now,
					}
					if _, err := store.EnqueueSemanticTask(ctx, task); err != nil {
						t.Fatal(err)
					}
				}
				claimed, found, err := store.ClaimSemanticTask(ctx, now.Add(time.Second), time.Minute, tc.taskType)
				if err != nil || !found {
					t.Fatalf("claim found=%v err=%v", found, err)
				}
				rows := map[string]meta.TenantExtractConfig{}
				if state == "disabled" {
					rows["terminal-batch\x00"+tc.mediaType] = meta.TenantExtractConfig{
						TenantID: "terminal-batch", MediaType: tc.mediaType, Enabled: false,
					}
				}
				resolver := newExtractConfigResolver(
					&fakeTenantExtractConfigReader{rows: rows},
					&fakeExtractDecryptor{}, nil, time.Minute,
				)
				t.Cleanup(resolver.Close)
				m := &tenantWorkerManager{extractResolver: resolver}
				target := &tenantTarget{tenantID: "terminal-batch", backend: &backend.Dat9Backend{}, store: store}
				m.processTask(ctx, target, claimed)
				var queued int
				if err := store.DB().QueryRowContext(ctx,
					`SELECT COUNT(*) FROM semantic_tasks WHERE task_type = ? AND status = ?`, tc.taskType, semantic.TaskQueued,
				).Scan(&queued); err != nil {
					t.Fatal(err)
				}
				if queued != 0 {
					t.Fatalf("queued %s tasks=%d, want batch discarded", tc.name, queued)
				}
			})
		}
	}
}

func TestUnconfiguredVideoWithCheckpointAcksAndDiscardsFreshPeers(t *testing.T) {
	initServerTenantSchema(t, testDSN)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	testtidb.ResetDB(t, store.DB())
	t.Cleanup(func() { _ = store.Close() })

	ctx := context.Background()
	now := time.Now().UTC()
	progress, err := json.Marshal(semantic.VideoExtractTaskPayload{VisualDone: true, VisualText: "scene"})
	if err != nil {
		t.Fatal(err)
	}
	for i, payload := range [][]byte{progress, nil, nil} {
		task := &semantic.Task{
			TaskID: fmt.Sprintf("video-progress-%d", i), TaskType: semantic.TaskTypeVideoExtractVisual,
			ResourceID: fmt.Sprintf("video-file-%d", i), ResourceVersion: 1,
			Status: semantic.TaskQueued, MaxAttempts: 5, AvailableAt: now,
			PayloadJSON: payload, CreatedAt: now.Add(time.Duration(i) * time.Millisecond), UpdatedAt: now,
		}
		if _, err := store.EnqueueSemanticTask(ctx, task); err != nil {
			t.Fatal(err)
		}
	}
	claimed, found, err := store.ClaimSemanticTask(ctx, now.Add(time.Second), time.Minute, semantic.TaskTypeVideoExtractVisual)
	if err != nil || !found || claimed.TaskID != "video-progress-0" {
		t.Fatalf("claim found=%v task=%+v err=%v", found, claimed, err)
	}
	resolver := newExtractConfigResolver(
		&fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{}},
		&fakeExtractDecryptor{}, nil, time.Minute,
	)
	t.Cleanup(resolver.Close)
	m := &tenantWorkerManager{extractResolver: resolver}
	m.processTask(ctx, &tenantTarget{tenantID: "video-progress", store: store}, claimed)
	var queued int
	if err := store.DB().QueryRowContext(ctx,
		`SELECT COUNT(*) FROM semantic_tasks WHERE task_type = ? AND status = ?`,
		semantic.TaskTypeVideoExtractVisual, semantic.TaskQueued,
	).Scan(&queued); err != nil {
		t.Fatal(err)
	}
	if queued != 0 {
		t.Fatalf("queued video tasks=%d, want fresh peers batch discarded", queued)
	}
}

// (c) A resolver error is a retryable failure and the extractor is never
// called (fail-closed: no process-default fallback on error).
func TestProcessImgExtractTaskResolveErrorRetries(t *testing.T) {
	spy := &spyImageExtractor{text: "spy"}
	def := defaultTestRuntime()
	def.Extractor = spy
	reader := &fakeTenantExtractConfigReader{err: errors.New("meta store down")}
	r := newExtractConfigResolver(reader, &fakeExtractDecryptor{}, def, time.Minute)
	t.Cleanup(r.Close)
	m := &tenantWorkerManager{extractResolver: r}

	outcome := m.processImgExtractTask(context.Background(), &tenantTarget{tenantID: "t1"}, imgExtractTestTask())
	if outcome.action != semanticTaskActionRetry || outcome.result != "resolve_error" {
		t.Fatalf("outcome = %+v, want retry/resolve_error", outcome)
	}
	if spy.calls != 0 {
		t.Fatalf("extractor calls = %d, want 0 on resolve error", spy.calls)
	}
}

func TestProcessImgExtractTaskRejectsZeroValueResolution(t *testing.T) {
	r := newExtractConfigResolver(
		&fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{}},
		&fakeExtractDecryptor{},
		nil,
		time.Minute,
	)
	t.Cleanup(r.Close)
	r.mu.Lock()
	r.cache["t-zero"] = extractConfigCacheEntry{
		resolved:  resolvedImageExtract{},
		expiresAt: time.Now().Add(time.Minute),
	}
	r.mu.Unlock()

	m := &tenantWorkerManager{extractResolver: r}
	outcome := m.processImgExtractTask(
		context.Background(),
		&tenantTarget{tenantID: "t-zero"},
		imgExtractTestTask(),
	)
	if outcome.action != semanticTaskActionRetry || outcome.result != "resolve_state_invalid" {
		t.Fatalf("outcome = %+v, want retry/resolve_state_invalid", outcome)
	}
}

// (d) Kill switch: an already-claimed task that has not called the provider
// is deferred with global_disabled without consuming an attempt (resolver not consulted), and no new
// img_extract_text claim succeeds while the task stays queued. With the
// switch off, the claim drains and the resolved tenant runtime writes back.
func TestProcessImgExtractTaskKillSwitch(t *testing.T) {
	initServerTenantSchema(t, testDSN)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	testtidb.ResetDB(t, store.DB())
	t.Cleanup(func() { _ = store.Close() })

	s3c, err := s3client.NewLocal(mustTempDir(t), "http://localhost:9091/s3")
	if err != nil {
		t.Fatal(err)
	}
	b, err := backend.NewWithS3ModeAndOptions(store, s3c, true, backend.Options{
		AsyncImageExtract: backend.AsyncImageExtractOptions{Capability: true},
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := b.Write("/img/kill-switch.png", fakeImagePNG, 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}

	spy := &spyImageExtractor{text: "tenant caption"}
	reader := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{
		"t-kill": enabledImageRow("t-kill"),
	}}
	resolver := newExtractConfigResolver(reader, &fakeExtractDecryptor{plaintext: "k"}, nil, time.Minute)
	t.Cleanup(resolver.Close)
	resolver.newOpenAIExtractor = func(backend.OpenAIImageTextExtractorConfig) (backend.ImageTextExtractor, error) {
		return spy, nil
	}

	kill := &atomic.Bool{}
	kill.Store(true)
	m := newTenantWorkerManager(b, nil, nil, nil, TenantWorkerOptions{}, 0)
	if m == nil {
		t.Fatal("expected viable worker manager")
	}
	m.setExtractRuntimeDeps(resolver, kill)
	target := &tenantTarget{
		tenantID:         "t-kill",
		backend:          b,
		store:            b.Store(),
		allowedTaskTypes: []semantic.TaskType{semantic.TaskTypeImgExtractText},
		release:          func() {},
	}
	ctx := context.Background()

	// In-flight (already claimed) task: deferred global_disabled, resolver and
	// extractor untouched.
	outcome := m.processImgExtractTask(ctx, target, imgExtractTestTask())
	if outcome.action != semanticTaskActionDefer || outcome.result != "global_disabled" {
		t.Fatalf("outcome = %+v, want defer/global_disabled", outcome)
	}
	if reader.callCount() != 0 || spy.calls != 0 {
		t.Fatalf("resolver calls = %d, extractor calls = %d, want 0/0", reader.callCount(), spy.calls)
	}

	// A task claimed just before the switch turns on must return to queued
	// without consuming the attempt added by ClaimSemanticTask.
	kill.Store(false)
	claimed, found, err := store.ClaimSemanticTask(ctx, time.Now().UTC(), time.Minute, semantic.TaskTypeImgExtractText)
	if err != nil || !found {
		t.Fatalf("ClaimSemanticTask: found=%v err=%v", found, err)
	}
	if claimed.AttemptCount != 1 {
		t.Fatalf("claimed attempt_count = %d, want 1", claimed.AttemptCount)
	}
	kill.Store(true)
	m.processTask(ctx, target, claimed)
	var status string
	var attemptCount int
	if err := store.DB().QueryRowContext(ctx,
		`SELECT status, attempt_count FROM semantic_tasks WHERE task_id = ?`, claimed.TaskID,
	).Scan(&status, &attemptCount); err != nil {
		t.Fatalf("query deferred task: %v", err)
	}
	if status != string(semantic.TaskQueued) || attemptCount != 0 {
		t.Fatalf("deferred task status/attempt_count = %s/%d, want queued/0", status, attemptCount)
	}

	// No new claim while the switch is on; the queued task stays queued.
	if claimed, _ := m.claimAndProcessOne(ctx, target); claimed {
		t.Fatal("claimed img_extract_text while the kill switch is on")
	}
	assertSemanticTaskCount(t, store, string(semantic.TaskQueued), 1)

	// Switch off: the task is claimed, resolved, extracted, and written back.
	kill.Store(false)
	if claimed, _ := m.claimAndProcessOne(ctx, target); !claimed {
		t.Fatal("no claim after the kill switch was lifted")
	}
	if spy.calls != 1 {
		t.Fatalf("extractor calls = %d, want 1", spy.calls)
	}
	assertSemanticTaskCount(t, store, string(semantic.TaskQueued), 0)
	nf, err := b.Store().Stat(ctx, "/img/kill-switch.png")
	if err != nil || nf.File == nil {
		t.Fatalf("stat /img/kill-switch.png: %v", err)
	}
	if nf.File.ContentText != "tenant caption" {
		t.Fatalf("content_text = %q, want %q", nf.File.ContentText, "tenant caption")
	}
}

// flippingExtractConfigReader runs flip on every read, letting a test change
// cluster state (e.g. the kill switch) in the middle of task resolution.
type flippingExtractConfigReader struct {
	inner *fakeTenantExtractConfigReader
	flip  func()
}

func (f *flippingExtractConfigReader) GetTenantExtractConfig(ctx context.Context, tenantID, mediaType string) (*meta.TenantExtractConfig, error) {
	f.flip()
	return f.inner.GetTenantExtractConfig(ctx, tenantID, mediaType)
}

// (e) Kill switch re-check: the switch flipping during resolve/file-load —
// after the task-entry check already passed — must still prevent the provider
// call; the task is deferred with global_disabled.
func TestProcessImgExtractTaskKillSwitchRecheckBeforeProviderCall(t *testing.T) {
	initServerTenantSchema(t, testDSN)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	testtidb.ResetDB(t, store.DB())
	t.Cleanup(func() { _ = store.Close() })

	s3c, err := s3client.NewLocal(mustTempDir(t), "http://localhost:9091/s3")
	if err != nil {
		t.Fatal(err)
	}
	b, err := backend.NewWithS3ModeAndOptions(store, s3c, true, backend.Options{
		AsyncImageExtract: backend.AsyncImageExtractOptions{Capability: true},
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := b.Write("/img/guard-flip.png", fakeImagePNG, 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}

	spy := &spyImageExtractor{text: "tenant caption"}
	kill := &atomic.Bool{}
	reader := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{
		"t-guard": enabledImageRow("t-guard"),
	}}
	// Flip the switch while the worker resolves the tenant config — after
	// the task-entry kill-switch check already passed.
	flipReader := &flippingExtractConfigReader{inner: reader, flip: func() { kill.Store(true) }}
	resolver := newExtractConfigResolver(flipReader, &fakeExtractDecryptor{plaintext: "k"}, nil, time.Minute)
	t.Cleanup(resolver.Close)
	resolver.newOpenAIExtractor = func(backend.OpenAIImageTextExtractorConfig) (backend.ImageTextExtractor, error) {
		return spy, nil
	}
	m := newTenantWorkerManager(b, nil, nil, nil, TenantWorkerOptions{}, 0)
	if m == nil {
		t.Fatal("expected viable worker manager")
	}
	m.setExtractRuntimeDeps(resolver, kill)
	target := &tenantTarget{
		tenantID:         "t-guard",
		backend:          b,
		store:            b.Store(),
		allowedTaskTypes: []semantic.TaskType{semantic.TaskTypeImgExtractText},
		release:          func() {},
	}

	ctx := context.Background()
	nf, err := b.Store().Stat(ctx, "/img/guard-flip.png")
	if err != nil || nf.File == nil {
		t.Fatalf("stat /img/guard-flip.png: %v", err)
	}
	task := &semantic.Task{
		TaskID:          "task-img-guard",
		TaskType:        semantic.TaskTypeImgExtractText,
		ResourceID:      nf.File.FileID,
		ResourceVersion: nf.File.Revision,
	}
	outcome := m.processImgExtractTask(ctx, target, task)
	if outcome.action != semanticTaskActionDefer || outcome.result != "global_disabled" {
		t.Fatalf("outcome = %+v, want defer/global_disabled", outcome)
	}
	if spy.calls != 0 {
		t.Fatalf("extractor calls = %d, want 0 (provider must not be called after the switch flipped)", spy.calls)
	}
}

func TestProcessImgExtractTaskDefaultRuntimeKillSwitchRecheckDefers(t *testing.T) {
	initServerTenantSchema(t, testDSN)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	testtidb.ResetDB(t, store.DB())
	t.Cleanup(func() { _ = store.Close() })

	spy := &spyImageExtractor{text: "default caption"}
	s3c, err := s3client.NewLocal(mustTempDir(t), "http://localhost:9091/s3")
	if err != nil {
		t.Fatal(err)
	}
	b, err := backend.NewWithS3ModeAndOptions(store, s3c, true, backend.Options{
		AsyncImageExtract: backend.AsyncImageExtractOptions{
			Capability: true,
			Enabled:    true,
			Extractor:  spy,
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := b.Write("/img/default-guard-flip.png", fakeImagePNG, 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	nf, err := b.Store().Stat(context.Background(), "/img/default-guard-flip.png")
	if err != nil || nf.File == nil {
		t.Fatalf("stat default-runtime image: %v", err)
	}

	kill := &atomic.Bool{}
	kill.Store(true)
	m := &tenantWorkerManager{imageExtractGlobalDisabled: kill}
	target := &tenantTarget{tenantID: "t-default-guard", backend: b, store: store}
	outcome := m.processImgExtractTaskWithDefaultRuntime(context.Background(), target, &semantic.Task{
		TaskID:          "task-default-guard",
		TaskType:        semantic.TaskTypeImgExtractText,
		ResourceID:      nf.File.FileID,
		ResourceVersion: nf.File.Revision,
	})
	if outcome.action != semanticTaskActionDefer || outcome.result != "global_disabled" {
		t.Fatalf("outcome = %+v, want defer/global_disabled", outcome)
	}
	if spy.calls != 0 {
		t.Fatalf("extractor calls = %d, want 0", spy.calls)
	}
}

func assertSemanticTaskCount(t *testing.T, store *datastore.Store, status string, want int) {
	t.Helper()
	var n int
	if err := store.DB().QueryRowContext(context.Background(),
		`SELECT COUNT(*) FROM semantic_tasks WHERE status = ?`, status).Scan(&n); err != nil {
		t.Fatal(err)
	}
	if n != want {
		t.Fatalf("semantic task count(status=%s) = %d, want %d", status, n, want)
	}
}

// Capability-only backends (no process-default extractor) must still register
// img_extract_text on every worker task-type path: the runtime is resolved
// per tenant at execution time.
func TestImgExtractTaskTypesRegisteredFromCapability(t *testing.T) {
	initServerTenantSchema(t, testDSN)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	testtidb.ResetDB(t, store.DB())
	t.Cleanup(func() { _ = store.Close() })

	s3c, err := s3client.NewLocal(mustTempDir(t), "http://localhost:9091/s3")
	if err != nil {
		t.Fatal(err)
	}
	capabilityOnly := backend.Options{
		AsyncImageExtract: backend.AsyncImageExtractOptions{Capability: true},
	}
	autoOpts := capabilityOnly
	autoOpts.DatabaseAutoEmbedding = true
	autoBackend, err := backend.NewWithS3ModeAndOptions(store, s3c, true, autoOpts)
	if err != nil {
		t.Fatal(err)
	}
	appBackend, err := backend.NewWithS3ModeAndOptions(store, s3c, true, capabilityOnly)
	if err != nil {
		t.Fatal(err)
	}
	if autoBackend.SupportsAsyncImageExtract() || appBackend.SupportsAsyncImageExtract() {
		t.Fatal("capability-only backend must not report a process-default extractor")
	}
	if !hasTaskType(autoBackend.AutoSemanticTaskTypes(), semantic.TaskTypeImgExtractText) {
		t.Fatal("AutoSemanticTaskTypes missing img_extract_text for capability-only backend")
	}
	m := newTenantWorkerManager(appBackend, nil, nil, nil, TenantWorkerOptions{}, 0)
	if m == nil {
		t.Fatal("expected viable worker manager")
	}
	m.extractResolver = &extractConfigResolver{}
	targetTypes := m.taskTypesForTarget(appBackend)
	if !hasTaskType(targetTypes, semantic.TaskTypeImgExtractText) {
		t.Fatal("taskTypesForTarget missing img_extract_text for capability-only backend")
	}
	if got := countTaskType(targetTypes, semantic.TaskTypeImgExtractText); got != 1 {
		t.Fatalf("taskTypesForTarget img_extract_text count = %d, want 1", got)
	}
	if !hasTaskType(m.fallbackExtractTaskTypes(), semantic.TaskTypeImgExtractText) {
		t.Fatal("fallbackExtractTaskTypes missing img_extract_text for capability-only backend")
	}
}

func TestPrepareAudioExtractRuntimeRestoresCheckpointAndFinalAttempt(t *testing.T) {
	payload, err := json.Marshal(semantic.AudioExtractTaskPayload{
		Path: "/a.wav", ContentType: "audio/wav", CompletedSegments: 2, Text: "prefix",
		Tags: map[string]string{"topic": "meeting"}, Structured: true,
	})
	if err != nil {
		t.Fatal(err)
	}
	task := &semantic.Task{AttemptCount: 5, MaxAttempts: 5, PayloadJSON: payload}
	runtime := prepareAudioExtractRuntime(context.Background(), &tenantTarget{}, task, backend.AudioExtractRuntime{})
	if !runtime.FinalAttempt || runtime.CheckpointState.CompletedSegments != 2 || runtime.CheckpointState.Text != "prefix" ||
		runtime.CheckpointState.Tags["topic"] != "meeting" || !runtime.CheckpointState.Structured {
		t.Fatalf("runtime=%+v, want restored final-attempt checkpoint", runtime)
	}
}

func TestPrepareVideoExtractRuntimeRestoresModalityCheckpoint(t *testing.T) {
	payload, err := json.Marshal(semantic.VideoExtractTaskPayload{
		Path: "/v.mp4", ContentType: "video/mp4", VisualDone: false,
		VisualCompletedSegments: 2, VisualSegmentCount: 4, VisualSegmentsText: "segments 1-2",
		VisualTags: map[string]string{"scene": "office"}, AudioCompletedSegments: 3, AudioText: "audio prefix",
		AudioTags: map[string]string{"speaker": "alice"},
	})
	if err != nil {
		t.Fatal(err)
	}
	task := &semantic.Task{AttemptCount: 1, MaxAttempts: 5, PayloadJSON: payload}
	runtime := prepareVideoExtractRuntime(context.Background(), &tenantTarget{}, task, backend.VideoExtractRuntime{})
	if runtime.FinalAttempt || runtime.CheckpointState.VisualDone || runtime.CheckpointState.VisualCompletedSegments != 2 ||
		runtime.CheckpointState.VisualSegmentCount != 4 || runtime.CheckpointState.VisualSegmentsText != "segments 1-2" ||
		runtime.CheckpointState.VisualTags["scene"] != "office" || runtime.CheckpointState.AudioCompletedSegments != 3 ||
		runtime.CheckpointState.AudioText != "audio prefix" || runtime.CheckpointState.AudioTags["speaker"] != "alice" {
		t.Fatalf("runtime=%+v, want restored non-final modality checkpoint", runtime)
	}
}

func TestMarshalSemanticCheckpointPayloadChecksCompletePayloadSize(t *testing.T) {
	tags := make(map[string]string, 30)
	for i := 0; i < 30; i++ {
		key := fmt.Sprintf("%02d%s", i, strings.Repeat("界", 62))
		tags[key] = strings.Repeat("值", 255)
	}
	_, err := marshalSemanticCheckpointPayload(semantic.AudioExtractTaskPayload{
		Path: "/" + strings.Repeat("\\", 4096), ContentType: "audio/mpeg",
		CompletedSegments: 1, Text: strings.Repeat("x", 40<<10), Tags: tags,
	})
	if !errors.Is(err, datastore.ErrSemanticTaskPayloadTooLarge) {
		t.Fatalf("marshal error=%v, want ErrSemanticTaskPayloadTooLarge", err)
	}
	visualTags := make(map[string]string, 15)
	audioTags := make(map[string]string, 15)
	for key, value := range tags {
		if len(visualTags) < 15 {
			visualTags[key] = value
		} else {
			audioTags[key] = value
		}
	}
	_, err = marshalSemanticCheckpointPayload(semantic.VideoExtractTaskPayload{
		Path: "/" + strings.Repeat("\\", 4096), ContentType: "video/mp4",
		VisualDone: true, VisualText: strings.Repeat("v", 20<<10), VisualTags: visualTags,
		AudioDone: true, AudioPresent: true, AudioCompletedSegments: 1,
		AudioText: strings.Repeat("a", 20<<10), AudioTags: audioTags,
	})
	if !errors.Is(err, datastore.ErrSemanticTaskPayloadTooLarge) {
		t.Fatalf("video marshal error=%v, want ErrSemanticTaskPayloadTooLarge", err)
	}
	encoded, err := marshalSemanticCheckpointPayload(semantic.AudioExtractTaskPayload{
		Path: "/audio.mp3", CompletedSegments: 1, Text: "short", Tags: map[string]string{"topic": "meeting"},
	})
	if err != nil {
		t.Fatal(err)
	}
	if len(encoded) > datastore.MaxSemanticTaskPayloadBytes {
		t.Fatalf("encoded bytes=%d, want <=%d", len(encoded), datastore.MaxSemanticTaskPayloadBytes)
	}
}

func TestPrepareVideoExtractRuntimeCountsProviderCallOnRetry(t *testing.T) {
	task := &semantic.Task{AttemptCount: 2, MaxAttempts: 5}
	runtime := prepareVideoExtractRuntime(context.Background(), &tenantTarget{backend: &backend.Dat9Backend{}}, task, backend.VideoExtractRuntime{})
	if runtime.BeforeExtract == nil {
		t.Fatal("retry runtime has no provider-call accounting hook")
	}
}

func TestProcessVideoExtractTaskDiscardsFreshUnconfiguredTask(t *testing.T) {
	r := newExtractConfigResolver(
		&fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{}},
		&fakeExtractDecryptor{}, nil, time.Minute,
	)
	t.Cleanup(r.Close)
	m := &tenantWorkerManager{extractResolver: r}
	outcome := m.processVideoExtractTask(context.Background(), &tenantTarget{tenantID: "t1"}, &semantic.Task{
		TaskType: semantic.TaskTypeVideoExtractVisual, AttemptCount: 1, MaxAttempts: 5,
	})
	if outcome.action != semanticTaskActionDiscard || outcome.result != "not_configured" {
		t.Fatalf("outcome=%+v, want discard/not_configured", outcome)
	}
}

func TestProcessVideoExtractTaskDiscardsUnconfiguredRetryWithoutProgress(t *testing.T) {
	r := newExtractConfigResolver(
		&fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{}},
		&fakeExtractDecryptor{}, nil, time.Minute,
	)
	t.Cleanup(r.Close)
	m := &tenantWorkerManager{extractResolver: r}
	outcome := m.processVideoExtractTask(context.Background(), &tenantTarget{tenantID: "t1"}, &semantic.Task{
		TaskType: semantic.TaskTypeVideoExtractVisual, AttemptCount: 2, MaxAttempts: 5,
	})
	if outcome.action != semanticTaskActionDiscard || outcome.result != "not_configured" {
		t.Fatalf("outcome=%+v, want discard/not_configured", outcome)
	}
}

func TestUnconfiguredVideoDiscardDoesNotConsumeLaterQuota(t *testing.T) {
	initServerTenantSchema(t, testDSN)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	testtidb.ResetDB(t, store.DB())
	t.Cleanup(func() { _ = store.Close() })
	s3c, err := s3client.NewLocal(mustTempDir(t), "http://localhost:9091/s3")
	if err != nil {
		t.Fatal(err)
	}
	b, err := backend.NewWithS3ModeAndOptions(store, s3c, true, backend.Options{
		TenantID: "t-video-quota",
		AsyncVideoExtract: backend.AsyncVideoExtractOptions{
			Capability:       true,
			MaxVideoLLMFiles: 1,
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(b.Close)

	if _, err := b.Write("/before-config.mp4", []byte("video-before-config"), 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	reader := &fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{}}
	resolver := newExtractConfigResolver(reader, &fakeExtractDecryptor{plaintext: "key"}, nil, time.Minute)
	t.Cleanup(resolver.Close)
	m := &tenantWorkerManager{extractResolver: resolver}
	target := &tenantTarget{tenantID: "t-video-quota", backend: b, store: store}
	claimed, found, err := store.ClaimSemanticTask(context.Background(), time.Now().UTC().Add(time.Second), time.Minute, semantic.TaskTypeVideoExtractVisual)
	if err != nil || !found {
		t.Fatalf("claim unconfigured video: found=%v err=%v", found, err)
	}
	m.processTask(context.Background(), target, claimed)

	var count int
	if err := store.DB().QueryRow(`SELECT COUNT(*) FROM semantic_tasks WHERE task_type = ?`, semantic.TaskTypeVideoExtractVisual).Scan(&count); err != nil {
		t.Fatal(err)
	}
	if count != 0 {
		t.Fatalf("unconfigured video task rows = %d, want 0 after discard", count)
	}

	reader.mu.Lock()
	reader.rows["t-video-quota\x00"+meta.ExtractMediaTypeVideo] = enabledMediaRow("t-video-quota", meta.ExtractMediaTypeVideo)
	reader.mu.Unlock()
	resolver.Evict("t-video-quota")
	if _, err := b.Write("/after-config.mp4", []byte("video-after-config"), 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	if err := store.DB().QueryRow(`SELECT COUNT(*) FROM semantic_tasks WHERE task_type = ?`, semantic.TaskTypeVideoExtractVisual).Scan(&count); err != nil {
		t.Fatal(err)
	}
	if count != 1 {
		t.Fatalf("video task rows after enabling config = %d, want 1", count)
	}
}

func TestVideoFallbackAudioRouting(t *testing.T) {
	taskPayload, err := json.Marshal(semantic.AudioExtractTaskPayload{
		Path: "/clip.mp4", ContentType: "video/mp4", VideoFallback: true,
	})
	if err != nil {
		t.Fatal(err)
	}
	task := &semantic.Task{TaskType: semantic.TaskTypeAudioExtractText, PayloadJSON: taskPayload}

	t.Run("video runtime owns audio", func(t *testing.T) {
		r := newExtractConfigResolverWithFallback(
			&fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{}},
			&fakeExtractDecryptor{}, nil, 0, time.Minute,
			ExtractResolverDefaults{
				Audio: &backend.AudioExtractRuntime{Extractor: stubAudioTextExtractor{}},
				Video: &backend.VideoExtractRuntime{Extractor: stubVideoTextExtractor{}},
			},
		)
		defer r.Close()
		m := &tenantWorkerManager{extractResolver: r}
		owned, err := m.videoOwnsFallbackAudio(context.Background(), &tenantTarget{tenantID: "t1"}, task)
		if err != nil || !owned {
			t.Fatalf("owned=%t err=%v, want video-owned fallback", owned, err)
		}
	})

	t.Run("visual-only video runtime leaves audio independent", func(t *testing.T) {
		r := newExtractConfigResolverWithFallback(
			&fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{}},
			&fakeExtractDecryptor{}, nil, 0, time.Minute,
			ExtractResolverDefaults{Video: &backend.VideoExtractRuntime{Extractor: stubVideoTextExtractor{}}},
		)
		defer r.Close()
		m := &tenantWorkerManager{extractResolver: r}
		owned, err := m.videoOwnsFallbackAudio(context.Background(), &tenantTarget{tenantID: "t1"}, task)
		if err != nil || owned {
			t.Fatalf("owned=%t err=%v, want visual-only video runtime to leave fallback audio independent", owned, err)
		}
	})

	t.Run("disabled video leaves audio independent", func(t *testing.T) {
		r := newExtractConfigResolver(
			&fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{
				"t1\x00" + meta.ExtractMediaTypeVideo: {TenantID: "t1", MediaType: meta.ExtractMediaTypeVideo, Enabled: false},
			}},
			&fakeExtractDecryptor{}, nil, time.Minute,
		)
		defer r.Close()
		m := &tenantWorkerManager{extractResolver: r}
		owned, err := m.videoOwnsFallbackAudio(context.Background(), &tenantTarget{tenantID: "t1"}, task)
		if err != nil || owned {
			t.Fatalf("owned=%t err=%v, want audio fallback to proceed", owned, err)
		}
	})

	t.Run("process default video runtime owns audio", func(t *testing.T) {
		initServerTenantSchema(t, testDSN)
		store, err := datastore.Open(testDSN)
		if err != nil {
			t.Fatal(err)
		}
		testtidb.ResetDB(t, store.DB())
		t.Cleanup(func() { _ = store.Close() })
		s3c, err := s3client.NewLocal(mustTempDir(t), "http://localhost:9091/s3")
		if err != nil {
			t.Fatal(err)
		}
		b, err := backend.NewWithS3ModeAndOptions(store, s3c, true, backend.Options{
			AsyncAudioExtract: backend.AsyncAudioExtractOptions{
				Enabled:   true,
				Extractor: stubAudioTextExtractor{},
			},
			AsyncVideoExtract: backend.AsyncVideoExtractOptions{
				Enabled:   true,
				Extractor: stubVideoTextExtractor{},
			},
		})
		if err != nil {
			t.Fatal(err)
		}
		t.Cleanup(b.Close)

		r := newExtractConfigResolver(
			&fakeTenantExtractConfigReader{rows: map[string]meta.TenantExtractConfig{}},
			&fakeExtractDecryptor{}, nil, time.Minute,
		)
		t.Cleanup(r.Close)
		m := &tenantWorkerManager{extractResolver: r}
		owned, err := m.videoOwnsFallbackAudio(context.Background(), &tenantTarget{tenantID: "t1", backend: b}, task)
		if err != nil || !owned {
			t.Fatalf("owned=%t err=%v, want process-default video runtime to own fallback audio", owned, err)
		}
	})
}

func hasTaskType(types []semantic.TaskType, want semantic.TaskType) bool {
	for _, tt := range types {
		if tt == want {
			return true
		}
	}
	return false
}

func countTaskType(types []semantic.TaskType, want semantic.TaskType) int {
	var count int
	for _, tt := range types {
		if tt == want {
			count++
		}
	}
	return count
}