fix(plex): make the HW->SW fallback supersede-check atomic; honest cull yield

Deep-review round 2 on the round-1 fix:

- Race (medium): the round-1 identity guard did the supersede check and the
  respawn in TWO separate _lock sections. A newer same-key session installed in
  the gap was still torn down by _spawn's unconditional pop+kill (the round-1
  comment even had the direction backwards). Fold the check into _spawn via a
  new replace_only= param: the identity test and the pop+replace now share ONE
  lock, so the respawn is a true no-op when superseded/culled.

- Honest states (low): the superseded branch returned current_session(key),
  which is None when the key was cap-culled during our out-of-lock wait; the
  route turns None into a dishonest 'Media file not found on disk' 404 for a
  file that exists. Yield 'current_session(key) or s' so a culled session
  degrades to the route's retryable 'not available yet' path instead.

- Hygiene: _want_hw docstring corrected (it is a test-only shim now; production
  decides via _select_encode) and the two stale _want_hw references refreshed
  (stream.py header, config.py plex_transcode_hw).

- tests: the cap-cull no-respawn/no-None sub-case; the _opencl_present except
  OSError branch (glob raising, not just empty); tightened a tautological -vf
  assertion. Backend gate green (ruff + full pytest).

backend/app/plex/stream.py

@@ -145,7 +145,7 @@ _HLS_TAIL = [
 
 # --- Transcode (playable="transcode") video re-encode -----------------------------------------
 # The remux path stream-copies the video (`-c:v copy`); a non-h264 source can't be copied and must be
-# re-encoded to H.264. Two encoders, chosen per-session by `_want_hw`:
+# re-encoded to H.264. The encoder + tonemap are chosen per-session by `_select_encode`:
 #   * VAAPI hardware (`h264_vaapi`) — the prod default once the iGPU is passed in. A full GPU pipeline
 #     (decode + scale + encode all on-device) offloads the CPU almost entirely, so several concurrent
 #     1080p transcodes are fine and the API stays responsive.
@@ -382,8 +382,8 @@ def _select_encode(db: DbSession, item: PlexItem, transcode: bool, src: Path) ->
 
 
 def _want_hw(db: DbSession, item: PlexItem, transcode: bool, src: Path) -> bool:
-    """The HW gate on its own (the `hw` half of `_select_encode`). Kept as a thin wrapper for the
-    callers/tests that only care whether a session runs on the GPU at all."""
+    """The HW gate on its own (the `hw` half of `_select_encode`). A readability shim for the tests that
+    assert only whether a session runs on the GPU (production decides via `_select_encode` directly)."""
     return _select_encode(db, item, transcode, src)[0]
 
 
@@ -587,7 +587,15 @@ def start_session(
         # multi-audio: the video rendition is variant v0 → seg_{i}_0.ts; single-audio → seg_{i}.ts
         return directory / (f"seg_{i}_0.ts" if is_multi else f"seg_{i}.ts")
 
-    def _spawn(hw: bool, tonemap: bool = False) -> HlsSession:
+    def _spawn(
+        hw: bool, tonemap: bool = False, *, replace_only: "HlsSession | None" = None
+    ) -> "HlsSession | None":
+        # `replace_only` makes the supersede identity-aware for the runtime fallback: the whole
+        # check-and-replace happens under ONE _lock, so a newer same-key session (a concurrent
+        # seek-restart) installed between the fallback's decision and here is never clobbered. When the
+        # registered session is no longer `replace_only`, we abort and return None (the caller yields to
+        # whoever is registered now). The INITIAL spawn passes replace_only=None and always supersedes —
+        # that is the seek-restart's intended last-writer-wins.
         # Software-fallback encode tuning — read only on the SW transcode path (the VAAPI path and
         # remux ignore preset/crf/max_height, and remux must not touch the DB for them).
         preset, crf, max_height = "veryfast", 20, 0
@@ -600,6 +608,8 @@ def start_session(
         # platform without fork (the API always runs in a Linux container in prod) it's simply omitted.
         preexec = _sw_child_preexec if (transcode and not hw and os.name == "posix") else None
         with _lock:
+            if replace_only is not None and _sessions.get(key) is not replace_only:
+                return None  # superseded/culled while we waited out of lock — don't clobber the newer one
             old = _sessions.pop(key, None)
             if old is not None:
                 _kill(old)
@@ -647,22 +657,21 @@ def start_session(
     if not ready and s.hw:
         # We waited on _seg(0) OUTSIDE _lock, so `s.proc` dying is ambiguous: it may have failed on its
         # own (VAProfile/init/tonemap refusal), OR a concurrent same-key _spawn (a rapid seek-restart) or
-        # _enforce_cap may have killed it to install/keep a newer session. Only fall back if THIS session
-        # is still the registered one — otherwise respawning would tear down the newer session and (via
-        # _spawn's supersede + _enforce_cap) breach the transcode cap. If superseded, yield to whoever is
-        # registered now. (A tiny window remains between this check and _spawn's own lock; that just makes
-        # the fallback lose to an even-newer seek, which is the correct seek-restart outcome anyway.)
-        with _lock:
-            superseded = _sessions.get(key) is not s
-        if superseded:
-            return current_session(key)
-        # Runtime HW→SW fallback: respawn on libx264 (no HW, no tonemap) so a failed GPU session never
-        # strands playback; the software path CPU-decodes and downconverts every source. s.hw is now
-        # False, so the route honestly reports "CPU" and the player's Preparing-stream state stays correct.
+        # _enforce_cap may have killed it to install/keep a newer session. Respawn on libx264 (no HW, no
+        # tonemap) so a genuinely-failed GPU session never strands playback — but pass `replace_only=s` so
+        # the respawn is a no-op if we've been superseded: the identity check and the pop+replace share
+        # ONE lock in _spawn, so we can never tear down the newer session. s.hw is now False, so the route
+        # honestly reports "CPU" and the player's Preparing-stream state stays correct.
         log.warning(
             "plex transcode: HW session produced no segments — falling back to software key=%s", key
         )
-        s = _spawn(False, False)
+        respawned = _spawn(False, False, replace_only=s)
+        if respawned is None:
+            # Superseded/culled mid-fallback — yield the session registered now (a newer seek-restart), or
+            # our own now-dead session if the key was cap-culled (its segment requests degrade to the
+            # route's retryable "not available yet", never a dishonest "file not found" from a None).
+            return current_session(key) or s
+        s = respawned
         ready = wait_for(_seg(0), timeout=25.0, proc=s.proc)
     if ready:
         # For a seek (start_s>0) learn the REAL start offset K from the first segment's absolute PTS,