package backend

import (
	"context"
	"database/sql"
	"encoding/binary"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"sort"
	"strings"
	"time"

	"github.com/pingcap/failpoint"

	"github.com/mem9-ai/drive9/pkg/datastore"
)

// AudioExtractRequest is the input to a pluggable audio->text extractor.
type AudioExtractRequest struct {
	TenantID    string
	FileID      string
	Path        string
	ContentType string
	Data        []byte
}

// AudioTextExtractor extracts searchable transcript text from audio bytes.
type AudioTextExtractor interface {
	ExtractAudioText(ctx context.Context, req AudioExtractRequest) (string, AudioExtractUsage, error)
}

// AudioExtractRuntime is the effective, tenant-scoped runtime used for one
// durable audio task. It is intentionally dependency-only: provider secrets
// never enter semantic task payloads or durable state.
type AudioExtractRuntime struct {
	Checkpoint          func(AudioExtractCheckpoint) error
	CheckpointState     AudioExtractCheckpoint
	FinalAttempt        bool
	Extractor           AudioTextExtractor
	Timeout             time.Duration
	MaxSourceBytes      int64
	MaxAudioBytes       int64
	MaxExtractTextBytes int
	// CheckpointTextMaxJSONBytes bounds the JSON-encoded transcript string
	// stored in a mutable semantic task payload. Zero uses the standalone
	// audio budget; video supplies a smaller per-modality budget.
	CheckpointTextMaxJSONBytes int
	Workspace                  *MediaExtractWorkspace
}

type AudioExtractCheckpoint struct {
	CompletedSegments int               `json:"completed_segments"`
	Text              string            `json:"text,omitempty"`
	Tags              map[string]string `json:"tags,omitempty"`
	Structured        bool              `json:"structured,omitempty"`
}

// ValidateAudioProvider performs one bounded real transcription request. It
// sends silence with the effective configured prompt; no tenant media is
// included.
func ValidateAudioProvider(ctx context.Context, protocol string, cfg AudioTextExtractorConfig) error {
	ownedClient := false
	if cfg.Client == nil {
		cfg.Client = NewTenantExtractorHTTPClient(cfg.Timeout)
		ownedClient = true
	}
	if ownedClient {
		defer cfg.Client.CloseIdleConnections()
	}
	extractor, err := NewAudioTextExtractorForProtocol(protocol, cfg)
	if err != nil {
		return err
	}
	const sampleRate, samples = 16000, 800
	data := make([]byte, 44+samples*2)
	copy(data[0:4], "RIFF")
	binary.LittleEndian.PutUint32(data[4:8], uint32(len(data)-8))
	copy(data[8:12], "WAVE")
	copy(data[12:16], "fmt ")
	binary.LittleEndian.PutUint32(data[16:20], 16)
	binary.LittleEndian.PutUint16(data[20:22], 1)
	binary.LittleEndian.PutUint16(data[22:24], 1)
	binary.LittleEndian.PutUint32(data[24:28], sampleRate)
	binary.LittleEndian.PutUint32(data[28:32], sampleRate*2)
	binary.LittleEndian.PutUint16(data[32:34], 2)
	binary.LittleEndian.PutUint16(data[34:36], 16)
	copy(data[36:40], "data")
	binary.LittleEndian.PutUint32(data[40:44], samples*2)
	_, _, err = extractor.ExtractAudioText(ctx, AudioExtractRequest{ContentType: "audio/wav", Path: "validation.wav", Data: data})
	if errors.Is(err, errAudioExtractEmptyText) {
		return nil
	}
	return err
}

var errAudioExtractPartial = errors.New("audio extraction partially completed")
var errAudioExtractEmptyText = errors.New("audio extraction returned empty text")

type audioExtractEmptyTextError struct {
	cause error
}

func (e *audioExtractEmptyTextError) Error() string { return e.cause.Error() }
func (e *audioExtractEmptyTextError) Unwrap() error { return e.cause }
func (e *audioExtractEmptyTextError) Is(target error) bool {
	return target == errAudioExtractEmptyText
}

func newAudioExtractEmptyTextError(cause error) error {
	return &audioExtractEmptyTextError{cause: cause}
}

const DefaultAudioExtractMaxTextBytes = 128 << 10

const (
	audioCheckpointTextMaxJSONBytes = 40 << 10
	videoCheckpointTextMaxJSONBytes = 20 << 10
	mp3ExtractBytesPerSecond        = int64(32_000 / 8)
	maxAudioExtractDurationSeconds  = int64(2 * 60 * 60)
	maxAudioExtractSegments         = 12
	maxAudioExtractDerivedBytes     = int64(40 << 20)
)

