package backend

import (
	"bytes"
	"context"
	"encoding/json"
	"io"
	"testing"
	"time"

	"github.com/c4pt0r/agfs/agfs-server/pkg/filesystem"
	"github.com/mem9-ai/drive9/pkg/embedding"
	"github.com/mem9-ai/drive9/pkg/s3client"
	"github.com/mem9-ai/drive9/pkg/semantic"
	"github.com/mem9-ai/drive9/pkg/tenant/schema"
)

type backendSemanticTaskRow struct {
	TaskType        string
	Status          string
	ResourceVersion int64
}

func loadSemanticTasksForFile(t *testing.T, b *Dat9Backend, fileID string) []backendSemanticTaskRow {
	t.Helper()
	rows, err := b.Store().DB().Query(`SELECT task_type, status, resource_version
		FROM semantic_tasks WHERE resource_id = ?
		ORDER BY resource_version, created_at, task_id`, fileID)
	if err != nil {
		t.Fatalf("query semantic tasks for %s: %v", fileID, err)
	}
	defer func() { _ = rows.Close() }()

	var tasks []backendSemanticTaskRow
	for rows.Next() {
		var task backendSemanticTaskRow
		if err := rows.Scan(&task.TaskType, &task.Status, &task.ResourceVersion); err != nil {
			t.Fatalf("scan semantic task for %s: %v", fileID, err)
		}
		tasks = append(tasks, task)
	}
	if err := rows.Err(); err != nil {
		t.Fatalf("iterate semantic tasks for %s: %v", fileID, err)
	}
	return tasks
}

func setStoredEmbeddingState(t *testing.T, b *Dat9Backend, fileID string, revision int64) {
	t.Helper()
	vector := embedding.FormatVector(make([]float32, schema.TiDBAutoEmbeddingDimensions))
	if _, err := b.Store().DB().Exec(`UPDATE files SET embedding = ?, embedding_revision = ? WHERE file_id = ?`, vector, revision, fileID); err != nil {
		t.Fatalf("set embedding state for %s: %v", fileID, err)
	}
	if _, err := b.Store().DB().Exec(`UPDATE semantic SET embedding = ?, embedding_revision = ? WHERE inode_id = ?`, vector, revision, fileID); err != nil {
		t.Fatalf("set semantic embedding state for %s: %v", fileID, err)
	}
}

func uploadAllPartsForPlan(t *testing.T, b *Dat9Backend, plan *UploadPlan, uploadID string, totalSize int64) {
	t.Helper()
	ctx := context.Background()
	upload, err := b.GetUpload(ctx, uploadID)
	if err != nil {
		t.Fatalf("get upload %s: %v", uploadID, err)
	}
	partData := make([]byte, totalSize)
	for i := range partData {
		partData[i] = byte(i % 251)
	}
	for _, part := range plan.Parts {
		start := int64(part.Number-1) * upload.PartSize
		end := start + part.Size
		if end > totalSize {
			end = totalSize
		}
		if _, err := b.S3().(*s3client.LocalS3Client).UploadPart(ctx, upload.S3UploadID, part.Number, bytes.NewReader(partData[start:end])); err != nil {
			t.Fatalf("upload part %d: %v", part.Number, err)
		}
	}
}

// completePartsForConfirmV2 builds the client part list for ConfirmUploadV2 after
// parts have been uploaded to S3 (same bytes as uploadAllPartsForPlan).
func completePartsForConfirmV2(t *testing.T, b *Dat9Backend, uploadID string) []CompletePart {
	t.Helper()
	ctx := context.Background()
	u, err := b.GetUpload(ctx, uploadID)
	if err != nil {
		t.Fatalf("get upload: %v", err)
	}
	parts, err := b.S3().ListParts(ctx, u.S3Key, u.S3UploadID)
	if err != nil {
		t.Fatalf("list parts: %v", err)
	}
	out := make([]CompletePart, len(parts))
	for i, p := range parts {
		out[i] = CompletePart{Number: p.Number, ETag: p.ETag}
	}
	return out
}

func mustFileForPath(t *testing.T, b *Dat9Backend, path string) (string, int64, *int64, string) {
	t.Helper()
	nf, err := b.Store().Stat(context.Background(), path)
	if err != nil {
		t.Fatalf("stat %s: %v", path, err)
	}
	if nf.File == nil {
		t.Fatalf("path %s has no file entity", path)
	}
	return nf.File.FileID, nf.File.Revision, nf.File.EmbeddingRevision, nf.File.ContentType
}

func TestWriteCreateEnqueuesEmbedTask(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{AppSemanticTasksEnabled: true})
	if _, err := b.Write("/data/file.txt", []byte("hello world"), 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}

	fileID, revision, embeddingRevision, _ := mustFileForPath(t, b, "/data/file.txt")
	if revision != 1 {
		t.Fatalf("revision=%d, want 1", revision)
	}
	if embeddingRevision != nil {
		t.Fatalf("embedding revision should be nil before worker, got %v", *embeddingRevision)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	if len(tasks) != 1 {
		t.Fatalf("semantic task count=%d, want 1", len(tasks))
	}
	if tasks[0].TaskType != string(semantic.TaskTypeEmbed) || tasks[0].Status != string(semantic.TaskQueued) || tasks[0].ResourceVersion != 1 {
		t.Fatalf("unexpected semantic task: %+v", tasks[0])
	}
}

func TestWriteCreateSkipsEmbedTaskWithoutTextSource(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{AppSemanticTasksEnabled: true})
	if _, err := b.Write("/data/blob.bin", []byte{0, 1, 2, 3}, 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}

	fileID, revision, embeddingRevision, _ := mustFileForPath(t, b, "/data/blob.bin")
	if revision != 1 {
		t.Fatalf("revision=%d, want 1", revision)
	}
	if embeddingRevision != nil {
		t.Fatalf("embedding revision should be nil before worker, got %v", *embeddingRevision)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	if len(tasks) != 0 {
		t.Fatalf("semantic task count=%d, want 0", len(tasks))
	}
}

func TestWriteCreateAutoEmbeddingSkipsEmbedTaskEvenWithTextSource(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{DatabaseAutoEmbedding: true})
	if _, err := b.Write("/data/file.txt", []byte("hello world"), 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}

	fileID, revision, embeddingRevision, _ := mustFileForPath(t, b, "/data/file.txt")
	if revision != 1 {
		t.Fatalf("revision=%d, want 1", revision)
	}
	if embeddingRevision != nil {
		t.Fatalf("embedding revision should remain nil on create, got %v", *embeddingRevision)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	if len(tasks) != 0 {
		t.Fatalf("semantic task count=%d, want 0", len(tasks))
	}
}

