merge main (0.78.2) into playback-class before release

# Conflicts:
#	backend/app/plex/stream.py

backend/tests/test_download_output.py

@@ -144,7 +144,7 @@ class TestCodecsFromStreams:
 
 
 class TestRunFfprobeJson:
-    """The single ffprobe-JSON invocation point (shared by probe_codecs + upload._probe). A failed
+    """The single ffprobe-JSON invocation point (shared by probe_codecs + probe.probe_media). A failed
     probe or unparseable output must degrade to None, not raise — the malformed-input safety the old
     parse_probe_streams string parser used to own now lives here."""
 

backend/tests/test_plex_playback_session.py

@@ -58,7 +58,7 @@ def _post(db, user, monkeypatch, rating_key, *, captured):
     def fake_start(db_, it, start, audio, aoff, multi, transcode=False):
         captured["transcode"] = transcode
         return SimpleNamespace(
-            media_start_s=start, entry="index.m3u8", transcode=transcode, hw=False
+            media_start_s=start, entry="index.m3u8", transcode=transcode, hw=False, tonemap=False
         )
 
     monkeypatch.setattr(plex_stream, "start_session", fake_start)

backend/tests/test_plex_stream.py

@@ -267,50 +267,206 @@ class TestTranscodeCommand:
         )
         assert cmd[cmd.index("-vf") + 1] == "scale_vaapi=format=nv12"  # not a software scale
 
+    def test_hdr_tonemap_uses_the_opencl_graph(self):
+        # The HDR/10-bit HW path: VAAPI decode → OpenCL tonemap → VAAPI encode. Gen 9.5 can't tonemap in
+        # the VAAPI VPP, so this is the only HW route for a 10-bit/HDR source.
+        cmd = stream._ffmpeg_cmd(
+            self.SRC, 0.0, self.OUT, None, transcode=True, hw=True, tonemap=True
+        )
+        assert _venc(cmd) == "h264_vaapi"
+        assert "libx264" not in cmd
+        # Named devices + an OpenCL device DERIVED from the VAAPI one, all before -i.
+        assert cmd[cmd.index("-init_hw_device") + 1] == f"vaapi=va:{stream._VAAPI_DEVICE}"
+        assert "opencl=ocl@va" in cmd
+        assert cmd[cmd.index("-hwaccel_device") + 1] == "va"
+        assert cmd.index("-init_hw_device") < cmd.index("-i")
+        # The filter graph runs on the derived OpenCL device and tone-maps HDR→SDR to 8-bit nv12.
+        assert cmd[cmd.index("-filter_hw_device") + 1] == "ocl"
+        vf = cmd[cmd.index("-vf") + 1]
+        assert "hwmap=derive_device=opencl" in vf
+        assert "tonemap_opencl=" in vf and "format=nv12" in vf
+        assert "hwmap=derive_device=vaapi:reverse=1" in vf
+        assert "scale_vaapi" not in vf  # the plain 8-bit path's filter must NOT be used here
+        assert (
+            cmd[cmd.index("-force_key_frames") + 1] == f"expr:gte(t,n_forced*{stream._SEG_SECONDS})"
+        )
+
 
 class TestWantHw:
     """VAAPI is used only where the render node was passed in AND (in auto) the source is HW-decodable,
-    with an admin `plex_transcode_hw` policy (auto/on/off) on top."""
+    with an admin `plex_transcode_hw` policy (auto/on/off) on top. A HW session on a 10-bit/HDR source
+    tone-maps on the GPU via OpenCL (tonemap=True) when the OpenCL runtime is present, else drops to
+    software — see the `_select_encode` cases below."""
+
+    SRC = Path("/media/movie.mkv")
 
     @staticmethod
     def _policy(monkeypatch, mode):
         monkeypatch.setattr(stream.sysconfig, "get_str", lambda db, k: mode)
 
+    @staticmethod
+    def _kind(monkeypatch, kind):
+        # Pin the source classifier so the tests don't shell out to ffprobe. `kind` is one of
+        # stream._VK_HWSAFE / _VK_HDR / _VK_SDR10.
+        monkeypatch.setattr(stream, "_probe_video_kind", lambda src: kind)
+
+    @staticmethod
+    def _hwsafe(monkeypatch, safe):
+        # Back-compat convenience for the hw-gate tests that only care about 8-bit-SDR vs not: True →
+        # _VK_HWSAFE, False → _VK_SDR10 (a non-HDR non-hwsafe class, so the hw half still resolves).
+        monkeypatch.setattr(
+            stream, "_probe_video_kind", lambda src: stream._VK_HWSAFE if safe else stream._VK_SDR10
+        )
+
+    @staticmethod
+    def _opencl(monkeypatch, present):
+        # Pin the OpenCL-runtime probe so the routing doesn't depend on the CI box's /etc/OpenCL.
+        monkeypatch.setattr(stream, "_opencl_present", lambda: present)
+
     def test_no_hw_when_not_transcoding(self, monkeypatch):
         self._policy(monkeypatch, "auto")
         monkeypatch.setattr(stream, "_render_node_present", lambda: True)
