fix(plex): an HLS artifact names the session it belongs to, and a paused viewer says so

Deep-review round 5, over round 4's own fixes.

- Sessions are keyed by rating_key alone while a session's DIRECTORY is per (key, offset, tag), and
  the artifact URL carried no session identity — so once a seek-restart or a second viewer of the
  same item took the registry slot, the old viewer's request for `seg_12.ts` was answered 200 with
  the new session's twelfth segment: bytes from a different `-ss` offset under `-copyts`, on a
  timeline the client is not on. Wrong content with a success status is the one failure a player
  cannot notice, and round 4's reclaim-on-404 could never fire on it. The id is in the path now (so
  the playlist's relative names resolve back into the same session), a mismatch is a 404, and the
  player reclaims at the playhead. The entry playlist's auto-start is gone with it: it could only
  ever begin at 0, silently discarding the viewer's position.

- Round 4 exempted paused viewers from the reaper by keeping EVERY finished session for the full
  ten-minute idle window — one whole segmented copy of a film per item, on the volume the download
  center shares. The signal it needed already existed: an open player checkpoints its position every
  10 s whether or not it is playing, so that route now touches the session and the periodic pass is
  back on the short rule.

- POST /session's opportunistic reap was never scoped to the key being started: it walked the whole
  registry with the 30 s finished rule, so pressing play on ANY item — even a direct-play one, the
  call ran before that early return — killed another viewer's paused session. It is gone; the
  periodic pass owns reaping.

- A segment that can never arrive no longer polls out the full 20 s: the producer's handle is passed
  to `wait_for`, so a finished ffmpeg answers at once. This is a sync route, so each such wait parked
  a threadpool worker AND a pooled DB connection, and an ordinary seek past the produced region was
  enough to do it.

- The DataError handler's re-raise branch — the half that keeps a genuine DataError a 500 — is
  pinned, so simplifying it to an unconditional 422 cannot pass in silence.

Six regression tests (two route-level, in the DB lane), each mutation-checked red on the assertion
it is named after. pytest 1124, vitest 999.

backend/tests/test_hls_reaper.py

@@ -1,21 +1,35 @@
-"""Idle HLS sessions are reaped on a timer, not only when the next session is requested.
+"""Idle HLS sessions are reaped on a timer, and a viewer who is merely PAUSED keeps theirs.
 
 plex/stream.py promises "a periodic reaper kills idle ones", but `reap_idle` ran only inside
 POST /session: after the last playback, an abandoned session (its ffmpeg still working through the
-film, its directory a whole segmented copy of it) lived until somebody pressed play again.
+film, its directory a whole segmented copy of it) lived until somebody pressed play again
+(deep-review round 4).
 
 And it must not take a PAUSED viewer's session: a remux finishes long before the film does, and a