func TestWriteCreateAutoEmbeddingImageEnqueuesImgExtractTaskWithoutLegacyQueue(t *testing.T) {
	extractor := &gatedImageExtractor{started: make(chan struct{}, 1), release: make(chan struct{})}
	b := newTestBackendWithOptions(t, Options{
		DatabaseAutoEmbedding: true,
		AsyncImageExtract: AsyncImageExtractOptions{
			Enabled:   true,
			Workers:   1,
			QueueSize: 8,
			Extractor: extractor,
		},
	})
	if _, err := b.Write("/img/create-auto.png", fakePNG, 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}

	select {
	case <-extractor.started:
		close(extractor.release)
		t.Fatal("legacy image queue should stay idle for auto create path")
	case <-time.After(200 * time.Millisecond):
	}

	fileID, revision, embeddingRevision, _ := mustFileForPath(t, b, "/img/create-auto.png")
	if revision != 1 {
		t.Fatalf("revision=%d, want 1", revision)
	}
	if embeddingRevision != nil {
		t.Fatalf("embedding revision should remain nil before durable worker, got %v", *embeddingRevision)
	}
	nf, err := b.Store().Stat(context.Background(), "/img/create-auto.png")
	if err != nil || nf.File == nil {
		t.Fatalf("stat /img/create-auto.png: %v", err)
	}
	if nf.File.ContentText != "" {
		t.Fatalf("content_text=%q, want empty before durable worker", nf.File.ContentText)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	if len(tasks) != 1 {
		t.Fatalf("semantic task count=%d, want 1", len(tasks))
	}
	if tasks[0].TaskType != string(semantic.TaskTypeImgExtractText) || tasks[0].Status != string(semantic.TaskQueued) || tasks[0].ResourceVersion != 1 {
		t.Fatalf("unexpected semantic task: %+v", tasks[0])
	}
}

func TestWriteCreateAutoEmbeddingAudioEnqueuesAudioExtractTask(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		DatabaseAutoEmbedding: true,
		AsyncAudioExtract: AsyncAudioExtractOptions{
			Enabled:   true,
			Extractor: &staticAudioExtractor{text: "x"},
		},
	})
	if _, err := b.Write("/tracks/create.mp3", []byte{0xff, 0xf3}, 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	fileID, revision, _, _ := mustFileForPath(t, b, "/tracks/create.mp3")
	if revision != 1 {
		t.Fatalf("revision=%d, want 1", revision)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	var audioSeen bool
	for _, tsk := range tasks {
		if tsk.TaskType == string(semantic.TaskTypeAudioExtractText) {
			audioSeen = true
			if tsk.Status != string(semantic.TaskQueued) || tsk.ResourceVersion != 1 {
				t.Fatalf("unexpected audio task: %+v", tsk)
			}
		}
	}
	if !audioSeen {
		t.Fatalf("expected audio_extract_text among %+v", tasks)
	}
	nf, err := b.Store().Stat(context.Background(), "/tracks/create.mp3")
	if err != nil || nf.File == nil {
		t.Fatal(err)
	}
	if nf.File.ContentText != "" {
		t.Fatalf("content_text should be empty before worker, got %q", nf.File.ContentText)
	}
}

func TestWriteVideoEnqueuesVisualAndFallbackAudioTasks(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		DatabaseAutoEmbedding: true,
		AsyncAudioExtract: AsyncAudioExtractOptions{
			Enabled: true, Extractor: &staticAudioExtractor{text: "audio"},
		},
		AsyncVideoExtract: AsyncVideoExtractOptions{
			Enabled: true, AllTenants: true, Extractor: &staticVideoExtractor{text: "visual"},
		},
	})
	if _, err := b.Write("/media/both.mp4", []byte{0x00, 0x00, 0x00, 0x18}, 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	fileID, _, _, _ := mustFileForPath(t, b, "/media/both.mp4")
	var audioPayload []byte
	rows, err := b.Store().DB().Query(`SELECT task_type, payload_json FROM semantic_tasks WHERE resource_id = ?`, fileID)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = rows.Close() }()
	seen := map[string]bool{}
	for rows.Next() {
		var taskType string
		var payload []byte
		if err := rows.Scan(&taskType, &payload); err != nil {
			t.Fatal(err)
		}
		seen[taskType] = true
		if taskType == string(semantic.TaskTypeAudioExtractText) {
			audioPayload = payload
		}
	}
	if err := rows.Err(); err != nil {
		t.Fatal(err)
	}
	if !seen[string(semantic.TaskTypeVideoExtractVisual)] || !seen[string(semantic.TaskTypeAudioExtractText)] {
		t.Fatalf("task types=%v, want video and fallback audio", seen)
	}
	var payload semantic.AudioExtractTaskPayload
	if err := json.Unmarshal(audioPayload, &payload); err != nil {
		t.Fatal(err)
	}
	if !payload.VideoFallback {
		t.Fatalf("audio payload=%+v, want video fallback marker", payload)
	}
}

func TestWriteVideoQuotaSkipLeavesAudioTaskIndependent(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		DatabaseAutoEmbedding: true,
		AsyncAudioExtract: AsyncAudioExtractOptions{
			Enabled: true, Extractor: &staticAudioExtractor{text: "audio"},
		},
		AsyncVideoExtract: AsyncVideoExtractOptions{
			Enabled: true, AllTenants: true, MaxVideoLLMFiles: 1,
			Extractor: &staticVideoExtractor{text: "visual"},
		},
	})
	if _, err := b.Write("/media/first.mp4", []byte{0x00, 0x00, 0x00, 0x18}, 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	if _, err := b.Write("/media/audio-only.mp4", []byte{0x00, 0x00, 0x00, 0x18}, 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	fileID, _, _, _ := mustFileForPath(t, b, "/media/audio-only.mp4")

	var taskType string
	var payloadJSON []byte
	if err := b.Store().DB().QueryRow(`SELECT task_type, payload_json FROM semantic_tasks WHERE resource_id = ?`, fileID).Scan(&taskType, &payloadJSON); err != nil {
		t.Fatal(err)
	}
	if taskType != string(semantic.TaskTypeAudioExtractText) {
		t.Fatalf("task type=%q, want audio task after video quota skip", taskType)
	}
	var payload semantic.AudioExtractTaskPayload
	if err := json.Unmarshal(payloadJSON, &payload); err != nil {
		t.Fatal(err)
	}
	if payload.VideoFallback {
		t.Fatalf("audio payload=%+v, must remain independent when no video task was admitted", payload)
	}
}

func TestWriteCreateAutoEmbeddingMP4EnqueuesAudioExtractTask(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		DatabaseAutoEmbedding: true,
		AsyncAudioExtract: AsyncAudioExtractOptions{
			Enabled:   true,
			Extractor: &staticAudioExtractor{text: "x"},
		},
	})
	if _, err := b.Write("/tracks/create.mp4", []byte{0x00, 0x00, 0x00, 0x18}, 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	fileID, revision, _, contentType := mustFileForPath(t, b, "/tracks/create.mp4")
	if revision != 1 {
		t.Fatalf("revision=%d, want 1", revision)
	}
	wantCT := detectContentType("/tracks/create.mp4", nil)
	if contentType != wantCT {
		t.Fatalf("content_type=%q, want %q", contentType, wantCT)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	var audioSeen bool
	for _, tsk := range tasks {
		if tsk.TaskType == string(semantic.TaskTypeAudioExtractText) {
			audioSeen = true
			if tsk.Status != string(semantic.TaskQueued) || tsk.ResourceVersion != 1 {
				t.Fatalf("unexpected audio task: %+v", tsk)
			}
		}
	}
	if !audioSeen {
		t.Fatalf("expected audio_extract_text among %+v", tasks)
	}
}

func TestWriteCreateAutoEmbeddingM4AEnqueuesAudioExtractTask(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		DatabaseAutoEmbedding: true,
		AsyncAudioExtract: AsyncAudioExtractOptions{
			Enabled:   true,
			Extractor: &staticAudioExtractor{text: "x"},
		},
	})
	if _, err := b.Write("/tracks/create.m4a", []byte{0x00, 0x00, 0x00}, 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	fileID, revision, _, contentType := mustFileForPath(t, b, "/tracks/create.m4a")
	if revision != 1 {
		t.Fatalf("revision=%d, want 1", revision)
	}
	wantCT := detectContentType("/tracks/create.m4a", nil)
	if contentType != wantCT {
		t.Fatalf("content_type=%q, want %q", contentType, wantCT)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	var audioSeen bool
	for _, tsk := range tasks {
		if tsk.TaskType == string(semantic.TaskTypeAudioExtractText) {
			audioSeen = true
			if tsk.Status != string(semantic.TaskQueued) || tsk.ResourceVersion != 1 {
				t.Fatalf("unexpected audio task: %+v", tsk)
			}
		}
	}
	if !audioSeen {
		t.Fatalf("expected audio_extract_text among %+v", tasks)
	}
}

func TestWriteOverwriteAutoEmbeddingAudioEnqueuesAudioExtractTask(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		DatabaseAutoEmbedding: true,
		AsyncAudioExtract: AsyncAudioExtractOptions{
			Enabled:   true,
			Extractor: &staticAudioExtractor{text: "x"},
		},
	})
	fileID := insertImageFileForExtractTest(t, b, "/tracks/ow.mp3", "audio/mpeg", []byte{1, 2, 3})
	setStoredEmbeddingState(t, b, fileID, 1)

	if _, err := b.Write("/tracks/ow.mp3", []byte{4, 5, 6}, 0, filesystem.WriteFlagTruncate); err != nil {
		t.Fatal(err)
	}
	_, revision, _, _ := mustFileForPath(t, b, "/tracks/ow.mp3")
	if revision != 2 {
		t.Fatalf("revision=%d, want 2", revision)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	var audioSeen bool
	for _, tsk := range tasks {
		if tsk.TaskType == string(semantic.TaskTypeAudioExtractText) && tsk.ResourceVersion == 2 {
			audioSeen = true
			if tsk.Status != string(semantic.TaskQueued) {
				t.Fatalf("unexpected audio task: %+v", tsk)
			}
		}
	}
	if !audioSeen {
		t.Fatalf("expected audio_extract_text for revision 2 among %+v", tasks)
	}
}