// AudioExtractAPIError describes an HTTP/API failure returned by an audio
// transcription provider.
type AudioExtractAPIError struct {
	Provider   string
	StatusCode int
	Message    string
}

func (e *AudioExtractAPIError) Error() string {
	provider := strings.TrimSpace(e.Provider)
	if provider == "" {
		provider = "audio extract"
	}
	return fmt.Sprintf("%s api status %d", provider, e.StatusCode)
}

// NonRetryableAudioExtract reports whether retrying the same audio bytes cannot
// reasonably succeed without changing the request, file, or provider account.
func (e *AudioExtractAPIError) NonRetryableAudioExtract() bool {
	switch e.StatusCode {
	case http.StatusBadRequest,
		http.StatusUnauthorized,
		http.StatusNotFound,
		http.StatusForbidden:
		return true
	default:
		return false
	}
}

type nonRetryableAudioExtractError interface {
	NonRetryableAudioExtract() bool
}

// IsNonRetryableAudioExtractError reports whether err should bypass normal
// semantic worker retry scheduling for an audio_extract_text task.
func IsNonRetryableAudioExtractError(err error) bool {
	var permanent nonRetryableAudioExtractError
	if errors.As(err, &permanent) {
		return permanent.NonRetryableAudioExtract()
	}
	return false
}

// AudioExtractTaskSpec carries the revision-scoped inputs needed to extract
// audio transcript text for one file version.
type AudioExtractTaskSpec struct {
	FileID      string
	Path        string
	ContentType string
	Revision    int64
}

// AudioExtractResult reports the outcome of one audio extraction attempt.
type AudioExtractResult string

const (
	AudioExtractResultRuntimeNotConfigured AudioExtractResult = "runtime_not_configured"
	AudioExtractResultGetFileError         AudioExtractResult = "get_file_error"
	AudioExtractResultFileNotFound         AudioExtractResult = "file_not_found"
	AudioExtractResultNotConfirmed         AudioExtractResult = "not_confirmed"
	AudioExtractResultNotAudio             AudioExtractResult = "not_audio"
	AudioExtractResultStale                AudioExtractResult = "stale"
	AudioExtractResultLoadError            AudioExtractResult = "load_error"
	AudioExtractResultTooLarge             AudioExtractResult = "too_large"
	AudioExtractResultTooLong              AudioExtractResult = "too_long"
	AudioExtractResultExtractError         AudioExtractResult = "extract_error"
	AudioExtractResultEmptyText            AudioExtractResult = "empty_text"
	AudioExtractResultUpdateError          AudioExtractResult = "update_error"
	AudioExtractResultWritten              AudioExtractResult = "written"
	AudioExtractResultPartialSuccess       AudioExtractResult = "partial_success"
	AudioExtractResultBudgetExhausted      AudioExtractResult = "budget_exhausted"
)

// SupportsAsyncAudioExtract reports whether this backend instance has a fully
// wired audio transcript runtime (enabled flag and non-nil extractor). Phase 2
// does not supply an implicit default extractor when Extractor is nil.
func (b *Dat9Backend) SupportsAsyncAudioExtract() bool {
	return b != nil && b.audioExtractEnabled && b.audioExtractor != nil
}

// SupportsAudioExtractTasks reports whether the backend can execute audio
// tasks either with its process default or a tenant runtime resolved later.
func (b *Dat9Backend) SupportsAudioExtractTasks() bool {
	return b != nil && (b.SupportsAsyncAudioExtract() || b.audioExtractCapability)
}

// ProcessAudioExtractTaskWithRuntime runs audio extraction with an explicitly
// resolved runtime. This is the multi-tenant counterpart to the legacy
// process-default method and does not require b.audioExtractor to be set.
func (b *Dat9Backend) ProcessAudioExtractTaskWithRuntime(ctx context.Context, task AudioExtractTaskSpec, rt AudioExtractRuntime) (AudioExtractResult, error) {
	if rt.MaxSourceBytes <= 0 {
		rt.MaxSourceBytes = b.audioExtractMaxSourceSize
	}
	if rt.MaxSourceBytes <= 0 {
		rt.MaxSourceBytes = DefaultAudioExtractMaxSourceBytes
	}
	if rt.Extractor == nil || rt.MaxAudioBytes <= 0 {
		return AudioExtractResultRuntimeNotConfigured, fmt.Errorf("async audio extract runtime not configured")
	}
	return b.processAudioExtractTask(ctx, task, rt)
}