-paused player with a full buffer fetches nothing — so "finished and unfetched for 30 s" (the rule a
-new session request applies) is not "abandoned".
+paused player with a full buffer fetches nothing, so "finished and unfetched for 30 s" does not mean
+"abandoned". Round 4 bought that safety by keeping EVERY finished session for the full ten-minute
+idle window — one segmented copy of a film per item, on the volume the download center shares. Round
+5 replaced the blanket retention with the signal that already existed: an open player checkpoints its
+position every 10 s whether or not it is playing, and that route says "still here".
+
+Both call sites are pinned here, because the bug each time was that only one of them moved: the
+periodic pass, and the fact that starting a session no longer sweeps OTHER viewers' sessions at all.
 """
 
 import asyncio
 import time
+from types import SimpleNamespace
 
 import pytest
+from fastapi.testclient import TestClient
 
 import app.main as app_main
+import app.state as state
+from app.auth import current_user
+from app.db import get_db
 from app.main import app
+from app.models import PlexItem, PlexLibrary, User
+from app.plex import paths as plex_paths
 from app.plex import stream
 
 
@@ -27,7 +41,7 @@ class _Configured:
         return False
 
 
-def test_idle_sessions_are_reaped_while_nobody_starts_a_new_one(monkeypatch):
+def test_the_periodic_pass_is_the_one_that_runs_and_it_uses_the_short_rule(monkeypatch):
     async def _no_sweep():
         return None
 
@@ -39,7 +53,9 @@ def test_idle_sessions_are_reaped_while_nobody_starts_a_new_one(monkeypatch):
     monkeypatch.setattr(app_main, "shutdown_scheduler", lambda: None)
     monkeypatch.setattr(app_main.plex_subtitles, "shutdown", lambda: None)
     monkeypatch.setattr(app_main, "_HLS_REAP_EVERY_S", 0.01)
-    monkeypatch.setattr(app_main.plex_stream, "reap_abandoned", lambda: reaped.append(1) or 0)
+    # `reap_idle`, i.e. the SHORT finished rule: a finished session nobody has said they still want
+    # is a whole copy of a film sitting on the download volume.
+    monkeypatch.setattr(app_main.plex_stream, "reap_idle", lambda: reaped.append(1) or 0)
 
     async def _cycle():
         async with app_main.lifespan(app):
@@ -79,12 +95,10 @@ def sessions(monkeypatch):
     return live
 
 
-def test_the_periodic_pass_keeps_a_paused_viewers_finished_session(tmp_path, sessions):
-    # A remux that finished a minute ago, its viewer paused with a full buffer since.
-    s = _Session(tmp_path, "paused", finished=True, idle_s=90)
-    sessions["12345"] = s
-    assert stream.reap_abandoned() == 0
-    assert "12345" in sessions and s.dir.exists() and not s.proc.killed
+def test_the_periodic_pass_reaps_a_finished_session_nobody_has_claimed(tmp_path, sessions):
+    sessions["12345"] = _Session(tmp_path, "done", finished=True, idle_s=45)
+    assert stream.reap_idle() == 1
+    assert sessions == {}
 
 
 def test_the_periodic_pass_still_reaps_what_nobody_has_touched_for_the_idle_limit(
@@ -96,11 +110,95 @@ def test_the_periodic_pass_still_reaps_what_nobody_has_touched_for_the_idle_limi
     sessions["2"] = _Session(
         tmp_path, "gone_live", finished=False, idle_s=stream._SESSION_IDLE_S + 5
     )
-    assert stream.reap_abandoned() == 2
+    assert stream.reap_idle() == 2
     assert sessions == {}
 
 
-def test_a_new_session_request_still_reaps_finished_sessions_quickly(tmp_path, sessions):
-    sessions["12345"] = _Session(tmp_path, "done", finished=True, idle_s=45)
-    assert stream.reap_idle() == 1
-    assert sessions == {}
+def test_touching_a_session_keeps_it_through_the_short_rule(tmp_path, sessions):
+    s = _Session(tmp_path, "paused", finished=True, idle_s=90)
+    sessions["12345"] = s
+    stream.touch_session("12345")
+    assert stream.reap_idle() == 0
+    assert "12345" in sessions and s.dir.exists() and not s.proc.killed
+
+
+# --- the two call sites, driven through the real routes (DB lane) -------------------------------
+
+
+def _library_item(db, rating_key: str, *, playable: str = "remux"):
+    lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
+    db.add(lib)
+    db.flush()
+    it = PlexItem(
+        rating_key=rating_key,
+        library_id=lib.id,
+        kind="movie",
+        title="A Movie",
+        file_path="/media/a.mkv",
+        playable=playable,
+    )
+    db.add(it)
+    db.commit()
+    return it
+
+
+def _client(db, user):
+    app.dependency_overrides[get_db] = lambda: db
+    app.dependency_overrides[current_user] = lambda: user
+    return TestClient(app)
+
+
+def test_a_paused_viewers_checkpoint_keeps_their_session_alive(db, monkeypatch, tmp_path, sessions):
+    """The whole reason the short rule is safe again: a paused player fetches nothing, so
+    `last_access` never moved on its own — but it does keep checkpointing its position."""
+    monkeypatch.setattr(state, "_configured_cache", True)
+    it = _library_item(db, "paused1")
+    user = User(email="paused@example.com")
+    db.add(user)
+    db.commit()
+    s = _Session(tmp_path, "paused", finished=True, idle_s=90)
+    sessions["paused1"] = s
+    try:
+        resp = _client(db, user).post(
+            f"/api/plex/item/{it.rating_key}/progress",
+            json={"position_seconds": 600, "duration_seconds": 3000, "final": False},
+        )
+    finally:
+        app.dependency_overrides.clear()
+    assert resp.status_code == 200, resp.text
+    assert stream.reap_idle() == 0, "the checkpoint did not say the viewer is still there"
+    assert "paused1" in sessions and not s.proc.killed
+
+
+def test_starting_a_session_leaves_another_viewers_paused_one_alone(
+    db, monkeypatch, tmp_path, sessions
+):
+    """The opportunistic sweep this route used to run was not scoped to the key being started: it
+    walked the WHOLE registry with the 30 s finished rule, so pressing play on any item killed a
+    paused viewer's session — the very case round 4 set out to protect (found in round 5)."""
+    monkeypatch.setattr(state, "_configured_cache", True)
+    monkeypatch.setattr(plex_paths, "local_media_path", lambda db_, fp: "/local/a.mkv")
+    started = _library_item(db, "starting")
+    user = User(email="starter@example.com")
+    db.add(user)
+    db.commit()
+    paused = _Session(tmp_path, "paused", finished=True, idle_s=90)
+    sessions["someone-else"] = paused
+
+    def fake_start(db_, item, start, audio=None, aoff=0.0, multi=False, transcode=False):
+        return SimpleNamespace(
+            id="sid",
+            media_start_s=start,
+            entry="index.m3u8",
+            transcode=transcode,
+            hw=False,
+            tonemap=False,
+        )
+
+    monkeypatch.setattr(stream, "start_session", fake_start)
+    try:
+        resp = _client(db, user).post(f"/api/plex/stream/{started.rating_key}/session")
+    finally:
+        app.dependency_overrides.clear()
+    assert resp.status_code == 200, resp.text
+    assert "someone-else" in sessions and not paused.proc.killed