func TestWriteCreateAutoEmbeddingSkipsAudioWhenExtractorUnset(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		DatabaseAutoEmbedding: true,
		AsyncAudioExtract: AsyncAudioExtractOptions{
			Enabled: true,
			// Extractor nil: Phase 2 does not wire a default extractor.
			Extractor: nil,
		},
	})
	if _, err := b.Write("/tracks/no-extractor.mp3", []byte{0xff, 0xf3}, 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	fileID, _, _, _ := mustFileForPath(t, b, "/tracks/no-extractor.mp3")
	tasks := loadSemanticTasksForFile(t, b, fileID)
	for _, tsk := range tasks {
		if tsk.TaskType == string(semantic.TaskTypeAudioExtractText) {
			t.Fatalf("unexpected audio task: %+v", tsk)
		}
	}
}

func TestWriteCreateAutoEmbeddingSkipsAudioForNonAudioFile(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		DatabaseAutoEmbedding: true,
		AsyncAudioExtract: AsyncAudioExtractOptions{
			Enabled:   true,
			Extractor: &staticAudioExtractor{text: "x"},
		},
	})
	if _, err := b.Write("/notes/plain.txt", []byte("hello"), 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	fileID, _, _, _ := mustFileForPath(t, b, "/notes/plain.txt")
	tasks := loadSemanticTasksForFile(t, b, fileID)
	for _, tsk := range tasks {
		if tsk.TaskType == string(semantic.TaskTypeAudioExtractText) {
			t.Fatalf("unexpected audio task: %+v", tsk)
		}
	}
}