// audioExtensionMIME maps path suffixes to canonical MIME types for the current
// audio_extract_text closed set. It is consulted only when content_type is
// missing or too generic.
//
// The shipped set stays small: MP3, WAV, MP4, and M4A (audio in MP4
// container). Broader formats remain deferred; see TODO(post-MVP audio) and
// TODO(WebM) below.
//
// TODO(post-MVP audio): Restore AAC, FLAC, OGG extensions (and MIME /
// alias allowlist entries) once the extractor and tests cover them.
//
//	".aac":  "audio/aac",
//	".ogg":  "audio/ogg",
//	".flac": "audio/flac",
//
// TODO(WebM): Phase 2 MVP excludes both audio/webm and video/webm. Go's
// mime.TypeByExtension(".webm") reports video/webm; supporting WebM means
// deciding extractor behavior for muxed vs audio-only payloads and possibly
// normalizing aliases—revisit after MVP.
var audioExtensionMIME = map[string]string{
	".mp3": "audio/mpeg",
	".wav": "audio/wav",
	".mp4": "audio/mp4",
	".m4a": "audio/mp4",
}

// allowedAudioMIME is the current closed set for durable audio_extract_text.
// MP4 container uploads are admitted here even though detectContentType()
// commonly reports video/mp4 for .mp4 paths.
//
// TODO(post-MVP audio): Re-add closed-set entries for AAC, OGG, FLAC
// (and further aliases) when those formats return to scope.
//
//	"audio/aac":   {},
//	"audio/ogg":   {},
//	"audio/flac":  {},
//
// Do not add audio/webm or video/webm until post-MVP WebM work (TODO(WebM)).
var allowedAudioMIME = map[string]struct{}{
	"audio/mpeg":  {},
	"audio/wav":   {},
	"audio/x-wav": {},
	"audio/mp4":   {},
	"audio/x-m4a": {},
}

func stripMIMEParams(ct string) string {
	ct = strings.TrimSpace(strings.ToLower(ct))
	if i := strings.Index(ct, ";"); i >= 0 {
		ct = strings.TrimSpace(ct[:i])
	}
	return ct
}

// normalizeStdlibAudioMIMEAliases maps Go mime.TypeByExtension and common
// platform MIME-info aliases onto the current allowlist keys (see
// allowedAudioMIME).
//
// Linux loads extension MIME from freedesktop globs2 before /etc/mime.types.
// Typical mappings include audio/vnd.wave (preferred in shared-mime-info2.3+),
// audio/wave, and audio/x-wav; only some of those are in allowedAudioMIME.
// Canonicalize WAV synonyms to audio/wav so enqueue and task payloads match
// across macOS vs Ubuntu CI. Also canonicalize MP4 container MIME variants so
// semantic routing can treat .mp4 uploads as audio_extract_text candidates
// without changing detectContentType().
//
// TODO(post-MVP audio): When AAC and FLAC are allowlisted again, restore
// further alias normalization (e.g. audio/x-aac -> audio/aac).
func normalizeStdlibAudioMIMEAliases(ct string) string {
	ct = stripMIMEParams(ct)
	switch ct {
	case "audio/wave", "audio/vnd.wave", "audio/x-pn-wav":
		return "audio/wav"
	case "video/mp4", "application/mp4":
		return "audio/mp4"
	case "audio/mp4a-latm":
		return "audio/mp4"
	default:
		return ct
	}
}

func isAllowedAudioMIME(ct string) bool {
	_, ok := allowedAudioMIME[normalizeStdlibAudioMIMEAliases(ct)]
	return ok
}

// audioMIMEAllowsPathFallback mirrors the conservative contract: only when
// content_type is absent or not meaningful do we consult the path extension.
func audioMIMEAllowsPathFallback(ct string) bool {
	switch stripMIMEParams(ct) {
	case "", "application/octet-stream", "text/plain":
		return true
	default:
		return false
	}
}

// effectiveAudioMIME resolves an allowlisted audio MIME from stored content_type
// and/or path. It does not sniff file bytes; detectContentType elsewhere is unchanged.
func effectiveAudioMIME(path, contentType string) string {
	rawStripped := stripMIMEParams(contentType)
	if isAllowedAudioMIME(contentType) {
		return normalizeStdlibAudioMIMEAliases(contentType)
	}
	if !audioMIMEAllowsPathFallback(rawStripped) {
		return ""
	}
	path = strings.ToLower(path)
	for ext, mime := range audioExtensionMIME {
		if strings.HasSuffix(path, ext) {
			return mime
		}
	}
	return ""
}

