feat(dub): second-pass ASR QC — flag lines whose dub drifts from target (Wave 3.3) (#370)

After a dub is generated, re-recognize the synthetic audio and compare what
the ASR heard against what we asked the TTS to say. Lines that drift are
flagged for the user to re-listen / re-dub — turning subtitle timing and
pronunciation from trusted math into measured truth, and doubling as an
automatic dub-quality check.

Design delta from pyvideotrans (which lets recognized text REPLACE the
subtitles wholesale): we keep the generated text authoritative and use the
second pass only for MEASUREMENT — a per-line drift score + measured
start/end that feed the incremental re-dub loop, never silently overwriting
the translation.

- services/dub_qc.py (pure, tested): word_error_rate (normalized token edit
  distance, case/punct-insensitive, script-agnostic) + score_dub (matches
  recognized segments to dub segments by time overlap, concatenates the
  hypothesis, scores drift, derives measured bounds).
- POST /dub/qc/{job_id}: runs the active ASR backend on the dubbed track in
  the GPU pool, annotates each segment with qc_drift/qc_flagged/
  qc_recognized/qc_measured_start-end (non-destructive — content untouched),
  persists, emits a qc_done job event. Opt-in, never fatal.
- Frontend: dubQc() API fn + a red 'Verify' badge on flagged segment rows
  (en.json keys; other locales fall back).

12 pure scoring tests (identical/substitution/empty/no-overlap/multi-segment
matching/measured-timing); endpoint validated in CI.

Spec 5 / parity program Wave 3.3.

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

backend/api/routers/dub_export.py

@@ -1,6 +1,7 @@
 import os
 import io
 import re
+import json
 import time
 import uuid
 import asyncio
@@ -854,6 +855,93 @@ async def dub_preview_segment(job_id: str, segment_index: int):
     return FileResponse(seg_path, media_type="audio/wav")
 
 
+# ── Second-pass ASR QC (Wave 3.3 / Spec 5) ───────────────────────────────────
+
+
+@router.post("/dub/qc/{job_id}")
+async def dub_qc_pass(job_id: str, lang: str = Query(None), drift_threshold: float = Query(0.5)):
+    """Re-recognize the dubbed audio and flag lines whose recognized text
+    drifts from the target text. Opt-in, never fatal: the dub is untouched —
+    this only annotates segments with a per-line drift score and a measured
+    start/end, surfaced as "verify this line" markers feeding incremental
+    re-dub. The generated text stays authoritative (design delta from
+    pyvideotrans, which overwrites subtitles)."""
+    from services import dub_qc
+    from services.dub_pipeline import put_job, save_job
+
+    job = _get_job(job_id)
+    if not job:
+        raise HTTPException(status_code=404, detail="Job not found")
+    tracks = job.get("dubbed_tracks", {})
+    if lang and lang in tracks:
+        wav_path = tracks[lang]["path"]
+    elif tracks:
+        wav_path = list(tracks.values())[0]["path"]
+    else:
+        raise HTTPException(status_code=400, detail="No dubbed audio track generated yet")
+    if not os.path.exists(wav_path):
+        raise HTTPException(status_code=404, detail="Dubbed audio file not found")
+
+    segments = job.get("segments") or []
+    if not segments:
+        raise HTTPException(status_code=400, detail="Job has no segments")
+
+    def _recognize():
+        from services.asr_backend import get_active_asr_backend
+        backend = get_active_asr_backend()
+        result = backend.transcribe(wav_path, word_timestamps=False)
+        return result.get("segments", []), backend.id
+
+    try:
+        from services.model_manager import _get_gpu_pool
+        loop = asyncio.get_running_loop()
+        recognized, engine_id = await loop.run_in_executor(_get_gpu_pool(), _recognize)
+    except Exception as e:
+        logger.exception("dub QC ASR pass failed for %s", job_id)
+        raise HTTPException(status_code=500, detail=f"QC transcription failed: {e}")
+
+    seg_ids = job.get("seg_order") or [s.get("id", i) for i, s in enumerate(segments)]
+    scored = dub_qc.score_dub(segments, recognized, drift_threshold=drift_threshold, seg_ids=seg_ids)
+
+    # Annotate each segment (non-destructive — content text untouched).
+    by_id = {q.seg_id: q for q in scored}
+    for i, s in enumerate(segments):
+        sid = str(seg_ids[i]) if i < len(seg_ids) else str(s.get("id", i))
+        q = by_id.get(sid)
+        if q is None:
+            continue
+        s["qc_drift"] = q.drift
+        s["qc_flagged"] = q.flagged
+        s["qc_recognized"] = q.recognized_text
+        if q.new_start is not None:
+            s["qc_measured_start"] = q.new_start
+            s["qc_measured_end"] = q.new_end
+    put_job(job_id, job)
+    save_job(job_id, job)
+
+    flagged = [q for q in scored if q.flagged]
+    payload = json.dumps({"event": "qc_done", "engine": engine_id,
+                          "flagged": len(flagged), "total": len(scored)})
+    try:
+        from core import job_store
+        job_store.append_event(job_id, f"data: {payload}\n\n")
+    except Exception:
+        pass
+
+    return {
+        "engine": engine_id,
+        "total": len(scored),
+        "flagged_count": len(flagged),
+        "drift_threshold": drift_threshold,
+        "segments": [
+            {"seg_id": q.seg_id, "drift": q.drift, "flagged": q.flagged,
+             "recognized_text": q.recognized_text,
+             "measured_start": q.new_start, "measured_end": q.new_end}
+            for q in scored
+        ],
+    }
+
+
 @router.get("/dub/download-audio/{job_id}")
 @router.get("/dub/download-audio/{job_id}/{filename}")
 async def dub_download_audio(job_id: str, lang: str = Query(None), preserve_bg: bool = Query(True), save_path: str = Query("")):

backend/services/dub_qc.py

@@ -0,0 +1,118 @@
+"""Second-pass ASR quality control for dubs (Wave 3.3 / Spec 5).
+
+After a dub is generated, re-recognize the synthetic audio and compare what
+the ASR *heard* against what we asked the TTS to *say*. Where the two drift
+apart, the line is flagged for the user to verify — turning subtitle timing
+and pronunciation from "trusted math" into "measured truth", and doubling as
+an automatic dub-quality check.
+
+Design delta from pyvideotrans (whose second pass lets recognized text
+*replace* the subtitles wholesale): we keep the GENERATED text authoritative
+for content and use the second pass for *measurement* — timing + a drift
+score that feeds the incremental re-dub loop, never silently overwriting the
+translation.
+
+Pure functions here (no ASR, no I/O) so the scoring is unit-testable; the
+pipeline stage that runs the ASR pass lives in the dub router.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+
+
+def _tokens(text: str) -> list[str]:
+    """Lowercase word tokens, punctuation stripped — the unit drift is scored
+    in. Script-agnostic: for no-space scripts each character is a token, which
+    still gives a sensible edit-distance ratio."""
+    text = (text or "").lower().strip()
+    if not text:
+        return []
+    words = re.findall(r"\w+", text, flags=re.UNICODE)
+    return words or list(text.replace(" ", ""))
+
+
+def _edit_distance(a: list[str], b: list[str]) -> int:
+    """Levenshtein distance between two token lists (iterative, O(len(a)*len(b))
+    time, O(len(b)) space)."""
+    if not a:
+        return len(b)
+    if not b:
+        return len(a)
+    prev = list(range(len(b) + 1))
+    for i, ta in enumerate(a, 1):
+        cur = [i]
+        for j, tb in enumerate(b, 1):
+            cost = 0 if ta == tb else 1
+            cur.append(min(prev[j] + 1, cur[j - 1] + 1, prev[j - 1] + cost))
+        prev = cur
+    return prev[-1]
+
+
+def word_error_rate(reference: str, hypothesis: str) -> float:
+    """Normalized token edit distance in [0.0, 1.0+].
+
+    0.0 = the ASR heard exactly the target text. ~1.0 = entirely different.
+    Can exceed 1.0 when the hypothesis is much longer than the reference
+    (insertions); callers clamp/threshold as needed. An empty reference with a
+    non-empty hypothesis scores 1.0 (everything is an insertion)."""
+    ref = _tokens(reference)
+    hyp = _tokens(hypothesis)
+    if not ref and not hyp:
+        return 0.0
+    if not ref:
+        return 1.0
+    return _edit_distance(ref, hyp) / len(ref)
+
+
+@dataclass
+class SegmentQC:
+    seg_id: str
+    target_text: str
+    recognized_text: str
+    drift: float           # word_error_rate(target, recognized)
+    flagged: bool          # drift >= threshold
+    new_start: float | None  # measured onset from the dubbed-audio recognition
+    new_end: float | None
+
+
+def _overlap(a0: float, a1: float, b0: float, b1: float) -> float:
+    return max(0.0, min(a1, b1) - max(a0, b0))
+
+
+def score_dub(
+    dub_segments: list[dict],
+    recognized: list[dict],
+    *,
+    drift_threshold: float = 0.5,
+    seg_ids: list | None = None,
+) -> list[SegmentQC]:
+    """Match the second-pass recognition to the dub segments and score drift.
+
+    ``dub_segments`` are the segments we generated (each {start, end, text});
+    ``recognized`` are the ASR result segments on the dubbed audio (each
+    {start, end, text}). Each dub segment is matched to the recognized
+    segment(s) it overlaps in time; their text is concatenated as the
+    hypothesis and scored against the dub segment's ``text``. The recognized
+    span's bounds become the measured start/end (subtitle-timing truth).
+    """
+    results: list[SegmentQC] = []
+    for i, seg in enumerate(dub_segments):
+        sid = str(seg_ids[i]) if (seg_ids and i < len(seg_ids)) else str(seg.get("id", i))
+        s0, s1 = float(seg.get("start", 0.0)), float(seg.get("end", 0.0))
+        hits = [r for r in recognized if _overlap(s0, s1, float(r.get("start", 0.0)), float(r.get("end", 0.0))) > 0]
+        hyp = " ".join((r.get("text") or "").strip() for r in hits).strip()
+        drift = word_error_rate(seg.get("text", ""), hyp)
+        new_start = min((float(r.get("start", 0.0)) for r in hits), default=None)
+        new_end = max((float(r.get("end", 0.0)) for r in hits), default=None)
+        results.append(SegmentQC(
+            seg_id=sid,
+            target_text=(seg.get("text") or "").strip(),
+            recognized_text=hyp,
+            drift=round(drift, 3),
+            flagged=drift >= drift_threshold,
+            new_start=new_start,
+            new_end=new_end,
+        ))
+    return results

frontend/src/api/dub.ts

@@ -103,3 +103,24 @@ export async function listDubHistory(): Promise<DubHistoryResponse> {
 export async function clearDubHistory(): Promise<Response> {
   return apiFetch('/dub/history', { method: 'DELETE' });
 }
+
+export interface DubQCResponse {
+  engine: string;
+  total: number;
+  flagged_count: number;
+  drift_threshold: number;
+  segments: {
+    seg_id: string; drift: number; flagged: boolean;
+    recognized_text: string; measured_start: number | null; measured_end: number | null;
+  }[];
+}
+
+/** Wave 3.3: second-pass ASR QC — re-recognize the dubbed audio and flag
+ *  lines whose recognized text drifts from the target. Non-destructive. */
+export async function dubQc(jobId: string, lang?: string, driftThreshold?: number): Promise<DubQCResponse> {
+  const qs = new URLSearchParams();
+  if (lang) qs.set('lang', lang);
+  if (driftThreshold != null) qs.set('drift_threshold', String(driftThreshold));
+  const suffix = qs.toString() ? `?${qs}` : '';
+  return apiPost<DubQCResponse>(`/dub/qc/${jobId}${suffix}`);
+}

frontend/src/components/DubSegmentRow.jsx

@@ -188,6 +188,17 @@ function DubSegmentRow({
             <fitBadge.Icon size={8} /> {fitBadge.label}
           </span>
         )}
+        {seg.qc_flagged && (
+          // Wave 3.3: second-pass ASR heard something different from the
+          // target text for this line — worth a re-listen / re-dub.
+          <span
+            className="seg-sync-badge"
+            style={{ color: '#fb4934' }}
+            title={t('segment.qc_verify_title', { heard: seg.qc_recognized || '' })}
+          >
+            <AlertCircle size={8} /> {t('segment.qc_verify')}
+          </span>
+        )}
         {seg.rate_ratio != null && Math.abs(seg.rate_ratio - 1.0) > 0.03 && (
           <span
             className="seg-rate-badge"

frontend/src/i18n/locales/en.json

@@ -661,6 +661,8 @@
     "speaker_title_detected": "Speaker — pick from detected, or type a custom name",
     "speaker_title_custom": "Speaker — type a name (no diarization clones detected)",
     "time_edit_title": "Click to edit start time (m:ss.s). Enter to commit, Esc to cancel.",
+    "qc_verify": "Verify",
+    "qc_verify_title": "Second-pass ASR heard: \"{{heard}}\" — re-listen or re-dub this line.",
     "fit_fits": "Fits",
     "fit_fits_title": "Natural-rate audio fit inside the slot.",
     "fit_overflows": "Overflows +{{seconds}}s",

tests/test_dub_qc.py

@@ -0,0 +1,90 @@
+"""Second-pass ASR QC scoring (Wave 3.3 / Spec 5) — pure, no ASR/main import."""
+
+import pytest
+
+from services.dub_qc import score_dub, word_error_rate
+
+
+# ── word_error_rate ──────────────────────────────────────────────────────────
+
+def test_wer_identical_is_zero():
+    assert word_error_rate("hello there world", "hello there world") == 0.0
+
+
+def test_wer_case_and_punctuation_insensitive():
+    assert word_error_rate("Hello, world!", "hello world") == 0.0
+
+
+def test_wer_one_substitution():
+    # 1 edit over 3 reference tokens.
+    assert word_error_rate("the cat sat", "the dog sat") == pytest.approx(1 / 3)
+
+
+def test_wer_empty_reference_with_hypothesis_is_one():
+    assert word_error_rate("", "something") == 1.0
+
+
+def test_wer_both_empty_is_zero():
+    assert word_error_rate("", "") == 0.0
+
+
+def test_wer_completely_different():
+    assert word_error_rate("alpha beta", "x y z") >= 1.0
+
+
+# ── score_dub ────────────────────────────────────────────────────────────────
+
+def _seg(start, end, text, sid=None):
+    d = {"start": start, "end": end, "text": text}
+    if sid is not None:
+        d["id"] = sid
+    return d
+
+
+def test_clean_dub_no_flags():
+    dub = [_seg(0, 3, "hello world", "a"), _seg(3, 6, "good morning", "b")]
+    recog = [_seg(0.1, 2.9, "hello world"), _seg(3.0, 5.8, "good morning")]
+    out = score_dub(dub, recog)
+    assert [q.flagged for q in out] == [False, False]
+    assert out[0].seg_id == "a"
+    assert all(q.drift == 0.0 for q in out)
+
+
+def test_drifted_segment_is_flagged():
+    dub = [_seg(0, 3, "the quarterly report is ready", "a")]
+    # ASR heard something quite different (mispronunciation / bad clone).
+    recog = [_seg(0, 3, "the quarter lee deport is read")]
+    out = score_dub(dub, recog, drift_threshold=0.5)
+    assert out[0].flagged is True
+    assert out[0].recognized_text == "the quarter lee deport is read"
+
+
+def test_measured_timing_from_recognition():
+    dub = [_seg(0.0, 5.0, "hello", "a")]
+    recog = [_seg(0.4, 1.2, "hello")]
+    out = score_dub(dub, recog)
+    assert out[0].new_start == pytest.approx(0.4)
+    assert out[0].new_end == pytest.approx(1.2)
+
+
+def test_segment_with_no_overlap_scores_full_drift():
+    dub = [_seg(0, 3, "spoken line", "a")]
+    recog = [_seg(10, 12, "elsewhere")]  # no time overlap
+    out = score_dub(dub, recog)
+    assert out[0].recognized_text == ""
+    assert out[0].drift == 1.0 and out[0].flagged
+    assert out[0].new_start is None
+
+
+def test_multiple_recognized_segments_concatenate():
+    dub = [_seg(0, 6, "one two three four", "a")]
+    recog = [_seg(0, 3, "one two"), _seg(3, 6, "three four")]
+    out = score_dub(dub, recog)
+    assert out[0].recognized_text == "one two three four"
+    assert out[0].drift == 0.0
+
+
+def test_seg_ids_override():
+    dub = [_seg(0, 3, "x")]
+    out = score_dub(dub, [_seg(0, 3, "x")], seg_ids=["custom"])
+    assert out[0].seg_id == "custom"