-        assert stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=False) is False
+        assert (
+            stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=False, src=self.SRC)
+            is False
+        )
 
     def test_no_hw_without_a_render_node(self, monkeypatch):
         self._policy(monkeypatch, "auto")
         monkeypatch.setattr(stream, "_render_node_present", lambda: False)
-        assert stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=True) is False
+        assert (
+            stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=True, src=self.SRC)
+            is False
+        )
 
-    def test_auto_uses_hw_for_a_decodable_source_with_the_render_node(self, monkeypatch):
+    def test_auto_uses_hw_for_a_decodable_8bit_source_with_the_render_node(self, monkeypatch):
         self._policy(monkeypatch, "auto")
+        self._hwsafe(monkeypatch, True)
         monkeypatch.setattr(stream, "_render_node_present", lambda: True)
-        assert stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=True) is True
+        assert (
+            stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=True, src=self.SRC)
+            is True
+        )
 
     def test_auto_falls_back_to_software_for_a_non_decodable_source(self, monkeypatch):
         # AV1 / MPEG-4 ASP (XviD) aren't decodable on Gen 9.5 → the GPU can't feed its own encoder.
         self._policy(monkeypatch, "auto")
+        self._hwsafe(monkeypatch, True)  # decodability gate must fire BEFORE the probe
         monkeypatch.setattr(stream, "_render_node_present", lambda: True)
-        assert stream._want_hw(None, _FakeItem(codec_video="av1"), transcode=True) is False
-        assert stream._want_hw(None, _FakeItem(codec_video="mpeg4"), transcode=True) is False
+        assert (
+            stream._want_hw(None, _FakeItem(codec_video="av1"), transcode=True, src=self.SRC)
+            is False
+        )
+        assert (
+            stream._want_hw(None, _FakeItem(codec_video="mpeg4"), transcode=True, src=self.SRC)
+            is False
+        )
+
+    def test_auto_hdr_source_falls_back_to_software_without_opencl(self, monkeypatch):
+        # A decodable HEVC that is true-HDR (PQ/HLG) can't use scale_vaapi=format=nv12 (Gen 9.5), and with
+        # no OpenCL runtime to tonemap on the GPU it drops to software libx264 (BUG 1's behaviour).
+        self._policy(monkeypatch, "auto")
+        self._kind(monkeypatch, stream._VK_HDR)
+        self._opencl(monkeypatch, False)
+        monkeypatch.setattr(stream, "_render_node_present", lambda: True)
+        assert stream._select_encode(
+            None, _FakeItem(codec_video="hevc"), transcode=True, src=self.SRC
+        ) == (False, False)
+
+    def test_auto_hdr_source_uses_hw_tonemap_with_opencl(self, monkeypatch):
+        # With the OpenCL runtime present, a true-HDR HEVC tone-maps on the GPU (VAAPI decode → OpenCL
+        # tonemap → VAAPI encode) instead of dropping to CPU — the HW HDR path.
+        self._policy(monkeypatch, "auto")
+        self._kind(monkeypatch, stream._VK_HDR)
+        self._opencl(monkeypatch, True)
+        monkeypatch.setattr(stream, "_render_node_present", lambda: True)
+        assert stream._select_encode(
+            None, _FakeItem(codec_video="hevc"), transcode=True, src=self.SRC
+        ) == (True, True)
+
+    def test_auto_10bit_SDR_source_uses_software_not_tonemap(self, monkeypatch):
+        # F1 regression: a 10-bit *SDR* (bt709) source is NOT HDR, so it must NOT get the hable HDR tone
+        # curve even when the OpenCL runtime is present — that would visibly alter an SDR picture. It takes
+        # software libx264 (an exact 10→8-bit downconvert), never (True, True).
+        self._policy(monkeypatch, "auto")
+        self._kind(monkeypatch, stream._VK_SDR10)
+        self._opencl(monkeypatch, True)
+        monkeypatch.setattr(stream, "_render_node_present", lambda: True)
+        assert stream._select_encode(
+            None, _FakeItem(codec_video="hevc"), transcode=True, src=self.SRC
+        ) == (False, False)
+
+    def test_auto_8bit_source_uses_plain_hw_not_tonemap(self, monkeypatch):
+        # An 8-bit SDR source takes the plain full-GPU scale path — never the tonemap graph — even where
+        # OpenCL is available.
+        self._policy(monkeypatch, "auto")
+        self._kind(monkeypatch, stream._VK_HWSAFE)
+        self._opencl(monkeypatch, True)
+        monkeypatch.setattr(stream, "_render_node_present", lambda: True)
+        assert stream._select_encode(
+            None, _FakeItem(codec_video="hevc"), transcode=True, src=self.SRC
+        ) == (True, False)
 
     def test_off_forces_software_even_for_a_decodable_source(self, monkeypatch):
         self._policy(monkeypatch, "off")
         monkeypatch.setattr(stream, "_render_node_present", lambda: True)
-        assert stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=True) is False
+        assert (
+            stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=True, src=self.SRC)
+            is False
+        )
 
-    def test_on_forces_hw_even_for_a_non_decodable_source(self, monkeypatch):
-        # The admin override trusts the GPU to decode it; skips the decodability gate (render node still
-        # required — HW is impossible without it).
+    def test_on_forces_hw_and_skips_both_source_gates(self, monkeypatch):
+        # The admin override trusts the GPU; skips the decodability AND the bit-depth gate (the runtime
+        # HW→SW fallback still backstops it). Render node still required — HW is impossible without it.
         self._policy(monkeypatch, "on")