func TestWriteOverwriteEnqueuesNextRevisionAndClearsEmbeddingState(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{AppSemanticTasksEnabled: true})
	if _, err := b.Write("/f.txt", []byte("old"), 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	fileID, _, _, _ := mustFileForPath(t, b, "/f.txt")
	setStoredEmbeddingState(t, b, fileID, 1)

	if _, err := b.Write("/f.txt", []byte("new"), 0, filesystem.WriteFlagTruncate); err != nil {
		t.Fatal(err)
	}

	_, revision, embeddingRevision, _ := mustFileForPath(t, b, "/f.txt")
	if revision != 2 {
		t.Fatalf("revision=%d, want 2", revision)
	}
	if embeddingRevision != nil {
		t.Fatalf("embedding revision should be cleared, got %v", *embeddingRevision)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	if len(tasks) != 2 {
		t.Fatalf("semantic task count=%d, want 2", len(tasks))
	}
	if tasks[1].ResourceVersion != 2 || tasks[1].TaskType != string(semantic.TaskTypeEmbed) {
		t.Fatalf("unexpected overwrite semantic task: %+v", tasks[1])
	}
}

func TestWriteOverwriteSkipsEmbedTaskWithoutTextSource(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{AppSemanticTasksEnabled: true})
	if _, err := b.Write("/f", []byte("old"), 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	fileID, _, _, _ := mustFileForPath(t, b, "/f")
	setStoredEmbeddingState(t, b, fileID, 1)

	if _, err := b.Write("/f", []byte{0, 1, 2, 3}, 0, filesystem.WriteFlagTruncate); err != nil {
		t.Fatal(err)
	}

	_, revision, embeddingRevision, _ := mustFileForPath(t, b, "/f")
	if revision != 2 {
		t.Fatalf("revision=%d, want 2", revision)
	}
	if embeddingRevision != nil {
		t.Fatalf("embedding revision should be cleared, got %v", *embeddingRevision)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	if len(tasks) != 1 {
		t.Fatalf("semantic task count=%d, want 1", len(tasks))
	}
	if tasks[0].ResourceVersion != 1 {
		t.Fatalf("unexpected semantic task history: %+v", tasks)
	}
}

func TestWriteOverwriteAutoEmbeddingSkipsEmbedTaskAndPreservesEmbeddingState(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{DatabaseAutoEmbedding: true})
	if _, err := b.Write("/f.txt", []byte("old"), 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	fileID, _, _, _ := mustFileForPath(t, b, "/f.txt")
	setStoredEmbeddingState(t, b, fileID, 1)

	if _, err := b.Write("/f.txt", []byte("new"), 0, filesystem.WriteFlagTruncate); err != nil {
		t.Fatal(err)
	}

	_, revision, embeddingRevision, _ := mustFileForPath(t, b, "/f.txt")
	if revision != 2 {
		t.Fatalf("revision=%d, want 2", revision)
	}
	if embeddingRevision == nil || *embeddingRevision != 1 {
		t.Fatalf("embedding revision should be preserved, got %v", embeddingRevision)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	if len(tasks) != 0 {
		t.Fatalf("semantic task count=%d, want 0", len(tasks))
	}
}

func TestWriteOverwriteAutoEmbeddingImageEnqueuesImgExtractTaskWithoutLegacyQueue(t *testing.T) {
	extractor := &gatedImageExtractor{started: make(chan struct{}, 1), release: make(chan struct{})}
	b := newTestBackendWithOptions(t, Options{
		DatabaseAutoEmbedding: true,
		AsyncImageExtract: AsyncImageExtractOptions{
			Enabled:   true,
			Workers:   1,
			QueueSize: 8,
			Extractor: extractor,
		},
	})
	fileID := insertImageFileForExtractTest(t, b, "/img/overwrite-auto.png", "image/png", []byte("old-image"))
	setStoredEmbeddingState(t, b, fileID, 1)

	if _, err := b.Write("/img/overwrite-auto.png", []byte("new-image"), 0, filesystem.WriteFlagTruncate); err != nil {
		t.Fatal(err)
	}

	select {
	case <-extractor.started:
		close(extractor.release)
		t.Fatal("legacy image queue should stay idle for auto overwrite path")
	case <-time.After(200 * time.Millisecond):
	}

	_, revision, embeddingRevision, _ := mustFileForPath(t, b, "/img/overwrite-auto.png")
	if revision != 2 {
		t.Fatalf("revision=%d, want 2", revision)
	}
	if embeddingRevision == nil || *embeddingRevision != 1 {
		t.Fatalf("embedding revision should be preserved, got %v", embeddingRevision)
	}
	nf, err := b.Store().Stat(context.Background(), "/img/overwrite-auto.png")
	if err != nil || nf.File == nil {
		t.Fatalf("stat /img/overwrite-auto.png: %v", err)
	}
	if nf.File.ContentText != "" {
		t.Fatalf("content_text=%q, want empty before durable worker", nf.File.ContentText)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	if len(tasks) != 1 {
		t.Fatalf("semantic task count=%d, want 1", len(tasks))
	}
	if tasks[0].TaskType != string(semantic.TaskTypeImgExtractText) || tasks[0].Status != string(semantic.TaskQueued) || tasks[0].ResourceVersion != 2 {
		t.Fatalf("unexpected semantic task: %+v", tasks[0])
	}
}

func TestWriteOverwriteDoesNotDeleteInlineMarkerObject(t *testing.T) {
	b := newTestBackend(t)
	if _, err := b.Write("/f", []byte("old"), 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	if err := b.S3().PutObject(context.Background(), "inline", bytes.NewReader([]byte("marker")), int64(len("marker")), s3client.EncryptionOpts{}); err != nil {
		t.Fatal(err)
	}
	large := bytes.Repeat([]byte("a"), 2<<20)
	if _, err := b.Write("/f", large, 0, filesystem.WriteFlagTruncate); err != nil {
		t.Fatal(err)
	}
	rc, err := b.S3().GetObject(context.Background(), "inline")
	if err != nil {
		t.Fatalf("inline marker object should survive overwrite cleanup: %v", err)
	}
	defer func() { _ = rc.Close() }()
	data, err := io.ReadAll(rc)
	if err != nil {
		t.Fatal(err)
	}
	if string(data) != "marker" {
		t.Fatalf("marker object=%q, want %q", data, "marker")
	}
}

func TestConfirmUploadSkipsEmbedTaskWithoutTextSource(t *testing.T) {
	b := newTestBackendWithS3AndOptions(t, Options{AppSemanticTasksEnabled: true})
	ctx := context.Background()
	totalSize := int64(2 << 20)
	plan, err := b.InitiateUpload(ctx, "/bigfile.txt", totalSize)
	if err != nil {
		t.Fatal(err)
	}
	uploadAllPartsForPlan(t, b, plan, plan.UploadID, totalSize)

	if err := b.ConfirmUpload(ctx, plan.UploadID); err != nil {
		t.Fatal(err)
	}

	fileID, revision, embeddingRevision, contentType := mustFileForPath(t, b, "/bigfile.txt")
	if revision != 1 {
		t.Fatalf("revision=%d, want 1", revision)
	}
	if embeddingRevision != nil {
		t.Fatalf("embedding revision should be nil before worker, got %v", *embeddingRevision)
	}
	if contentType != detectContentType("/bigfile.txt", nil) {
		t.Fatalf("content_type=%q, want %q", contentType, detectContentType("/bigfile.txt", nil))
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	if len(tasks) != 0 {
		t.Fatalf("semantic task count=%d, want 0", len(tasks))
	}
}

func TestConfirmUploadOverwriteSkipsEmbedTaskWithoutTextSourceAndRebindsUpload(t *testing.T) {
	b := newTestBackendWithS3AndOptions(t, Options{AppSemanticTasksEnabled: true})
	ctx := context.Background()
	if _, err := b.Write("/report.txt", []byte("old body"), 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	fileID, _, _, _ := mustFileForPath(t, b, "/report.txt")
	setStoredEmbeddingState(t, b, fileID, 1)

	totalSize := int64(2 << 20)
	plan, err := b.InitiateUpload(ctx, "/report.txt", totalSize)
	if err != nil {
		t.Fatal(err)
	}
	pendingUpload, err := b.GetUpload(ctx, plan.UploadID)
	if err != nil {
		t.Fatal(err)
	}
	pendingFileID := pendingUpload.FileID
	uploadAllPartsForPlan(t, b, plan, plan.UploadID, totalSize)

	if err := b.ConfirmUpload(ctx, plan.UploadID); err != nil {
		t.Fatal(err)
	}

	confirmedFileID, revision, embeddingRevision, contentType := mustFileForPath(t, b, "/report.txt")
	if confirmedFileID != fileID {
		t.Fatalf("overwrite should preserve inode file_id=%q, got %q", fileID, confirmedFileID)
	}
	if revision != 2 {
		t.Fatalf("revision=%d, want 2", revision)
	}
	if embeddingRevision != nil {
		t.Fatalf("embedding revision should be cleared, got %v", *embeddingRevision)
	}
	if contentType != detectContentType("/report.txt", nil) {
		t.Fatalf("content_type=%q, want %q", contentType, detectContentType("/report.txt", nil))
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	if len(tasks) != 1 {
		t.Fatalf("semantic task count=%d, want 1", len(tasks))
	}
	if tasks[0].ResourceVersion != 1 || tasks[0].TaskType != string(semantic.TaskTypeEmbed) {
		t.Fatalf("unexpected overwrite upload semantic tasks: %+v", tasks)
	}
	upload, err := b.GetUpload(ctx, plan.UploadID)
	if err != nil {
		t.Fatal(err)
	}
	if upload.FileID != fileID {
		t.Fatalf("upload file_id=%q, want surviving inode %q", upload.FileID, fileID)
	}
	var deletedStatus string
	if err := b.Store().DB().QueryRow(`SELECT status FROM files WHERE file_id = ?`, pendingFileID).Scan(&deletedStatus); err != nil {
		t.Fatal(err)
	}
	if deletedStatus != "DELETED" {
		t.Fatalf("pending upload file status=%q, want DELETED", deletedStatus)
	}
}

func TestConfirmUploadOverwriteAutoEmbeddingSkipsEmbedTaskAndPreservesEmbeddingState(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{DatabaseAutoEmbedding: true})
	ctx := context.Background()
	if _, err := b.Write("/report.txt", []byte("old body"), 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	fileID, _, _, _ := mustFileForPath(t, b, "/report.txt")
	setStoredEmbeddingState(t, b, fileID, 1)

	totalSize := int64(2 << 20)
	plan, err := b.InitiateUpload(ctx, "/report.txt", totalSize)
	if err != nil {
		t.Fatal(err)
	}
	pendingUpload, err := b.GetUpload(ctx, plan.UploadID)
	if err != nil {
		t.Fatal(err)
	}
	pendingFileID := pendingUpload.FileID
	uploadAllPartsForPlan(t, b, plan, plan.UploadID, totalSize)

	if err := b.ConfirmUpload(ctx, plan.UploadID); err != nil {
		t.Fatal(err)
	}

	confirmedFileID, revision, embeddingRevision, contentType := mustFileForPath(t, b, "/report.txt")
	if confirmedFileID != fileID {
		t.Fatalf("overwrite should preserve inode file_id=%q, got %q", fileID, confirmedFileID)
	}
	if revision != 2 {
		t.Fatalf("revision=%d, want 2", revision)
	}
	if embeddingRevision == nil || *embeddingRevision != 1 {
		t.Fatalf("embedding revision should be preserved, got %v", embeddingRevision)
	}
	if contentType != detectContentType("/report.txt", nil) {
		t.Fatalf("content_type=%q, want %q", contentType, detectContentType("/report.txt", nil))
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	if len(tasks) != 0 {
		t.Fatalf("semantic task count=%d, want 0", len(tasks))
	}
	upload, err := b.GetUpload(ctx, plan.UploadID)
	if err != nil {
		t.Fatal(err)
	}
	if upload.FileID != fileID {
		t.Fatalf("upload file_id=%q, want surviving inode %q", upload.FileID, fileID)
	}
	var deletedStatus string
	if err := b.Store().DB().QueryRow(`SELECT status FROM files WHERE file_id = ?`, pendingFileID).Scan(&deletedStatus); err != nil {
		t.Fatal(err)
	}
	if deletedStatus != "DELETED" {
		t.Fatalf("pending upload file status=%q, want DELETED", deletedStatus)
	}
}

func TestConfirmUploadAutoEmbeddingImageEnqueuesImgExtractTaskWithoutLegacyQueue(t *testing.T) {
	extractor := &gatedImageExtractor{started: make(chan struct{}, 1), release: make(chan struct{})}
	b := newTestBackendWithOptions(t, Options{
		DatabaseAutoEmbedding: true,
		AsyncImageExtract: AsyncImageExtractOptions{
			Enabled:   true,
			Workers:   1,
			QueueSize: 8,
			Extractor: extractor,
		},
	})
	ctx := context.Background()
	totalSize := int64(2 << 20)
	plan, err := b.InitiateUpload(ctx, "/img/upload-auto.png", totalSize)
	if err != nil {
		t.Fatal(err)
	}
	uploadAllPartsForPlan(t, b, plan, plan.UploadID, totalSize)

	if err := b.ConfirmUpload(ctx, plan.UploadID); err != nil {
		t.Fatal(err)
	}

	select {
	case <-extractor.started:
		close(extractor.release)
		t.Fatal("legacy image queue should stay idle for auto upload create path")
	case <-time.After(200 * time.Millisecond):
	}

	fileID, revision, embeddingRevision, contentType := mustFileForPath(t, b, "/img/upload-auto.png")
	if revision != 1 {
		t.Fatalf("revision=%d, want 1", revision)
	}
	if embeddingRevision != nil {
		t.Fatalf("embedding revision should be nil before durable worker, got %v", *embeddingRevision)
	}
	if contentType != detectContentType("/img/upload-auto.png", nil) {
		t.Fatalf("content_type=%q, want %q", contentType, detectContentType("/img/upload-auto.png", nil))
	}
	nf, err := b.Store().Stat(ctx, "/img/upload-auto.png")
	if err != nil || nf.File == nil {
		t.Fatalf("stat /img/upload-auto.png: %v", err)
	}
	if nf.File.ContentText != "" {
		t.Fatalf("content_text=%q, want empty before durable worker", nf.File.ContentText)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	if len(tasks) != 1 {
		t.Fatalf("semantic task count=%d, want 1", len(tasks))
	}
	if tasks[0].TaskType != string(semantic.TaskTypeImgExtractText) || tasks[0].Status != string(semantic.TaskQueued) || tasks[0].ResourceVersion != 1 {
		t.Fatalf("unexpected semantic task: %+v", tasks[0])
	}
}

func TestConfirmUploadAutoEmbeddingEnqueuesAudioExtractTaskForMP3(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		DatabaseAutoEmbedding: true,
		AsyncAudioExtract: AsyncAudioExtractOptions{
			Enabled:   true,
			Extractor: &staticAudioExtractor{text: "x"},
		},
	})
	ctx := context.Background()
	totalSize := int64(2 << 20)
	plan, err := b.InitiateUpload(ctx, "/upload/clip.mp3", totalSize)
	if err != nil {
		t.Fatal(err)
	}
	uploadAllPartsForPlan(t, b, plan, plan.UploadID, totalSize)
	if err := b.ConfirmUpload(ctx, plan.UploadID); err != nil {
		t.Fatal(err)
	}
	fileID, revision, _, _ := mustFileForPath(t, b, "/upload/clip.mp3")
	if revision != 1 {
		t.Fatalf("revision=%d, want 1", revision)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	var audioSeen bool
	for _, tsk := range tasks {
		if tsk.TaskType == string(semantic.TaskTypeAudioExtractText) {
			audioSeen = true
			if tsk.Status != string(semantic.TaskQueued) || tsk.ResourceVersion != 1 {
				t.Fatalf("unexpected audio task: %+v", tsk)
			}
		}
	}
	if !audioSeen {
		t.Fatalf("expected audio_extract_text for upload completion .mp3 among %+v", tasks)
	}
}

func TestConfirmUploadAutoEmbeddingEnqueuesAudioExtractTaskForWAV(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		DatabaseAutoEmbedding: true,
		AsyncAudioExtract: AsyncAudioExtractOptions{
			Enabled:   true,
			Extractor: &staticAudioExtractor{text: "x"},
		},
	})
	ctx := context.Background()
	totalSize := int64(2 << 20)
	plan, err := b.InitiateUpload(ctx, "/upload/clip.wav", totalSize)
	if err != nil {
		t.Fatal(err)
	}
	uploadAllPartsForPlan(t, b, plan, plan.UploadID, totalSize)
	if err := b.ConfirmUpload(ctx, plan.UploadID); err != nil {
		t.Fatal(err)
	}
	fileID, revision, _, contentType := mustFileForPath(t, b, "/upload/clip.wav")
	if revision != 1 {
		t.Fatalf("revision=%d, want 1", revision)
	}
	wantCT := detectContentType("/upload/clip.wav", nil)
	if contentType != wantCT {
		t.Fatalf("content_type=%q, want %q", contentType, wantCT)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	var audioSeen bool
	for _, tsk := range tasks {
		if tsk.TaskType == string(semantic.TaskTypeAudioExtractText) {
			audioSeen = true
			if tsk.Status != string(semantic.TaskQueued) || tsk.ResourceVersion != 1 {
				t.Fatalf("unexpected audio task: %+v", tsk)
			}
		}
	}
	if !audioSeen {
		t.Fatalf("expected audio_extract_text for upload completion .wav among %+v", tasks)
	}
}

func TestConfirmUploadAutoEmbeddingEnqueuesAudioExtractTaskForMP4(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		DatabaseAutoEmbedding: true,
		AsyncAudioExtract: AsyncAudioExtractOptions{
			Enabled:   true,
			Extractor: &staticAudioExtractor{text: "x"},
		},
	})
	ctx := context.Background()
	totalSize := int64(2 << 20)
	plan, err := b.InitiateUpload(ctx, "/upload/clip.mp4", totalSize)
	if err != nil {
		t.Fatal(err)
	}
	uploadAllPartsForPlan(t, b, plan, plan.UploadID, totalSize)
	if err := b.ConfirmUpload(ctx, plan.UploadID); err != nil {
		t.Fatal(err)
	}
	fileID, revision, _, contentType := mustFileForPath(t, b, "/upload/clip.mp4")
	if revision != 1 {
		t.Fatalf("revision=%d, want 1", revision)
	}
	wantCT := detectContentType("/upload/clip.mp4", nil)
	if contentType != wantCT {
		t.Fatalf("content_type=%q, want %q", contentType, wantCT)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	var audioSeen bool
	for _, tsk := range tasks {
		if tsk.TaskType == string(semantic.TaskTypeAudioExtractText) {
			audioSeen = true
			if tsk.Status != string(semantic.TaskQueued) || tsk.ResourceVersion != 1 {
				t.Fatalf("unexpected audio task: %+v", tsk)
			}
		}
	}
	if !audioSeen {
		t.Fatalf("expected audio_extract_text for upload completion .mp4 among %+v", tasks)
	}
}

func TestConfirmUploadOverwriteAutoEmbeddingEnqueuesAudioExtractTask(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		DatabaseAutoEmbedding: true,
		AsyncAudioExtract: AsyncAudioExtractOptions{
			Enabled:   true,
			Extractor: &staticAudioExtractor{text: "x"},
		},
	})
	ctx := context.Background()
	fileID := insertImageFileForExtractTest(t, b, "/upload/ow.mp3", "audio/mpeg", []byte{1, 2, 3})
	setStoredEmbeddingState(t, b, fileID, 1)

	totalSize := int64(2 << 20)
	plan, err := b.InitiateUpload(ctx, "/upload/ow.mp3", totalSize)
	if err != nil {
		t.Fatal(err)
	}
	uploadAllPartsForPlan(t, b, plan, plan.UploadID, totalSize)
	if err := b.ConfirmUpload(ctx, plan.UploadID); err != nil {
		t.Fatal(err)
	}
	confirmedID, revision, _, _ := mustFileForPath(t, b, "/upload/ow.mp3")
	if confirmedID != fileID {
		t.Fatalf("overwrite should preserve file_id=%q, got %q", fileID, confirmedID)
	}
	if revision != 2 {
		t.Fatalf("revision=%d, want 2", revision)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	var audioRev2 bool
	for _, tsk := range tasks {
		if tsk.TaskType == string(semantic.TaskTypeAudioExtractText) && tsk.ResourceVersion == 2 {
			audioRev2 = true
			if tsk.Status != string(semantic.TaskQueued) {
				t.Fatalf("unexpected audio task: %+v", tsk)
			}
		}
	}
	if !audioRev2 {
		t.Fatalf("expected audio_extract_text for revision 2 among %+v", tasks)
	}
}

func TestConfirmUploadAutoEmbeddingEnqueuesAudioExtractTaskForM4A(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		DatabaseAutoEmbedding: true,
		AsyncAudioExtract: AsyncAudioExtractOptions{
			Enabled:   true,
			Extractor: &staticAudioExtractor{text: "x"},
		},
	})
	ctx := context.Background()
	totalSize := int64(2 << 20)
	plan, err := b.InitiateUpload(ctx, "/upload/clip.m4a", totalSize)
	if err != nil {
		t.Fatal(err)
	}
	uploadAllPartsForPlan(t, b, plan, plan.UploadID, totalSize)
	if err := b.ConfirmUpload(ctx, plan.UploadID); err != nil {
		t.Fatal(err)
	}
	fileID, revision, _, contentType := mustFileForPath(t, b, "/upload/clip.m4a")
	if revision != 1 {
		t.Fatalf("revision=%d, want 1", revision)
	}
	wantCT := detectContentType("/upload/clip.m4a", nil)
	if contentType != wantCT {
		t.Fatalf("content_type=%q, want %q", contentType, wantCT)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	var audioSeen bool
	for _, tsk := range tasks {
		if tsk.TaskType == string(semantic.TaskTypeAudioExtractText) {
			audioSeen = true
			if tsk.Status != string(semantic.TaskQueued) || tsk.ResourceVersion != 1 {
				t.Fatalf("unexpected audio task: %+v", tsk)
			}
		}
	}
	if !audioSeen {
		t.Fatalf("expected audio_extract_text for upload completion .m4a among %+v", tasks)
	}
}

func TestConfirmUploadAutoEmbeddingSkipsAudioWhenAsyncAudioDisabled(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		DatabaseAutoEmbedding: true,
		AsyncAudioExtract: AsyncAudioExtractOptions{
			Enabled:   false,
			Extractor: &staticAudioExtractor{text: "x"},
		},
	})
	ctx := context.Background()
	totalSize := int64(2 << 20)
	plan, err := b.InitiateUpload(ctx, "/upload/no-audio-async.mp3", totalSize)
	if err != nil {
		t.Fatal(err)
	}
	uploadAllPartsForPlan(t, b, plan, plan.UploadID, totalSize)
	if err := b.ConfirmUpload(ctx, plan.UploadID); err != nil {
		t.Fatal(err)
	}
	fileID, _, _, _ := mustFileForPath(t, b, "/upload/no-audio-async.mp3")
	tasks := loadSemanticTasksForFile(t, b, fileID)
	for _, tsk := range tasks {
		if tsk.TaskType == string(semantic.TaskTypeAudioExtractText) {
			t.Fatalf("unexpected audio task when async audio disabled: %+v", tsk)
		}
	}
}

// TestConfirmUploadV2AutoEmbeddingEnqueuesAudioViaSharedFinalize checks that
// ConfirmUploadV2 enqueues audio_extract_text for TiDB auto-embedding by going
// through the shared finalizeUpload path.
//
// The fixture intentionally uses InitiateUpload (v1) to create the multipart
// upload and upload parts; only the completion step uses ConfirmUploadV2. That
// does not exercise the full InitiateUploadV2 / v2 presign lifecycle (covered
// in upload_test.go); it specifically locks shared finalize behavior so v1 and
// v2 completion stay aligned for semantic enqueue.
func TestConfirmUploadV2AutoEmbeddingEnqueuesAudioViaSharedFinalize(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		DatabaseAutoEmbedding: true,
		AsyncAudioExtract: AsyncAudioExtractOptions{
			Enabled:   true,
			Extractor: &staticAudioExtractor{text: "x"},
		},
	})
	ctx := context.Background()
	totalSize := int64(2 << 20)
	plan, err := b.InitiateUpload(ctx, "/upload/shared-finalize-v2-clip.mp3", totalSize)
	if err != nil {
		t.Fatal(err)
	}
	uploadAllPartsForPlan(t, b, plan, plan.UploadID, totalSize)
	parts := completePartsForConfirmV2(t, b, plan.UploadID)
	if err := b.ConfirmUploadV2(ctx, plan.UploadID, parts); err != nil {
		t.Fatal(err)
	}
	fileID, revision, _, _ := mustFileForPath(t, b, "/upload/shared-finalize-v2-clip.mp3")
	if revision != 1 {
		t.Fatalf("revision=%d, want 1", revision)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	var audioSeen bool
	for _, tsk := range tasks {
		if tsk.TaskType == string(semantic.TaskTypeAudioExtractText) {
			audioSeen = true
			if tsk.Status != string(semantic.TaskQueued) || tsk.ResourceVersion != 1 {
				t.Fatalf("unexpected audio task: %+v", tsk)
			}
		}
	}
	if !audioSeen {
		t.Fatalf("ConfirmUploadV2 should enqueue audio via shared finalize; got %+v", tasks)
	}
}

func TestConfirmUploadOverwriteAutoEmbeddingImageEnqueuesImgExtractTaskWithoutLegacyQueue(t *testing.T) {
	extractor := &gatedImageExtractor{started: make(chan struct{}, 1), release: make(chan struct{})}
	b := newTestBackendWithOptions(t, Options{
		DatabaseAutoEmbedding: true,
		AsyncImageExtract: AsyncImageExtractOptions{
			Enabled:   true,
			Workers:   1,
			QueueSize: 8,
			Extractor: extractor,
		},
	})
	ctx := context.Background()
	fileID := insertImageFileForExtractTest(t, b, "/img/upload-overwrite-auto.png", "image/png", []byte("old-image"))
	setStoredEmbeddingState(t, b, fileID, 1)

	totalSize := int64(2 << 20)
	plan, err := b.InitiateUpload(ctx, "/img/upload-overwrite-auto.png", totalSize)
	if err != nil {
		t.Fatal(err)
	}
	pendingUpload, err := b.GetUpload(ctx, plan.UploadID)
	if err != nil {
		t.Fatal(err)
	}
	pendingFileID := pendingUpload.FileID
	uploadAllPartsForPlan(t, b, plan, plan.UploadID, totalSize)

	if err := b.ConfirmUpload(ctx, plan.UploadID); err != nil {
		t.Fatal(err)
	}

	select {
	case <-extractor.started:
		close(extractor.release)
		t.Fatal("legacy image queue should stay idle for auto upload overwrite path")
	case <-time.After(200 * time.Millisecond):
	}

	confirmedFileID, revision, embeddingRevision, contentType := mustFileForPath(t, b, "/img/upload-overwrite-auto.png")
	if confirmedFileID != fileID {
		t.Fatalf("overwrite should preserve inode file_id=%q, got %q", fileID, confirmedFileID)
	}
	if revision != 2 {
		t.Fatalf("revision=%d, want 2", revision)
	}
	if embeddingRevision == nil || *embeddingRevision != 1 {
		t.Fatalf("embedding revision should be preserved, got %v", embeddingRevision)
	}
	if contentType != detectContentType("/img/upload-overwrite-auto.png", nil) {
		t.Fatalf("content_type=%q, want %q", contentType, detectContentType("/img/upload-overwrite-auto.png", nil))
	}
	nf, err := b.Store().Stat(ctx, "/img/upload-overwrite-auto.png")
	if err != nil || nf.File == nil {
		t.Fatalf("stat /img/upload-overwrite-auto.png: %v", err)
	}
	if nf.File.ContentText != "" {
		t.Fatalf("content_text=%q, want empty before durable worker", nf.File.ContentText)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	if len(tasks) != 1 {
		t.Fatalf("semantic task count=%d, want 1", len(tasks))
	}
	if tasks[0].TaskType != string(semantic.TaskTypeImgExtractText) || tasks[0].Status != string(semantic.TaskQueued) || tasks[0].ResourceVersion != 2 {
		t.Fatalf("unexpected semantic task: %+v", tasks[0])
	}
	upload, err := b.GetUpload(ctx, plan.UploadID)
	if err != nil {
		t.Fatal(err)
	}
	if upload.FileID != fileID {
		t.Fatalf("upload file_id=%q, want surviving inode %q", upload.FileID, fileID)
	}
	var deletedStatus string
	if err := b.Store().DB().QueryRow(`SELECT status FROM files WHERE file_id = ?`, pendingFileID).Scan(&deletedStatus); err != nil {
		t.Fatal(err)
	}
	if deletedStatus != "DELETED" {
		t.Fatalf("pending upload file status=%q, want DELETED", deletedStatus)
	}
}

func TestRenameDoesNotCreateAdditionalSemanticTasks(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{AppSemanticTasksEnabled: true})
	if _, err := b.Write("/old.txt", []byte("data"), 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	fileID, _, _, _ := mustFileForPath(t, b, "/old.txt")
	if err := b.Rename("/old.txt", "/new.txt"); err != nil {
		t.Fatal(err)
	}

	newFileID, _, _, _ := mustFileForPath(t, b, "/new.txt")
	if newFileID != fileID {
		t.Fatalf("rename should preserve file_id=%q, got %q", fileID, newFileID)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	if len(tasks) != 1 {
		t.Fatalf("semantic task count=%d, want 1", len(tasks))
	}
}

func TestCopyFileDoesNotCreateAdditionalSemanticTasks(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{AppSemanticTasksEnabled: true})
	if _, err := b.Write("/src.txt", []byte("shared"), 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	fileID, _, _, _ := mustFileForPath(t, b, "/src.txt")
	if err := b.CopyFile("/src.txt", "/dst.txt"); err != nil {
		t.Fatal(err)
	}

	dstFileID, _, _, _ := mustFileForPath(t, b, "/dst.txt")
	if dstFileID != fileID {
		t.Fatalf("copy should preserve file_id=%q, got %q", fileID, dstFileID)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	if len(tasks) != 1 {
		t.Fatalf("semantic task count=%d, want 1", len(tasks))
	}
}

func TestShouldEnqueueEmbedForRevisionWithSynchronousText(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{AppSemanticTasksEnabled: true})
	if !b.shouldEnqueueEmbedForRevision("/docs/a.txt", "text/plain", "hello world", "") {
		t.Fatal("expected synchronous text content to enqueue embed work")
	}
}

func TestShouldEnqueueEmbedForRevisionWithAsyncImageSource(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		AppSemanticTasksEnabled: true,
		AsyncImageExtract: AsyncImageExtractOptions{
			Enabled:   true,
			Workers:   1,
			QueueSize: 8,
			Extractor: &staticImageExtractor{text: "caption"},
		},
	})
	if b.shouldEnqueueEmbedForRevision("/img/a.png", "application/octet-stream", "", "") {
		t.Fatal("image without semantic text must wait for extraction writeback before enqueueing embed work")
	}
}

func TestNewEmbedTaskCarriesGeneration(t *testing.T) {
	task := newEmbedTask("task-embed", "file-embed", 3, 9, 11, true, time.Now())
	var payload semantic.EmbedTaskPayload
	if err := json.Unmarshal(task.PayloadJSON, &payload); err != nil {
		t.Fatal(err)
	}
	if payload.Generation != 9 || payload.ConfigRevision != 11 || !payload.TenantManaged {
		t.Fatalf("payload=%+v, want tenant-managed generation 9 config revision 11", payload)
	}
}

func TestEnqueueEmbeddingBackfillTaskDeduplicatesConfigRevision(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		AppSemanticTasksEnabled:    true,
		AppEmbeddingGeneration:     5,
		AppEmbeddingConfigRevision: 8,
		QueryEmbedding:             QueryEmbeddingOptions{Client: embedding.NopClient{}},
	})
	created, err := b.EnqueueEmbeddingBackfillTask(context.Background())
	if err != nil || !created {
		t.Fatalf("first enqueue created=%v err=%v", created, err)
	}
	created, err = b.EnqueueEmbeddingBackfillTask(context.Background())
	if err != nil || created {
		t.Fatalf("duplicate enqueue created=%v err=%v", created, err)
	}
	var taskType string
	var resourceVersion int64
	var payloadJSON []byte
	if err := b.Store().DB().QueryRow(`SELECT task_type, resource_version, payload_json
		FROM semantic_tasks WHERE resource_id = ?`, embeddingBackfillResourceID).
		Scan(&taskType, &resourceVersion, &payloadJSON); err != nil {
		t.Fatal(err)
	}
	if taskType != string(semantic.TaskTypeEmbedBackfill) || resourceVersion != 8 {
		t.Fatalf("task type=%q version=%d", taskType, resourceVersion)
	}
	var payload semantic.EmbedBackfillTaskPayload
	if err := json.Unmarshal(payloadJSON, &payload); err != nil {
		t.Fatal(err)
	}
	if payload.Generation != 5 || payload.ConfigRevision != 8 || payload.AfterInodeID != "" {
		t.Fatalf("payload=%+v", payload)
	}
}

