feat: embedded subtitles in the video player (MNP 3.3)

MSE decodes no in-band text track, so a subtitle cannot ride inside the
fragmented MP4 the player is fed. The node extracts one track whole, converts
it to WebVTT and caches it under its own hash; the client pulls that blob
through the ordinary file_req/chunk path and hangs a <track> on the video
element — the same indirection as a TMDB poster or an audio transcode, which
is what makes a film's subtitles extracted once in the life of the file rather
than once per viewing. Whole-file also makes the cues absolute, so a seek and
an audio-language change both leave the track untouched.

**The ordinal counts every subtitle stream, including the ones never listed.**
Only text codecs are offered: a bitmap track (PGS, VOBSUB — about a fifth of a
real library) has no path to WebVTT without OCR, and one extracted anyway
yields a header with no cues, which is a menu entry that shows nothing and
reports no error. Numbering the survivors of that filter would give a
PGS/SRT/SRT file the ordinals 0 and 1 for its text tracks and `-map 0:s:0`
would then extract the PGS — the same trap `AudioTrack.ordinal` exists for,
one level deeper. A fixture whose first subtitle stream cannot be decoded pins
it, and the handler checks membership of the probed list, never a range.

Additive and MINOR: the selector is drawn from `subtitle_tracks` in the node's
own `stream_init` and from no version number, so `subtitle_req` is never sent
to a peer that would not answer it. The floor stays at 3.0.

Also here: a failed extraction never touches playback, a superseded reply
cannot install its blob over a newer choice, and `_languageName` is shared
with the audio labels — lifted by both label harnesses, since a lift that
names one function stops covering the rule the moment logic moves out of it.

Tests: 9 node (tracks told apart by the words in the extracted cues, not by
tags), 10 client. Full suite green: 1545 node/common, 1252 hub.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UGY17EPph5LsLzePPXhUVc

3 files changed, 549 insertions, 0 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/media_probe.py b/packages/meshbay-node/src/meshbay_node/media_probe.py
index 7858ebe..06b1e28 100644
--- a/packages/meshbay-node/src/meshbay_node/media_probe.py
+++ b/packages/meshbay-node/src/meshbay_node/media_probe.py
@@ -40,6 +40,37 @@ class AudioTrack:
     channels: int | None
 
 