+        self._hwsafe(monkeypatch, False)  # would veto in auto; "on" ignores it
+        self._opencl(
+            monkeypatch, False
+        )  # pin: tonemap-half is irrelevant here, keep it deterministic
         monkeypatch.setattr(stream, "_render_node_present", lambda: True)
-        assert stream._want_hw(None, _FakeItem(codec_video="av1"), transcode=True) is True
+        assert (
+            stream._want_hw(None, _FakeItem(codec_video="av1"), transcode=True, src=self.SRC)
+            is True
+        )
         monkeypatch.setattr(stream, "_render_node_present", lambda: False)
-        assert stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=True) is False
+        assert (
+            stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=True, src=self.SRC)
+            is False
+        )
+
+    def test_on_still_picks_the_tonemap_graph_for_an_hdr_source(self, monkeypatch):
+        # "on" trusts the GPU but must still route a true-HDR source through the tonemap graph (plain
+        # scale_vaapi would hit the VPP wall) when OpenCL is present.
+        self._policy(monkeypatch, "on")
+        self._kind(monkeypatch, stream._VK_HDR)
+        self._opencl(monkeypatch, True)
+        monkeypatch.setattr(stream, "_render_node_present", lambda: True)
+        assert stream._select_encode(
+            None, _FakeItem(codec_video="hevc"), transcode=True, src=self.SRC
+        ) == (True, True)
+
+    def test_on_10bit_SDR_uses_plain_hw_not_tonemap(self, monkeypatch):
+        # F1 regression under "on": a 10-bit SDR source is NOT HDR, so even with OpenCL present "on" must
+        # pick plain HW (True, False) — the HDR tone curve must never touch an SDR picture.
+        self._policy(monkeypatch, "on")
+        self._kind(monkeypatch, stream._VK_SDR10)
+        self._opencl(monkeypatch, True)
+        monkeypatch.setattr(stream, "_render_node_present", lambda: True)
+        assert stream._select_encode(
+            None, _FakeItem(codec_video="hevc"), transcode=True, src=self.SRC
+        ) == (True, False)
+
+    def test_on_without_opencl_is_plain_hw(self, monkeypatch):
+        # "on" with no OpenCL runtime → plain HW, tonemap=False, regardless of source class (the runtime
+        # HW→SW fallback backstops a source the plain path can't actually handle). Pins the tonemap half
+        # of the "on" tuple, which test_on_forces_hw_and_skips_both_source_gates only checks via _want_hw.
+        self._policy(monkeypatch, "on")
+        self._kind(monkeypatch, stream._VK_HDR)  # even HDR: no runtime → can't tonemap on the GPU
+        self._opencl(monkeypatch, False)
+        monkeypatch.setattr(stream, "_render_node_present", lambda: True)
+        assert stream._select_encode(
+            None, _FakeItem(codec_video="hevc"), transcode=True, src=self.SRC
+        ) == (True, False)
 
 
 class TestColdStartWarmup:
@@ -333,7 +489,7 @@ class TestColdStartWarmup:
     def test_warmup_waits_for_the_first_few_segments(self, fake_env, monkeypatch):
         waited = []
 
-        def spy(path, timeout=20.0):
+        def spy(path, timeout=20.0, proc=None):
             waited.append(path.name)
             return True
 
@@ -346,7 +502,7 @@ class TestColdStartWarmup:
     def test_warmup_stops_at_the_first_segment_that_isnt_ready(self, fake_env, monkeypatch):
         waited = []
 
-        def spy(path, timeout=20.0):
+        def spy(path, timeout=20.0, proc=None):
             waited.append(path.name)
             return path.name != "seg_1.ts"  # seg_1 never appears → warm-up gives up there
 
@@ -360,7 +516,7 @@ class TestColdStartWarmup:
         # still returned so the client's own segment requests can drive it, as before.
         waited = []
 
-        def spy(path, timeout=20.0):
+        def spy(path, timeout=20.0, proc=None):
             waited.append(path.name)
             return False
 
@@ -398,7 +554,7 @@ class TestTranscodeStartSession:
             "get_str",
             lambda db, k: {"plex_transcode_preset": "slow"}.get(k, "auto"),
         )
-        monkeypatch.setattr(stream, "wait_for", lambda path, timeout=20.0: True)
+        monkeypatch.setattr(stream, "wait_for", lambda path, timeout=20.0, proc=None: True)
         _CapturingProc.calls = []
         monkeypatch.setattr(stream.subprocess, "Popen", _CapturingProc)
         return tmp_path
@@ -406,7 +562,9 @@ class TestTranscodeStartSession:
     def test_software_transcode_threads_the_knobs_into_the_command(
         self, capturing_env, monkeypatch
     ):
-        monkeypatch.setattr(stream, "_want_hw", lambda db, item, transcode: False)
+        monkeypatch.setattr(
+            stream, "_select_encode", lambda db, item, transcode, src: (False, False)
+        )
         s = stream.start_session(
             db=None, item=_FakeItem(codec_video="mpeg4"), start_s=0.0, transcode=True
         )
