From b3198276f765c4b84f791d3c4bf486d0b1b4220b Mon Sep 17 00:00:00 2001 From: Quick <31828688+Quick104@users.noreply.github.com> Date: Thu, 18 Jun 2026 12:16:27 -0400 Subject: [PATCH] [codex] fix(subtitles): stream live transcribe_translate cues (#177) * fix(subtitles): stream live transcribe_translate cues * fix(subtitles): harden live AI transcription --- ...8-ai-subtitle-live-translate-interleave.md | 311 ++++++++++++ internal/api/router.go | 1 + internal/config/config.go | 5 +- internal/config/db_loader.go | 5 + internal/playback/audio_extract.go | 253 +++++++++- internal/subtitles/ai/config.go | 11 +- internal/subtitles/ai/service.go | 149 ++++-- internal/subtitles/ai/service_test.go | 442 +++++++++++++++++- internal/subtitles/ai/transcriber.go | 241 ++++++++-- internal/subtitles/ai/transcriber_test.go | 281 ++++++++++- 10 files changed, 1590 insertions(+), 109 deletions(-) create mode 100644 docs/superpowers/plans/2026-06-18-ai-subtitle-live-translate-interleave.md diff --git a/docs/superpowers/plans/2026-06-18-ai-subtitle-live-translate-interleave.md b/docs/superpowers/plans/2026-06-18-ai-subtitle-live-translate-interleave.md new file mode 100644 index 00000000..56d51a66 --- /dev/null +++ b/docs/superpowers/plans/2026-06-18-ai-subtitle-live-translate-interleave.md @@ -0,0 +1,311 @@ +# Plan: interleave `transcribe_translate` so live AI subtitles follow the playhead + +Issue: https://github.com/Silo-Server/silo-server/issues/154 +Area: `internal/subtitles/ai` (+ `internal/playback` extraction, `internal/config`) +Type: bug fix to an existing (shipped) capability — makes the chained AI-subtitle +path deliver its designed live behavior. No new client-facing contract. + +Commands assume the repository root is the cwd. + +## Problem (confirmed) + +`Service.runTranscribe` (`internal/subtitles/ai/service.go`) runs the chained +`transcribe_translate` kind as two **global, sequential** stages: + +1. Transcribe the **entire** file (progress 5%→70%). Transcript cues are not + streamed for the chained kind — + `streamTranscript := streaming && job.Kind == JobKindTranscribe` + (`service.go:417`), so it is `false` for `transcribe_translate`. +2. Only then translate **all** cues (70%→95%), streaming each translated batch + as it lands (`service.go:482-493`). + +Because translation cannot start until step 1 finishes for the whole file, the +first translated cue streams minutes into the job, long past the playhead. + +### Two distinct bottlenecks (both must be addressed) + +- **Transcription stage is global.** Translation waits for the whole-file ASR + pass. This is the dominant term for a feature film (~20 min ASR) and the + primary cause. Fixed by interleaving translation per chunk (Phase 1). +- **Extraction is global and precedes any callback.** `Transcribe` calls + `t.extract(...)` (`transcriber.go:124`) before the `onChunk` loop, and + `ExtractAudioChunks` blocks on a single whole-track ffmpeg pass + (`internal/playback/audio_extract.go:34,65`). So even with per-chunk + interleave and small chunks, the first cue still waits for full-file + extraction (~1–2 min for a 2h film). Fixed by playhead-first incremental + extraction (Phase 2). + +The transcriber already exposes the per-chunk seam: `Transcribe` invokes +`onChunk(cues, done, total)` per chunk, playhead-first (`transcriber.go:111-174`), +and plain `transcribe` already streams through it (`service.go:426-429`). + +Secondary defect found while confirming: `WhisperTranscriber.SetExtraction` +clamps any `chunkSeconds` outside `[60, 600]` to the **default 600** +(`transcriber.go:100-102`), so a small live chunk value silently becomes 10 min +instead of clamping to the floor. + +## Goal & honest latency targets + +Translated cues stream playhead-first for a session-attached +`transcribe_translate` job, instead of waiting for whole-file transcription. + +Time-to-first-translated-cue, by phase: + +- **Today:** `extract(full) + transcribe(full) + translate(first batch)` — many + minutes (≈ full ASR pass). +- **After Phase 1 (interleave, existing full extraction):** + `extract(full) + transcribe(first chunk) + translate(first chunk)` — removes + the whole-file ASR pass from the critical path; ≈ extraction time (~1–2 min for + a 2h film), then cues follow the playhead. Already usable for live after a + one-time wait. +- **After Phase 2 (playhead-first incremental extraction):** + `extract(one chunk near playhead) + transcribe + translate` — single-digit + seconds. This is what the issue means by "within seconds." + +Phase 1 delivers most of the value with low risk; Phase 2 closes the remaining +extraction latency. Ship Phase 1 first; do not claim "within seconds" until +Phase 2 lands. + +### Non-goals (this plan) + +- Pacing/throttling to the playhead (rolling window). Optional in the issue; + follow-up below. +- Cross-chunk translation-context continuity beyond batch context neighbours. + +## Decision: interleave only when streaming + +**Session-attached (live) jobs** use the interleaved per-chunk path. +**Background (non-streaming) `transcribe_translate` jobs keep the current +whole-file transcribe-then-translate-at-end path unchanged.** Rationale: the +issue is live-only; this preserves background output byte-for-byte (full +cross-file translation context, no incremental-extraction cost) and keeps the +two behaviors cleanly separated by the existing `streaming` boolean. The +shared transcript-storage step is factored so it is not duplicated. + +--- + +## Phase 1 — interleave translation per chunk (streaming jobs) + +### 1a. `internal/subtitles/ai/service.go` — `runTranscribe` + +For the chained kind **when `streaming`**, replace the kind-specific `onChunk` +body (`service.go:423-430`) with per-chunk translate-and-stream: + +- Capture a translate error in a closure: `var translateErr error` and an + accumulator `var translatedAccum []SubtitleCue`. +- In `onChunk(chunk, done, total)`: + - If `translateErr != nil` or `len(chunk) == 0`, return (skip silence; stop + translating after a prior failure). Do **not** abort transcription. + - Else call `s.translator.Translate` on **that chunk's** cues. On error: set + `translateErr`, emit `TranslationFailed` once, and return (let transcription + continue). On success: stream translated cues via `TranslationCues` and + append to `translatedAccum`. + - Progress: single 5%→95% band keyed on chunk `done/total` + (message "Transcribing & translating"). +- Keep `onChunk` signature unchanged (no error return). Aborting on translate + failure via the callback is intentionally avoided — see 1b. + +After `Transcribe` returns: + +- **Always store the transcript track** (unchanged, `service.go:447-466`, + including the transcript `SubtitleReady` broadcast). This must run regardless + of `translateErr`. +- If `translateErr != nil`: `finishWithError(ctx, job, translateErr)`. The + transcript is already persisted (cache preserved), and the live session + already received `TranslationFailed`. +- Else: assemble the translated track from `translatedAccum` (`sortCuesByStart` + + `StoreSubtitle`, `service.go:499-514`), `CompleteJob`, and + `TranslationCompleted` (`service.go:515-522`). +- Remove the whole-file translate-at-end (`service.go:479-497`) for the + streaming path only. + +Non-streaming chained jobs fall through to the existing whole-file path, +untouched. + +### 1b. Transcript-on-translation-failure (resolves review P1) + +Today the full transcript is stored before translation starts +(`service.go:447` then `:482`), so a translation failure still leaves a +transcript/cache row and an already-broadcast `SubtitleReady`. The interleaved +path preserves this exactly because: + +- a per-chunk translate failure sets `translateErr` but lets the ASR pass run to + completion (it does **not** return an error from `onChunk`), so `Transcribe` + returns the full cue set normally; +- the transcript is then stored on the same post-transcribe path as today, + before the job is failed for the translated track. + +An ASR failure or context cancellation still returns an error from `Transcribe` +and yields no transcript — identical to today. + +### 1c. Live progress semantics (resolves review P2) + +`SubtitleTranslationCuesPayload.Done/Total` is documented as overall progress +(`internal/playback/realtime.go:172-181`). For live `transcribe_translate`: + +- `TranslationCues` carries **`done = chunks completed, total = total chunk + count`** — the same chunk-granular convention plain `transcribe` already emits + (`service.go:427-428`). Monotonic and stable across chunks. +- `TranslationStarted.TotalCues = 0` (indeterminate), consistent with the + transcribe path today (`service.go:412`), since the cue total is unknown before + ASR completes. +- Cues carry absolute Start/End, so the client places them regardless of + arrival order; unpause is driven by cue arrival near the playhead, not by + `Done/Total`. Confirm with client teams that `Done/Total` is treated as a + fraction, not absolute cue indices (clients already receive chunk counts from + the plain `transcribe` path, so this should hold). + +### 1d. Small live chunks + clamp fix + +- Add `TranscribeJobRequest.ChunkSeconds int` (0 = configured default). In + `Transcribe`, use `req.ChunkSeconds` when `> 0`, else the atomic default + (`transcriber.go:121-122`); validate/clamp identically to `SetExtraction`. +- Fix the clamp in `SetExtraction` (`transcriber.go:99-105`): clamp + out-of-range to the nearest bound, not to the 600 default. +- Lower `minASRChunkSeconds` from 60 to 15 (`transcriber.go:26`); update the + doc comment about request count / boundary word-clip tradeoff. +- Config: add `subtitle_ai.live_asr_chunk_seconds` (default 30): + - `internal/config/config.go`: new field beside `ASRChunkSeconds` + (`config.go:274-277`). + - `internal/config/db_loader.go`: load it next to `asr_chunk_seconds` + (`db_loader.go:472-476`). + - `internal/subtitles/ai/config.go`: add `LiveASRChunkSeconds int` to the + service `Config` so the service can set `req.ChunkSeconds` for streaming jobs. + - `internal/api/router.go`: plumb through `effectiveSubtitleAIConfig` and the + `OnConfigChange` `UpdateConfig` (`router.go:1024-1055`). + +### 1e. Phase 1 tests + +- `service_test.go`: fake `Transcriber` emitting several chunks playhead-first + via `onChunk`; fake `Translator` recording calls. For a streaming + `transcribe_translate` job assert: one `Translate` per non-empty chunk; first + `TranslationCues` streamed before the last chunk is transcribed; empty chunks + skipped; final stored translated track = all translated cues sorted by start; + `Done/Total` is chunk-granular and monotonic. +- Failure case: a translate error on chunk N sets the failure but transcription + completes, the transcript track **is** stored, `TranslationFailed` is emitted, + and the job ends via `finishWithError`. +- Non-streaming `transcribe_translate`: unchanged whole-file path; correct final + track; no `TranslationCues`. +- `transcriber_test.go`: `SetExtraction` clamp fix and `ChunkSeconds` override + (incl. the new 15s floor). + +--- + +## Phase 2 — playhead-first incremental extraction (true seconds latency) + +Phase 1 still waits on full-file extraction before the first cue. Phase 2 +removes that by extracting and consuming chunks incrementally, seeked to the +playhead. + +### 2a. New incremental extractor in `internal/playback` + +Add a streaming, seek-based extractor alongside `ExtractAudioChunks`: + +``` +ExtractAudioChunksFrom(ctx, filePath, audioTrackIndex, dir, ffmpegPath, + startSec float64, chunkSeconds int, + onSegment func(AudioChunk) error) error +``` + +- Runs one ffmpeg pass with `-ss -i ... -f segment + -segment_list segments.csv` covering `startSec`→end. +- Consumes segments **as they complete**: poll the segment-list CSV (ffmpeg + appends `filename,start,end` when it closes each segment) on a short interval; + for each new row, invoke `onSegment(AudioChunk{Path, Start: startSec + csvStart})`. + Reconcile any trailing rows after `cmd.Wait()` so the final segment is not + missed. +- Honors `ctx` (CommandContext kills ffmpeg on cancel) and `onSegment` errors + (stop the pass). Caller owns `dir`; segments are deleted by the consumer after + ASR to cap disk at one extraction. +- Preserves exact per-segment starts from the CSV, so timing does not accumulate + drift within a pass (only the initial `-ss` seek is packet-accurate, sub-second, + and is further corrected by the existing `ProbeAudioStartOffset`). + +Rejected alternative — **per-chunk seek extraction** (`-ss s -t d` per chunk): +simpler control flow but O(n) ffmpeg spawns and packet-aligned per-chunk seeks +that can clip words at every seam. The single-pass streaming consumer keeps the +existing segment-muxer boundary quality and far fewer process spawns. + +### 2b. Transcriber incremental mode + +- Add `TranscribeJobRequest.Incremental bool`, set by the service for streaming + jobs. +- When `Incremental`, `Transcribe` drives playhead-first in two streaming passes + (reproducing today's `chunkOrderForPosition` order): pass A `startSec = + playhead → end`, then pass B `0 → playhead`. Each completed segment is fed + straight into the existing per-chunk ASR loop body (transcribe → `onChunk`), + which Phase 1 already wires to translate+stream. The full cue set is still + accumulated and returned for transcript storage. +- When not `Incremental` (background, and plain transcribe unless we opt it in), + keep the existing whole-file `ExtractAudioChunks` + `chunkOrderForPosition` + path. + +Note: timing composition (`-ss` reset PTS + segment CSV start + probe offset) +must be verified with a fixture; see tests. + +### 2c. Phase 2 tests + +- `audio_extract` test (or transcriber test with an injected extractor) covering + incremental emission order, exact-start mapping with a `startSec` offset, and + ctx-cancel mid-pass. +- Transcriber test asserting `Incremental` yields playhead-first chunk order + across the two passes and returns the complete cue set for transcript storage. +- Confirm first `onChunk` fires after ~one chunk of audio, not after full-file + extraction (timing/ordering assertion via a fake extractor, not wall-clock). + +--- + +## Multi-repo / client impact + +None expected to the contract. Reuses the existing notifier protocol +(`TranslationStarted` / `TranslationCues` / `TranslationCompleted` / +`TranslationFailed`, `notifier.go`) under the same translated-track `trackKey`. +Translated cues simply begin arriving far earlier, and `Done/Total` keeps the +chunk-granular meaning plain `transcribe` already uses (1c). Worth a smoke test +on one `silo-android` / `silo-apple` client to confirm the live track renders +cues arriving before the transcript track completes, and that `Done/Total` is +treated as a fraction. + +## Risks & tradeoffs + +- **Translation context at chunk boundaries (live only).** Per-chunk translation + resets the batch context-neighbour window at each ~30s boundary — accepted + quality tradeoff for live; background jobs are unchanged. +- **More/smaller ASR requests for live jobs.** 30s chunks mean more requests; + per-chunk timeout (`chunkSeconds * asrChunkTimeoutFactor`) and per-spawn + overhead scale; the 15s floor bounds the worst case. +- **Incremental extraction reliability (Phase 2).** Segment-list polling + the + `-ss` seek path need fixtures for boundary timing, the final-segment race, and + cancel/cleanup. Single ffmpeg pass per direction keeps cost low. +- **Boundary word-clips.** Smaller live chunks add more fixed-length boundaries + where a straddling word can clip — already a documented v1 limitation; the + silence-aligned follow-up would address it. + +## Out of scope / follow-ups + +- **Pace to playback (issue item 3):** rolling window ahead of the playhead to + bound work; separate change once interleave + incremental extraction land. +- **Cross-chunk translation context:** feed the previous chunk's tail source + cues as untranslated context into each chunk's translate call. +- **Silence-aligned chunk boundaries:** reduce boundary word-clips that smaller + live chunks make more frequent. + +## v1 scope note + +`docs/architecture/v1-scope.md` is **NOT LOCKED**; the issue carries +`v1-proposed`. This is a **bug fix** to an already-shipped capability (no new +user-facing capability, endpoint, or response field — only an internal config +key and corrected streaming behavior), so it proceeds under the "bug fixes +proceed normally" clause. Link the PR to the AI-subtitle capability item per the +v1 PR requirements. + +## Pre-push checklist + +- `make lint` +- `cd web && pnpm run lint && pnpm run format:check` (run only if web touched — + none expected) +- `make verify-local-paths` +- Go tests for `internal/subtitles/ai` and `internal/playback` in a + libvips-capable container (a bare-host `go test ./...` silently skips CGO + packages). diff --git a/internal/api/router.go b/internal/api/router.go index f770bda3..8a5d2cd4 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -2915,6 +2915,7 @@ func effectiveSubtitleAIConfig(cfg *config.Config) (subtitleai.Config, string) { ASRModel: cfg.AI.ASRModel, BatchSize: cfg.SubtitleAI.BatchSize, ContextNeighbors: cfg.SubtitleAI.ContextNeighbors, + LiveASRChunkSeconds: cfg.SubtitleAI.LiveASRChunkSeconds, TranscribeQuotaJobs: cfg.SubtitleAI.TranscribeQuotaJobs, TranscribeQuotaPeriod: cfg.SubtitleAI.TranscribeQuotaPeriod, }, disabledGateway diff --git a/internal/config/config.go b/internal/config/config.go index 403a78dd..690c1e8a 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -272,9 +272,12 @@ type SubtitleAIConfig struct { BatchSize int `yaml:"-"` ContextNeighbors int `yaml:"-"` // ASRChunkSeconds is the audio chunk length per transcription request - // (60..600). Shorter chunks bound Whisper timestamp drift on long files; + // (15..600). Shorter chunks bound Whisper timestamp drift on long files; // longer chunks mean fewer requests and fewer boundary word-clips. ASRChunkSeconds int `yaml:"-"` + // LiveASRChunkSeconds is the smaller per-request chunk length for + // session-attached live subtitle jobs (15..600). + LiveASRChunkSeconds int `yaml:"-"` // TranscribeQuotaJobs caps how many transcription jobs each non-admin user // may start per rolling quota period (0 = unlimited). TranscribeQuotaJobs int `yaml:"-"` diff --git a/internal/config/db_loader.go b/internal/config/db_loader.go index bd9b7b42..a4d5303b 100644 --- a/internal/config/db_loader.go +++ b/internal/config/db_loader.go @@ -474,6 +474,11 @@ func LoadFromDB(m map[string]string) (*Config, error) { return nil, err } cfg.SubtitleAI.ASRChunkSeconds = subtitleAIChunkSeconds + subtitleAILiveChunkSeconds, err := intOr(m, "subtitle_ai.live_asr_chunk_seconds", 30) + if err != nil { + return nil, err + } + cfg.SubtitleAI.LiveASRChunkSeconds = subtitleAILiveChunkSeconds // A bad quota row must not block startup: fall back to "no quota" / // the daily window with a warning instead of failing the config load. transcribeQuotaJobs, err := intOr(m, "subtitle_ai.transcribe_quota_jobs", 0) diff --git a/internal/playback/audio_extract.go b/internal/playback/audio_extract.go index b157f067..8397a12e 100644 --- a/internal/playback/audio_extract.go +++ b/internal/playback/audio_extract.go @@ -6,6 +6,7 @@ import ( "encoding/csv" "encoding/json" "fmt" + "io" "math" "os" "os/exec" @@ -13,6 +14,7 @@ import ( "sort" "strconv" "strings" + "time" ) // AudioChunk is one extracted piece of an audio track. Start is the chunk's @@ -40,25 +42,10 @@ func ExtractAudioChunks(ctx context.Context, filePath string, audioTrackIndex in } listPath := filepath.Join(dir, "segments.csv") - args := []string{ - "-i", filePath, - "-vn", "-sn", "-dn", - "-map", fmt.Sprintf("0:a:%d", audioTrackIndex), - "-ac", "1", - "-ar", "16000", - "-c:a", "pcm_s16le", - "-f", "segment", - "-segment_time", strconv.Itoa(chunkSeconds), - "-segment_list", listPath, - "-segment_list_type", "csv", - "-y", filepath.Join(dir, "chunk%05d.wav"), - } + args := audioChunkExtractionArgs(filePath, audioTrackIndex, listPath, + filepath.Join(dir, "chunk%05d.wav"), 0, chunkSeconds) - ffmpeg := "ffmpeg" - if ffmpegPath != "" { - ffmpeg = ffmpegPath - } - cmd := exec.CommandContext(ctx, ffmpeg, args...) + cmd := exec.CommandContext(ctx, audioExtractionFFmpegBinary(ffmpegPath), args...) var stderr bytes.Buffer cmd.Stderr = &stderr @@ -96,6 +83,126 @@ func ExtractAudioChunks(ctx context.Context, filePath string, audioTrackIndex in return chunks, nil } +// ExtractAudioChunksFrom extracts one audio track starting at startSec and calls +// onSegment as each WAV segment is closed by ffmpeg. onSegment is synchronous: +// if it blocks on ASR/translation, ffmpeg may continue extracting later chunks +// into dir, so callers should remove processed chunks promptly. The caller owns +// dir and cleanup. Segment starts are absolute media positions. +func ExtractAudioChunksFrom( + ctx context.Context, + filePath string, + audioTrackIndex int, + dir, ffmpegPath string, + startSec float64, + chunkSeconds int, + onSegment func(AudioChunk) error, +) error { + if chunkSeconds <= 0 { + chunkSeconds = 600 + } + if audioTrackIndex < 0 { + audioTrackIndex = 0 + } + if startSec < 0 { + startSec = 0 + } + + listPath := filepath.Join(dir, "segments.csv") + args := audioChunkExtractionArgs(filePath, audioTrackIndex, listPath, + filepath.Join(dir, "chunk%05d.wav"), startSec, chunkSeconds) + + cmdCtx, cancel := context.WithCancel(ctx) + defer cancel() + + cmd := exec.CommandContext(cmdCtx, audioExtractionFFmpegBinary(ffmpegPath), args...) + var stderr bytes.Buffer + cmd.Stderr = &stderr + + if err := cmd.Start(); err != nil { + return fmt.Errorf("start ffmpeg audio extraction: %w", err) + } + + waitCh := make(chan error, 1) + go func() { + waitCh <- cmd.Wait() + }() + + seen := map[string]bool{} + tailer := &segmentListTailer{} + emit := func(flush bool) error { + return emitNewSegmentListEntries(listPath, dir, startSec, seen, tailer, flush, onSegment) + } + + ticker := time.NewTicker(100 * time.Millisecond) + defer ticker.Stop() + + for { + select { + case waitErr := <-waitCh: + segmentErr := emit(true) + if segmentErr != nil { + return segmentErr + } + if waitErr != nil { + if ctx.Err() != nil { + return ctx.Err() + } + return fmt.Errorf("ffmpeg audio extraction failed: %w (stderr: %s)", + waitErr, truncateStderr(stderr.String())) + } + if len(seen) == 0 { + return fmt.Errorf("ffmpeg produced no audio chunks for track %d", audioTrackIndex) + } + return nil + case <-ticker.C: + if err := emit(false); err != nil { + cancel() + <-waitCh + return err + } + case <-ctx.Done(): + cancel() + <-waitCh + return ctx.Err() + } + } +} + +func audioChunkExtractionArgs( + filePath string, + audioTrackIndex int, + listPath string, + outPattern string, + startSec float64, + chunkSeconds int, +) []string { + args := []string{} + if startSec > 0 { + args = append(args, "-ss", strconv.FormatFloat(startSec, 'f', 3, 64)) + } + args = append(args, + "-i", filePath, + "-vn", "-sn", "-dn", + "-map", fmt.Sprintf("0:a:%d", audioTrackIndex), + "-ac", "1", + "-ar", "16000", + "-c:a", "pcm_s16le", + "-f", "segment", + "-segment_time", strconv.Itoa(chunkSeconds), + "-segment_list", listPath, + "-segment_list_type", "csv", + "-y", outPattern, + ) + return args +} + +func audioExtractionFFmpegBinary(ffmpegPath string) string { + if ffmpegPath != "" { + return ffmpegPath + } + return "ffmpeg" +} + // parseSegmentList reads ffmpeg's CSV segment list (filename,start,end per // line) into a filename → start map. Best effort: a missing or malformed list // yields an empty map and callers fall back to nominal chunk starts. @@ -123,6 +230,116 @@ func parseSegmentList(listPath string) map[string]float64 { return starts } +func emitNewSegmentListEntries( + listPath, dir string, + startOffset float64, + seen map[string]bool, + tailer *segmentListTailer, + flush bool, + onSegment func(AudioChunk) error, +) error { + entries := tailer.read(listPath, flush) + for _, entry := range entries { + key := filepath.Base(entry.name) + if seen[key] { + continue + } + seen[key] = true + path := entry.name + if !filepath.IsAbs(path) { + path = filepath.Join(dir, path) + } + if onSegment != nil { + if err := onSegment(AudioChunk{Path: path, Start: startOffset + entry.start}); err != nil { + return err + } + } + } + return nil +} + +type segmentListEntry struct { + name string + start float64 +} + +type segmentListTailer struct { + offset int64 + pending string +} + +func (t *segmentListTailer) read(listPath string, flush bool) []segmentListEntry { + f, err := os.Open(listPath) + if err != nil { + return nil + } + defer f.Close() + + info, err := f.Stat() + if err == nil && info.Size() < t.offset { + t.offset = 0 + t.pending = "" + } + if _, err := f.Seek(t.offset, io.SeekStart); err != nil { + return nil + } + data, err := io.ReadAll(f) + if err != nil { + return nil + } + t.offset += int64(len(data)) + + text := t.pending + string(data) + if text == "" { + return nil + } + lineEnd := strings.LastIndexByte(text, '\n') + if lineEnd < 0 { + if !flush { + t.pending = text + return nil + } + t.pending = "" + return parseSegmentListEntries(strings.NewReader(text)) + } + + complete := text[:lineEnd+1] + t.pending = text[lineEnd+1:] + if flush && t.pending != "" { + complete += t.pending + t.pending = "" + } + return parseSegmentListEntries(strings.NewReader(complete)) +} + +func parseSegmentListEntries(r io.Reader) []segmentListEntry { + reader := csv.NewReader(r) + var out []segmentListEntry + for { + row, err := reader.Read() + if err == io.EOF { + break + } + if err != nil { + // Keep completed rows; malformed rows are ignored best-effort. + break + } + if len(row) < 2 { + continue + } + start, err := strconv.ParseFloat(strings.TrimSpace(row[1]), 64) + if err != nil || start < 0 { + continue + } + name := strings.TrimSpace(row[0]) + if name == "" { + continue + } + out = append(out, segmentListEntry{name: name, start: start}) + } + return out +} + // maxPlausibleAudioDelay caps the probed audio/container start delta; beyond // this it is far more likely a probing artifact than a real stream delay. const maxPlausibleAudioDelay = 30.0 diff --git a/internal/subtitles/ai/config.go b/internal/subtitles/ai/config.go index f384084b..9283e7f5 100644 --- a/internal/subtitles/ai/config.go +++ b/internal/subtitles/ai/config.go @@ -12,11 +12,12 @@ type Config struct { TranslateEnabled bool // TranscribeEnabled gates Whisper ASR generation (transcribe / // transcribe_translate jobs). - TranscribeEnabled bool - ChatModel string // chat-completions model used for translation - ASRModel string // audio-transcription model used for ASR - BatchSize int // cues per translation request - ContextNeighbors int // preceding source cues sent as untranslated context + TranscribeEnabled bool + ChatModel string // chat-completions model used for translation + ASRModel string // audio-transcription model used for ASR + BatchSize int // cues per translation request + ContextNeighbors int // preceding source cues sent as untranslated context + LiveASRChunkSeconds int // smaller ASR chunks for session-attached live jobs // TranscribeQuotaJobs caps how many transcription jobs (transcribe / // transcribe_translate) each non-admin user may start per rolling quota // period; 0 means unlimited. diff --git a/internal/subtitles/ai/service.go b/internal/subtitles/ai/service.go index 476fbf61..112e8dae 100644 --- a/internal/subtitles/ai/service.go +++ b/internal/subtitles/ai/service.go @@ -309,7 +309,7 @@ func (s *Service) run(ctx context.Context, job *Job) { TargetLanguage: job.TargetLanguage, }, func(batch []SubtitleCue, done, total int) { // Map cue progress into the 5%..95% band; push the batch live. - _ = s.repo.UpdateProgress(ctx, job.ID, JobStatusRunning, 0.05+0.9*float64(done)/float64(total), "Translating") + _ = s.repo.UpdateProgress(ctx, job.ID, JobStatusRunning, progressInBand(0.05, 0.9, done, total), "Translating") if streaming { s.notifier.TranslationCues(ctx, job.SessionID, job.MediaFileID, job.ID, trackKey, toStreamCues(batch), done, total) @@ -325,34 +325,9 @@ func (s *Service) run(ctx context.Context, job *Job) { // Persist in chronological order regardless of the playhead-first order used // for translation/streaming. - sortCuesByStart(translated) - sub, err := s.store.StoreSubtitle(storeCtx, subtitles.StoreSubtitleRequest{ - MediaFileID: job.MediaFileID, - UserID: job.RequestedBy, - Provider: providerTranslated, - Language: job.TargetLanguage, - Format: subtitles.FormatSRT, - ReleaseName: releaseName, - Data: SerializeSRT(translated), - }) - if err != nil { - s.finishWithError(ctx, job, fmt.Errorf("store translated subtitle: %w", err)) + if !s.finishTranslatedTrack(ctx, storeCtx, job, translated, releaseName, trackKey, streaming) { return } - - if err := s.repo.CompleteJob(storeCtx, job.ID, sub.ID); err != nil { - s.logger.Warn("failed to mark subtitle ai job complete", "job", job.ID, "error", err) - } - - if s.notifier != nil { - if streaming { - s.notifier.TranslationCompleted(storeCtx, job.SessionID, job.MediaFileID, job.ID, trackKey, - sub.ID, job.TargetLanguage, releaseName) - } - // Broadcast to other sessions watching this file (a no-op for the - // streaming requester, which already tracks its live track). - s.notifier.SubtitleReady(storeCtx, job.MediaFileID, sub.ID, job.TargetLanguage, releaseName) - } } // runTranscribe executes transcribe / transcribe_translate jobs: extract the @@ -412,17 +387,37 @@ func (s *Service) runTranscribe(ctx context.Context, job *Job) { s.notifier.TranslationStarted(ctx, job.SessionID, job.MediaFileID, job.ID, trackKey, liveLang, liveLabel, 0) } - // Only a plain transcribe streams transcript cues — the chained kind - // streams the translated cues from its translation stage instead. + cfg := s.config() + + // Plain transcribe streams transcript cues. A live chained job translates + // each non-empty ASR chunk immediately and streams translated cues from the + // same callback, while background chained jobs keep the whole-file path. streamTranscript := streaming && job.Kind == JobKindTranscribe + streamTranslated := streaming && job.Kind == JobKindTranscribeTranslate + liveChunkSeconds := 0 + if streaming { + liveChunkSeconds = cfg.LiveASRChunkSeconds + } + liveTranslate := liveTranslateState{sourceLanguage: hint} cues, detected, err := s.transcriber.Transcribe(ctx, TranscribeJobRequest{ FilePath: file.FilePath, AudioTrackIndex: audioIdx, LanguageHint: hint, StartPosition: job.StartPosition, - }, func(chunk []SubtitleCue, done, total int) { + DurationSeconds: float64(file.Duration), + ChunkSeconds: liveChunkSeconds, + Incremental: streaming, + }, func(chunk []SubtitleCue, chunkLanguage string, done, total int) { + if liveTranslate.sourceLanguage == "" && chunkLanguage != "" { + liveTranslate.sourceLanguage = chunkLanguage + } + if streamTranslated { + s.translateLiveChunk(ctx, job, trackKey, chunk, done, total, &liveTranslate) + return + } + // Transcription occupies the 5%..70% progress band (chunk granularity). - _ = s.repo.UpdateProgress(ctx, job.ID, JobStatusRunning, 0.05+0.65*float64(done)/float64(total), "Transcribing") + _ = s.repo.UpdateProgress(ctx, job.ID, JobStatusRunning, progressInBand(0.05, 0.65, done, total), "Transcribing") if streamTranscript { s.notifier.TranslationCues(ctx, job.SessionID, job.MediaFileID, job.ID, trackKey, toStreamCues(chunk), done, total) @@ -476,6 +471,16 @@ func (s *Service) runTranscribe(ctx context.Context, job *Job) { return } + if streamTranslated { + if liveTranslate.err != nil { + s.finishWithErrorNotify(ctx, job, liveTranslate.err, !liveTranslate.failedNotified) + return + } + translatedLabel := translatedReleaseName(language, job.TargetLanguage) + s.finishTranslatedTrack(ctx, storeCtx, job, liveTranslate.cues, translatedLabel, trackKey, true) + return + } + // transcribe_translate: run the transcript through the regular translator. // Cues already arrive playhead-first from chunk ordering, so the viewer's // region translates (and streams) first here too. @@ -485,7 +490,7 @@ func (s *Service) runTranscribe(ctx context.Context, job *Job) { TargetLanguage: job.TargetLanguage, }, func(batch []SubtitleCue, done, total int) { // Translation occupies the 70%..95% band. - _ = s.repo.UpdateProgress(ctx, job.ID, JobStatusRunning, 0.7+0.25*float64(done)/float64(total), "Translating") + _ = s.repo.UpdateProgress(ctx, job.ID, JobStatusRunning, progressInBand(0.7, 0.25, done, total), "Translating") if streaming { s.notifier.TranslationCues(ctx, job.SessionID, job.MediaFileID, job.ID, trackKey, toStreamCues(batch), done, total) @@ -497,31 +502,86 @@ func (s *Service) runTranscribe(ctx context.Context, job *Job) { } storeCtx = context.WithoutCancel(ctx) - sortCuesByStart(translated) translatedLabel := translatedReleaseName(language, job.TargetLanguage) + s.finishTranslatedTrack(ctx, storeCtx, job, translated, translatedLabel, trackKey, streaming) +} + +type liveTranslateState struct { + sourceLanguage string + err error + failedNotified bool + cues []SubtitleCue +} + +func (s *Service) translateLiveChunk( + ctx context.Context, + job *Job, + trackKey string, + chunk []SubtitleCue, + done, total int, + state *liveTranslateState, +) { + _ = s.repo.UpdateProgress(ctx, job.ID, JobStatusRunning, + progressInBand(0.05, 0.9, done, total), "Transcribing & translating") + if state.err != nil || len(chunk) == 0 { + return + } + translatedChunk, err := s.translator.Translate(ctx, TranslateRequest{ + Cues: chunk, + SourceLanguage: state.sourceLanguage, + TargetLanguage: job.TargetLanguage, + }, func(batch []SubtitleCue, _, _ int) { + s.notifier.TranslationCues(ctx, job.SessionID, job.MediaFileID, job.ID, trackKey, + toStreamCues(batch), done, total) + }) + if err != nil { + state.err = err + if !state.failedNotified { + s.notifier.TranslationFailed(ctx, job.SessionID, job.MediaFileID, job.ID, + trackKey, llm.Truncate(err.Error(), 500)) + state.failedNotified = true + } + return + } + state.cues = append(state.cues, translatedChunk...) +} + +func (s *Service) finishTranslatedTrack( + ctx context.Context, + storeCtx context.Context, + job *Job, + cues []SubtitleCue, + releaseName string, + trackKey string, + streaming bool, +) bool { + sortCuesByStart(cues) sub, err := s.store.StoreSubtitle(storeCtx, subtitles.StoreSubtitleRequest{ MediaFileID: job.MediaFileID, UserID: job.RequestedBy, Provider: providerTranslated, Language: job.TargetLanguage, Format: subtitles.FormatSRT, - ReleaseName: translatedLabel, - Data: SerializeSRT(translated), + ReleaseName: releaseName, + Data: SerializeSRT(cues), }) if err != nil { s.finishWithError(ctx, job, fmt.Errorf("store translated subtitle: %w", err)) - return + return false } + if err := s.repo.CompleteJob(storeCtx, job.ID, sub.ID); err != nil { s.logger.Warn("failed to mark subtitle ai job complete", "job", job.ID, "error", err) } + if s.notifier != nil { if streaming { s.notifier.TranslationCompleted(storeCtx, job.SessionID, job.MediaFileID, job.ID, trackKey, - sub.ID, job.TargetLanguage, translatedLabel) + sub.ID, job.TargetLanguage, releaseName) } - s.notifier.SubtitleReady(storeCtx, job.MediaFileID, sub.ID, job.TargetLanguage, translatedLabel) + s.notifier.SubtitleReady(storeCtx, job.MediaFileID, sub.ID, job.TargetLanguage, releaseName) } + return true } // defaultAudioTrackIndex returns the index of the default-flagged audio @@ -544,6 +604,10 @@ func transcribedReleaseName(lang string) string { } func (s *Service) finishWithError(ctx context.Context, job *Job, err error) { + s.finishWithErrorNotify(ctx, job, err, true) +} + +func (s *Service) finishWithErrorNotify(ctx context.Context, job *Job, err error, notify bool) { status := JobStatusFailed msg := llm.Truncate(err.Error(), 500) // Only a genuine cancellation (user cancel, or shutdown via cancel) becomes @@ -555,7 +619,7 @@ func (s *Service) finishWithError(ctx context.Context, job *Job, err error) { if dbErr := s.repo.FailJob(context.WithoutCancel(ctx), job.ID, status, msg); dbErr != nil { s.logger.Warn("failed to record subtitle ai job failure", "job", job.ID, "error", dbErr) } - if job.SessionID != "" && s.notifier != nil { + if notify && job.SessionID != "" && s.notifier != nil { s.notifier.TranslationFailed(context.WithoutCancel(ctx), job.SessionID, job.MediaFileID, job.ID, liveTrackKey(job.ID), msg) } @@ -564,6 +628,13 @@ func (s *Service) finishWithError(ctx context.Context, job *Job, err error) { } } +func progressInBand(start, width float64, done, total int) float64 { + if total <= 0 { + return start + } + return start + width*float64(done)/float64(total) +} + // loadSource resolves the combined player subtitle index to translatable cues, // mirroring the index space used by the playback subtitle endpoints // (external → embedded → downloaded). Embedded tracks are extracted to SRT via diff --git a/internal/subtitles/ai/service_test.go b/internal/subtitles/ai/service_test.go index 243a8580..61f78597 100644 --- a/internal/subtitles/ai/service_test.go +++ b/internal/subtitles/ai/service_test.go @@ -3,12 +3,16 @@ package ai import ( "context" "errors" + "fmt" + "strings" "sync" "testing" "time" "github.com/Silo-Server/silo-server/internal/ai/jobrunner" "github.com/Silo-Server/silo-server/internal/models" + "github.com/Silo-Server/silo-server/internal/playback" + "github.com/Silo-Server/silo-server/internal/subtitles" ) // recordingRepo is a JobRepository that records ResetStaleJobs calls and @@ -21,6 +25,20 @@ type recordingRepo struct { inserts int quotaUsed int // returned by CountTranscribeJobsByUserSince lastJob Job + completed []int + failures []recordedFailure + progress []recordedProgress +} + +type recordedFailure struct { + id int64 + status JobStatus + message string +} + +type recordedProgress struct { + progress float64 + message string } // InsertJob mirrors the Postgres repo's atomic quota guard: a non-nil quota @@ -47,12 +65,25 @@ func (r *recordingRepo) CountTranscribeJobsByUserSince(context.Context, int, tim defer r.mu.Unlock() return r.quotaUsed, nil } -func (r *recordingRepo) UpdateProgress(context.Context, int64, JobStatus, float64, string) error { +func (r *recordingRepo) UpdateProgress(_ context.Context, _ int64, _ JobStatus, progress float64, message string) error { + r.mu.Lock() + defer r.mu.Unlock() + r.progress = append(r.progress, recordedProgress{progress: progress, message: message}) return nil } -func (r *recordingRepo) CompleteJob(context.Context, int64, int) error { return nil } -func (r *recordingRepo) FailJob(context.Context, int64, JobStatus, string) error { return nil } -func (r *recordingRepo) Heartbeat(context.Context, int64) error { return nil } +func (r *recordingRepo) CompleteJob(_ context.Context, _ int64, subtitleID int) error { + r.mu.Lock() + defer r.mu.Unlock() + r.completed = append(r.completed, subtitleID) + return nil +} +func (r *recordingRepo) FailJob(_ context.Context, id int64, status JobStatus, message string) error { + r.mu.Lock() + defer r.mu.Unlock() + r.failures = append(r.failures, recordedFailure{id: id, status: status, message: message}) + return nil +} +func (r *recordingRepo) Heartbeat(context.Context, int64) error { return nil } func (r *recordingRepo) ResetStaleJobs(_ context.Context, before time.Time, _ string) (int64, error) { r.mu.Lock() @@ -73,7 +104,7 @@ func (r *recordingRepo) snapshot() (int, time.Time) { type stubTranscriber struct{} func (stubTranscriber) Transcribe(context.Context, TranscribeJobRequest, - func([]SubtitleCue, int, int)) ([]SubtitleCue, string, error) { + TranscribeChunkCallback) ([]SubtitleCue, string, error) { return nil, "", nil } @@ -221,3 +252,404 @@ func TestRecoverReapsStaleJobsImmediately(t *testing.T) { t.Errorf("stale cutoff = %v, want ~%v (now-staleJobThreshold)", before, want) } } + +type runTranscribeFileResolver struct { + file *models.MediaFile +} + +func (r runTranscribeFileResolver) GetByID(context.Context, int) (*models.MediaFile, error) { + return r.file, nil +} + +type chunkedTranscriber struct { + chunks [][]SubtitleCue + detected string + trace *[]string + lastReq TranscribeJobRequest +} + +func (t *chunkedTranscriber) Transcribe(_ context.Context, req TranscribeJobRequest, + onChunk TranscribeChunkCallback) ([]SubtitleCue, string, error) { + t.lastReq = req + var all []SubtitleCue + for i, chunk := range t.chunks { + appendTrace(t.trace, fmt.Sprintf("transcribe:%d", i)) + if onChunk != nil { + total := len(t.chunks) + if req.Incremental { + total = progressTotal(i+1, estimatedChunkTotal(req.DurationSeconds, req.ChunkSeconds)) + } + onChunk(chunk, t.detected, i+1, total) + } + all = append(all, chunk...) + } + return all, t.detected, nil +} + +type recordingTranslator struct { + calls []TranslateRequest + errOnCall int + trace *[]string +} + +func (t *recordingTranslator) Translate(_ context.Context, req TranslateRequest, + onBatch func([]SubtitleCue, int, int)) ([]SubtitleCue, error) { + t.calls = append(t.calls, TranslateRequest{ + Cues: cloneTestCues(req.Cues), + SourceLanguage: req.SourceLanguage, + TargetLanguage: req.TargetLanguage, + }) + call := len(t.calls) + if t.errOnCall == call { + return nil, errors.New("translator chunk failed") + } + + out := cloneTestCues(req.Cues) + for i := range out { + for j, line := range out[i].Lines { + out[i].Lines[j] = req.TargetLanguage + ":" + line + } + } + if len(out) > 0 { + appendTrace(t.trace, "translate:"+strings.Join(out[0].Lines, " ")) + } + if onBatch != nil { + onBatch(out, len(out), len(out)) + } + return out, nil +} + +type recordingSubtitleStore struct { + stored []subtitles.StoreSubtitleRequest +} + +func (s *recordingSubtitleStore) StoreSubtitle(_ context.Context, req subtitles.StoreSubtitleRequest) (*subtitles.DownloadedSubtitle, error) { + s.stored = append(s.stored, req) + return &subtitles.DownloadedSubtitle{ + ID: len(s.stored), + MediaFileID: req.MediaFileID, + Provider: req.Provider, + Language: req.Language, + Format: req.Format, + ReleaseName: req.ReleaseName, + }, nil +} + +func (s *recordingSubtitleStore) GetSubtitleContent(context.Context, int) (*subtitles.DownloadedSubtitle, []byte, error) { + return nil, nil, errors.New("not implemented") +} + +type notifierEvent struct { + kind string + cues []playback.StreamCue + done int + total int + message string +} + +type recordingNotifier struct { + events []notifierEvent + trace *[]string +} + +func (n *recordingNotifier) SubtitleReady(_ context.Context, _ int, subtitleID int, language, _ string) { + n.events = append(n.events, notifierEvent{kind: fmt.Sprintf("ready:%d:%s", subtitleID, language)}) +} + +func (n *recordingNotifier) TranslationStarted(context.Context, string, int, int64, string, string, string, int) { + n.events = append(n.events, notifierEvent{kind: "started"}) +} + +func (n *recordingNotifier) TranslationCues(_ context.Context, _ string, _ int, _ int64, _ string, + cues []playback.StreamCue, done, total int) { + n.events = append(n.events, notifierEvent{kind: "cues", cues: cues, done: done, total: total}) + if len(cues) > 0 { + appendTrace(n.trace, "stream:"+cues[0].Text) + } +} + +func (n *recordingNotifier) TranslationCompleted(context.Context, string, int, int64, string, int, string, string) { + n.events = append(n.events, notifierEvent{kind: "completed"}) +} + +func (n *recordingNotifier) TranslationFailed(_ context.Context, _ string, _ int, _ int64, _, message string) { + n.events = append(n.events, notifierEvent{kind: "failed", message: message}) +} + +func TestRunTranscribeTranslateStreamingInterleavesTranslationPerChunk(t *testing.T) { + var trace []string + transcriber := &chunkedTranscriber{ + detected: "en", + trace: &trace, + chunks: [][]SubtitleCue{ + {testCue(60, "second")}, + {}, + {testCue(0, "first")}, + }, + } + translator := &recordingTranslator{trace: &trace} + store := &recordingSubtitleStore{} + notifier := &recordingNotifier{trace: &trace} + repo := &recordingRepo{} + svc := newRunTranscribeService(repo, translator, transcriber, store, notifier, Config{LiveASRChunkSeconds: 30}) + + svc.runTranscribe(context.Background(), &Job{ + ID: 7, + MediaFileID: 10, + Kind: JobKindTranscribeTranslate, + SourceIndex: -1, + TargetLanguage: "es", + SessionID: "session-1", + StartPosition: 60, + }) + + if !transcriber.lastReq.Incremental { + t.Fatalf("streaming request did not enable incremental extraction") + } + if transcriber.lastReq.ChunkSeconds != 30 { + t.Fatalf("chunk override = %d, want 30", transcriber.lastReq.ChunkSeconds) + } + if got := len(translator.calls); got != 2 { + t.Fatalf("Translate calls = %d, want one per non-empty chunk", got) + } + if strings.Join(translator.calls[0].Cues[0].Lines, " ") != "second" { + t.Errorf("first translated chunk = %q, want playhead chunk", translator.calls[0].Cues[0].Lines) + } + if streamIdx, lastChunkIdx := traceIndex(trace, "stream:es:second"), traceIndex(trace, "transcribe:2"); streamIdx < 0 || lastChunkIdx < 0 || streamIdx > lastChunkIdx { + t.Fatalf("first translated stream did not precede final ASR chunk: %v", trace) + } + + cueEvents := cueNotifierEvents(notifier.events) + if got := len(cueEvents); got != 2 { + t.Fatalf("TranslationCues events = %d, want 2", got) + } + if cueEvents[0].done != 1 || cueEvents[0].total != 6 || cueEvents[1].done != 3 || cueEvents[1].total != 6 { + t.Fatalf("chunk progress = (%d/%d, %d/%d), want 1/6 then 3/6", + cueEvents[0].done, cueEvents[0].total, cueEvents[1].done, cueEvents[1].total) + } + if len(store.stored) != 2 { + t.Fatalf("stored subtitles = %d, want transcript and translation", len(store.stored)) + } + if store.stored[0].Provider != providerTranscribed || store.stored[1].Provider != providerTranslated { + t.Fatalf("stored providers = %q, %q", store.stored[0].Provider, store.stored[1].Provider) + } + translated, err := ParseCues(store.stored[1].Data) + if err != nil { + t.Fatalf("parse stored translation: %v", err) + } + if len(translated) != 2 || strings.Join(translated[0].Lines, " ") != "es:first" || + strings.Join(translated[1].Lines, " ") != "es:second" { + t.Fatalf("stored translation not sorted/complete: %#v", translated) + } + if got := fmt.Sprint(repo.completed); got != "[2]" { + t.Fatalf("completed subtitle IDs = %s, want [2]", got) + } +} + +func TestRunTranscribeTranslateStreamingFailureStoresTranscript(t *testing.T) { + var trace []string + transcriber := &chunkedTranscriber{ + detected: "en", + trace: &trace, + chunks: [][]SubtitleCue{ + {testCue(0, "one")}, + {testCue(30, "two")}, + {testCue(60, "three")}, + }, + } + translator := &recordingTranslator{errOnCall: 2, trace: &trace} + store := &recordingSubtitleStore{} + notifier := &recordingNotifier{trace: &trace} + repo := &recordingRepo{} + svc := newRunTranscribeService(repo, translator, transcriber, store, notifier, Config{LiveASRChunkSeconds: 30}) + + svc.runTranscribe(context.Background(), &Job{ + ID: 8, + MediaFileID: 10, + Kind: JobKindTranscribeTranslate, + SourceIndex: -1, + TargetLanguage: "es", + SessionID: "session-1", + }) + + if got := len(translator.calls); got != 2 { + t.Fatalf("Translate calls = %d, want calls stop after failure", got) + } + if traceIndex(trace, "transcribe:2") < 0 { + t.Fatalf("ASR did not continue through final chunk after translate failure: %v", trace) + } + if len(store.stored) != 1 || store.stored[0].Provider != providerTranscribed { + t.Fatalf("stored subtitles = %#v, want transcript only", store.stored) + } + if len(repo.failures) != 1 || repo.failures[0].status != JobStatusFailed { + t.Fatalf("failures = %#v, want one failed job", repo.failures) + } + if got := countNotifierKind(notifier.events, "failed"); got != 1 { + t.Fatalf("TranslationFailed events = %d, want 1", got) + } + if len(repo.completed) != 0 { + t.Fatalf("completed despite translation failure: %v", repo.completed) + } +} + +func TestRunTranscribeTranslateNonStreamingKeepsWholeFileTranslation(t *testing.T) { + transcriber := &chunkedTranscriber{ + detected: "en", + chunks: [][]SubtitleCue{ + {testCue(30, "later")}, + {testCue(0, "earlier")}, + }, + } + translator := &recordingTranslator{} + store := &recordingSubtitleStore{} + notifier := &recordingNotifier{} + repo := &recordingRepo{} + svc := newRunTranscribeService(repo, translator, transcriber, store, notifier, Config{LiveASRChunkSeconds: 30}) + + svc.runTranscribe(context.Background(), &Job{ + ID: 9, + MediaFileID: 10, + Kind: JobKindTranscribeTranslate, + SourceIndex: -1, + TargetLanguage: "es", + }) + + if transcriber.lastReq.Incremental { + t.Fatalf("background request unexpectedly enabled incremental extraction") + } + if transcriber.lastReq.ChunkSeconds != 0 { + t.Fatalf("background chunk override = %d, want 0", transcriber.lastReq.ChunkSeconds) + } + if got := len(translator.calls); got != 1 { + t.Fatalf("Translate calls = %d, want one whole-file call", got) + } + if got := len(translator.calls[0].Cues); got != 2 { + t.Fatalf("whole-file call cues = %d, want 2", got) + } + if got := countNotifierKind(notifier.events, "cues"); got != 0 { + t.Fatalf("non-streaming TranslationCues events = %d, want 0", got) + } + if len(store.stored) != 2 || store.stored[1].Provider != providerTranslated { + t.Fatalf("stored subtitles = %#v, want transcript and translated track", store.stored) + } +} + +func TestRunTranscribeTranslateStreamingUsesDetectedLanguageForLiveChunks(t *testing.T) { + transcriber := &chunkedTranscriber{ + detected: "ja", + chunks: [][]SubtitleCue{ + {testCue(0, "konnichiwa")}, + }, + } + translator := &recordingTranslator{} + store := &recordingSubtitleStore{} + notifier := &recordingNotifier{} + repo := &recordingRepo{} + file := &models.MediaFile{ + ID: 10, + FilePath: "/media/movie.mkv", + Duration: 180, + AudioTracks: []models.AudioTrack{{Default: true}}, + } + svc := newRunTranscribeServiceWithFile(repo, translator, transcriber, store, notifier, + Config{LiveASRChunkSeconds: 30}, file) + + svc.runTranscribe(context.Background(), &Job{ + ID: 10, + MediaFileID: 10, + Kind: JobKindTranscribeTranslate, + SourceIndex: -1, + TargetLanguage: "es", + SessionID: "session-1", + }) + + if got := len(translator.calls); got != 1 { + t.Fatalf("Translate calls = %d, want one live chunk call", got) + } + if got := translator.calls[0].SourceLanguage; got != "ja" { + t.Fatalf("live chunk source language = %q, want detected ja", got) + } +} + +func newRunTranscribeService( + repo *recordingRepo, + translator Translator, + transcriber Transcriber, + store *recordingSubtitleStore, + notifier Notifier, + cfg Config, +) *Service { + file := &models.MediaFile{ + ID: 10, + FilePath: "/media/movie.mkv", + Duration: 180, + AudioTracks: []models.AudioTrack{{Language: "eng", Default: true}}, + } + return newRunTranscribeServiceWithFile(repo, translator, transcriber, store, notifier, cfg, file) +} + +func newRunTranscribeServiceWithFile( + repo *recordingRepo, + translator Translator, + transcriber Transcriber, + store *recordingSubtitleStore, + notifier Notifier, + cfg Config, + file *models.MediaFile, +) *Service { + return NewService(context.Background(), cfg, repo, translator, transcriber, store, nil, + runTranscribeFileResolver{file: file}, notifier, "", nil, nil) +} + +func testCue(startSeconds float64, text string) SubtitleCue { + start := time.Duration(startSeconds * float64(time.Second)) + return SubtitleCue{ + Start: start, + End: start + 2*time.Second, + Lines: []string{text}, + } +} + +func cloneTestCues(cues []SubtitleCue) []SubtitleCue { + out := make([]SubtitleCue, len(cues)) + for i, cue := range cues { + out[i] = cue + out[i].Lines = append([]string(nil), cue.Lines...) + } + return out +} + +func appendTrace(trace *[]string, value string) { + if trace != nil { + *trace = append(*trace, value) + } +} + +func traceIndex(trace []string, value string) int { + for i, got := range trace { + if got == value { + return i + } + } + return -1 +} + +func cueNotifierEvents(events []notifierEvent) []notifierEvent { + var out []notifierEvent + for _, event := range events { + if event.kind == "cues" { + out = append(out, event) + } + } + return out +} + +func countNotifierKind(events []notifierEvent, kind string) int { + count := 0 + for _, event := range events { + if event.kind == kind { + count++ + } + } + return count +} diff --git a/internal/subtitles/ai/transcriber.go b/internal/subtitles/ai/transcriber.go index 025d1f3a..cfea3af9 100644 --- a/internal/subtitles/ai/transcriber.go +++ b/internal/subtitles/ai/transcriber.go @@ -2,7 +2,9 @@ package ai import ( "context" + "errors" "fmt" + "math" "os" "path/filepath" "strings" @@ -21,9 +23,11 @@ const ( // 10 minutes of 16 kHz mono WAV is ~19 MB — under typical 25 MB API // limits. Shorter chunks bound Whisper's within-chunk timestamp drift at // the cost of more requests and more boundary word-clips; the operator - // tunes this via subtitle_ai.asr_chunk_seconds. + // tunes this via subtitle_ai.asr_chunk_seconds. Live jobs may use smaller + // chunks through subtitle_ai.live_asr_chunk_seconds to reduce time to first + // cue; the 15s floor bounds request count and boundary word-clips. defaultASRChunkSeconds = 600 - minASRChunkSeconds = 60 + minASRChunkSeconds = 15 // Per-chunk request timeout: 3× realtime accommodates local Whisper // servers on modest hardware. asrChunkTimeoutFactor = 3 @@ -46,6 +50,9 @@ type TranscribeJobRequest struct { AudioTrackIndex int // resolved 0-based audio stream index LanguageHint string // ISO 639-1; "" lets the model detect StartPosition float64 // seconds; chunks are processed playhead-first + DurationSeconds float64 // optional media duration for incremental progress + ChunkSeconds int // optional per-job override; 0 uses configured default + Incremental bool // stream extraction chunks instead of extracting whole file first } // Transcriber converts an audio track into subtitle cues. The built-in @@ -55,9 +62,14 @@ type TranscribeJobRequest struct { // callers can report progress and stream cues live. type Transcriber interface { Transcribe(ctx context.Context, req TranscribeJobRequest, - onChunk func(cues []SubtitleCue, done, total int)) ([]SubtitleCue, string, error) + onChunk TranscribeChunkCallback) ([]SubtitleCue, string, error) } +// TranscribeChunkCallback receives each chunk's cues, the detected language for +// that chunk when the ASR endpoint reports one, and chunk progress. A total of +// 0 means the total is unknown/indeterminate. +type TranscribeChunkCallback func(cues []SubtitleCue, language string, done, total int) + // audioTranscriber is the slice of llm.Client the transcriber needs. type audioTranscriber interface { Transcribe(ctx context.Context, req llm.TranscribeRequest) (*llm.Transcription, error) @@ -76,18 +88,20 @@ type WhisperTranscriber struct { ffmpegPath atomic.Pointer[string] chunkSeconds atomic.Int32 // extract and probeOffset are playback helpers, injectable for tests. - extract func(ctx context.Context, filePath string, audioTrackIndex int, dir, ffmpegPath string, chunkSeconds int) ([]playback.AudioChunk, error) - probeOffset func(ctx context.Context, filePath string, audioTrackIndex int, ffmpegPath string) float64 + extract func(ctx context.Context, filePath string, audioTrackIndex int, dir, ffmpegPath string, chunkSeconds int) ([]playback.AudioChunk, error) + incrementalExtract func(ctx context.Context, filePath string, audioTrackIndex int, dir, ffmpegPath string, startSec float64, chunkSeconds int, onSegment func(playback.AudioChunk) error) error + probeOffset func(ctx context.Context, filePath string, audioTrackIndex int, ffmpegPath string) float64 } // NewWhisperTranscriber builds a transcriber backed by the shared AI client. -// chunkSeconds outside [minASRChunkSeconds, defaultASRChunkSeconds] falls -// back to the default (longer chunks would exceed upload limits). +// chunkSeconds outside [minASRChunkSeconds, defaultASRChunkSeconds] clamps +// to the nearest bound (longer chunks would exceed upload limits). func NewWhisperTranscriber(client *llm.Client, ffmpegPath string, chunkSeconds int) *WhisperTranscriber { t := &WhisperTranscriber{ - client: client, - extract: playback.ExtractAudioChunks, - probeOffset: playback.ProbeAudioStartOffset, + client: client, + extract: playback.ExtractAudioChunks, + incrementalExtract: playback.ExtractAudioChunksFrom, + probeOffset: playback.ProbeAudioStartOffset, } t.SetExtraction(ffmpegPath, chunkSeconds) return t @@ -95,13 +109,10 @@ func NewWhisperTranscriber(client *llm.Client, ffmpegPath string, chunkSeconds i // SetExtraction updates the ffmpeg path and ASR chunk duration used by jobs // started afterwards. Safe for concurrent use; out-of-range chunk durations -// fall back to the default, matching construction. +// clamp to the nearest supported bound, matching per-request overrides. func (t *WhisperTranscriber) SetExtraction(ffmpegPath string, chunkSeconds int) { - if chunkSeconds < minASRChunkSeconds || chunkSeconds > defaultASRChunkSeconds { - chunkSeconds = defaultASRChunkSeconds - } t.ffmpegPath.Store(&ffmpegPath) - t.chunkSeconds.Store(int32(chunkSeconds)) + t.chunkSeconds.Store(int32(clampASRChunkSeconds(chunkSeconds))) } // Transcribe implements Transcriber. The returned cues are NOT sorted (they @@ -109,7 +120,7 @@ func (t *WhisperTranscriber) SetExtraction(ffmpegPath string, chunkSeconds int) // endpoint's report for the first processed chunk, normalized to an ISO code // where possible. func (t *WhisperTranscriber) Transcribe(ctx context.Context, req TranscribeJobRequest, - onChunk func(cues []SubtitleCue, done, total int)) ([]SubtitleCue, string, error) { + onChunk TranscribeChunkCallback) ([]SubtitleCue, string, error) { dir, err := os.MkdirTemp("", "silo-asr-*") if err != nil { return nil, "", fmt.Errorf("create ASR temp dir: %w", err) @@ -120,50 +131,41 @@ func (t *WhisperTranscriber) Transcribe(ctx context.Context, req TranscribeJobRe // even if the config reloads mid-job. ffmpegPath := *t.ffmpegPath.Load() chunkSeconds := int(t.chunkSeconds.Load()) - - chunks, err := t.extract(ctx, req.FilePath, req.AudioTrackIndex, dir, ffmpegPath, chunkSeconds) - if err != nil { - return nil, "", err + if req.ChunkSeconds > 0 { + chunkSeconds = req.ChunkSeconds } + chunkSeconds = clampASRChunkSeconds(chunkSeconds) // Audio streams can start after the container's timeline origin (TS // remuxes especially); Whisper times are relative to the first audio // sample, so the delta is a constant sync error unless added back. startOffset := t.probeOffset(ctx, req.FilePath, req.AudioTrackIndex, ffmpegPath) + if req.Incremental { + return t.transcribeIncremental(ctx, req, dir, ffmpegPath, chunkSeconds, startOffset, onChunk) + } + + chunks, err := t.extract(ctx, req.FilePath, req.AudioTrackIndex, dir, ffmpegPath, chunkSeconds) + if err != nil { + return nil, "", err + } + order := chunkOrderForPosition(chunks, req.StartPosition) timeout := time.Duration(chunkSeconds*asrChunkTimeoutFactor) * time.Second var all []SubtitleCue detected := "" for done, idx := range order { - if err := ctx.Err(); err != nil { - return nil, "", err - } - data, err := os.ReadFile(chunks[idx].Path) - if err != nil { - return nil, "", fmt.Errorf("read audio chunk: %w", err) - } - tr, err := t.client.Transcribe(ctx, llm.TranscribeRequest{ - Filename: filepath.Base(chunks[idx].Path), - Audio: data, - Language: req.LanguageHint, - Timeout: timeout, - }) + cues, lang, err := t.transcribeChunk(ctx, chunks[idx], req.LanguageHint, timeout, startOffset) if err != nil { return nil, "", fmt.Errorf("transcribe chunk %d/%d: %w", idx+1, len(chunks), err) } - // Each chunk is read exactly once; deleting it as we go caps disk usage - // at one extraction rather than extraction + retranscription leftovers. - _ = os.Remove(chunks[idx].Path) - if detected == "" { - detected = normalizeDetectedLanguage(tr.Language) + detected = lang } - cues := cuesFromSegments(tr.Segments, chunks[idx].Start+startOffset) all = append(all, cues...) if onChunk != nil { - onChunk(cues, done+1, len(order)) + onChunk(cues, lang, done+1, len(order)) } } @@ -173,6 +175,165 @@ func (t *WhisperTranscriber) Transcribe(ctx context.Context, req TranscribeJobRe return all, detected, nil } +var errStopIncrementalPass = errors.New("stop incremental ASR pass") + +func (t *WhisperTranscriber) transcribeIncremental( + ctx context.Context, + req TranscribeJobRequest, + dir, ffmpegPath string, + chunkSeconds int, + startOffset float64, + onChunk TranscribeChunkCallback, +) ([]SubtitleCue, string, error) { + timeout := time.Duration(chunkSeconds*asrChunkTimeoutFactor) * time.Second + pivot := incrementalPivotStart(req.StartPosition, req.DurationSeconds, chunkSeconds) + total := estimatedChunkTotal(req.DurationSeconds, chunkSeconds) + + var all []SubtitleCue + detected := "" + done := 0 + process := func(chunk playback.AudioChunk, chunkOffset float64) error { + cues, lang, err := t.transcribeChunk(ctx, chunk, req.LanguageHint, timeout, chunkOffset) + if err != nil { + return fmt.Errorf("transcribe chunk at %.3fs: %w", chunk.Start, err) + } + if detected == "" { + detected = lang + } + all = append(all, cues...) + done++ + if onChunk != nil { + onChunk(cues, lang, done, progressTotal(done, total)) + } + return nil + } + + firstPassDone := done + firstPassOffset := audioOffsetForIncrementalPass(pivot, startOffset) + if err := t.runIncrementalPass(ctx, req, dir, ffmpegPath, pivot, chunkSeconds, func(chunk playback.AudioChunk) error { + return process(chunk, firstPassOffset) + }); err != nil { + if ctx.Err() != nil || pivot == 0 || done != firstPassDone { + return nil, "", err + } + } + if pivot > 0 { + wrap := func(chunk playback.AudioChunk) error { + if chunk.Start >= pivot { + return errStopIncrementalPass + } + return process(chunk, startOffset) + } + if err := t.runIncrementalPass(ctx, req, dir, ffmpegPath, 0, chunkSeconds, wrap); err != nil { + if !errors.Is(err, errStopIncrementalPass) { + return nil, "", err + } + } + } + + if len(all) == 0 { + return nil, detected, fmt.Errorf("no speech recognized in the audio track") + } + return all, detected, nil +} + +func (t *WhisperTranscriber) runIncrementalPass( + ctx context.Context, + req TranscribeJobRequest, + dir, ffmpegPath string, + startSec float64, + chunkSeconds int, + onSegment func(playback.AudioChunk) error, +) error { + passDir := filepath.Join(dir, fmt.Sprintf("from-%.3f", startSec)) + if err := os.MkdirAll(passDir, 0o755); err != nil { + return fmt.Errorf("create ASR incremental temp dir: %w", err) + } + return t.incrementalExtract(ctx, req.FilePath, req.AudioTrackIndex, passDir, ffmpegPath, startSec, chunkSeconds, onSegment) +} + +func (t *WhisperTranscriber) transcribeChunk( + ctx context.Context, + chunk playback.AudioChunk, + languageHint string, + timeout time.Duration, + startOffset float64, +) ([]SubtitleCue, string, error) { + if err := ctx.Err(); err != nil { + return nil, "", err + } + data, err := os.ReadFile(chunk.Path) + if err != nil { + return nil, "", fmt.Errorf("read audio chunk: %w", err) + } + tr, err := t.client.Transcribe(ctx, llm.TranscribeRequest{ + Filename: filepath.Base(chunk.Path), + Audio: data, + Language: languageHint, + Timeout: timeout, + }) + if err != nil { + return nil, "", err + } + // Each chunk is read exactly once; deleting it as we go caps disk usage + // at one extraction rather than extraction + retranscription leftovers. + _ = os.Remove(chunk.Path) + + return cuesFromSegments(tr.Segments, chunk.Start+startOffset), normalizeDetectedLanguage(tr.Language), nil +} + +func clampASRChunkSeconds(chunkSeconds int) int { + switch { + case chunkSeconds < minASRChunkSeconds: + return minASRChunkSeconds + case chunkSeconds > defaultASRChunkSeconds: + return defaultASRChunkSeconds + default: + return chunkSeconds + } +} + +func incrementalPivotStart(startSeconds, durationSeconds float64, chunkSeconds int) float64 { + if startSeconds <= 0 || chunkSeconds <= 0 { + return 0 + } + if durationSeconds > 0 && startSeconds >= durationSeconds { + startSeconds = math.Max(0, durationSeconds-0.001) + } + pivot := math.Floor(startSeconds/float64(chunkSeconds)) * float64(chunkSeconds) + if pivot < 0 { + return 0 + } + return pivot +} + +func estimatedChunkTotal(durationSeconds float64, chunkSeconds int) int { + if durationSeconds <= 0 || chunkSeconds <= 0 { + return 0 + } + return int(math.Ceil(durationSeconds / float64(chunkSeconds))) +} + +func progressTotal(done, total int) int { + if total <= 0 { + return 0 + } + if total < done { + return done + } + return total +} + +func audioOffsetForIncrementalPass(passStartSec, audioStartOffset float64) float64 { + if passStartSec <= 0 { + return audioStartOffset + } + if audioStartOffset <= passStartSec { + return 0 + } + return audioStartOffset - passStartSec +} + // chunkOrderForPosition orders chunk indexes so the chunk containing // startSeconds is processed first, then forward, then wrapping to the start — // the viewer's current region fills first, mirroring translation's diff --git a/internal/subtitles/ai/transcriber_test.go b/internal/subtitles/ai/transcriber_test.go index 528e0512..721dde11 100644 --- a/internal/subtitles/ai/transcriber_test.go +++ b/internal/subtitles/ai/transcriber_test.go @@ -117,7 +117,7 @@ func TestTranscribeProcessesChunksPlayheadFirst(t *testing.T) { var chunkOrder []string cues, _, err := tr.Transcribe(context.Background(), TranscribeJobRequest{ FilePath: "/x.mkv", StartPosition: 1300, // inside chunk 2 - }, func(chunk []SubtitleCue, done, total int) { + }, func(chunk []SubtitleCue, _ string, done, total int) { if total != 3 { t.Errorf("total = %d, want 3", total) } @@ -157,6 +157,67 @@ func TestTranscribePassesHintAndChunkSizedTimeout(t *testing.T) { } } +func TestSetExtractionClampsChunkSecondsToNearestBound(t *testing.T) { + tr := &WhisperTranscriber{} + + tr.SetExtraction("", 1) + if got := int(tr.chunkSeconds.Load()); got != minASRChunkSeconds { + t.Fatalf("low clamp = %d, want %d", got, minASRChunkSeconds) + } + + tr.SetExtraction("", defaultASRChunkSeconds+1) + if got := int(tr.chunkSeconds.Load()); got != defaultASRChunkSeconds { + t.Fatalf("high clamp = %d, want %d", got, defaultASRChunkSeconds) + } + + tr.SetExtraction("", 45) + if got := int(tr.chunkSeconds.Load()); got != 45 { + t.Fatalf("in-range chunk seconds = %d, want 45", got) + } +} + +func TestTranscribeUsesRequestChunkSecondsOverride(t *testing.T) { + var dir string + var extractedChunkSeconds int + client := &fakeASRClient{ + language: "en", + perChunk: map[string][]llm.TranscriptionSegment{ + "chunk00000.wav": {{Start: 0, End: 1, Text: "hi"}}, + }, + } + tr := &WhisperTranscriber{ + client: client, + extract: func(_ context.Context, _ string, _ int, d, _ string, chunkSeconds int) ([]playback.AudioChunk, error) { + dir = d + extractedChunkSeconds = chunkSeconds + p := filepath.Join(d, "chunk00000.wav") + if err := os.WriteFile(p, []byte("RIFF"), 0o644); err != nil { + return nil, err + } + return []playback.AudioChunk{{Path: p, Start: 0}}, nil + }, + probeOffset: func(context.Context, string, int, string) float64 { return 0 }, + } + tr.SetExtraction("", 600) + + _, _, err := tr.Transcribe(context.Background(), TranscribeJobRequest{ + FilePath: "/x.mkv", + ChunkSeconds: 10, + }, nil) + if err != nil { + t.Fatalf("Transcribe: %v", err) + } + if extractedChunkSeconds != minASRChunkSeconds { + t.Fatalf("extract chunk seconds = %d, want clamped override %d", extractedChunkSeconds, minASRChunkSeconds) + } + if want := time.Duration(minASRChunkSeconds*asrChunkTimeoutFactor) * time.Second; client.requests[0].Timeout != want { + t.Fatalf("timeout = %v, want %v", client.requests[0].Timeout, want) + } + if _, statErr := os.Stat(dir); !os.IsNotExist(statErr) { + t.Errorf("temp dir %s not cleaned up", dir) + } +} + func TestTranscribeAllSilentChunksFailsClearly(t *testing.T) { var dir string client := &fakeASRClient{language: "en", perChunk: map[string][]llm.TranscriptionSegment{}} @@ -326,6 +387,215 @@ func TestChunkOrderForPosition(t *testing.T) { } } +func TestTranscribeIncrementalStreamsPlayheadFirstAcrossTwoPasses(t *testing.T) { + var trace []string + var starts []float64 + client := &fakeASRClient{ + language: "en", + perChunk: map[string][]llm.TranscriptionSegment{ + "inc1200.wav": {{Start: 0, End: 1, Text: "pivot"}}, + "inc1800.wav": {{Start: 0, End: 1, Text: "tail"}}, + "inc0.wav": {{Start: 0, End: 1, Text: "head"}}, + "inc600.wav": {{Start: 0, End: 1, Text: "middle"}}, + }, + } + tr := &WhisperTranscriber{ + client: client, + incrementalExtract: func(_ context.Context, _ string, _ int, dir, _ string, startSec float64, _ int, + onSegment func(playback.AudioChunk) error) error { + starts = append(starts, startSec) + segmentStarts := []float64{1200, 1800} + if startSec == 0 { + segmentStarts = []float64{0, 600, 1200} + } + for _, segStart := range segmentStarts { + name := fmt.Sprintf("inc%.0f.wav", segStart) + p := filepath.Join(dir, name) + if err := os.WriteFile(p, []byte("RIFF"), 0o644); err != nil { + return err + } + trace = append(trace, fmt.Sprintf("segment:%.0f", segStart)) + if err := onSegment(playback.AudioChunk{Path: p, Start: segStart}); err != nil { + return err + } + trace = append(trace, fmt.Sprintf("after:%.0f", segStart)) + } + return nil + }, + probeOffset: func(context.Context, string, int, string) float64 { return 0 }, + } + tr.SetExtraction("", 600) + + var progress []string + cues, _, err := tr.Transcribe(context.Background(), TranscribeJobRequest{ + FilePath: "/x.mkv", + StartPosition: 1300, + DurationSeconds: 2400, + Incremental: true, + }, func(chunk []SubtitleCue, _ string, done, total int) { + progress = append(progress, fmt.Sprintf("%d/%d", done, total)) + trace = append(trace, "chunk:"+strings.Join(chunk[0].Lines, " ")) + }) + if err != nil { + t.Fatalf("Transcribe: %v", err) + } + if got := fmt.Sprint(starts); got != "[1200 0]" { + t.Fatalf("incremental pass starts = %s, want [1200 0]", got) + } + var requestOrder []string + for _, req := range client.requests { + requestOrder = append(requestOrder, req.Filename) + } + if got := strings.Join(requestOrder, ","); got != "inc1200.wav,inc1800.wav,inc0.wav,inc600.wav" { + t.Fatalf("request order = %s", got) + } + if got := strings.Join(progress, ","); got != "1/4,2/4,3/4,4/4" { + t.Fatalf("progress = %s, want chunk totals", got) + } + if chunkIdx, nextSegmentIdx := traceIndexString(trace, "chunk:pivot"), traceIndexString(trace, "segment:1800"); chunkIdx < 0 || nextSegmentIdx < 0 || chunkIdx > nextSegmentIdx { + t.Fatalf("first onChunk did not fire before the next extracted segment: %v", trace) + } + if len(cues) != 4 { + t.Fatalf("returned cues = %d, want complete transcript", len(cues)) + } +} + +func TestTranscribeIncrementalFallsBackToStartWhenSeekedPassProducesNoAudio(t *testing.T) { + client := &fakeASRClient{ + language: "en", + perChunk: map[string][]llm.TranscriptionSegment{ + "head.wav": {{Start: 1, End: 2, Text: "head"}}, + }, + } + tr := &WhisperTranscriber{ + client: client, + incrementalExtract: func(_ context.Context, _ string, _ int, dir, _ string, startSec float64, _ int, + onSegment func(playback.AudioChunk) error) error { + if startSec > 0 { + return fmt.Errorf("ffmpeg produced no audio chunks for track 0") + } + p := filepath.Join(dir, "head.wav") + if err := os.WriteFile(p, []byte("RIFF"), 0o644); err != nil { + return err + } + return onSegment(playback.AudioChunk{Path: p, Start: 0}) + }, + probeOffset: func(context.Context, string, int, string) float64 { return 0 }, + } + tr.SetExtraction("", 600) + + cues, _, err := tr.Transcribe(context.Background(), TranscribeJobRequest{ + FilePath: "/x.mkv", + StartPosition: 1300, + DurationSeconds: 1800, + Incremental: true, + }, nil) + if err != nil { + t.Fatalf("Transcribe: %v", err) + } + if len(cues) != 1 || strings.Join(cues[0].Lines, " ") != "head" { + t.Fatalf("fallback cues = %#v, want head cue from start pass", cues) + } + if len(client.requests) != 1 || client.requests[0].Filename != "head.wav" { + t.Fatalf("ASR requests = %#v, want only fallback chunk", client.requests) + } +} + +func TestTranscribeIncrementalDoesNotDoubleApplyAudioOffsetAfterSeek(t *testing.T) { + client := &fakeASRClient{ + language: "en", + perChunk: map[string][]llm.TranscriptionSegment{ + "seek.wav": {{Start: 1, End: 2, Text: "seek"}}, + "head.wav": {{Start: 1, End: 2, Text: "head"}}, + }, + } + tr := &WhisperTranscriber{ + client: client, + incrementalExtract: func(_ context.Context, _ string, _ int, dir, _ string, startSec float64, _ int, + onSegment func(playback.AudioChunk) error) error { + name := "seek.wav" + chunkStart := 600.0 + if startSec == 0 { + name = "head.wav" + chunkStart = 0 + } + p := filepath.Join(dir, name) + if err := os.WriteFile(p, []byte("RIFF"), 0o644); err != nil { + return err + } + if err := onSegment(playback.AudioChunk{Path: p, Start: chunkStart}); err != nil { + return err + } + if startSec == 0 { + stop := filepath.Join(dir, "seek.wav") + if err := os.WriteFile(stop, []byte("RIFF"), 0o644); err != nil { + return err + } + return onSegment(playback.AudioChunk{Path: stop, Start: 600}) + } + return nil + }, + probeOffset: func(context.Context, string, int, string) float64 { return 1.25 }, + } + tr.SetExtraction("", 600) + + cues, _, err := tr.Transcribe(context.Background(), TranscribeJobRequest{ + FilePath: "/x.mkv", + StartPosition: 600, + DurationSeconds: 1200, + Incremental: true, + }, nil) + if err != nil { + t.Fatalf("Transcribe: %v", err) + } + if len(cues) != 2 { + t.Fatalf("cues = %d, want seek and head", len(cues)) + } + if want := 601 * time.Second; cues[0].Start != want { + t.Fatalf("seek cue start = %v, want %v without re-adding audio offset", cues[0].Start, want) + } + if want := time.Duration(2.25 * float64(time.Second)); cues[1].Start != want { + t.Fatalf("head cue start = %v, want %v with audio offset", cues[1].Start, want) + } +} + +func TestTranscribeIncrementalKeepsProgressIndeterminateWhenDurationUnknown(t *testing.T) { + client := &fakeASRClient{ + language: "en", + perChunk: map[string][]llm.TranscriptionSegment{ + "head.wav": {{Start: 0, End: 1, Text: "head"}}, + }, + } + tr := &WhisperTranscriber{ + client: client, + incrementalExtract: func(_ context.Context, _ string, _ int, dir, _ string, _ float64, _ int, + onSegment func(playback.AudioChunk) error) error { + p := filepath.Join(dir, "head.wav") + if err := os.WriteFile(p, []byte("RIFF"), 0o644); err != nil { + return err + } + return onSegment(playback.AudioChunk{Path: p, Start: 0}) + }, + probeOffset: func(context.Context, string, int, string) float64 { return 0 }, + } + tr.SetExtraction("", 600) + + var gotDone, gotTotal int + _, _, err := tr.Transcribe(context.Background(), TranscribeJobRequest{ + FilePath: "/x.mkv", + Incremental: true, + }, func(_ []SubtitleCue, _ string, done, total int) { + gotDone = done + gotTotal = total + }) + if err != nil { + t.Fatalf("Transcribe: %v", err) + } + if gotDone != 1 || gotTotal != 0 { + t.Fatalf("progress = %d/%d, want 1/0 for indeterminate total", gotDone, gotTotal) + } +} + // Cue timing must use the muxer-reported chunk start (not index*chunkSeconds) // plus the probed audio start offset, or sync drifts on long files and // delayed-audio containers. @@ -358,6 +628,15 @@ func TestTranscribeUsesExactChunkStartsAndAudioOffset(t *testing.T) { } } +func traceIndexString(values []string, target string) int { + for i, value := range values { + if value == target { + return i + } + } + return -1 +} + func TestNormalizeDetectedLanguage(t *testing.T) { cases := map[string]string{ "english": "en",