+# Subtitle codecs ffmpeg can convert to WebVTT, which is the only thing MSE
+# can be given. An allow-list rather than a bitmap deny-list: the cost of
+# wrongly excluding an exotic text codec is a track nobody can pick, and the
+# cost of wrongly including a bitmap one is a track that is picked and then
+# displays nothing, with no error to lead anyone back here.
+TEXT_SUBTITLE_CODECS = frozenset({
+    "subrip", "srt", "ass", "ssa", "mov_text", "webvtt", "text",
+    "subviewer", "subviewer1", "sami", "realtext", "stl", "jacosub",
+    "microdvd", "mpl2", "vplayer", "pjs",
+})
+
+
+@dataclass(frozen=True)
+class SubtitleTrack:
+    """
+    One selectable subtitle track, guaranteed convertible to WebVTT.
+
+    **`ordinal` counts every subtitle stream, including the bitmap ones this
+    list does not carry**, because that is what `-map 0:s:<n>` counts. The
+    same trap as `AudioTrack.ordinal` one level deeper: filtering the list and
+    numbering the survivors would give a file whose streams are PGS, SRT, SRT
+    the ordinals 0 and 1 for its two text tracks, and `-map 0:s:0` would then
+    extract the PGS stream — which produces an empty WebVTT rather than an
+    error, so the viewer gets a subtitle track with no subtitles in it.
+    """
+    ordinal: int
+    language: str | None
+    title: str | None
+    codec_name: str | None
+
+
 @dataclass
 class VideoProbe:
     """
@@ -57,6 +88,7 @@ class VideoProbe:
     height: int | None
     raw_codec_name: str | None
     audio_tracks: list[AudioTrack] = field(default_factory=list)
+    subtitle_tracks: list[SubtitleTrack] = field(default_factory=list)
 
 
 async def probe_video(path: str) -> VideoProbe:
@@ -93,6 +125,14 @@ async def probe_video(path: str) -> VideoProbe:
     group does not want. `has_audio` stays as the single question the muxing
     decisions ask, and is now `bool(audio_tracks)`.
 
+    **Only text subtitle tracks are reported.** A library's embedded subtitles
+    are roughly four-fifths text (subrip, ass) and one-fifth bitmap (PGS,
+    VOBSUB); a bitmap track has no path to WebVTT without OCR, so listing one
+    would offer a choice that silently displays nothing. A file whose only
+    subtitles are bitmap therefore reports none at all and gets no selector,
+    exactly like a file with no subtitles — which is a true statement about
+    what this node can serve, not a concealed failure.
+
     width/height come from the same ffprobe call (one extra `-show_entries`
     field, no second process spawn) — resolution is deliberately never
     guessed from the filename (docs/mediacenter.md §3.5).
@@ -116,6 +156,8 @@ async def probe_video(path: str) -> VideoProbe:
     width: int | None = None
     height: int | None = None
     audio_tracks: list[AudioTrack] = []
+    subtitle_tracks: list[SubtitleTrack] = []
+    subtitle_streams_seen = 0
     for s in info.get("streams", []):
         if s.get("codec_type") == "video" and not v_codec:
             cn = s.get("codec_name", "")
@@ -142,6 +184,20 @@ async def probe_video(path: str) -> VideoProbe:
                 codec_name=s.get("codec_name") or None,
                 channels=s.get("channels"),
             ))
+        elif s.get("codec_type") == "subtitle":
+            # Counted before the filter, never after — see SubtitleTrack.
+            ordinal = subtitle_streams_seen
+            subtitle_streams_seen += 1
+            codec_name = (s.get("codec_name") or "").strip() or None
+            if codec_name not in TEXT_SUBTITLE_CODECS:
+                continue
+            tags = s.get("tags") or {}
+            subtitle_tracks.append(SubtitleTrack(
+                ordinal=ordinal,
+                language=(tags.get("language") or "").strip() or None,
+                title=(tags.get("title") or "").strip() or None,
+                codec_name=codec_name,
+            ))
 
     has_audio = bool(audio_tracks)
     codec = None
@@ -155,4 +211,5 @@ async def probe_video(path: str) -> VideoProbe:
         height=height,
         raw_codec_name=raw_codec_name,
         audio_tracks=audio_tracks,
+        subtitle_tracks=subtitle_tracks,
     )
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index 67ca380..f858a53 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -287,6 +287,15 @@ BROWSER_INCOMPATIBLE_AUDIO_EXTS = frozenset({".wma", ".mpc"})
 # bounded generously so one slow/huge outlier can't pin a transcode slot
 # (shared with video, MAX_CONCURRENT_TRANSCODES above) indefinitely.
 AUDIO_TRANSCODE_TIMEOUT_SECS = 120
+# Extracting one subtitle track is a demux and a text conversion, not an
+# encode: measured at ~1.2 s for a full film. The bound is generous against a
+# pathological container rather than against the work itself, and it is short
+# next to the audio one because nothing here decodes a media stream.
+SUBTITLE_EXTRACT_TIMEOUT_SECS = 60
+# A subtitle file is text; a film's is ~96 KB. Anything past this is not a
+# subtitle track, it is an ffmpeg that found something else to write, and it
+# would sit in the media cache for ever.
+SUBTITLE_MAX_BYTES = 8 * 1024 * 1024
 # Bundle fetches are served in the pre-proof window (C4). Bounded and audited
 # until the native client removes remote keypair bundles entirely.
 MAX_PRE_PROOF_FETCHES = 4
@@ -675,6 +684,8 @@ class WebRTCPeerSession:
                 self._spawn(self._do_music_meta_request(msg))
             elif mtype == MNP.AUDIO_TRANSCODE_REQ:
                 self._spawn(self._do_audio_transcode_request(msg))
+            elif mtype == MNP.SUBTITLE_REQ:
+                self._spawn(self._do_subtitle_request(msg))
             elif mtype == MNP.MEMBER_UNPIN:
                 self._do_member_unpin(msg)
             elif mtype == MNP.GEK_ROTATE:
@@ -4016,6 +4027,87 @@ class WebRTCPeerSession:
                    "file_id": file_id, "hash": transcode_hash,
                    "size": len(blob), "mime": "audio/mp4"})
 
+    async def _do_subtitle_request(self, msg: dict) -> None:
+        """
+        One embedded subtitle track, extracted whole-file to WebVTT and served
+        back through the ordinary file_req/chunk path — the same indirection
+        as `_do_audio_transcode_request` above, and cached the same way, so a
+        film's subtitles are extracted once in the life of the file rather
+        than once per viewing.
+
+        The ordinal is validated against `probe_video`'s *filtered* list and
+        then used as the ffmpeg `-map 0:s:<n>` argument, which is only correct
+        because `SubtitleTrack.ordinal` counts every subtitle stream including
+        the bitmap ones the list omits (see media_probe.py). Checking
+        membership rather than range is what makes that hold: a bitmap
+        ordinal is in range and is not in the list, and extracting it would
+        produce an empty WebVTT — a subtitle track with no subtitles in it,
+        which reports no error anywhere.
+        """
+        ctx = self._group_ctx()
+        file_id = msg.get("file_id", "")
+        entry = ctx["index"].get_entry(file_id)
+        if not entry:
+            self._send({"type": "error", "detail": "File not found"})
+            return
+        file_path = entry_abs_path(ctx["roots"], entry)
+        if file_path is None:
+            self._send({"type": "error", "detail": ROOT_NOT_SERVED})
+            return
+        if not file_path.exists():
+            self._send({"type": "error", "detail": "File not on disk"})
+            return
+
+        media_cache = self._ctx.get("media_cache")
+        if media_cache is None:
+            self._send({"type": "error", "detail": "Subtitles unavailable"})
+            return
+
+        try:
+            ordinal = int(msg.get("track", 0) or 0)
+        except (TypeError, ValueError):
+            ordinal = -1
+
+        synthetic_id = f"subtitle:{entry.id}:{ordinal}"
+        cached_hash = await media_cache.get_thumb_hash_by_file_id(synthetic_id)
+        if cached_hash is not None:
+            blob = await media_cache.get_thumb(cached_hash)
+            if blob is not None:
+                self._send({"type": MNP.SUBTITLE_RESP, "v": MNP_VERSION,
+                            "file_id": file_id, "track": ordinal,
+                            "hash": cached_hash, "size": len(blob),
+                            "mime": "text/vtt"})
+                return
+            # Cached hash but the blob was pruned: fall through and extract
+            # again, same as a cold cache.
+
+        probe = await _probe_video(str(file_path))
+        if not any(tr.ordinal == ordinal for tr in probe.subtitle_tracks):
+            # Not a range check — see this method's docstring.
+            self._send({"type": "error", "detail": "No such subtitle track"})
+            return
+
+        sem = self._transcode_semaphore()
+        if sem.locked() and sem._value <= 0:
+            self._send({"type": "error", "detail": "Server busy, retry shortly"})
+            return
+        async with sem:
+            try:
+                blob = await _extract_subtitle_to_webvtt(file_path, ordinal)
+            except Exception as e:
+                log.warning("Subtitle extract failed for %s track %d: %s",
+                            entry.id[:12], ordinal, e)
+                self._send({"type": "error", "detail": f"Subtitle extraction failed: {e}"})
+                return
+
+        subtitle_hash = blake3.blake3(blob).hexdigest()
+        await media_cache.put_thumb(subtitle_hash, synthetic_id, blob)
+        self._audit("subtitle_extract", f"{entry.name} [{ordinal}]")
+        self._send({"type": MNP.SUBTITLE_RESP, "v": MNP_VERSION,
+                    "file_id": file_id, "track": ordinal,
+                    "hash": subtitle_hash, "size": len(blob),
+                    "mime": "text/vtt"})
+
     async def _do_music_meta_request(self, msg: dict) -> None:
         """
         docs/musicbay.md §4.3: MusicBrainz metadata for one track, resolved