// isAudioContentType reports whether contentType (and optional path fallback) is
// in the Phase-2 audio allowlist.
func isAudioContentType(contentType, path string) bool {
	return effectiveAudioMIME(path, contentType) != ""
}

func isSupportedAudioForSemanticTask(path, contentType string) bool {
	return isAudioContentType(contentType, path)
}

// errAudioExtractSourceTooLarge signals the object exceeds MaxAudioBytes; it is
// distinct from an empty payload so the handler does not report oversize for 0-byte inputs.
var errAudioExtractSourceTooLarge = errors.New("audio extract source exceeds configured max bytes")

// ErrAudioExtractDurationTooLong signals a decodable audio source exceeds the
// supported two-hour transcript timeline.
var ErrAudioExtractDurationTooLong = errors.New("audio extract duration exceeds supported timeline")

// resolvedAudioMIMEForHandler combines the stored file Content-Type with the durable
// task payload hint. When the DB value is generic (empty, octet-stream, text/plain),
// we may still resolve audio via path extension or the payload MIME—so a task enqueued
// with an audio hint is not dropped just because the inode row kept a generic type.
func resolvedAudioMIMEForHandler(path, storedContentType, payloadContentType string) string {
	stored := strings.TrimSpace(storedContentType)
	payload := strings.TrimSpace(payloadContentType)
	if m := effectiveAudioMIME(path, stored); m != "" {
		return m
	}
	if audioMIMEAllowsPathFallback(stored) {
		return effectiveAudioMIME(path, payload)
	}
	return ""
}

// resolvedAudioExtractContentType chooses the MIME sent to the pluggable
// extractor after the handler has already decided the revision is a supported
// audio source.
//
// Semantic routing canonicalizes MP4 containers to audio/mp4 for stable
// eligibility checks, but OpenAI-compatible ASR providers often expect the
// original part-level container MIME (commonly video/mp4) for .mp4 uploads.
func resolvedAudioExtractContentType(storedContentType, payloadContentType, resolvedAudioMIME string) string {
	resolved := strings.TrimSpace(resolvedAudioMIME)
	if resolved == "" {
		return ""
	}
	for _, candidate := range []string{storedContentType, payloadContentType} {
		switch stripMIMEParams(candidate) {
		case "video/mp4", "application/mp4", "audio/x-m4a":
			return stripMIMEParams(candidate)
		}
	}
	return resolved
}

// ProcessAudioExtractTask runs transcript extraction for one durable
// audio_extract_text task. Terminal business outcomes return a nil error; runtime
// misconfiguration and transient failures return a retryable error.
func (b *Dat9Backend) ProcessAudioExtractTask(ctx context.Context, task AudioExtractTaskSpec) (AudioExtractResult, error) {
	runtime, ok := b.DefaultAudioExtractRuntime()
	if !ok {
		return AudioExtractResultRuntimeNotConfigured, fmt.Errorf("async audio extract runtime not configured")
	}
	return b.processAudioExtractTask(ctx, task, runtime)
}

// DefaultAudioExtractRuntime returns the process-default audio dependencies.
func (b *Dat9Backend) DefaultAudioExtractRuntime() (AudioExtractRuntime, bool) {
	if !b.SupportsAsyncAudioExtract() {
		return AudioExtractRuntime{}, false
	}
	return AudioExtractRuntime{
		Extractor: b.audioExtractor, Timeout: b.audioExtractTimeout,
		MaxSourceBytes: b.audioExtractMaxSourceSize, MaxAudioBytes: b.audioExtractMaxSize, MaxExtractTextBytes: b.maxAudioExtractTextBytes,
		Workspace: b.mediaExtractWorkspace,
	}, true
}