@@ -421,25 +579,292 @@ class TestTranscodeStartSession:
             assert kwargs.get("preexec_fn") is stream._sw_child_preexec
 
     def test_hardware_transcode_uses_the_gpu_and_never_nices(self, capturing_env, monkeypatch):
-        monkeypatch.setattr(stream, "_want_hw", lambda db, item, transcode: True)
+        monkeypatch.setattr(
+            stream, "_select_encode", lambda db, item, transcode, src: (True, False)
+        )
         s = stream.start_session(
             db=None, item=_FakeItem(codec_video="hevc"), start_s=0.0, transcode=True
         )
-        assert s is not None and s.transcode is True and s.hw is True
+        assert s is not None and s.transcode is True and s.hw is True and s.tonemap is False
+        assert len(_CapturingProc.calls) == 1  # first segment arrived — no HW→SW fallback respawn
         (cmd, *_), kwargs = _CapturingProc.calls[0]
         assert _venc(cmd) == "h264_vaapi"
         assert "libx264" not in cmd
-        assert (
-            "-vf" not in cmd or cmd[cmd.index("-vf") + 1] == "scale_vaapi=format=nv12"
-        )  # no SW cap
+        # Plain-HW path unconditionally emits the nv12 downconvert (a disjunction on "-vf not in cmd"
+        # would silently pass if the filter were dropped — assert it is actually present).
+        assert cmd[cmd.index("-vf") + 1] == "scale_vaapi=format=nv12"  # no SW cap
         assert kwargs.get("preexec_fn") is None  # GPU encode is cheap — never niced
 