@@ -6143,6 +6235,21 @@ class WebRTCPeerSession:
                 for tr in probe.audio_tracks
             ],
             "audio_track": audio_track if has_audio else None,
+            # Same discovery-from-the-answer shape as `audio_tracks`: a node
+            # too old to enumerate sends no list, the client shows no selector
+            # and never sends `subtitle_req` to a peer that would answer
+            # "unknown message type". Text tracks only — a bitmap one has no
+            # WebVTT to offer (media_probe.py), so it is absent here rather
+            # than present and unplayable.
+            "subtitle_tracks": [
+                {
+                    "i": tr.ordinal,
+                    "lang": tr.language,
+                    "title": tr.title,
+                    "codec": tr.codec_name,
+                }
+                for tr in probe.subtitle_tracks
+            ],
         })
 
         # A client that says nothing gets the old behaviour, which is why this
@@ -6360,6 +6467,64 @@ async def _transcode_audio_to_aac(file_path: Path) -> bytes:
         tmp_path.unlink(missing_ok=True)
 
 
+async def _extract_subtitle_to_webvtt(file_path: Path, ordinal: int) -> bytes:
+    """
+    One subtitle track out of a container, whole, as WebVTT.
+
+    Whole-file rather than following the stream, which is what makes the
+    result reusable: the cues carry the source's own absolute timestamps, so
+    the same extraction serves every seek, every audio-language change and
+    every later viewing, and the `<track>` the client attaches never has to be
+    rebuilt. It is also the only shape the cache makes sense in — a segment
+    keyed on a seek position would be a different blob every time.
+
+    `-map 0:s:<ordinal>` counts subtitle streams (see media_probe.py), and
+    `-c:s webvtt` converts subrip/ass to text; a bitmap codec reaching here
+    would produce an empty file rather than an error, which is why the caller
+    checks membership of the probed text list first and never a range.
+
+    Written to a temp file rather than read off a pipe: the caller wants one
+    complete blob to hash and cache, and there is nothing to gain from
+    streaming a hundred kilobytes.
+    """
+    fd, tmp_name = tempfile.mkstemp(suffix=".vtt")
+    os.close(fd)
+    tmp_path = Path(tmp_name)
+    try:
+        proc = await asyncio.create_subprocess_exec(
+            platform.ffmpeg_cmd(), "-hide_banner", "-loglevel", "error", "-y",
+            "-i", str(file_path),
+            "-map", f"0:s:{ordinal}", "-c:s", "webvtt",
+            "-f", "webvtt", str(tmp_path),
+            stdout=asyncio.subprocess.DEVNULL,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        try:
+            _, stderr = await asyncio.wait_for(
+                proc.communicate(), timeout=SUBTITLE_EXTRACT_TIMEOUT_SECS)
+        except asyncio.TimeoutError:
+            proc.kill()
+            await proc.wait()
+            raise RuntimeError(f"ffmpeg timed out after {SUBTITLE_EXTRACT_TIMEOUT_SECS}s")
+        if proc.returncode != 0:
+            raise RuntimeError(
+                f"ffmpeg exited {proc.returncode}: {stderr.decode(errors='replace')[:300]}")
+        size = tmp_path.stat().st_size
+        if size > SUBTITLE_MAX_BYTES:
+            raise RuntimeError(f"subtitle track is {size} bytes, over the {SUBTITLE_MAX_BYTES} cap")
+        blob = tmp_path.read_bytes()
+        # A WebVTT file that is only its header has no cues in it. That is what
+        # a bitmap track extracted by mistake produces, and what a text track
+        # whose stream is empty produces; either way there is nothing to show,
+        # and an empty track attached to the player is worse than none — it
+        # appears in the menu and does nothing when picked.
+        if len(blob.strip()) <= len(b"WEBVTT"):
+            raise RuntimeError("extracted subtitle contains no cues")
+        return blob
+    finally:
+        tmp_path.unlink(missing_ok=True)
+
+
 class WebRTCTransport:
     """
     Manages WebRTC peer connections for browser clients.
diff --git a/packages/meshbay-node/tests/test_stream_subtitle_tracks.py b/packages/meshbay-node/tests/test_stream_subtitle_tracks.py
new file mode 100644
index 0000000..cafb2e3
--- /dev/null
+++ b/packages/meshbay-node/tests/test_stream_subtitle_tracks.py
@@ -0,0 +1,327 @@
+"""
+The viewer picks a subtitle track, and only the ones that can be shown.
+
+MSE decodes no in-band text track, so a subtitle cannot ride inside the
+fragmented MP4 the player is fed: it is extracted whole, converted to WebVTT,
+cached under its own hash and pulled through the ordinary chunk path. Whole,
+because that makes the cue timestamps absolute — a seek re-extracts nothing
+and the `<track>` survives every restart of the MediaSource underneath it.
+
+Two things here are about *not* offering something. Roughly a fifth of the
+subtitle streams in a real library are bitmap (PGS, VOBSUB) and have no path
+to WebVTT without OCR; a bitmap track extracted anyway yields a WebVTT with a
+header and no cues, which is a subtitle track that appears in the menu and
+does nothing. So they are not listed — and, because they still occupy a
+position in `-map 0:s:<n>`, the ordinal of the tracks that *are* listed is not
+their position in the list. That is the whole trap, and it is the same one
+`AudioTrack.ordinal` exists for, one level deeper.
+
+**The fixture's unusable stream is TTML, not bitmap, and that is deliberate.**
+ffmpeg refuses to encode text to bitmap, so a PGS stream cannot be synthesised
+here at all; TTML is a stream this ffmpeg has no decoder for, which is the
+same branch — `codec_name not in TEXT_SUBTITLE_CODECS` — reached by exactly
+the same route. The real bitmap codec names are asserted against the allow-list
+directly, where no fixture is needed.
+
+Tracks are told apart by **the words in the extracted cues**, never by their
+language tags: a tag only proves the node copied a string it was handed.
+"""
+
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_common.crypto import generate_gek
+from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.media_probe import TEXT_SUBTITLE_CODECS
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession, _probe_video
+
+from conftest import needs_subprocess, one_root
+
+_HAVE_FFMPEG = shutil.which("ffmpeg") and shutil.which("ffprobe")
+# `asyncio` is per-test rather than on the module: one test here needs no
+# event loop, and a module-wide mark on a synchronous function is a warning
+# that reads as a broken test every time the suite runs.
+pytestmark = [
+    pytest.mark.skipif(not _HAVE_FFMPEG, reason="ffmpeg/ffprobe not installed"),
+    needs_subprocess,
+]
+
+# Ordinal 0 is the unusable one and is never listed; 1 and 2 are the text
+# tracks. The words differ per track because that is what the assertions read.
+_CUE_WORD = {1: "francaise", 2: "English"}
+
+_SRT_FR = """1
+00:00:01,000 --> 00:00:03,000
+Ceci est la piste francaise.
+"""
+
+_SRT_EN = """1
+00:00:01,000 --> 00:00:03,000
+This is the English track.
+"""
+
+
+def _make_subtitled_clip(path: Path) -> None:
+    """~6 s of video, then three subtitle streams: TTML, then two text ones.
+
+    The video and audio are muxed first, so the subtitle streams sit at
+    container indices 2, 3 and 4 while their subtitle *ordinals* are 0, 1 and
+    2 — and the first ordinal belongs to a stream that is never listed, so the
+    listed tracks are 1 and 2 and never 0 and 1.
+    """
+    tmp = path.parent
+    fr, en = tmp / "fr.srt", tmp / "en.srt"
+    fr.write_text(_SRT_FR, encoding="utf-8")
+    en.write_text(_SRT_EN, encoding="utf-8")
+    base = tmp / "base.mp4"
+    subprocess.run(
+        ["ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
+         "-f", "lavfi", "-i", "testsrc=size=320x240:rate=10:duration=6",
+         "-f", "lavfi", "-i", "sine=duration=6",
+         "-c:v", "libx264", "-preset", "ultrafast", "-c:a", "aac",
+         "-shortest", str(base)],
+        check=True, capture_output=True)
+    subprocess.run(
+        ["ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
+         "-i", str(base), "-i", str(fr), "-i", str(en),
+         "-map", "0:v", "-map", "0:a", "-map", "1", "-map", "1", "-map", "2",
+         "-c:v", "copy", "-c:a", "copy",
+         "-c:s:0", "ttml", "-c:s:1", "mov_text", "-c:s:2", "mov_text",
+         "-metadata:s:s:0", "language=fre",
+         "-metadata:s:s:1", "language=fre",
+         "-metadata:s:s:2", "language=eng",
+         str(path)],
+        check=True, capture_output=True)
+
+
+class _FakeMediaCache:
+    """The three methods `_do_subtitle_request` uses, and a count of the puts.
+
+    A double rather than the real cache because what is under test is the
+    handler's use of it — that it looks before extracting, and extracts once.
+    """
+
+    def __init__(self):
+        self.blobs: dict[str, bytes] = {}
+        self.by_file_id: dict[str, str] = {}
+        self.puts = 0
+
+    async def get_thumb_hash_by_file_id(self, file_id: str) -> str | None:
+        return self.by_file_id.get(file_id)
+
+    async def get_thumb(self, thumb_hash: str) -> bytes | None:
+        return self.blobs.get(thumb_hash)
+
+    async def put_thumb(self, thumb_hash: str, file_id: str, blob: bytes) -> None:
+        self.puts += 1
+        self.blobs[thumb_hash] = blob
+        self.by_file_id[file_id] = thumb_hash
+
+
+def _session(video_path: Path, gek: bytes):
+    import blake3
+    file_bytes = video_path.read_bytes()
+    file_id = blake3.blake3(file_bytes).hexdigest()
+
+    sk_node = Ed25519PrivateKey.generate()
+    index = GroupIndex(group_id="g" * 32, sk_node=sk_node, gek=gek)
+    from meshbay_common.protocol import IndexEntry
+    index.add_entry(IndexEntry(
+        id=file_id, name=video_path.name, path=video_path.parent.name,
+        size=len(file_bytes), type="video", added_at=0))
+
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = {
+        "roots": one_root(video_path.parent),
+        "index": index,
+        "gek": gek,
+        "sk_node": sk_node,
+        "max_concurrent_streams": 4,
+        "media_cache": _FakeMediaCache(),
+    }
+    session._group_id = None
+    session._user_id = "tester"
+    session._stream_stopped = False
+    session._stream_keepalives = 0
+    session.sent = []
+    session._send = session.sent.append
+    session._audit = lambda *a, **k: None
+    return session, file_id
+
+
+async def _ask_for(session, file_id: str, track: int) -> dict:
+    before = len(session.sent)
+    await session._do_subtitle_request({"file_id": file_id, "track": track})
+    replies = session.sent[before:]
+    assert len(replies) == 1, f"expected one reply, got {replies}"
+    return replies[0]
+
+
+@pytest.mark.asyncio
+async def test_probe_lists_only_text_tracks_and_numbers_them_by_stream(tmp_path):
+    """The trap this feature is one wrong line away from.
+
+    Numbering the survivors of the filter would give the two text tracks the
+    ordinals 0 and 1, and `-map 0:s:0` would then extract the stream that
+    cannot be decoded — which produces an empty WebVTT, not an error.
+    """
+    clip = tmp_path / "clip.mp4"
+    _make_subtitled_clip(clip)
+
+    probe = await _probe_video(str(clip))
+
+    assert [tr.ordinal for tr in probe.subtitle_tracks] == [1, 2], (
+        "the listed tracks must keep their position among all subtitle "
+        "streams, not be renumbered from zero")
+    assert [tr.language for tr in probe.subtitle_tracks] == ["fre", "eng"]
+    assert all(tr.codec_name == "mov_text" for tr in probe.subtitle_tracks)
+
+    # The fixture really does carry a subtitle stream that is not listed, and
+    # really does put the subtitles at container indices of their own — or the
+    # assertion above distinguishes nothing.
+    raw = subprocess.run(
+        ["ffprobe", "-v", "error", "-select_streams", "s",
+         "-show_entries", "stream=index,codec_name", "-of", "csv=p=0", str(clip)],
+        check=True, capture_output=True, text=True)
+    rows = [line.split(",") for line in raw.stdout.split()]
+    assert [int(r[0]) for r in rows] == [2, 3, 4]
+    assert [r[1] for r in rows] == ["ttml", "mov_text", "mov_text"]
+
+
+def test_bitmap_codecs_are_not_offered():
+    """The 20 % no amount of ffmpeg turns into text.
+
+    Asserted against the allow-list rather than a fixture because ffmpeg
+    cannot encode text to bitmap, so a PGS or VOBSUB stream cannot be built
+    here — while the names ffprobe reports for them are fixed and are what the
+    filter is actually matched against.
+    """
+    for codec in ("hdmv_pgs_subtitle", "dvd_subtitle", "dvb_subtitle", "xsub"):
+        assert codec not in TEXT_SUBTITLE_CODECS
+    # And the two that make up four-fifths of a real library are.
+    assert "subrip" in TEXT_SUBTITLE_CODECS
+    assert "ass" in TEXT_SUBTITLE_CODECS
+
+
+@pytest.mark.asyncio
+async def test_stream_init_announces_the_tracks(tmp_path):
+    """How a client discovers this node can do subtitles at all.
+
+    From the answer, never from a version number: a node too old to enumerate
+    sends no list, the client draws no selector and never asks.
+    """
+    clip = tmp_path / "clip.mp4"
+    _make_subtitled_clip(clip)
+    gek = generate_gek()
+    session, file_id = _session(clip, gek)
+
+    await session._stream_video_inner(
+        {"file_id": file_id, "start": 0, "credits": 0})
+
+    init = next(m for m in session.sent if m.get("type") == "stream_init")
+    assert [tr["i"] for tr in init["subtitle_tracks"]] == [1, 2]
+    assert [tr["lang"] for tr in init["subtitle_tracks"]] == ["fre", "eng"]
+
+
+@pytest.mark.parametrize("track", [1, 2])
+@pytest.mark.asyncio
+async def test_the_requested_track_is_the_one_extracted(tmp_path, track):
+    """Read out of the cues, not out of the reply's language tag."""
+    clip = tmp_path / "clip.mp4"
+    _make_subtitled_clip(clip)
+    gek = generate_gek()
+    session, file_id = _session(clip, gek)
+
+    reply = await _ask_for(session, file_id, track)
+
+    assert reply["type"] == "subtitle_resp"
+    assert reply["track"] == track
+    assert reply["mime"] == "text/vtt"
+    vtt = session._ctx["media_cache"].blobs[reply["hash"]].decode("utf-8")
+    assert vtt.startswith("WEBVTT")
+    assert _CUE_WORD[track] in vtt
+    other = _CUE_WORD[1 if track == 2 else 2]
+    assert other not in vtt, (
+        f"track {track} carries the other track's words, so the ordinal was "
+        "mapped to the wrong stream")
+
+
+@pytest.mark.asyncio
+async def test_a_track_that_cannot_be_decoded_is_refused_not_served_empty(tmp_path):
+    """Ordinal 0 exists in the container and is not in the list.
+
+    A viewer cannot ask for it through the interface, which draws its menu
+    from the list — but the ordinal travels on the wire, and a reply carrying
+    a WebVTT with no cues in it would be a track that appears and shows
+    nothing, with no error anywhere to lead back here.
+    """
+    clip = tmp_path / "clip.mp4"
+    _make_subtitled_clip(clip)
+    gek = generate_gek()
+    session, file_id = _session(clip, gek)
+
+    reply = await _ask_for(session, file_id, 0)
+
+    assert reply["type"] == "error"
+    assert session._ctx["media_cache"].puts == 0, (
+        "nothing may be cached for a track that could not be extracted")
+
+
+@pytest.mark.asyncio
+async def test_an_ordinal_past_the_end_is_refused(tmp_path):
+    clip = tmp_path / "clip.mp4"
+    _make_subtitled_clip(clip)
+    gek = generate_gek()
+    session, file_id = _session(clip, gek)
+
+    reply = await _ask_for(session, file_id, 9)
+
+    assert reply["type"] == "error"
+
+
+@pytest.mark.asyncio
+async def test_a_second_request_is_served_from_the_cache(tmp_path):
+    """The reason the extraction is whole-file rather than per-seek.
+
+    A film's subtitles are extracted once in the life of the file: the second
+    viewing, the second seek and the second sitting all answer from the cache,
+    and ffmpeg runs exactly once.
+    """
+    clip = tmp_path / "clip.mp4"
+    _make_subtitled_clip(clip)
+    gek = generate_gek()
+    session, file_id = _session(clip, gek)
+
+    first = await _ask_for(session, file_id, 1)
+    second = await _ask_for(session, file_id, 1)
+
+    assert first["hash"] == second["hash"]
+    assert session._ctx["media_cache"].puts == 1, (
+        "the second request re-extracted instead of reading the cache")
+
+
+@pytest.mark.asyncio
+async def test_the_result_is_fetched_through_the_ordinary_chunk_path(tmp_path):
+    """The reply names a cache hash, not a new transfer mechanism.
+
+    Same indirection as an audio transcode or a TMDB poster — and it has to
+    actually resolve, or the client is handed a hash it cannot pull.
+    """
+    clip = tmp_path / "clip.mp4"
+    _make_subtitled_clip(clip)
+    gek = generate_gek()
+    session, file_id = _session(clip, gek)
+
+    reply = await _ask_for(session, file_id, 2)
+    chunk = await session._try_serve_thumbnail(reply["hash"], 0, gek)
+
+    assert chunk is not None, "the hash in the reply resolves to nothing"
+    key = chunk_key_aes(gek, bytes.fromhex(reply["hash"]), 0)
+    plain = decrypt_chunk_aes(key, chunk["nonce"], chunk["ct"])
+    assert plain.decode("utf-8").startswith("WEBVTT")
+    assert _CUE_WORD[2] in plain.decode("utf-8")