feat(dub): regenerate subtitle timeline on the fitted timeline (Wave 3.1) (#371)

Smart Fit Phase A (planner) + the export-side video retime + audio stretch
already shipped (#347 + dub_export stretch filter). The last piece of
Spec 1 was the subtitle timeline: under stretch_video the dubbed audio
plays at FITTED positions, but the standalone SRT/VTT export still used the
original segment times — so external subtitles drifted against the dubbed
video.

- services/fitted_subtitles.py (pure, tested): map_time_to_fitted() +
  fitted_cues() remap original cue times onto the same per-chunk
  {orig→new, stretch_ratio} plan the video stretch uses, with a
  monotonicity guard.
- dub_export SRT + VTT endpoints: when a job used stretch_video, cues are
  regenerated from the plan (subtitles track actual dub placement); no
  plan → original times, unchanged. New optional ?lang= selects the track.

7 pure tests (chunk-bound mapping, linear interpolation, unit-rate tail,
fitted cues, monotonicity, empty-plan identity).

Spec 1 (remaining) / parity program Wave 3.1.

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

backend/api/routers/dub_export.py

@@ -1025,9 +1025,21 @@ def _pick_subtitle_text(seg: dict, dual: bool) -> str:
 # (#309). The frontend's JSON-envelope save flow stays for binary exports.
 
 
+def _fitted_cue_times(job: dict, lang: str | None) -> list | None:
+    """Per-segment (start, end) on the fitted timeline when this job used
+    stretch_video; None to use the original segment times. (Wave 3.1.)"""
+    tracks = job.get("dubbed_tracks", {})
+    lc = lang if (lang and lang in tracks) else (next(iter(tracks), None))
+    entry = _video_stretch_plan_for(job, lc) if lc else None
+    if not entry:
+        return None
+    from services.fitted_subtitles import fitted_cues
+    return fitted_cues(job.get("segments", []), entry["plan"])
+
+
 @router.get("/dub/srt/{job_id}")
 @router.get("/dub/srt/{job_id}/{filename}")
-async def dub_export_srt(job_id: str, dual: bool = False):
+async def dub_export_srt(job_id: str, dual: bool = False, lang: str = None):
     job = _get_job(job_id)
     if not job:
         raise HTTPException(status_code=404, detail="Job not found")
@@ -1036,12 +1048,16 @@ async def dub_export_srt(job_id: str, dual: bool = False):
     if not segments:
         raise HTTPException(status_code=400, detail="No transcript segments available")
 
+    # Wave 3.1: under stretch_video the dubbed audio plays at fitted
+    # positions, so regenerate the cue timeline from the same stretch plan
+    # ("subtitles track actual dub placement"). No plan → original times.
+    cues = _fitted_cue_times(job, lang)
+
     srt_lines = []
     for i, seg in enumerate(segments):
-        start_ts = _format_srt_time(seg["start"])
-        end_ts = _format_srt_time(seg["end"])
+        s, e = cues[i] if cues else (seg["start"], seg["end"])
         srt_lines.append(f"{i + 1}")
-        srt_lines.append(f"{start_ts} --> {end_ts}")
+        srt_lines.append(f"{_format_srt_time(s)} --> {_format_srt_time(e)}")
         srt_lines.append(_pick_subtitle_text(seg, dual))
         srt_lines.append("")
 
@@ -1064,7 +1080,7 @@ def _format_vtt_time(seconds):
 
 @router.get("/dub/vtt/{job_id}")
 @router.get("/dub/vtt/{job_id}/{filename}")
-async def dub_export_vtt(job_id: str, dual: bool = False):
+async def dub_export_vtt(job_id: str, dual: bool = False, lang: str = None):
     job = _get_job(job_id)
     if not job:
         raise HTTPException(status_code=404, detail="Job not found")
@@ -1073,12 +1089,14 @@ async def dub_export_vtt(job_id: str, dual: bool = False):
     if not segments:
         raise HTTPException(status_code=400, detail="No transcript segments available")
 
+    # Wave 3.1: fitted timeline under stretch_video (see SRT export).
+    cues = _fitted_cue_times(job, lang)
+
     vtt_lines = ["WEBVTT", ""]
     for i, seg in enumerate(segments):
-        start_ts = _format_vtt_time(seg["start"])
-        end_ts = _format_vtt_time(seg["end"])
+        s, e = cues[i] if cues else (seg["start"], seg["end"])
         vtt_lines.append(str(i + 1))
-        vtt_lines.append(f"{start_ts} --> {end_ts}")
+        vtt_lines.append(f"{_format_vtt_time(s)} --> {_format_vtt_time(e)}")
         vtt_lines.append(_pick_subtitle_text(seg, dual))
         vtt_lines.append("")
 

backend/services/fitted_subtitles.py

@@ -0,0 +1,65 @@
+"""Map subtitle cues onto the Smart-Fit timeline (Wave 3.1 / Spec 1).
+
+When a dub uses ``stretch_video`` mode, the video is re-timed per segment so
+the dubbed audio fits (see fit_planner + the export stretch filter). The
+dubbed audio therefore plays at *fitted* positions, not the original
+timestamps. A subtitle file exported with the original times would drift
+against the dubbed video — so we regenerate the cue timeline from the same
+plan the video stretch uses ("subtitles track actual dub placement", the
+last piece of Spec 1).
+
+Pure functions — no I/O — so the remapping is unit-testable. The plan is the
+persisted ``video_stretch_plan`` list of
+``{orig_start, orig_end, new_start, new_end, stretch_ratio}`` chunks.
+"""
+
+from __future__ import annotations
+
+
+def map_time_to_fitted(t: float, plan: list[dict]) -> float:
+    """Map a time on the original timeline to its position on the fitted one.
+
+    Finds the plan chunk whose original span contains ``t`` and interpolates
+    linearly into that chunk's fitted span (a chunk's stretch is uniform).
+    Before the first chunk maps 1:1; after the last chunk the trailing offset
+    is carried at 1:1 (the planner runs gaps/tail at rate 1.0). Empty plan ⇒
+    identity.
+    """
+    if not plan:
+        return t
+    for chunk in plan:
+        o0 = float(chunk.get("orig_start", 0.0))
+        o1 = float(chunk.get("orig_end", 0.0))
+        n0 = float(chunk.get("new_start", o0))
+        n1 = float(chunk.get("new_end", o1))
+        if t < o0:
+            # In a gap before this chunk — carry the offset at 1:1 from the
+            # previous chunk's fitted end (or from 0 for the very first).
+            return n0 - (o0 - t)
+        if o0 <= t <= o1:
+            span = o1 - o0
+            if span <= 0:
+                return n0
+            return n0 + (t - o0) / span * (n1 - n0)
+    # Past the last chunk: 1:1 tail from its fitted end.
+    last = plan[-1]
+    return float(last.get("new_end", 0.0)) + (t - float(last.get("orig_end", 0.0)))
+
+
+def fitted_cues(segments: list[dict], plan: list[dict]) -> list[tuple[float, float]]:
+    """Return ``[(start, end), ...]`` for each segment on the fitted timeline.
+
+    Monotonicity guard: a cue's end is never before its start, and successive
+    starts never go backwards (rounding across chunk seams can't produce a
+    non-monotone SRT).
+    """
+    out: list[tuple[float, float]] = []
+    prev_end = 0.0
+    for seg in segments:
+        s = map_time_to_fitted(float(seg.get("start", 0.0)), plan)
+        e = map_time_to_fitted(float(seg.get("end", 0.0)), plan)
+        s = max(s, prev_end if out else 0.0)
+        e = max(e, s)
+        out.append((s, e))
+        prev_end = e
+    return out

tests/test_fitted_subtitles.py

@@ -0,0 +1,52 @@
+"""Fitted-timeline subtitle remapping (Wave 3.1 / Spec 1) — pure, no I/O."""
+
+import pytest
+
+from services.fitted_subtitles import fitted_cues, map_time_to_fitted
+
+# A 2-chunk plan: chunk 0 [0,4]→[0,6] (1.5× slow), chunk 1 [4,8]→[6,10] (1× — gap).
+PLAN = [
+    {"orig_start": 0.0, "orig_end": 4.0, "new_start": 0.0, "new_end": 6.0, "stretch_ratio": 1.5},
+    {"orig_start": 4.0, "orig_end": 8.0, "new_start": 6.0, "new_end": 10.0, "stretch_ratio": 1.0},
+]
+
+
+def test_empty_plan_is_identity():
+    assert map_time_to_fitted(3.7, []) == 3.7
+
+
+def test_chunk_start_and_end_map_to_fitted_bounds():
+    assert map_time_to_fitted(0.0, PLAN) == pytest.approx(0.0)
+    assert map_time_to_fitted(4.0, PLAN) == pytest.approx(6.0)
+    assert map_time_to_fitted(8.0, PLAN) == pytest.approx(10.0)
+
+
+def test_midpoint_interpolates_linearly():
+    # halfway through chunk 0 (t=2 of [0,4]) → halfway of [0,6] = 3.0
+    assert map_time_to_fitted(2.0, PLAN) == pytest.approx(3.0)
+
+
+def test_time_past_last_chunk_carries_at_unit_rate():
+    assert map_time_to_fitted(9.0, PLAN) == pytest.approx(11.0)  # 10 + (9-8)
+
+
+def test_fitted_cues_uses_new_timeline():
+    segs = [{"start": 0.0, "end": 2.0}, {"start": 4.0, "end": 6.0}]
+    cues = fitted_cues(segs, PLAN)
+    assert cues[0] == (pytest.approx(0.0), pytest.approx(3.0))
+    # seg 2: 4.0→6.0, 6.0→8.0
+    assert cues[1] == (pytest.approx(6.0), pytest.approx(8.0))
+
+
+def test_fitted_cues_are_monotone():
+    # Overlapping/odd inputs must still produce a non-decreasing cue stream.
+    segs = [{"start": 3.9, "end": 4.1}, {"start": 4.0, "end": 4.0}]
+    cues = fitted_cues(segs, PLAN)
+    for s, e in cues:
+        assert e >= s
+    assert cues[1][0] >= cues[0][1] - 1e-9
+
+
+def test_empty_plan_cues_match_original():
+    segs = [{"start": 1.0, "end": 2.5}]
+    assert fitted_cues(segs, []) == [(pytest.approx(1.0), pytest.approx(2.5))]