fix(plex): the subtitle extractor's seams — doubled cues, a truncated track cached as whole, a poll that respawned ffmpeg

Deep-review of the bug-hunt round (R1). The background extractor from ed4ab4e1 held, but its seams
did not:

- Window + full pass doubled every cue in the first hour. merge_vtt de-duplicated on the raw timing
  text, and the two passes spell a timestamp differently (shift_vtt writes HH:MM:SS.mmm, ffmpeg's
  webvtt muxer drops the hour below 1 h; shift_vtt(-0.0) never normalised the full pass). Timing is
  now parsed and re-emitted canonically, keyed with the cue text so two cues sharing a time survive.
- A full pass that exited non-zero was published as complete: "ready" forever, subtitles ending
  where ffmpeg died, and no eviction to ever retry. It now covers only up to its last cue, is never
  complete, and records the failure memo.
- The window's dedup identity held the live playhead, so every 1 s poll queued another read of the
  file. It is keyed on the track now, and a queued job re-checks coverage before reading anything.
- A sidecar track waited behind an unrelated embedded full pass (the full job is per item). Its
  window job now fetches it whole — one small request.
- The <track> fetch ran the scheduler at position 0: an unrequested window over the head of the file,
  and a refusal of a window that was on disk. It is a pure read of the cache now.
- publish() is serialised (a lost update became permanent once complete was set); the pool is shut
  down and running ffmpeg children killed at app shutdown (they held interpreter exit for minutes);
  ?at=inf is a 422, not a 500; and both subtitle routes use current_user instead of the cookie-only
  poster gate, which let a suspended or epoch-revoked session keep scheduling reads (pre-existing).

_extract_and_publish finally has tests of its own — every earlier test stubbed it out. 19 mutations,
each red on the assertion that names it.

backend/app/main.py

@@ -27,6 +27,7 @@ from app import state
 from app.config import settings
 from app.db import SessionLocal
 from app.plex import stream as plex_stream