func TestAppEmbeddingRuntimeRequiresConfigRevision(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		AppSemanticTasksEnabled: true,
		AppEmbeddingGeneration:  5,
		QueryEmbedding:          QueryEmbeddingOptions{Client: embedding.NopClient{}},
	})
	if b.SupportsAppEmbeddingTasks() {
		t.Fatal("zero config revision must not expose a ready app embedding runtime")
	}
	if _, ready := b.AppEmbeddingRuntime(); ready {
		t.Fatal("zero config revision runtime must not be ready")
	}
	created, err := b.EnqueueEmbeddingBackfillTask(context.Background())
	if err != nil || created {
		t.Fatalf("zero config revision backfill created=%v err=%v, want no-op", created, err)
	}
}

func TestShouldEnqueueEmbedForRevisionWithoutTextSource(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{AppSemanticTasksEnabled: true})
	if b.shouldEnqueueEmbedForRevision("/bin/a.bin", "application/octet-stream", "", "") {
		t.Fatal("generic binary object should not enqueue embed work without text source")
	}
}

func TestNewImgExtractTaskCarriesPayloadHints(t *testing.T) {
	now := time.Now().UTC()
	task, err := newImgExtractTask("task-1", "file-1", 7, "/img/a.png", "image/png", now)
	if err != nil {
		t.Fatal(err)
	}
	if task.TaskType != semantic.TaskTypeImgExtractText {
		t.Fatalf("task type=%q, want %q", task.TaskType, semantic.TaskTypeImgExtractText)
	}
	if task.ResourceID != "file-1" || task.ResourceVersion != 7 {
		t.Fatalf("unexpected task identity: %+v", task)
	}
	var payload semantic.ImgExtractTaskPayload
	if err := json.Unmarshal(task.PayloadJSON, &payload); err != nil {
		t.Fatal(err)
	}
	if payload.Path != "/img/a.png" || payload.ContentType != "image/png" {
		t.Fatalf("unexpected payload: %+v", payload)
	}
}