+    def test_hdr_tonemap_transcode_threads_the_opencl_graph_and_flags_the_session(
+        self, capturing_env, monkeypatch
+    ):
+        # A HW tonemap session (10-bit/HDR + OpenCL) must build the OpenCL graph in the argv and carry
+        # tonemap=True on the session, still on the GPU (hw=True) and never niced.
+        monkeypatch.setattr(stream, "_select_encode", lambda db, item, transcode, src: (True, True))
+        s = stream.start_session(
+            db=None, item=_FakeItem(codec_video="hevc"), start_s=0.0, transcode=True
+        )
+        assert s is not None and s.transcode is True and s.hw is True and s.tonemap is True
+        (cmd, *_), kwargs = _CapturingProc.calls[0]
+        assert _venc(cmd) == "h264_vaapi"
+        assert "opencl=ocl@va" in cmd and cmd[cmd.index("-filter_hw_device") + 1] == "ocl"
+        assert "tonemap_opencl=" in cmd[cmd.index("-vf") + 1]
+        assert kwargs.get("preexec_fn") is None
+
+    def test_a_hw_session_that_produces_nothing_falls_back_to_software(
+        self, capturing_env, monkeypatch
+    ):
+        # BUG 1 backstop: _want_hw picked HW but the VAAPI session emits no first segment (a VAProfile
+        # refusal the gates didn't catch). start_session must respawn on libx264 and report hw=False so
+        # the item still plays and the route/player honestly show CPU.
+        monkeypatch.setattr(
+            stream, "_select_encode", lambda db, item, transcode, src: (True, False)
+        )
+        seen = {"n": 0}
+
+        def flaky_wait(path, timeout=20.0, proc=None):
+            seen["n"] += 1
+            return seen["n"] > 1  # the first seg-0 wait (HW) fails; every wait after succeeds (SW)
+
+        monkeypatch.setattr(stream, "wait_for", flaky_wait)
+        s = stream.start_session(
+            db=None, item=_FakeItem(codec_video="hevc"), start_s=0.0, transcode=True
+        )
+        assert (
+            s is not None and s.transcode is True and s.hw is False
+        )  # honest CPU after the fallback
+        assert len(_CapturingProc.calls) == 2  # the HW attempt, then the SW respawn
+        (hw_cmd, *_), _ = _CapturingProc.calls[0]
+        (sw_cmd, *_), sw_kwargs = _CapturingProc.calls[1]
+        assert _venc(hw_cmd) == "h264_vaapi"
+        assert _venc(sw_cmd) == "libx264"
+        if os.name == "posix":
+            assert sw_kwargs.get("preexec_fn") is stream._sw_child_preexec  # the SW child is niced
+
+    def test_a_hw_TONEMAP_session_that_produces_nothing_falls_back_to_software(
+        self, capturing_env, monkeypatch
+    ):
+        # The fallback gate is `not ready and s.hw`, shared by plain-HW and tonemap sessions — but only
+        # the plain-HW variant was covered. Pin the tonemap variant too: a (True, True) session that
+        # emits nothing (an OpenCL/tonemap init refusal) must respawn on libx264 and end hw=False,
+        # tonemap=False, so an HDR item still plays on CPU.
+        monkeypatch.setattr(stream, "_select_encode", lambda db, item, transcode, src: (True, True))
+        seen = {"n": 0}
+
+        def flaky_wait(path, timeout=20.0, proc=None):
+            seen["n"] += 1
+            return (
+                seen["n"] > 1
+            )  # the first seg-0 wait (HW tonemap) fails; every wait after succeeds
+
+        monkeypatch.setattr(stream, "wait_for", flaky_wait)
+        s = stream.start_session(
+            db=None, item=_FakeItem(codec_video="hevc"), start_s=0.0, transcode=True
+        )
+        assert s is not None and s.transcode is True and s.hw is False and s.tonemap is False
+        assert len(_CapturingProc.calls) == 2  # the HW tonemap attempt, then the SW respawn
+        (hw_cmd, *_), _ = _CapturingProc.calls[0]
+        (sw_cmd, *_), _ = _CapturingProc.calls[1]
+        assert (
+            "tonemap_opencl=" in hw_cmd[hw_cmd.index("-vf") + 1]
+        )  # first attempt built the OpenCL graph
+        assert _venc(sw_cmd) == "libx264"  # respawn is plain software, no tonemap
+
+    def test_a_superseded_hw_session_does_not_respawn_over_the_newer_one(
+        self, capturing_env, monkeypatch
+    ):
+        # F2/race: start_session waits on seg_0 OUTSIDE _lock. If a concurrent same-key start (a rapid
+        # seek-restart) or _enforce_cap replaces _sessions[key] while we wait, our proc dying must NOT
+        # trigger the SW fallback — that would tear down the NEWER session and breach the transcode cap.
+        # The identity guard must detect it's been superseded and yield the currently-registered session.
+        monkeypatch.setattr(
+            stream, "_select_encode", lambda db, item, transcode, src: (True, False)
+        )
+
+        class _Newer:  # stands in for the session a concurrent _spawn registered
+            hw = True
+            tonemap = False
+            transcode = True
+            last_access = 0.0
+
+        newer = _Newer()
+
+        def wait_then_supersede(path, timeout=20.0, proc=None):
+            # Simulate the concurrent replacement happening during our out-of-lock wait, then report the
+            # (now-killed) HW proc produced nothing.
+            stream._sessions["12345"] = newer
+            return False
+
+        monkeypatch.setattr(stream, "wait_for", wait_then_supersede)
+        s = stream.start_session(
+            db=None, item=_FakeItem(codec_video="hevc"), start_s=0.0, transcode=True
+        )
+        # No SW respawn — only the initial HW _spawn ran; the newer session was left intact.
+        assert len(_CapturingProc.calls) == 1
+        assert s is newer  # yielded the currently-registered session, did not clobber it
+        assert stream._sessions["12345"] is newer
+
+    def test_a_cap_culled_hw_session_does_not_respawn_and_is_not_reported_missing(
+        self, capturing_env, monkeypatch
+    ):
+        # F2/race, the cull sub-case: while we wait on seg_0 out of lock, cross-key cap pressure lets
+        # _enforce_cap evict THIS session's key entirely (no replacement). The fallback must still not
+        # respawn (that would re-breach the cap the cull just enforced), and must NOT return None — a None
+        # is turned by the route into a dishonest "file not found" 404 for a file that exists. It yields
+        # our own (now-dead) session instead, so the client's segment requests degrade to the retryable
+        # "not available yet" path.
+        monkeypatch.setattr(
+            stream, "_select_encode", lambda db, item, transcode, src: (True, False)
+        )
+        captured = {}
+
+        def wait_then_cull(path, timeout=20.0, proc=None):
+            captured["s"] = stream._sessions[
+                "12345"
+            ]  # the HW session the initial _spawn just registered
+            del stream._sessions[
+                "12345"
+            ]  # simulate _enforce_cap dropping the LRU under cap pressure
+            return False
+
+        monkeypatch.setattr(stream, "wait_for", wait_then_cull)
+        s = stream.start_session(
+            db=None, item=_FakeItem(codec_video="hevc"), start_s=0.0, transcode=True
+        )
+        assert (
+            len(_CapturingProc.calls) == 1
+        )  # no SW respawn — the cull is respected, cap not re-breached
+        assert (
+            s is captured["s"]
+        )  # our own session yielded, NOT None (which the route would 404 as missing)
+        assert "12345" not in stream._sessions  # the culled key stays gone
+
     def test_remux_reads_no_transcode_knobs_and_stream_copies(self, capturing_env, monkeypatch):
         # A remux start must stream-copy the video and spawn no niced child (the transcode knobs and
         # the preexec nice are gated behind `transcode=True`).
-        monkeypatch.setattr(stream, "_want_hw", lambda db, item, transcode: False)
+        monkeypatch.setattr(
+            stream, "_select_encode", lambda db, item, transcode, src: (False, False)
+        )
         s = stream.start_session(db=None, item=_FakeItem(), start_s=0.0)  # transcode defaults False
         assert s is not None and s.transcode is False
         (cmd, *_), kwargs = _CapturingProc.calls[0]
         assert _venc(cmd) == "copy"
         assert kwargs.get("preexec_fn") is None