func (b *Dat9Backend) processAudioExtractTask(ctx context.Context, task AudioExtractTaskSpec, rt AudioExtractRuntime) (AudioExtractResult, error) {
	start := time.Now()
	var result AudioExtractResult
	defer func() {
		b.recordTenantOperation("audio_extract", "process", legacyAudioExtractMetricResult(result), time.Since(start))
	}()

	if b.monthlyLLMCostExceededCheck(ctx) {
		b.recordTenantOperation("llm_cost_budget", "process_skip", "budget_exhausted", 0)
		result = AudioExtractResultBudgetExhausted
		return result, nil
	}

	f, err := b.store.GetFile(ctx, task.FileID)
	if err != nil {
		if errors.Is(err, datastore.ErrNotFound) {
			result = AudioExtractResultFileNotFound
			return result, nil
		}
		result = AudioExtractResultGetFileError
		return result, fmt.Errorf("get file: %w", err)
	}
	if f.Status != datastore.StatusConfirmed {
		result = AudioExtractResultNotConfirmed
		return result, nil
	}
	if task.Revision > 0 && f.Revision != task.Revision {
		result = AudioExtractResultStale
		return result, nil
	}
	if f.SizeBytes > rt.MaxSourceBytes {
		result = AudioExtractResultTooLarge
		return result, nil
	}
	resolvedMIME := resolvedAudioMIMEForHandler(task.Path, f.ContentType, task.ContentType)
	if resolvedMIME == "" {
		result = AudioExtractResultNotAudio
		return result, nil
	}
	extractorContentType := resolvedAudioExtractContentType(f.ContentType, task.ContentType, resolvedMIME)
	if extractorContentType == "" {
		extractorContentType = resolvedMIME
	}

	var taskWorkspace *MediaExtractTaskWorkspace
	var sourcePath string
	var data []byte
	if rt.Workspace != nil {
		taskWorkspace, err = rt.Workspace.NewTask(ctx, task.FileID)
		if err != nil {
			result = AudioExtractResultLoadError
			return result, fmt.Errorf("create audio extract workspace: %w", err)
		}
		defer func() { _ = taskWorkspace.Close() }()
		sourcePath, _, err = b.stageMediaFile(ctx, taskWorkspace, f, "source"+audioMIMEToExt(extractorContentType), rt.MaxSourceBytes)
		if err != nil {
			if errors.Is(err, ErrMediaExtractSourceTooLarge) {
				result = AudioExtractResultTooLarge
				return result, nil
			}
			result = AudioExtractResultLoadError
			return result, fmt.Errorf("stage audio source: %w", err)
		}
	} else {
		maxInMemoryBytes := rt.MaxSourceBytes
		if f.StorageType == datastore.StorageS3 && rt.MaxAudioBytes < maxInMemoryBytes {
			maxInMemoryBytes = rt.MaxAudioBytes
		}
		data, err = b.loadAudioBytesForExtract(ctx, f, maxInMemoryBytes)
		if err != nil {
			if errors.Is(err, errAudioExtractSourceTooLarge) && maxInMemoryBytes == rt.MaxSourceBytes {
				result = AudioExtractResultTooLarge
				return result, nil
			}
			if errors.Is(err, errAudioExtractSourceTooLarge) {
				result = AudioExtractResultLoadError
				return result, ErrMediaExtractWorkspaceRequired
			}
			result = AudioExtractResultLoadError
			return result, fmt.Errorf("load audio bytes: %w", err)
		}
	}

	extractCtx, cancelExtract := context.WithTimeout(ctx, effectiveAudioExtractTimeout(rt.Timeout, b.audioExtractTimeout))
	defer cancelExtract()
	text, audioUsage, tags, structured, err := b.extractAudioDataWithTags(extractCtx, task, extractorContentType, data, sourcePath, taskWorkspace, rt)
	b.recordAudioExtractUsage(task.FileID, audioUsage)
	if err != nil {
		if errors.Is(err, ErrAudioExtractDurationTooLong) {
			result = AudioExtractResultTooLong
			return result, nil
		}
		if !errors.Is(err, errAudioExtractPartial) || strings.TrimSpace(text) == "" {
			result = AudioExtractResultExtractError
			return result, fmt.Errorf("extract audio text: %w", err)
		}
		if !rt.FinalAttempt {
			result = AudioExtractResultExtractError
			return result, fmt.Errorf("extract audio text: %w", err)
		}
		result = AudioExtractResultPartialSuccess
	}
	text = sanitizeExtractedText(text, rt.MaxExtractTextBytes)
	emptyText := text == ""
	emptyUnstructured := emptyText && !structured

	// Always revision-gate writeback: UpdateFileSearchTextTx treats expectedRevision<=0
	// as an unscoped UPDATE: use the loaded row revision when the task omits ResourceVersion.
	expectedRevision := task.Revision
	if expectedRevision == 0 {
		expectedRevision = f.Revision
	}

	var updated bool
	err = b.store.InTx(ctx, func(tx *sql.Tx) error {
		if err := injectedAudioExtractWritebackError("audioExtractWritebackUpdateFileSearchTextError"); err != nil {
			return err
		}
		var txErr error
		updated, txErr = b.updateMediaExtractTextTx(tx, task.FileID, expectedRevision, text)
		if txErr != nil {
			return txErr
		}
		if !updated {
			return nil
		}
		if txErr = b.store.DeleteFileTagsBySourceTx(tx, task.FileID, datastore.TagSourceAudioExtract); txErr != nil {
			return txErr
		}
		skipped, txErr := b.store.InsertFileTagsWithSourceTx(tx, task.FileID, datastore.TagSourceAudioExtract, tags)
		if txErr != nil {
			return txErr
		}
		for range skipped {
			b.recordTenantOperation("audio_extract", "attr_skip", "user_tag_conflict", 0)
		}
		if emptyText {
			// No replacement embed task follows an empty app-managed result.
			if !b.UsesDatabaseAutoEmbedding() {
				return b.store.ClearFileEmbeddingStateTx(tx, task.FileID)
			}
			return nil
		}
		// Bridge an embed task in app-embedding mode so the transcribed text
		// gets embedded for semantic search. In auto-embedding mode the DB
		// generates the vector automatically, so no bridge is needed.
		if b.UsesDatabaseAutoEmbedding() || !b.appSemanticTasksEnabled {
			return nil
		}
		_, txErr = b.store.EnsureSemanticTaskQueuedTx(tx, newEmbedTask(
			b.genID(), task.FileID, expectedRevision, b.appEmbeddingGeneration,
			b.appEmbeddingConfigRevision, b.appEmbeddingTenantManaged, time.Now().UTC(),
		))
		return txErr
	})
	if err != nil {
		result = AudioExtractResultUpdateError
		return result, fmt.Errorf("update file search text: %w", err)
	}
	if !updated {
		result = AudioExtractResultStale
		return result, nil
	}
	if emptyUnstructured {
		result = AudioExtractResultEmptyText
		return result, nil
	}
	if result != AudioExtractResultPartialSuccess {
		result = AudioExtractResultWritten
	}
	return result, nil
}