func TestShouldEnqueueEmbedForRevisionWithDescriptionOnly(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{AppSemanticTasksEnabled: true})
	if !b.shouldEnqueueEmbedForRevision("/bin/a.bin", "application/octet-stream", "", "some description") {
		t.Fatal("expected non-empty description to enqueue embed work")
	}
}

func TestShouldEnqueueEmbedForRevision_DisabledSkipsTextContent(t *testing.T) {
	b := newTestBackend(t) // AppSemanticTasksEnabled defaults to false
	if b.shouldEnqueueEmbedForRevision("/docs/a.txt", "text/plain", "hello world", "") {
		t.Fatal("should not enqueue embed task when app semantic tasks disabled")
	}
}

func TestShouldEnqueueEmbedForRevision_DisabledSkipsDescription(t *testing.T) {
	b := newTestBackend(t)
	if b.shouldEnqueueEmbedForRevision("/bin/a.bin", "application/octet-stream", "", "some description") {
		t.Fatal("should not enqueue embed task for description when app semantic tasks disabled")
	}
}

func TestShouldEnqueueEmbedForRevision_DisabledSkipsImage(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		AsyncImageExtract: AsyncImageExtractOptions{
			Enabled:   true,
			Workers:   1,
			QueueSize: 8,
			Extractor: &staticImageExtractor{text: "caption"},
		},
	})
	if b.shouldEnqueueEmbedForRevision("/img/a.png", "image/png", "", "") {
		t.Fatal("should not enqueue embed task for image when app semantic tasks disabled")
	}
}