+
+
+class TestProbeVideoKind:
+    """The ffprobe-based source classifier that routes the transcode pipeline: 8-bit SDR → plain VAAPI;
+    true-HDR (PQ/HLG) → the OpenCL tonemap graph; 10-bit *SDR* → software (an exact downconvert, NEVER
+    the HDR tone curve — the F1 distinction). Exercises the REAL function body (not a monkeypatch)."""
+
+    @staticmethod
+    def _probe(monkeypatch, stdout):
+        class _R:
+            def __init__(self):
+                self.stdout = stdout
+
+        monkeypatch.setattr(stream.subprocess, "run", lambda *a, **k: _R())
+
+    def test_8bit_sdr_is_hwsafe(self, monkeypatch):
+        self._probe(monkeypatch, "pix_fmt=yuv420p\nbits_per_raw_sample=8\ncolor_transfer=bt709\n")
+        assert stream._probe_video_kind(Path("/x")) == stream._VK_HWSAFE
+
+    def test_10bit_sdr_pix_fmt_is_sdr10(self, monkeypatch):
+        # yuv420p10le with a non-HDR transfer (bt2020-10 is NOT PQ/HLG) = 10-bit SDR → software, not
+        # tonemap. Signalled by the pix_fmt suffix even when bits_per_raw_sample is absent.
+        self._probe(monkeypatch, "pix_fmt=yuv420p10le\ncolor_transfer=bt2020-10\n")
+        assert stream._probe_video_kind(Path("/x")) == stream._VK_SDR10
+
+    def test_bits_per_raw_sample_over_8_sdr_is_sdr10(self, monkeypatch):
+        self._probe(monkeypatch, "pix_fmt=yuv420p\nbits_per_raw_sample=10\ncolor_transfer=bt709\n")
+        assert stream._probe_video_kind(Path("/x")) == stream._VK_SDR10
+
+    def test_pq_transfer_is_hdr(self, monkeypatch):
+        # PQ (smpte2084) is true HDR — needs a real tonemap, even if only 8-bit-tagged.
+        self._probe(
+            monkeypatch, "pix_fmt=yuv420p10le\nbits_per_raw_sample=10\ncolor_transfer=smpte2084\n"
+        )
+        assert stream._probe_video_kind(Path("/x")) == stream._VK_HDR
+
+    def test_hlg_transfer_is_hdr(self, monkeypatch):
+        # HLG (arib-std-b67) is the other HDR transfer — previously untested; it must classify as HDR.
+        self._probe(
+            monkeypatch,
+            "pix_fmt=yuv420p10le\nbits_per_raw_sample=10\ncolor_transfer=arib-std-b67\n",
+        )
+        assert stream._probe_video_kind(Path("/x")) == stream._VK_HDR
+
+    def test_empty_probe_output_is_hwsafe(self, monkeypatch):
+        # A probe that returns nothing (no streams / stripped metadata) → assume hwsafe; the runtime
+        # HW→SW fallback backstops a wrong guess.
+        self._probe(monkeypatch, "")
+        assert stream._probe_video_kind(Path("/x")) == stream._VK_HWSAFE
+
+    def test_probe_failure_assumes_hwsafe(self, monkeypatch):
+        # ffprobe missing/erroring → assume hwsafe; the runtime HW→SW fallback is the real backstop, so a
+        # probe miss costs at most one wasted HW attempt, never a stranded item.
+        def boom(*a, **k):
+            raise OSError("no ffprobe")
+
+        monkeypatch.setattr(stream.subprocess, "run", boom)
+        assert stream._probe_video_kind(Path("/x")) == stream._VK_HWSAFE
+
+
+class TestOpenclPresent:
+    """The OpenCL-runtime probe (_opencl_present) — the precondition for the HW HDR tonemap graph. Drives
+    the REAL glob against a temp ICD dir instead of monkeypatching the function, so a regression to the
+    ICD path / glob / OSError guard would be caught (this routes every HDR source on prod)."""
+
+    def test_true_when_an_icd_file_is_registered(self, tmp_path, monkeypatch):
+        monkeypatch.setattr(stream, "_OPENCL_ICD_DIR", tmp_path)
+        (tmp_path / "intel.icd").write_text("libintelocl.so\n")
+        assert stream._opencl_present() is True
+
+    def test_false_when_the_dir_is_empty(self, tmp_path, monkeypatch):
+        monkeypatch.setattr(stream, "_OPENCL_ICD_DIR", tmp_path)
+        assert stream._opencl_present() is False
+
+    def test_false_and_no_raise_when_the_dir_is_missing(self, tmp_path, monkeypatch):
+        # A missing dir: Path.glob yields nothing (no raise) → False.
+        monkeypatch.setattr(stream, "_OPENCL_ICD_DIR", tmp_path / "nope")
+        assert stream._opencl_present() is False
+
+    def test_false_when_glob_raises_oserror(self, monkeypatch):
+        # The `except OSError` guard proper: an unreadable ICD dir (permission denied) makes glob RAISE,
+        # not return empty. That must be swallowed to False, never propagated up through _select_encode
+        # into start_session (which would crash HDR routing).
+        class _BoomDir:
+            def glob(self, pattern):
+                raise OSError("permission denied")
+
+        monkeypatch.setattr(stream, "_OPENCL_ICD_DIR", _BoomDir())
+        assert stream._opencl_present() is False
+
+
+class TestWaitForProcFastFail:
+    """wait_for(proc=…) must stop as soon as the ffmpeg proc exits (a VAAPI init refusal dies in ~1s
+    producing nothing) instead of blocking the whole timeout — that fast-fail is what keeps the HW→SW
+    fallback from turning into a 25s stall. Exercises the REAL wait_for (every other test monkeypatches
+    it wholesale, so this branch would otherwise never run)."""
+
+    class _DeadProc:
+        def poll(self):
+            return 1  # already exited non-zero
+
+    def test_returns_true_when_the_segment_exists_despite_a_dead_proc(self, tmp_path):
+        seg = tmp_path / "seg_0.ts"
+        seg.write_bytes(b"x")
+        # The proc exited but DID write a first segment → ready. (The final existence re-check.)
+        assert stream.wait_for(seg, timeout=5.0, proc=self._DeadProc()) is True
+
+    def test_returns_false_fast_when_a_dead_proc_wrote_nothing(self, tmp_path):
+        # Missing segment + dead proc → give up immediately, not after `timeout`. Assert it's fast by
+        # using a long timeout the test would visibly hang on if the fast-fail branch regressed.
+        seg = tmp_path / "seg_0.ts"
+        assert stream.wait_for(seg, timeout=30.0, proc=self._DeadProc()) is False
+
+    def test_empty_segment_from_a_dead_proc_is_not_ready(self, tmp_path):
+        # A zero-byte segment is not a real segment (size > 0 required).
+        seg = tmp_path / "seg_0.ts"
+        seg.write_bytes(b"")
+        assert stream.wait_for(seg, timeout=30.0, proc=self._DeadProc()) is False