func (b *Dat9Backend) extractAudioDataWithTags(ctx context.Context, task AudioExtractTaskSpec, contentType string, data []byte, sourcePath string, workspace *MediaExtractTaskWorkspace, rt AudioExtractRuntime) (string, AudioExtractUsage, map[string]string, bool, error) {
	// Keep small/unknown test payloads on the existing direct path. Segment only
	// when FFmpeg can prove the input is a decodable audio container.
	segmentThreshold := rt.MaxAudioBytes
	if segmentThreshold <= 0 {
		segmentThreshold = 8 << 20
	}
	sourceSize := int64(len(data))
	if sourcePath != "" {
		info, err := os.Stat(sourcePath)
		if err != nil {
			return "", AudioExtractUsage{}, nil, false, err
		}
		sourceSize = info.Size()
	}
	duration, probeErr := probeAudioDuration(ctx, sourcePath, data, contentType)
	if probeErr != nil {
		// Production runtimes always stage audio in the media workspace. Fail
		// closed there so an unavailable or failed probe cannot bypass the
		// two-hour boundary. The memory-only path is retained for lightweight
		// embedders and tests whose synthetic payloads are not media containers.
		if sourcePath != "" {
			return "", AudioExtractUsage{}, nil, false, fmt.Errorf("probe audio duration: %w", probeErr)
		}
	} else if duration > float64(maxAudioExtractDurationSeconds) {
		return "", AudioExtractUsage{}, nil, false, ErrAudioExtractDurationTooLong
	}
	if sourceSize <= segmentThreshold {
		if sourcePath != "" {
			var err error
			data, err = readBoundedFile(sourcePath, segmentThreshold)
			if err != nil {
				return "", AudioExtractUsage{}, nil, false, err
			}
		}
		text, usage, err := b.extractAudioSegment(ctx, task, contentType, data, rt)
		if errors.Is(err, errAudioExtractEmptyText) {
			usage.successfulCalls++
			return "", usage, nil, false, nil
		}
		if err == nil {
			usage.successfulCalls++
		}
		writeback := prepareMediaExtractWriteback(text, rt.MaxExtractTextBytes)
		if strings.TrimSpace(writeback.text) != "" && duration > 0 {
			writeback.text = audioTimeRangeHeader(0, duration) + "\n" + writeback.text
		}
		return writeback.text, usage, writeback.tags, writeback.structured, err
	}
	segments, err := splitAudioWithFFmpeg(ctx, sourcePath, data, contentType, workspace, segmentThreshold)
	if err != nil {
		return "", AudioExtractUsage{}, nil, false, err
	}
	return b.extractAudioSegmentsWithTags(ctx, task, segments, segmentThreshold, duration, rt, true, true)
}

func resumeAudioCheckpoint(checkpoint AudioExtractCheckpoint, segmentCount int) (int, string) {
	text := checkpoint.Text
	start := checkpoint.CompletedSegments
	if start < 0 || start > segmentCount {
		return 0, ""
	}
	return start, text
}

