refactor(player): the Plex session machine is a tested engine, one request per item in flight

PlexPlayer's session logic (generation, pending play intent, clock ownership, session offset, reclaim
budget, listener teardown) moves into lib/hlsEngine.ts, a plain class with hls.js behind lib/hlsPort.ts
and its own unit tests.

- A seek, track change or reclaim asked for while a session request is out replaces the queued
  target; it is sent once the in-flight answer settles. A burst of 20 presses past the buffer was 20
  requests and 20 ffmpeg spawns; it is now 2-3, and plays again ~2 s sooner.
- The player stops its server session when it leaves the item, closes or the page unloads
  (POST /api/plex/stream/{rk}/session/{sid}/stop, identity-aware). A left transcode used to run to
  the end of the film first.
- A restart asked for after a new session attached but before it played no longer reads the emptied
  media as the viewer's pause.

backend/app/plex/stream.py

@@ -1117,6 +1117,23 @@ def reap_idle(finished_idle_s: float = _FINISHED_IDLE_S) -> int:
     return dropped
 
 
+def stop_session(user_id: int, key: str, session_id: str) -> bool:
+    """Stop this viewer's session for `key` now, because the player that asked for it is gone (left
+    the item, closed, unloaded the page). Left to `reap_idle`, a transcode ran on to the end of the
+    film first — measured: ~4 cores for a minute after leaving a resumed 480p film.
+
+    Only while `session_id` is still the registered one: the player names the session it played, and
+    a seek-restart or a second tab may have replaced it meanwhile — that newer session is somebody's
+    current playback, not this request's to end."""
+    with _lock:
+        s = _sessions.get((user_id, key))
+        if s is None or s.id != session_id:
+            return False
+        _kill(s)
+        _sessions.pop((user_id, key), None)
+    return True
+
+
 def touch_session(user_id: int, key: str) -> None:
     """Say this viewer is still there for `key`, without fetching anything. `last_access` used to move
     only on an HLS file fetch, which a paused player does not do — so the only way to keep a paused

backend/app/routes/plex/playback.py

@@ -637,12 +637,27 @@ def stream_session(
         # The session id is part of the path (not a query param) so the playlist's RELATIVE segment
         # names resolve back into the same session — see `stream_hls_file` for why it has to.
         "url": f"/api/plex/stream/{it.rating_key}/hls/{s.id}/{s.entry}",
+        # What the player names when it no longer wants the session (POST …/session/{id}/stop).
+        "id": s.id,
         "transcode": s.transcode,
         "hw": s.hw,
         "tonemap": s.tonemap,
     }
 
 
+@router.post("/stream/{rating_key}/session/{session_id}/stop")
+def stream_session_stop(
+    rating_key: str,
+    session_id: str,
+    user: User = Depends(current_user),
+) -> dict:
+    """The player no longer wants this session (it left the item, closed, or the page unloaded):
+    its ffmpeg stops now instead of at the idle reaper. A POST, so the player can send it as a
+    keepalive beacon from a page being unloaded. Stops nothing but the caller's own session, and
+    only while it is still the one registered for the item (see `stop_session`)."""
+    return {"ok": True, "stopped": plex_stream.stop_session(user.id, rating_key, session_id)}
+
+
 @router.get("/stream/{rating_key}/file")
 def stream_file(
     rating_key: str,

backend/tests/test_plex_playback_session.py

@@ -96,6 +96,8 @@ def test_remux_item_stays_a_stream_copy_session(db, monkeypatch, override_dep):
     assert body["mode"] == "hls"
     assert body["transcode"] is False
     assert captured["transcode"] is False
+    # The id the player names when it stops the session (POST …/session/{id}/stop).
+    assert body["id"] == "sid"
 
 
 def test_direct_item_never_opens_a_session(db, monkeypatch, override_dep):

backend/tests/test_plex_session_stop.py

@@ -0,0 +1,101 @@
+"""A player that no longer wants its HLS session stops it, instead of leaving ffmpeg to run on.
+
+Left to the idle reaper, a transcode ran to the end of the film after the viewer had gone (measured on
+localdev: ~4 cores for a minute after leaving a resumed 480p film). The player names the session it
+played; the server stops it only while it is still that viewer's registered session for the item, so
+a stop that arrives after a seek-restart (or from a second tab) cannot end the newer playback.
+
+No `db` fixture: the route reads nothing from the database (`current_user` is overridden wholesale).
+"""
+
+import time
+
+import pytest
+from fastapi.testclient import TestClient
+
+import app.state as state
+from app.auth import current_user
+from app.main import app
+from app.models import User
+from app.plex import stream as plex_stream
+
+VIEWER = 41
+
+
+class _Proc:
+    def __init__(self):
+        self.terminated = False
+
+    def poll(self):
+        return None
+
+    def terminate(self):
+        self.terminated = True
+
+    def wait(self, timeout=None):
+        return 0
+
+
+class _Session:
+    def __init__(self, session_id: str, directory):
+        self.id = session_id
+        self.dir = directory
+        self.proc = _Proc()
+        self.last_access = time.time()
+
+
+@pytest.fixture
+def sessions(monkeypatch):
+    live: dict = {}
+    monkeypatch.setattr(plex_stream, "_sessions", live)
+    return live
+
+
+@pytest.fixture
+def client(monkeypatch, override_dep):
+    monkeypatch.setattr(state, "_configured_cache", True)
+    user = User(id=VIEWER, email="stop@example.com")
+    override_dep(current_user, lambda: user)
+    return TestClient(app)
+
+
+def _session(tmp_path, sid: str) -> _Session:
+    d = tmp_path / sid
+    d.mkdir()
+    return _Session(sid, d)
+
+
+def test_the_named_session_is_stopped_and_its_directory_goes(client, tmp_path, sessions):
+    s = _session(tmp_path, "sid-a")
+    sessions[(VIEWER, "m1")] = s
+    r = client.post("/api/plex/stream/m1/session/sid-a/stop", json={})
+    assert r.status_code == 200, r.text
+    assert r.json() == {"ok": True, "stopped": True}
+    assert s.proc.terminated
+    assert sessions == {}
+    assert not s.dir.exists()
+
+
+def test_a_session_that_replaced_the_named_one_plays_on(client, tmp_path, sessions):
+    """The stop of the session a seek-restart replaced arrives after the new one registered."""
+    newer = _session(tmp_path, "sid-b")
+    sessions[(VIEWER, "m1")] = newer
+    r = client.post("/api/plex/stream/m1/session/sid-a/stop", json={})
+    assert r.json() == {"ok": True, "stopped": False}
+    assert not newer.proc.terminated
+    assert sessions == {(VIEWER, "m1"): newer}
+
+
+def test_another_viewers_session_is_not_this_viewers_to_stop(client, tmp_path, sessions):
+    theirs = _session(tmp_path, "sid-a")
+    sessions[(7, "m1")] = theirs
+    r = client.post("/api/plex/stream/m1/session/sid-a/stop", json={})
+    assert r.json() == {"ok": True, "stopped": False}
+    assert not theirs.proc.terminated
+
+
+def test_a_stop_with_no_session_left_is_not_an_error(client, sessions):
+    """The reaper or a cap may have dropped it first; the beacon must not fail loudly for that."""
+    r = client.post("/api/plex/stream/m1/session/sid-a/stop", json={})
+    assert r.status_code == 200, r.text
+    assert r.json() == {"ok": True, "stopped": False}