backend/tests/test_upload.py

@@ -7,9 +7,9 @@ guards that keep an upload out of paths that assume a re-fetchable source (edit
 
 DB-backed (real Postgres) — skipped when no database is reachable. `settings.download_root` is
 pointed at a per-test tmp dir so ingest's on-disk moves are isolated. The ingest/route tests
-monkeypatch `_probe` so they don't shell out to ffprobe; the classifier's OWN parsing (image →
-file, cover-art skip, dims, duration) is covered directly by `TestProbeClassifier` below, which
-stubs one level down (`run_ffprobe_json`) and exercises the real `_probe`.
+monkeypatch the shared `probe.probe_media` so they don't shell out to ffprobe; the classifier's OWN
+parsing (image → file, cover-art skip, dims, duration) is covered directly by `TestProbeClassifier`
+below, which stubs one level down (`run_ffprobe_json`) and exercises the real `probe_media`.
 """
 
 from pathlib import Path
@@ -18,6 +18,7 @@ import pytest
 
 from app.config import settings
 from app.downloads import mediakind, quota, storage
+from app.downloads import probe as probemod
 from app.downloads import upload as uploadmod
 from app.models import DownloadQuota, MediaAsset
 
@@ -42,7 +43,9 @@ def _root(tmp_path, monkeypatch):
 
 
 def _probe(monkeypatch, result):
-    monkeypatch.setattr(uploadmod, "_probe", lambda _p: result)
+    # upload ingest now calls the shared `probe.probe_media` (same ground-truth ffprobe the download
+    # worker uses); patch it at that shared home.
+    monkeypatch.setattr(probemod, "probe_media", lambda _p: result)
 
 
 def test_ingest_generic_file_is_download_only(db, make_user, _root, monkeypatch):
@@ -325,16 +328,16 @@ def test_ingest_cleans_up_placed_files_on_commit_failure(db, make_user, _root, m
     assert db.query(MediaAsset).count() == 0  # the flushed asset rolled back — no dangling row
 
 
-# --- classifier: the REAL _probe parsing (stubs ffprobe one level down) ----------------------
-# The ingest/route tests above stub _probe wholesale; these exercise _probe's OWN logic — image
+# --- classifier: the REAL probe_media parsing (stubs ffprobe one level down) ------------------
+# The ingest/route tests above stub probe_media wholesale; these exercise its OWN logic — image
 # detection, the cover-art skip, dimension/duration parsing — by faking only run_ffprobe_json, so a
 # regression in the classifier the force-download guarantee leans on is actually caught.
 
 
 class TestProbeClassifier:
     def _probe_with(self, monkeypatch, data):
-        monkeypatch.setattr(uploadmod.editmod, "run_ffprobe_json", lambda *a, **k: data)
-        return uploadmod._probe(Path("x"))
+        monkeypatch.setattr(probemod.editmod, "run_ffprobe_json", lambda *a, **k: data)
+        return probemod.probe_media(Path("x"))
 
     def test_probe_failure_is_a_generic_file(self, monkeypatch):
         # run_ffprobe_json returns None (ffprobe failed / not media) → all None → derive() == FILE.
@@ -392,3 +395,109 @@ class TestProbeClassifier:
         }
         v, a, w, h, d = self._probe_with(monkeypatch, data)
         assert d is None and mediakind.derive(v, a) == mediakind.AUDIO
+
+
+# --- resolve_media_facts: probe-of-the-file wins over yt-dlp's info (the FB regression fix) ---------
+# The download worker's finalize must classify from the ACTUAL file, not the extractor's `info` dict.
+# yt-dlp omits vcodec/acodec/dims for some sources (Facebook), and deriving media_kind from those
+# nulls turned a real H.264 video into a download-only 'file' (no in-app play). These pin that the
+# probe is authoritative and `info` is only a gap-filler.
+
+
+class TestResolveMediaFacts:
+    # These stub run_ffprobe_json (one level below probe_media) so the REAL probe parsing + the
+    # probe-vs-info precedence are exercised together — the precedence keys off whether ffprobe RAN.
+    @staticmethod
+    def _ffprobe(monkeypatch, data):
+        monkeypatch.setattr(probemod.editmod, "run_ffprobe_json", lambda *a, **k: data)
+
+    def test_probe_wins_when_info_lacks_codecs(self, monkeypatch):
+        # THE original regression: info has no vcodec/acodec/dims (Facebook), but the file probes as
+        # h264/aac.
+        self._ffprobe(
+            monkeypatch,
+            {
+                "format": {"format_name": "mov,mp4,m4a"},
+                "streams": [
+                    {"codec_type": "video", "codec_name": "h264", "width": 720, "height": 1280},
+                    {"codec_type": "audio", "codec_name": "aac"},
+                ],
+            },
+        )
+        v, a, w, h = probemod.resolve_media_facts(Path("x"), {"width": None, "height": None})
+        assert (v, a, w, h) == ("h264", "aac", 720, 1280)
+        assert mediakind.derive(v, a) == mediakind.VIDEO  # NOT 'file'
+
+    def test_probe_wins_over_a_DIFFERING_populated_info(self, monkeypatch):
+        # Both the probe AND info are populated but DISAGREE (info carries a pre-remux/pre-merge codec).
+        # The finished file on disk is ground truth, so the probe wins outright — pins the precedence a
+        # `info or probe` inversion would silently break.
+        self._ffprobe(
+            monkeypatch,
+            {
+                "format": {"format_name": "mov,mp4,m4a"},
+                "streams": [
+                    {"codec_type": "video", "codec_name": "h264", "width": 1920, "height": 1080},
+                    {"codec_type": "audio", "codec_name": "aac"},
+                ],
+            },
+        )
+        v, a, w, h = probemod.resolve_media_facts(
+            Path("x"), {"vcodec": "vp9", "acodec": "opus", "width": 640, "height": 480}
+        )
+        assert (v, a, w, h) == ("h264", "aac", 1920, 1080)  # probe, not info
+
+    def test_cover_art_audio_download_is_not_resurrected_to_video_by_info(self, monkeypatch):
+        # THE bug this fix targets: an audio-mode download from a progressive source embeds cover art,
+        # and yt-dlp's info still carries the source's video codec. The probe correctly finds only audio
+        # (the attached_pic is skipped); info's vcodec must NOT resurrect a video track → media_kind
+        # stays audio, not a broken in-app 'video'.
+        self._ffprobe(
+            monkeypatch,
+            {
+                "format": {"format_name": "mp3", "duration": "180.0"},
+                "streams": [
+                    {
+                        "codec_type": "video",
+                        "codec_name": "mjpeg",
+                        "disposition": {"attached_pic": 1},
+                    },
+                    {"codec_type": "audio", "codec_name": "mp3"},
+                ],
+            },
+        )
+        v, a, w, h = probemod.resolve_media_facts(
+            Path("x"), {"vcodec": "avc1", "acodec": "mp4a", "width": 1280, "height": 720}
+        )
+        assert v is None and a == "mp3"  # info's avc1 did NOT win
+        assert mediakind.derive(v, a) == mediakind.AUDIO  # NOT VIDEO
+
+    def test_still_image_download_is_not_resurrected_to_video_by_info(self, monkeypatch):
+        # A downloaded still image: the probe returns no vcodec (image-format guard). info's image
+        # "vcodec" must not turn it into a broken 'video' → stays a generic file.
+        self._ffprobe(
+            monkeypatch,
+            {
+                "format": {"format_name": "png_pipe"},
+                "streams": [{"codec_type": "video", "codec_name": "png", "width": 8, "height": 8}],
+            },
+        )
+        v, a, w, h = probemod.resolve_media_facts(
+            Path("x"), {"vcodec": "png", "width": 8, "height": 8}
+        )
+        assert v is None and mediakind.derive(v, a) == mediakind.FILE
+
+    def test_info_fills_in_only_when_ffprobe_could_not_read_the_file(self, monkeypatch):
+        # ffprobe failed entirely (None) → fall back to info so we never regress the working path.
+        self._ffprobe(monkeypatch, None)
+        v, a, w, h = probemod.resolve_media_facts(
+            Path("x"), {"vcodec": "vp9", "acodec": "opus", "width": 1920, "height": 1080}
+        )
+        assert (v, a, w, h) == ("vp9", "opus", 1920, 1080)
+
+    def test_info_none_codec_string_is_treated_as_absent(self, monkeypatch):
+        # yt-dlp writes the literal "none" for a missing stream — it must not become a codec name.
+        self._ffprobe(monkeypatch, None)  # probe failed → info path, where the "none" scrub applies
+        v, a, w, h = probemod.resolve_media_facts(Path("x"), {"vcodec": "none", "acodec": "aac"})
+        assert (v, a) == (None, "aac")
+        assert mediakind.derive(v, a) == mediakind.AUDIO