// App-embedding mode image write paths now enqueue durable img_extract_text
// tasks in the same transaction (same as auto-embedding mode), instead of the
// legacy in-process channel. The following tests verify the three write paths.

func TestWriteCreateAppEmbeddingImageEnqueuesImgExtractTask(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		AppSemanticTasksEnabled: true,
		AsyncImageExtract: AsyncImageExtractOptions{
			Enabled:   true,
			Extractor: &staticImageExtractor{text: "cat on sofa"},
		},
	})
	if _, err := b.Write("/img/create-app.png", fakePNG, 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	fileID, revision, _, _ := mustFileForPath(t, b, "/img/create-app.png")
	if revision != 1 {
		t.Fatalf("revision=%d, want 1", revision)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	// Expect one img_extract_text task (and optionally one embed task).
	var imgTasks, embedTasks int
	for _, task := range tasks {
		switch task.TaskType {
		case string(semantic.TaskTypeImgExtractText):
			imgTasks++
		case string(semantic.TaskTypeEmbed):
			embedTasks++
		}
	}
	if imgTasks != 1 {
		t.Fatalf("img_extract_text task count=%d, want 1 (all tasks: %+v)", imgTasks, tasks)
	}
	// App mode also enqueues an embed task for image revisions (it no-ops until
	// content_text is written by the extract worker, which then bridges a fresh
	// embed task). This is existing behavior, not changed by this PR.
	_ = embedTasks
}

func TestWriteOverwriteAppEmbeddingImageEnqueuesImgExtractTask(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		AppSemanticTasksEnabled: true,
		AsyncImageExtract: AsyncImageExtractOptions{
			Enabled:   true,
			Extractor: &staticImageExtractor{text: "updated caption"},
		},
	})
	fileID := insertImageFileForExtractTest(t, b, "/img/overwrite-app.png", "image/png", []byte("old-image"))

	if _, err := b.Write("/img/overwrite-app.png", []byte("new-image"), 0, filesystem.WriteFlagTruncate); err != nil {
		t.Fatal(err)
	}
	_, revision, _, _ := mustFileForPath(t, b, "/img/overwrite-app.png")
	if revision != 2 {
		t.Fatalf("revision=%d, want 2", revision)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	var imgTasks int
	for _, task := range tasks {
		if task.TaskType == string(semantic.TaskTypeImgExtractText) {
			if task.ResourceVersion == revision {
				imgTasks++
			}
		}
	}
	if imgTasks != 1 {
		t.Fatalf("img_extract_text task count for revision %d =%d, want 1", revision, imgTasks)
	}
}