+from app.plex import subtitles as plex_subtitles
 from app.routes import (
     admin,
     audit as audit_routes,
@@ -90,6 +91,9 @@ async def lifespan(app: FastAPI):
         with suppress(asyncio.CancelledError):
             await sweep
         shutdown_scheduler()
+        # Kills a running subtitle extraction: its worker would otherwise hold interpreter exit for
+        # the rest of the read (see plex_subtitles.shutdown).
+        plex_subtitles.shutdown()
 
 
 async def _sweep_hls_orphans() -> None:

backend/app/plex/subtitles.py

@@ -245,31 +245,73 @@ def covers(ranges: list[list[float]], at: float) -> bool:
 def merge_vtt(*sources: str) -> str:
     """Union of the cue blocks of several absolute-time VTTs, ordered by start time.
 
-    Two extraction passes over the same file emit identical cues for the region they share, so
-    de-duplication keys off the TIMING line — the one part of a cue that identifies it uniquely and
-    that neither pass can renumber (ffmpeg's webvtt muxer writes no cue identifiers)."""
+    Two extraction passes over the same file emit identical cues for the region they share, so they
+    are de-duplicated — but NOT on the raw timing text. The two passes do not spell a timestamp the
+    same way: a window comes back through `shift_vtt`, which always writes `HH:MM:SS.mmm`, while a
+    full pass is ffmpeg's own output, and its webvtt muxer drops the hour below one hour
+    (`10:00.123`). Keyed on the text, every cue of a window in the first hour survived twice and the
+    viewer saw each line stacked on itself. So each timing line is parsed and re-emitted in one
+    canonical form, and the key is that canonical timing PLUS the cue text: two different cues that
+    happen to share a time (two speakers, a sign over dialogue) are both kept."""
     seen: set[str] = set()
     cues: list[tuple[float, str]] = []
     for src in sources:
         for block in re.split(r"\n[ \t]*\n", src or ""):
-            b = block.strip()
-            if not b or "-->" not in b:
+            canon = _canonical_cue(block.strip())
+            if canon is None:
                 continue  # header / NOTE / STYLE — the merged file gets its own single header
-            timing = next(ln for ln in b.split("\n") if "-->" in ln)
-            if timing in seen:
+            start, key, text = canon
+            if key in seen:
                 continue
-            seen.add(timing)
-            try:
-                start = _vtt_ts_to_s(timing.partition("-->")[0].strip())
-            except (ValueError, IndexError):
-                continue
-            cues.append((start, b))
+            seen.add(key)
+            cues.append((start, text))
     cues.sort(key=lambda c: c[0])
     return "WEBVTT\n\n" + "\n\n".join(b for _, b in cues) + "\n"
 
 
+def _canonical_cue(block: str) -> tuple[float, str, str] | None:
+    """(start seconds, identity, the block with its timing line canonicalised) — or None for a block
+    that is not a cue. The identity leaves out anything before the timing line (an SRT-style cue
+    number), which is a label, not content."""
+    lines = block.split("\n")
+    for i, ln in enumerate(lines):
+        if "-->" not in ln:
+            continue
+        left, _, right = ln.partition("-->")
+        rparts = right.strip().split(None, 1)
+        settings = (" " + rparts[1]) if len(rparts) > 1 else ""
+        try:
+            st = _vtt_ts_to_s(left.strip())
+            en = _vtt_ts_to_s(rparts[0])
+        except (ValueError, IndexError):
+            return None
+        lines[i] = f"{_vtt_s_to_ts(st)} --> {_vtt_s_to_ts(en)}{settings}"
+        return st, "\n".join(lines[i:]), "\n".join(lines)
+    return None
+
+
+def _last_cue_end(vtt: str) -> float:
+    """Where the last cue of a VTT ends — how far a pass that died early provably got."""
+    end = 0.0
+    for ln in vtt.split("\n"):
+        if "-->" not in ln:
+            continue
+        try:
+            end = max(end, _vtt_ts_to_s(ln.partition("-->")[2].strip().split(None, 1)[0]))
+        except (ValueError, IndexError):
+            continue
+    return end
+
+
 def publish(key: str, vtt: str, covered: tuple[float, float], complete: bool) -> None:
-    """Merge a freshly extracted region into the cached track and widen its covered ranges."""
+    """Merge a freshly extracted region into the cached track and widen its covered ranges.
+
+    A read-merge-write, so it runs under `_PUBLISH_LOCK`: two jobs publish into the same keys (a
+    window and the full pass, and every embedded track of the file rides each of them), and without
+    the lock the later writer drops whatever the earlier one merged — permanently, once `complete` is
+    set. It is also the only writer of the cache files, which is what keeps their fixed `.tmp`
+    staging names safe."""
+    with _PUBLISH_LOCK:
         old = cache_read(key)
         old_ranges, was_complete = read_meta(key)
         cache_write(key, merge_vtt(old or "", vtt))
@@ -298,6 +340,9 @@ _POOL = ThreadPoolExecutor(max_workers=3, thread_name_prefix="plex-subs")
 _LOCK = threading.Lock()
 _INFLIGHT: set[tuple] = set()
 _FAILURES: dict[str, tuple[str, float]] = {}
+_PUBLISH_LOCK = threading.Lock()
+# The ffmpeg children currently running, so a shutdown can kill them (see `shutdown`).
+_PROCS: set[subprocess.Popen] = set()
 
 
 class SubtitleUnavailable(Exception):
@@ -328,6 +373,20 @@ def reset_state() -> None:
         _FAILURES.clear()
 
 
+def shutdown() -> None:
+    """Stop extracting, for the app's shutdown. The pool's workers are non-daemon threads and Python
+    joins them at interpreter exit — so a worker still inside a full pass (minutes on a big file, up
+    to EXTRACT_TIMEOUT_S) held the process open after uvicorn had stopped, until the container runtime
+    SIGKILLed it. Queued jobs are dropped and running ffmpeg children are killed, which returns their
+    workers at once. Nothing that matters is lost: a publish is an atomic rename, and an interrupted
+    track is simply extracted again on the next poll."""
+    _POOL.shutdown(wait=False, cancel_futures=True)
+    with _LOCK:
+        procs = list(_PROCS)
+    for p in procs:
+        p.kill()
+
+
 def _fail(key: str, code: str) -> None:
     with _LOCK:
         _FAILURES[key] = (code, time.monotonic())
@@ -362,25 +421,57 @@ def ensure(rating_key: str, sub_ord: int, at: float = 0.0) -> dict:
     return {"state": "partial" if covered else "preparing", "code": None, "gen": generation(key)}
 
 
+def cached_track(rating_key: str, sub_ord: int) -> tuple[str | None, str | None]:
+    """What the <track> fetch can serve RIGHT NOW, scheduling nothing: the cached VTT (whatever it
+    covers), else the recent failure that explains why there is none.
+
+    Scheduling belongs to `ensure` — the status poll, which knows where the viewer is. The fetch does
+    not: when it scheduled too, it judged readiness at position 0, so a viewer who resumed an hour in
+    got an unrequested window over the head of the file (plus a remount when it landed), and with a
+    failure memo alive it was refused a window that WAS on disk while the poll called it showable."""
+    key = cache_key(rating_key, sub_ord)
+    hit = cache_read(key)
+    return (hit, None) if hit is not None else (None, _recent_failure(key))
+
+
 def _schedule(rating_key: str, sub_ord: int, at: float, want_window: bool) -> None:
     """Kick off the jobs this track still needs: the window that makes it watchable now, and the one
     full pass that makes it complete. Each is deduplicated by identity, so a 1 s status poll (or a
-    second viewer) joins the running job instead of starting another ffmpeg."""
+    second viewer) joins the running job instead of starting another ffmpeg.
+
+    A window's identity is its TRACK, never its position. The poll sends the live playhead, so a
+    position-keyed window minted a new identity every second and queued one more read of the file per
+    poll until the first window landed. One window per track is in flight at a time; a seek past it
+    is picked up by the next poll after it publishes. The full pass is per ITEM, because one read
+    fans out to every embedded text track of the file."""
     jobs: list[tuple[tuple, float | None]] = []
     if want_window:
-        w_start = max(0.0, at - WINDOW_PRE_S)
-        jobs.append((("win", rating_key, round(w_start)), w_start))
+        jobs.append((("win", rating_key, sub_ord), max(0.0, at - WINDOW_PRE_S)))
     jobs.append((("full", rating_key), None))
     for ident, window_start in jobs:
         with _LOCK:
             if ident in _INFLIGHT:
                 continue
             _INFLIGHT.add(ident)
-        _POOL.submit(_run, ident, rating_key, sub_ord, window_start)
+        _POOL.submit(_run, ident, rating_key, sub_ord, window_start, at)
 
 
-def _run(ident: tuple, rating_key: str, sub_ord: int, window_start: float | None) -> None:
+def _already_done(key: str, window_start: float | None, at: float) -> bool:
+    """A queued job looks again before it reads anything. While it waited, another job may have
+    covered it: every embedded track rides each pass over its file, so a window for one language also
+    publishes the others."""
+    if not _vtt_path(key).exists():
+        return False
+    ranges, complete = read_meta(key)
+    return complete or (window_start is not None and covers(ranges, at))
+
+
+def _run(
+    ident: tuple, rating_key: str, sub_ord: int, window_start: float | None, at: float
+) -> None:
     try:
+        if _already_done(cache_key(rating_key, sub_ord), window_start, at):
+            return
         _extract_and_publish(rating_key, sub_ord, window_start)
     except SubtitleUnavailable as e:
         _fail(cache_key(rating_key, sub_ord), e.code)
@@ -422,9 +513,11 @@ def _extract_and_publish(rating_key: str, sub_ord: int, window_start: float | No
         want = subs[sub_ord]
         if _is_bitmap(want):
             raise SubtitleUnavailable("bitmap")
-        if want.get("key"):  # external sidecar → one fetch, always the complete track
-            if window_start is not None:
-                return  # nothing to window — the full job serves it in a single fetch
+        if want.get("key"):
+            # An external sidecar is one small fetch from Plex, never a read of the media file, so a
+            # window job fetches the whole thing too. It must not be left to the full job: that
+            # identity is per ITEM (the embedded fan-out shares it), so a sidecar picked while an
+            # embedded track's full pass was running sat behind that pass for minutes.
             vtt = _sidecar_vtt(db, want)
             publish(cache_key(rating_key, sub_ord), vtt, (0.0, _FULL_END_S), True)
             return
@@ -443,13 +536,22 @@ def _extract_and_publish(rating_key: str, sub_ord: int, window_start: float | No
     start = window_start or 0.0
     length = (WINDOW_LEN_S + WINDOW_PRE_S) if window_start is not None else None
     end = _FULL_END_S if length is None else start + length
-    for emb_ord, vtt in _ffmpeg_extract(str(src), sorted(targets), window_start, length).items():
+    texts, clean = _ffmpeg_extract(str(src), sorted(targets), window_start, length)
+    for emb_ord, vtt in texts.items():
+        key = cache_key(rating_key, targets[emb_ord])
+        # A pass that exited non-zero holds only what ffmpeg reached before it died. Claim no more
+        # than its last cue, never call it complete, and remember the failure: published as complete,
+        # the track read "ready" for good while its subtitles stopped wherever ffmpeg did (the cache
+        # has no eviction, so nothing would ever have extracted it again).
+        covered_end = end if clean else start + _last_cue_end(vtt)
         publish(
-            cache_key(rating_key, targets[emb_ord]),
+            key,
             shift_vtt(vtt, -start),  # zero-based window clock → absolute cue times
-            (start, end),
-            length is None,
+            (start, covered_end),
+            clean and length is None,
         )
+        if not clean:
+            _fail(key, "extract_failed")
 
 
 def _sidecar_vtt(db: Session, stream: dict) -> str:
@@ -470,14 +572,16 @@ def _sidecar_vtt(db: Session, stream: dict) -> str:
 
 def _ffmpeg_extract(
     src: str, emb_ords: list[int], start: float | None, length: float | None
-) -> dict[int, str]:
+) -> tuple[dict[int, str], bool]:
     """Extract several embedded subtitle streams to WebVTT in ONE ffmpeg run (= one read of the file).
+    Returns the texts and whether ffmpeg exited cleanly — a non-zero exit can still leave partial
+    output, which is worth publishing but must never be taken for the whole track.
 
     `start`/`length` bound the read to a window. Without `-copyts` the cues come out on a zero-based
     clock (the caller shifts them back) — which is exactly why `-copyts` is absent: with it ffmpeg
     ignores `-t` and reads the whole container anyway (measured)."""
     if not emb_ords:
-        return {}
+        return {}, True
     with tempfile.TemporaryDirectory(prefix="plexsub-") as tmpdir:
         cmd = ["ffmpeg", "-nostdin", "-loglevel", "error"]
         if start:
@@ -489,7 +593,22 @@ def _ffmpeg_extract(
             if length is not None:
                 cmd += ["-t", f"{length:.3f}"]
             cmd += ["-f", "webvtt", "-y", str(out)]
-        r = subprocess.run(cmd, capture_output=True, timeout=EXTRACT_TIMEOUT_S)
+        # Popen rather than run(): the child is registered so `shutdown` can kill it.
+        proc = subprocess.Popen(
+            cmd, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE
+        )
+        with _LOCK:
+            _PROCS.add(proc)
+        try:
+            try:
+                _, err = proc.communicate(timeout=EXTRACT_TIMEOUT_S)
+            except subprocess.TimeoutExpired:
+                proc.kill()
+                proc.communicate()
+                raise
+        finally:
+            with _LOCK:
+                _PROCS.discard(proc)
         got = {
             o: p.read_text(encoding="utf-8", errors="replace")
             for o, p in outs.items()
@@ -497,7 +616,7 @@ def _ffmpeg_extract(
         }
     if not got:
         raise SubtitleUnavailable("extract_failed")
-    if r.returncode != 0:
-        # Partial output with a non-zero exit still beats nothing (a later pass widens it) — say so.
-        log.warning("ffmpeg subtitle pass exited %s: %s", r.returncode, (r.stderr or b"")[:200])
-    return got
+    clean = proc.returncode == 0
+    if not clean:
+        log.warning("ffmpeg subtitle pass exited %s: %s", proc.returncode, (err or b"")[:200])
+    return got, clean

backend/app/routes/plex/playback.py

@@ -10,6 +10,7 @@ from fastapi import (
     BackgroundTasks,
     Depends,
     HTTPException,
+    Query,
     Request,
     Response,
 )
@@ -87,40 +88,52 @@ def image(rating_key: str, request: Request, variant: str = "thumb") -> Response
     return imgcache.response(data, ct)
 
 
+# Both subtitle routes authenticate with the full `current_user`, not the cookie-only
+# `_require_session` the image proxies use. That lighter gate exists for bursts of low-sensitivity
+# poster fetches and skips the is_active / is_suspended / session-epoch checks — acceptable for a
+# poster, not for a route that starts reads of the media library: a suspended or "log out
+# everywhere"-revoked cookie kept scheduling them. They are not bursty (one poll a second, one fetch
+# per track), so the DB user-load costs nothing that matters.
+
+
 @router.get("/subtitle/{rating_key}/{sub_ord}/status")
-def subtitle_status(rating_key: str, sub_ord: int, request: Request, at: float = 0.0) -> dict:
+def subtitle_status(
+    rating_key: str,
+    sub_ord: int,
+    user: User = Depends(current_user),
+    # Bounded: a non-finite position used to reach the scheduler's arithmetic and 500.
+    at: float = Query(0.0, ge=0.0, allow_inf_nan=False),
+) -> dict:
     """Is this subtitle track showable yet — and if not, get it moving. Extraction runs in the
     background (see plex/subtitles.py for the measurements that forced it there), so the player polls
     this while it waits instead of hanging on a <track> that will not load for a minute or more.
 
     `at` is the viewer's absolute position: the extractor windows around it, so what gets read first
     is the part actually about to be watched."""
-    _require_session(request)
     return subtitles.ensure(rating_key, sub_ord, at)
 
 
 @router.get("/subtitle/{rating_key}/{sub_ord}")
 def subtitle(
-    rating_key: str, sub_ord: int, request: Request, offset: float = 0.0, at: float = 0.0
+    rating_key: str, sub_ord: int, user: User = Depends(current_user), offset: float = 0.0
 ) -> Response:
     """Serve the ord-th subtitle track as WebVTT for the browser's <track> — so choosing a subtitle
     overlays it directly (no HLS re-mux / session restart). Handles EXTERNAL sidecar subs (fetched
     from Plex via the stream `key` — the old `-map 0:s` broke on these since they're not in the file)
     AND embedded text subs (extracted from the local file). Image subs (PGS/VobSub) → 415. Disk-cached.
 
-    NEVER extracts inline. A cold track answers 202 and the extraction proceeds in the background; the
-    player mounts the <track> only once /status says there is something to show, so in practice this
-    is only ever hit warm (or on a cache eviction race, which the 202 sends back to polling)."""
-    _require_session(request)
+    A pure READ: it serves whatever is cached and schedules nothing — the status poll owns that,
+    because only it knows where the viewer is (see `subtitles.cached_track`). The player mounts the
+    <track> only once /status says there is something to show, so in practice this is only ever hit
+    warm; on a cache eviction race the 202 sends the player back to polling."""
     # The cached VTT holds ABSOLUTE cue times (per rk+ord); the per-session cue-shift is applied on
     # the way out, so a resume/seek to any offset reuses the same track (no cache-per-offset explosion).
-    st = subtitles.ensure(rating_key, sub_ord, at)
-    if st["state"] == "failed":
+    hit, code = subtitles.cached_track(rating_key, sub_ord)
+    if hit is None and code:
         raise HTTPException(
-            status_code=subtitles.UNAVAILABLE_STATUS.get(st["code"], 422),
-            detail={"code": "subtitle", "reason": st["code"]},
+            status_code=subtitles.UNAVAILABLE_STATUS.get(code, 422),
+            detail={"code": "subtitle", "reason": code},
         )
-    hit = subtitles.cache_read(subtitles.cache_key(rating_key, sub_ord))
     if hit is None:
         return Response(
             status_code=202, media_type="application/json", content='{"state":"preparing"}'

backend/tests/test_plex_subtitle_auth.py

@@ -0,0 +1,44 @@
+"""The subtitle routes must refuse a session the rest of the app refuses.
+
+They used to sit behind `_require_session`, the cookie-only gate the poster proxies use — a signed
+cookie was enough, with none of `current_user`'s checks (is_active, is_suspended, the session epoch
+that "log out everywhere" bumps). Fine for a poster; not for a route that starts reads of the media
+library. Found by the 2026-09 bug-hunt deep-review; the gate itself pre-dates that round.
+"""
+
+from fastapi.testclient import TestClient
+
+import app.auth as auth
+import app.state as app_state
+from app.db import get_db
+from app.main import app
+from app.plex import subtitles
+from app.routes.plex import playback
+
+
+class _NoPool:
+    def __init__(self) -> None:
+        self.jobs: list = []
+
+    def submit(self, fn, *args) -> None:
+        self.jobs.append((fn, args))
+
+
+def test_a_suspended_account_cannot_drive_the_subtitle_extractor(
+    db, make_user, override_dep, monkeypatch, tmp_path
+):
+    user = make_user(is_suspended=True)
+    override_dep(get_db, lambda: db)
+    monkeypatch.setattr(app_state, "_configured_cache", True)
+    # A validly SIGNED cookie for that account — which is all the old gate looked at.
+    for mod in (auth, playback):
+        monkeypatch.setattr(mod, "resolved_user_id", lambda request: (user.id, True))
+    monkeypatch.setattr(subtitles, "CACHE_DIR", tmp_path)
+    pool = _NoPool()
+    monkeypatch.setattr(subtitles, "_POOL", pool)
+    subtitles.reset_state()
+
+    client = TestClient(app)
+    assert client.get("/api/plex/subtitle/98614/0/status?at=600").status_code == 401
+    assert client.get("/api/plex/subtitle/98614/0").status_code == 401
+    assert pool.jobs == [], "a refused session still scheduled an extraction"

backend/tests/test_plex_subtitle_prepare.py

@@ -12,21 +12,27 @@ thread sat with it), and picking the second language cost the whole read again.
 
 These tests pin the three things that fixes it: the request never waits for ffmpeg, the extraction
 publishes a window around the playhead before the whole track, and one ffmpeg run covers every
-embedded text track. Pure — no Postgres, no ffmpeg.
+embedded text track — plus what the deep-review of that fix found (cues doubled at the window/full
+seam, a truncated pass cached as complete, a poll that respawned ffmpeg every second, a sidecar stuck
+behind an unrelated pass). Pure — no Postgres, no ffmpeg (the auth half is in
+test_plex_subtitle_auth.py, which needs the DB).
 """
 
 import subprocess
+import sys
 import threading
 import time
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
+from types import SimpleNamespace
 
 import pytest
 from fastapi.testclient import TestClient
 
 import app.state as app_state
+from app.auth import current_user
 from app.main import app
 from app.plex import subtitles
-from app.routes.plex import playback
 
 
 @pytest.fixture(autouse=True)
@@ -39,10 +45,10 @@ def _isolated_cache(tmp_path, monkeypatch):
 
 
 @pytest.fixture
-def client(monkeypatch) -> TestClient:
-    # The subtitle endpoints authenticate off the signed session cookie directly (not a FastAPI
-    # dependency), so there is no dependency_overrides hook to use.
-    monkeypatch.setattr(playback, "resolved_user_id", lambda request: (1, None))
+def client(monkeypatch, override_dep) -> TestClient:
+    # Neither route reads the user; the real `current_user` (and what it rejects) is pinned in
+    # test_plex_subtitle_auth.py against the DB.
+    override_dep(current_user, lambda: SimpleNamespace(id=1))
     monkeypatch.setattr(app_state, "_configured_cache", True)  # pass the instance-setup gate, no DB
     return TestClient(app)
 
@@ -92,15 +98,15 @@ def test_a_cold_track_answers_at_once_instead_of_waiting_for_ffmpeg(client, monk
     monkeypatch.setattr(subtitles, "_extract_and_publish", _slow)
     try:
         t0 = time.monotonic()
-        resp = client.get("/api/plex/subtitle/98614/0")
+        st = client.get("/api/plex/subtitle/98614/0/status")
         elapsed = time.monotonic() - t0
-        assert resp.status_code == 202, resp.text
+        assert st.status_code == 200 and st.json()["state"] == "preparing", st.text
         assert elapsed < 2.0, f"the request waited {elapsed:.1f}s on the extraction"
         assert started.wait(timeout=5), "the extraction never started in the background"
-        # …and the status endpoint answers just as fast while that extraction is still running.
+        # …and the <track> fetch answers just as fast while that extraction is still running.
         t0 = time.monotonic()
-        st = client.get("/api/plex/subtitle/98614/0/status")
-        assert st.status_code == 200 and st.json()["state"] == "preparing"
+        resp = client.get("/api/plex/subtitle/98614/0")
+        assert resp.status_code == 202, resp.text
         assert time.monotonic() - t0 < 2.0
     finally:
         release.set()
@@ -114,7 +120,7 @@ def test_a_burst_of_polls_starts_one_window_and_one_full_pass(monkeypatch):
     for _ in range(5):
         subtitles.ensure("98614", 0, at=600.0)
     assert len(pool.jobs) == 2  # one window (at 600 s) + the one full pass
-    # submit(_run, ident, rating_key, sub_ord, window_start) — the last arg is what distinguishes them
+    # submit(_run, ident, rating_key, sub_ord, window_start, at) — window_start tells them apart
     assert [a[3] for _, a in pool.jobs] == [570.0, None]  # the window first, then the full pass
 
 
@@ -191,18 +197,33 @@ def test_an_unshowable_track_says_why_and_maps_to_a_status(client, monkeypatch):
 # --- the ffmpeg pass itself ---------------------------------------------------------------------
 
 
-def _capture_ffmpeg(monkeypatch) -> list[list[str]]:
+class _FakeFfmpeg:
+    """Stands in for the ffmpeg child: writes the outputs its argv asked for, exits with `returncode`."""
+
+    def __init__(self, cmd: list[str], returncode: int) -> None:
+        self.cmd, self.returncode = cmd, returncode
+
+    def communicate(self, timeout=None):
+        for i, arg in enumerate(self.cmd):
+            if arg == "-y":
+                Path(self.cmd[i + 1]).write_text(
+                    _vtt(_cue("00:01.000", "00:02.000", "hi")), "utf-8"
+                )
+        return b"", b""
+
+    def kill(self) -> None:
+        pass
+
+
+def _capture_ffmpeg(monkeypatch, returncode: int = 0) -> list[list[str]]:
     """Record the argv of each ffmpeg run and satisfy it by writing the outputs it asked for."""
     runs: list[list[str]] = []
 
-    def _fake_run(cmd, **kwargs):
+    def _fake_popen(cmd, **kwargs):
         runs.append(cmd)
-        for i, arg in enumerate(cmd):
-            if arg == "-y":
-                Path(cmd[i + 1]).write_text(_vtt(_cue("00:01.000", "00:02.000", "hi")), "utf-8")
-        return subprocess.CompletedProcess(cmd, 0, b"", b"")
+        return _FakeFfmpeg(cmd, returncode)
 
-    monkeypatch.setattr(subtitles.subprocess, "run", _fake_run)
+    monkeypatch.setattr(subtitles.subprocess, "Popen", _fake_popen)
     return runs
 
 
@@ -210,12 +231,20 @@ def test_one_ffmpeg_run_extracts_every_embedded_text_track(monkeypatch):
     """The read is the cost and it is SHARED. Extracting one track at a time is what made picking
     Hungarian cost a full pass and then switching to English cost another one."""
     runs = _capture_ffmpeg(monkeypatch)
-    out = subtitles._ffmpeg_extract("/plex-media/f.mkv", [0, 1], None, None)
+    out, clean = subtitles._ffmpeg_extract("/plex-media/f.mkv", [0, 1], None, None)
     assert len(runs) == 1, "each extra ffmpeg run is another full read of the file"
-    assert set(out) == {0, 1}
+    assert set(out) == {0, 1} and clean
     assert runs[0].count("-map") == 2
 
 
+def test_a_non_zero_exit_is_reported_even_when_it_left_output(monkeypatch):
+    """ffmpeg can die partway and still leave a VTT behind. That text is worth keeping — but the
+    caller has to be TOLD, or it publishes a truncated track as the whole thing."""
+    _capture_ffmpeg(monkeypatch, returncode=1)
+    out, clean = subtitles._ffmpeg_extract("/plex-media/f.mkv", [0], None, None)
+    assert out and not clean
+
+
 def test_a_windowed_pass_bounds_the_read_and_never_uses_copyts(monkeypatch):
     """`-copyts` is the trap: with it ffmpeg IGNORES -t and reads the whole container anyway
     (measured — a [0, 600 s] window still took the full 145 s). So the window is extracted on a
@@ -261,3 +290,230 @@ def test_a_cached_vtt_from_an_older_build_counts_as_complete(tmp_path):
     Reading one as "covers nothing" would re-extract every already-cached track after the upgrade."""
     (tmp_path / "sub_98614_0.vtt").write_text(_vtt(_cue("00:01.000", "00:02.000", "x")), "utf-8")
     assert subtitles.read_meta("sub_98614_0") == ([[0.0, 1_000_000_000.0]], True)
+
+
+# --- the deep-review round: the seams between the window and the full pass ----------------------
+
+
+def test_a_window_and_the_full_pass_do_not_double_a_cue():
+    """THE doubled-subtitle bug. The two passes spell the same timestamp differently — a window comes
+    back through shift_vtt (always HH:MM:SS.mmm), a full pass is ffmpeg's own output, whose webvtt
+    muxer drops the hour below one hour — so de-duplicating on the text kept both, and every cue of a
+    window in the first hour was on screen twice."""
+    window = subtitles.shift_vtt(_vtt(_cue("00:30.123", "00:32.000", "hello")), -570.0)
+    full = _vtt(_cue("10:00.123", "10:02.000", "hello"))  # ffmpeg's spelling of the same cue
+    merged = subtitles.merge_vtt(window, full)
+    assert merged.count("-->") == 1, merged
+    assert "00:10:00.123 --> 00:10:02.000" in merged  # one canonical spelling in the served file
+
+
+def test_two_different_cues_at_the_same_time_are_both_kept():
+    """De-duplication is by timing AND text: a sign and a line of dialogue can share a timestamp."""
+    both = _vtt(_cue("00:10.000", "00:12.000", "sign"), _cue("00:10.000", "00:12.000", "dialogue"))
+    merged = subtitles.merge_vtt(both, both)
+    assert merged.count("-->") == 2 and "sign" in merged and "dialogue" in merged
+
+
+class _NullSession:
+    def __enter__(self):
+        return None
+
+    def __exit__(self, *exc):
+        return False
+
+
+# A file's subtitle streams as Plex lists them: a downloaded sidecar first (no `index` — it is not in
+# the file), then the embedded ones, one of them a bitmap track that can never become text.
+_STREAMS = [
+    {"key": "/library/streams/1", "codec": "srt"},  # sub_ord 0 — sidecar
+    {"index": 3, "codec": "subrip"},  # sub_ord 1 — embedded ordinal 0 (Hungarian)
+    {"index": 4, "codec": "pgs"},  # sub_ord 2 — embedded ordinal 1, bitmap
+    {"index": 5, "codec": "subrip"},  # sub_ord 3 — embedded ordinal 2 (English)
+]
+
+
+def _stub_item(monkeypatch, *, clean: bool = True, cues=("00:05.000", "00:07.000")):
+    """Everything around `_extract_and_publish` stubbed, so the function itself is what runs."""
+    monkeypatch.setattr(subtitles, "SessionLocal", _NullSession)
+    monkeypatch.setattr(
+        subtitles, "_text_streams", lambda db, rk: (SimpleNamespace(file_path="/f.mkv"), _STREAMS)
+    )
+    monkeypatch.setattr(
+        subtitles.plex_paths, "local_media_path", lambda db, fp: "/plex-media/f.mkv"
+    )
+    monkeypatch.setattr(
+        subtitles, "_sidecar_vtt", lambda db, s: _vtt(_cue("00:01.000", "00:03.000", "side"))
+    )
+    calls: list[tuple] = []
+
+    def _extract(src, emb_ords, start, length):
+        calls.append((list(emb_ords), start, length))
+        return {o: _vtt(_cue(*cues, f"track {o}")) for o in emb_ords}, clean
+
+    monkeypatch.setattr(subtitles, "_ffmpeg_extract", _extract)
+    return calls
+
+
+def test_one_window_publishes_every_embedded_text_track_at_absolute_time(monkeypatch):
+    """What `_extract_and_publish` itself is for — every other test stubs it out. One read over the
+    file's EMBEDDED text streams (the sidecar is not in the file and must not shift the `-map`
+    ordinals; the bitmap track is skipped), each published under its own sub_ord, and a window's
+    zero-based cues shifted back to where they sit in the film."""
+    calls = _stub_item(monkeypatch)
+    subtitles._extract_and_publish("98614", 3, 3570.0)
+    assert calls == [([0, 2], 3570.0, 330.0)], "one ffmpeg run over embedded ordinals 0 and 2"
+    english = subtitles.cache_read(subtitles.cache_key("98614", 3)) or ""
+    assert "00:59:35.000 --> 00:59:37.000" in english, english  # 3570 + 5 s
+    assert "track 2" in english
+    assert "track 0" in (subtitles.cache_read(subtitles.cache_key("98614", 1)) or "")
+    for never in (0, 2):  # the sidecar and the bitmap track come from nowhere in this pass
+        assert subtitles.cache_read(subtitles.cache_key("98614", never)) is None
+
+
+def test_a_full_pass_that_died_is_not_cached_as_the_whole_track(monkeypatch):
+    """A truncated full pass used to be published as complete: `ensure` answered "ready" forever, the
+    player stopped polling, and — the cache has no eviction — the subtitles simply ended wherever
+    ffmpeg had died, in every later session too. Now it covers what it provably reached, and the
+    rest of the film honestly reads as failed."""
+    _stub_item(monkeypatch, clean=False, cues=("39:58.000", "40:00.000"))
+    monkeypatch.setattr(subtitles, "_POOL", _ManualPool())
+    subtitles._extract_and_publish("98614", 3, None)
+    key = subtitles.cache_key("98614", 3)
+    assert subtitles.read_meta(key) == ([[0.0, 2400.0]], False)
+    assert subtitles.ensure("98614", 3, at=100.0)["state"] == "partial"
+    after = subtitles.ensure("98614", 3, at=3000.0)
+    assert (after["state"], after["code"]) == ("failed", "extract_failed")
+
+
+def test_polls_that_follow_the_playhead_join_one_window(monkeypatch):
+    """The poll sends the LIVE position, so it moves a second per poll. A window keyed on that
+    position was a new job every second — another read of the file per poll until the first landed."""
+    pool = _ManualPool()
+    monkeypatch.setattr(subtitles, "_POOL", pool)
+    for at in (600.0, 601.0, 602.0, 603.0, 604.0):
+        subtitles.ensure("98614", 1, at=at)
+    assert len(pool.jobs) == 2, "one window + the one full pass"
+
+
+def test_a_queued_window_that_was_covered_meanwhile_reads_nothing(monkeypatch):
+    """Every embedded track rides each pass over its file, so by the time a queued window for one
+    language runs, another language's window may already have published it."""
+    pool = _ManualPool()
+    monkeypatch.setattr(subtitles, "_POOL", pool)
+    calls = _stub_item(monkeypatch)
+    subtitles.ensure("98614", 3, at=600.0)
+    subtitles.publish(
+        subtitles.cache_key("98614", 3),
+        _vtt(_cue("10:00.000", "10:02.000", "x")),
+        (570.0, 900.0),
+        False,
+    )
+    pool.run_next()  # the window job
+    assert calls == []
+
+
+def test_a_sidecar_does_not_wait_behind_an_embedded_full_pass(monkeypatch):
+    """The full pass is one job per ITEM (the embedded fan-out shares it). A sidecar picked while an
+    embedded track's pass was running used to find that identity taken and wait out the whole read —
+    minutes — for a track that is one small fetch."""
+    pool = _ManualPool()
+    monkeypatch.setattr(subtitles, "_POOL", pool)
+    _stub_item(monkeypatch)
+    subtitles._INFLIGHT.add(("full", "98614"))  # an embedded track's full pass is running
+    subtitles.ensure("98614", 0, at=600.0)
+    pool.run_all()
+    assert subtitles.ensure("98614", 0, at=600.0)["state"] == "ready"
+
+
+def test_publishes_into_one_track_do_not_lose_each_other(monkeypatch):
+    """`publish` is a read-merge-write. Two jobs publishing into the same track at once — the window
+    and the full pass both fan out to every embedded track — used to let the later writer drop the
+    earlier one's cues. Forced interleaving: the first publisher pauses right after its read."""
+    key = subtitles.cache_key("98614", 0)
+    real_read = subtitles.cache_read
+    paused, resume = threading.Event(), threading.Event()
+    reads = {"n": 0}
+
+    def _read(k):
+        reads["n"] += 1
+        out = real_read(k)
+        if reads["n"] == 1:
+            paused.set()
+            resume.wait(timeout=5)
+        return out
+
+    monkeypatch.setattr(subtitles, "cache_read", _read)
+    first = threading.Thread(
+        target=subtitles.publish,
+        args=(key, _vtt(_cue("00:10.000", "00:12.000", "alpha")), (0.0, 330.0), False),
+    )
+    first.start()
+    assert paused.wait(timeout=5)
+    second = threading.Thread(
+        target=subtitles.publish,
+        args=(key, _vtt(_cue("10:00.000", "10:02.000", "bravo")), (570.0, 900.0), False),
+    )
+    second.start()
+    time.sleep(0.3)  # without the lock, the second publisher finishes in here
+    resume.set()
+    first.join(timeout=5)
+    second.join(timeout=5)
+    merged = real_read(key) or ""
+    assert "alpha" in merged and "bravo" in merged, merged
+    assert subtitles.read_meta(key)[0] == [[0.0, 330.0], [570.0, 900.0]]
+
+
+def test_shutdown_kills_a_running_extraction(monkeypatch):
+    """The pool's workers are non-daemon threads that Python joins at exit, so one still inside a
+    full pass held a restarting process open for the rest of the read. Stand-in child: a real
+    process that would sleep for a minute."""
+    real_popen = subprocess.Popen
+    monkeypatch.setattr(
+        subtitles.subprocess,
+        "Popen",
+        lambda cmd, **kw: real_popen([sys.executable, "-c", "import time; time.sleep(60)"], **kw),
+    )
+    monkeypatch.setattr(subtitles, "_POOL", ThreadPoolExecutor(max_workers=1))
+    finished = threading.Event()
+
+    def _extract():
+        try:
+            subtitles._ffmpeg_extract("/plex-media/f.mkv", [0], None, None)
+        except subtitles.SubtitleUnavailable:
+            pass  # killed before it wrote anything — exactly what a shutdown does
+        finally:
+            finished.set()
+
+    threading.Thread(target=_extract, daemon=True).start()
+    deadline = time.monotonic() + 10
+    while not subtitles._PROCS and time.monotonic() < deadline:
+        time.sleep(0.02)
+    assert subtitles._PROCS, "the extraction never registered its child"
+    subtitles.shutdown()
+    assert finished.wait(timeout=10), "the extraction was still running after shutdown"
+
+
+def test_the_track_fetch_serves_what_is_cached_and_schedules_nothing(client, monkeypatch):
+    """The <track> URL carries no position, and it used to run the scheduler anyway — judging
+    readiness at position 0. A viewer resumed an hour in got an unrequested extraction over the head
+    of the file, and with a failure memo alive the fetch refused a window that WAS on disk while the
+    status poll (which knows the position) called the track showable: silently blank subtitles."""
+    pool = _ManualPool()
+    monkeypatch.setattr(subtitles, "_POOL", pool)
+    assert client.get("/api/plex/subtitle/98614/0").status_code == 202
+    assert pool.jobs == [], "the fetch scheduled an extraction"
+    key = subtitles.cache_key("98614", 0)
+    subtitles.publish(key, _vtt(_cue("59:54.187", "59:58.650", "window")), (3570.0, 3900.0), False)
+    subtitles._fail(key, "extract_failed")  # the full pass behind that window failed
+    resp = client.get("/api/plex/subtitle/98614/0")
+    assert resp.status_code == 200 and "window" in resp.text
+    assert pool.jobs == []
+
+
+@pytest.mark.parametrize("at", ["inf", "-1", "nan"])
+def test_an_impossible_position_is_rejected(client, monkeypatch, at):
+    """`?at=inf` reached the scheduler's arithmetic and came back as a 500 with a traceback."""
+    pool = _ManualPool()
+    monkeypatch.setattr(subtitles, "_POOL", pool)
+    assert client.get(f"/api/plex/subtitle/98614/0/status?at={at}").status_code == 422
+    assert pool.jobs == []

frontend/src/lib/api/plex.ts

@@ -650,7 +650,8 @@ export const plexApi = {
   // cue times onto the HLS session's zero-based clock so subtitles line up with speech after a seek.
   // `gen` is the cached track's generation: extraction publishes a window around the playhead first
   // and widens it in the background, so the URL has to change when it grows or the browser would keep
-  // serving the first, short VTT from its own cache.
+  // serving the first, short VTT from its own cache. It deliberately carries NO position: the fetch
+  // is a pure read of what is cached — scheduling is the status poll's job, since only it knows `at`.
   plexSubtitleUrl: (id: string, ord: number, offset = 0, gen = 0): string => {
     const p = new URLSearchParams();
     if (offset) p.set("offset", String(offset));