func effectiveAudioCheckpointTextBudget(rt AudioExtractRuntime) int {
	if rt.CheckpointTextMaxJSONBytes > 0 {
		return rt.CheckpointTextMaxJSONBytes
	}
	return audioCheckpointTextMaxJSONBytes
}

func completeAudioCheckpoint(completedSegments int, text string, maxJSONBytes int) AudioExtractCheckpoint {
	return AudioExtractCheckpoint{CompletedSegments: completedSegments, Text: boundCheckpointText(text, maxJSONBytes)}
}

func (b *Dat9Backend) checkpointAudioProgress(component string, rt AudioExtractRuntime, completedSegments int, combined *strings.Builder, tags map[string]string, structured bool) error {
	if rt.Checkpoint == nil {
		return nil
	}
	fullText := combined.String()
	checkpoint := completeAudioCheckpoint(completedSegments, fullText, effectiveAudioCheckpointTextBudget(rt))
	checkpoint.Tags = cloneMediaExtractTags(tags)
	checkpoint.Structured = structured
	if checkpoint.Text != fullText {
		// Advancing the cursor with only a truncated prefix would make the
		// omitted transcript unrecoverable after a retry. Keep the last safe
		// durable cursor and preserve the full same-attempt output in memory.
		b.recordTenantOperation(component, "checkpoint", "payload_oversize", 0)
		return nil
	}
	if err := rt.Checkpoint(checkpoint); err != nil {
		if errors.Is(err, datastore.ErrSemanticTaskPayloadTooLarge) {
			b.recordTenantOperation(component, "checkpoint", "payload_oversize", 0)
			return nil
		}
		return err
	}
	return nil
}

// boundCheckpointText retains the longest UTF-8 prefix whose JSON string
// encoding fits maxJSONBytes. Accounting for JSON escaping is necessary
// because task payload limits apply after json.Marshal.
func boundCheckpointText(text string, maxJSONBytes int) string {
	if maxJSONBytes < 2 {
		return ""
	}
	if encoded, err := json.Marshal(text); err == nil && len(encoded) <= maxJSONBytes {
		return text
	}
	runes := []rune(text)
	low, high := 0, len(runes)
	for low < high {
		mid := low + (high-low+1)/2
		encoded, err := json.Marshal(string(runes[:mid]))
		if err == nil && len(encoded) <= maxJSONBytes {
			low = mid
		} else {
			high = mid - 1
		}
	}
	return string(runes[:low])
}

func (b *Dat9Backend) extractAudioSegment(ctx context.Context, task AudioExtractTaskSpec, contentType string, data []byte, rt AudioExtractRuntime) (string, AudioExtractUsage, error) {
	taskCtx, cancel := context.WithTimeout(ctx, effectiveAudioExtractTimeout(rt.Timeout, b.audioExtractTimeout))
	defer cancel()
	return rt.Extractor.ExtractAudioText(taskCtx, AudioExtractRequest{
		TenantID: b.tenantID, FileID: task.FileID, Path: task.Path, ContentType: contentType, Data: data,
	})
}

func effectiveAudioExtractTimeout(runtimeTimeout, defaultTimeout time.Duration) time.Duration {
	if runtimeTimeout > 0 {
		return runtimeTimeout
	}
	if defaultTimeout > 0 {
		return defaultTimeout
	}
	return defaultAudioExtractTimeout
}

func splitAudioWithFFmpeg(ctx context.Context, sourcePath string, data []byte, contentType string, workspace *MediaExtractTaskWorkspace, maxSegmentBytes int64) ([]string, error) {
	if workspace == nil || workspace.Dir() == "" {
		return nil, fmt.Errorf("audio segmentation requires a task workspace")
	}
	tmpDir := filepath.Join(workspace.Dir(), "segments")
	if err := os.MkdirAll(tmpDir, 0o700); err != nil {
		return nil, err
	}
	inPath := sourcePath
	if inPath == "" {
		inPath = filepath.Join(tmpDir, "input"+audioMIMEToExt(contentType))
		if err := os.WriteFile(inPath, data, 0o600); err != nil {
			return nil, err
		}
	}
	outPattern := filepath.Join(tmpDir, "segment_%03d.mp3")
	cmd := exec.CommandContext(ctx, "ffmpeg", audioFFmpegArgs(inPath, outPattern, maxSegmentBytes)...)
	if err := cmd.Run(); err != nil {
		if ctx.Err() != nil {
			return nil, ctx.Err()
		}
		return nil, err
	}
	paths, err := filepath.Glob(filepath.Join(tmpDir, "segment_*.mp3"))
	if err != nil || len(paths) == 0 {
		return nil, fmt.Errorf("no audio segments produced")
	}
	sort.Strings(paths)
	if err := validateAudioSegments(paths); err != nil {
		return nil, err
	}
	return paths, nil
}