// Capability mode: a resolver-backed multi-tenant server (Capability=true, no
// process-default extractor) still enqueues durable img_extract_text tasks;
// the worker resolves a per-tenant runtime at claim time.

func TestWriteCreateCapabilityImageEnqueuesImgExtractTaskWithoutDefaultExtractor(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		AppSemanticTasksEnabled: true,
		AsyncImageExtract: AsyncImageExtractOptions{
			Capability: true, // Enabled intentionally false: no default extractor
		},
	})
	if b.SupportsAsyncImageExtract() {
		t.Fatal("capability-only backend should not support the process-default extract runtime")
	}
	if _, err := b.Write("/img/create-cap.png", fakePNG, 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	fileID, revision, _, _ := mustFileForPath(t, b, "/img/create-cap.png")
	if revision != 1 {
		t.Fatalf("revision=%d, want 1", revision)
	}
	tasks := loadSemanticTasksForFile(t, b, fileID)
	var imgTasks int
	for _, task := range tasks {
		if task.TaskType == string(semantic.TaskTypeImgExtractText) {
			imgTasks++
		}
	}
	if imgTasks != 1 {
		t.Fatalf("img_extract_text task count=%d, want 1 (all tasks: %+v)", imgTasks, tasks)
	}
}

func TestShouldEnqueueEmbedForRevisionWithImageCapability(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		AppSemanticTasksEnabled: true,
		AsyncImageExtract: AsyncImageExtractOptions{
			Capability: true,
		},
	})
	if b.shouldEnqueueEmbedForRevision("/img/a.png", "application/octet-stream", "", "") {
		t.Fatal("image capability without semantic text must not enqueue embed work")
	}
	if b.shouldEnqueueEmbedForRevision("/bin/a.bin", "application/octet-stream", "", "") {
		t.Fatal("non-image path should not enqueue embed work even with capability")
	}
}

func TestConfirmUploadAppEmbeddingImageEnqueuesImgExtractTask(t *testing.T) {
	b := newTestBackendWithOptions(t, Options{
		AppSemanticTasksEnabled: true,
		AsyncImageExtract: AsyncImageExtractOptions{
			Enabled:   true,
			Extractor: &staticImageExtractor{text: "uploaded image caption"},
		},
	})
	ctx := context.Background()
	totalSize := int64(2 << 20)
	plan, err := b.InitiateUpload(ctx, "/img/upload-app.png", totalSize)
	if err != nil {
		t.Fatal(err)
	}
	uploadAllPartsForPlan(t, b, plan, plan.UploadID, totalSize)
	if err := b.ConfirmUpload(ctx, plan.UploadID); err != nil {
		t.Fatal(err)
	}
	fileID, _, _, _ := mustFileForPath(t, b, "/img/upload-app.png")
	tasks := loadSemanticTasksForFile(t, b, fileID)
	var imgTasks int
	for _, task := range tasks {
		if task.TaskType == string(semantic.TaskTypeImgExtractText) {
			imgTasks++
		}
	}
	if imgTasks != 1 {
		t.Fatalf("img_extract_text task count=%d, want 1 (all tasks: %+v)", imgTasks, tasks)
	}
}