func audioFFmpegArgs(input, output string, maxSegmentBytes int64) []string {
	segmentSeconds := min(audioSegmentSeconds(maxSegmentBytes), int64(10*60))
	return []string{"-v", "error", "-i", input, "-vn", "-ac", "1", "-ar", "16000", "-c:a", "libmp3lame", "-b:a", "32k", "-f", "segment", "-segment_time", fmt.Sprint(segmentSeconds), "-reset_timestamps", "1", output}
}

func probeAudioDuration(ctx context.Context, sourcePath string, data []byte, contentType string) (float64, error) {
	if sourcePath != "" {
		return probeVideoDuration(ctx, "ffmpeg", sourcePath)
	}
	tmp, err := os.CreateTemp("", "drive9-audio-probe-*"+audioMIMEToExt(contentType))
	if err != nil {
		return 0, err
	}
	path := tmp.Name()
	defer func() { _ = os.Remove(path) }()
	if _, err = tmp.Write(data); err != nil {
		_ = tmp.Close()
		return 0, err
	}
	if err = tmp.Close(); err != nil {
		return 0, err
	}
	return probeVideoDuration(ctx, "ffmpeg", path)
}

func validateAudioSegments(paths []string) error {
	if len(paths) > maxAudioExtractSegments {
		return ErrMediaExtractDerivedTooLarge
	}
	var total int64
	for _, path := range paths {
		info, err := os.Stat(path)
		if err != nil {
			return err
		}
		total += info.Size()
		if total > maxAudioExtractDerivedBytes {
			return ErrMediaExtractDerivedTooLarge
		}
	}
	return nil
}

func audioSegmentSeconds(maxSegmentBytes int64) int64 {
	if maxSegmentBytes <= 4096+mp3ExtractBytesPerSecond {
		return 1
	}
	seconds := (maxSegmentBytes - 4096) / mp3ExtractBytesPerSecond
	if seconds > 600 {
		return 600
	}
	return max(seconds, int64(1))
}

func audioMIMEToExt(mime string) string {
	switch stripMIMEParams(mime) {
	case "audio/mpeg":
		return ".mp3"
	case "audio/wav", "audio/x-wav":
		return ".wav"
	case "audio/mp4", "audio/x-m4a":
		return ".m4a"
	default:
		return ".audio"
	}
}

func legacyAudioExtractMetricResult(result AudioExtractResult) string {
	switch result {
	case AudioExtractResultFileNotFound, AudioExtractResultGetFileError:
		return "get_file_error"
	case AudioExtractResultTooLarge:
		return "skip_too_large"
	case AudioExtractResultWritten:
		return "ok"
	default:
		return string(result)
	}
}

func injectedAudioExtractWritebackError(name string) error {
	var injected error
	failpoint.Inject(name, func(val failpoint.Value) {
		switch v := val.(type) {
		case string:
			injected = errors.New(v)
		case bool:
			if v {
				injected = fmt.Errorf("injected failpoint %s", name)
			}
		}
	})
	return injected
}

func (b *Dat9Backend) loadAudioBytesForExtract(ctx context.Context, f *datastore.File, maxSize int64) ([]byte, error) {
	if f.ContentLayout == datastore.ContentLayoutAppendLog {
		return nil, ErrAppendLogUnsupported
	}
	if maxSize > 0 && f.SizeBytes > maxSize {
		return nil, errAudioExtractSourceTooLarge
	}
	if f.StorageType == datastore.StorageDB9 {
		if maxSize > 0 && int64(len(f.ContentBlob)) > maxSize {
			return nil, errAudioExtractSourceTooLarge
		}
		return append([]byte(nil), f.ContentBlob...), nil
	}
	if f.StorageType != datastore.StorageS3 {
		return nil, fmt.Errorf("unsupported storage type: %s", f.StorageType)
	}
	if b.s3 == nil {
		return nil, fmt.Errorf("s3 client not configured")
	}
	rc, err := b.s3.GetObject(ctx, f.StorageRef)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rc.Close() }()

	reader := io.Reader(rc)
	if maxSize > 0 {
		reader = io.LimitReader(rc, maxSize+1)
	}
	data, err := io.ReadAll(reader)
	if err != nil {
		return nil, err
	}
	if maxSize > 0 && int64(len(data)) > maxSize {
		return nil, errAudioExtractSourceTooLarge
	}
	return data, nil
}