refactor(plex): split routes/plex.py into an admin/browse/playback package

R12 S1 step B. The 2011-line routes/plex.py becomes an app/routes/plex/
package:

- admin.py     — connection test, catalog sync, owner watch-link config (5)
- browse.py    — libraries, facets, collections, playlists, show/item, people (23)
- playback.py  — image/subtitle proxies, HLS streaming, watch-state writes (10)
- _shared.py   — the two genuinely cross-module helpers (_item_or_404, log)
- __init__.py  — assembles `router` from the three sub-routers; the app wiring
                 (plex_routes.router in main.py) is unchanged.

Mechanically generated and proven behaviour-equivalent: every function/class
source relocated verbatim (get_source_segment equality) and the (method, path)
route set is identical to before. Endpoints keep the /api/plex prefix on each
sub-router. No SQL change (db.query untouched; 2.0 select() is the next step).

Gate: ruff clean, pytest 193 passed (pure-logic + DB lane).

backend/app/routes/plex/__init__.py

@@ -0,0 +1,13 @@
+"""Plex route package — assembles the admin/browse/playback sub-routers.
+
+Split out of the former single routes/plex.py (R12 S1). `router` keeps the same shape the app
+wiring expects (`plex_routes.router`), so main.py is unchanged.
+"""
+from fastapi import APIRouter
+
+from app.routes.plex import admin, browse, playback
+
+router = APIRouter()
+router.include_router(admin.router)
+router.include_router(browse.router)
+router.include_router(playback.router)

backend/app/routes/plex/_shared.py

@@ -0,0 +1,16 @@
+"""Shared helpers, serializers and constants for the plex route package."""
+
+import logging
+from fastapi import HTTPException
+from sqlalchemy.orm import Session
+from app.models import (
+    PlexItem,
+)
+
+log = logging.getLogger("siftlode.plex")
+
+def _item_or_404(db: Session, rating_key: str) -> PlexItem:
+    it = db.query(PlexItem).filter_by(rating_key=str(rating_key)).first()
+    if it is None:
+        raise HTTPException(status_code=404, detail="Unknown Plex item")
+    return it

backend/app/routes/plex/admin.py

@@ -0,0 +1,143 @@
+"""Plex admin: connection test, catalog sync, and owner watch-link config."""
+
+from datetime import datetime, timezone
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel
+from sqlalchemy.orm import Session
+from app import sysconfig
+from app.auth import current_user
+from app.db import get_db
+from app.models import (
+    PlexItem,
+    PlexLink,
+    User,
+)
+from app.plex import paths as plex_paths
+from app.plex import sync as plex_sync
+from app.plex import watch_sync as plex_watch
+from app.plex.client import PlexClient, PlexError, PlexNotConfigured
+from app.routes.admin import admin_user
+from app.utils import iso
+
+router = APIRouter(prefix="/api/plex", tags=["plex"])
+
+@router.post("/test")
+def test_connection(_: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict:
+    """Admin: verify the configured Plex server URL + token, returning the server name and the
+    movie/show sections (for the config library-picker). Also probes ONE real media file through
+    the local mount, so a missing bind-mount / wrong path-map is caught here rather than only
+    surfacing as a dead player later."""
+    try:
+        with PlexClient(db) as plex:
+            info = plex.server_info()
+            sections = plex.sections()
+            media_check = _media_check(db, plex, sections)
+    except PlexNotConfigured as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    except PlexError as e:
+        raise HTTPException(status_code=422, detail=f"Plex connection failed: {e}")
+    return {
+        "ok": True,
+        "server_name": info.get("friendlyName") or info.get("title1") or "Plex",
+        "version": info.get("version"),
+        "sections": [
+            {"key": str(s.get("key")), "title": s.get("title"), "type": s.get("type"), "count": s.get("count")}
+            for s in sections
+            if s.get("type") in ("movie", "show")
+        ],
+        "media_check": media_check,
+    }
+
+def _sample_plex_path(db: Session, plex: PlexClient, sections: list[dict]) -> str | None:
+    """A real Plex-side media file path to validate the mount + path-map against. Prefers the
+    already-mirrored catalog (cheap); before the first sync, pulls one movie leaf from an enabled
+    section straight from Plex."""
+    row = db.query(PlexItem.file_path).filter(PlexItem.file_path.isnot(None)).first()
+    if row and row[0]:
+        return row[0]
+    wanted = plex_sync._enabled_section_keys(db)
+    for s in sections:
+        if s.get("type") != "movie":
+            continue
+        key = str(s.get("key"))
+        if wanted is not None and key not in wanted:
+            continue
+        items, _ = plex.section_items(key, 1, 0, 1)
+        if items:
+            fp = plex_sync._media_facts(items[0]).get("file_path")
+            if fp:
+                return fp
+    return None
+
+def _media_check(db: Session, plex: PlexClient, sections: list[dict]) -> dict:
+    """Resolve a sample Plex file path to its local mount path and confirm it's readable, so the
+    admin sees at Test time whether the physical media is actually reachable (bind-mount +
+    path-map correct). ``checked=False`` when no sample is available (empty library)."""
+    plex_path = _sample_plex_path(db, plex, sections)
+    if not plex_path:
+        return {"checked": False}
+    local_path = plex_paths.map_path(db, plex_path)
+    ok = plex_paths.local_media_path(db, plex_path) is not None
+    return {"checked": True, "ok": ok, "plex_path": plex_path, "local_path": local_path}
+
+@router.post("/sync")
+def trigger_sync(_: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict:
+    """Admin: mirror the enabled Plex sections into the local catalog. Synchronous (can take a
+    while on a large library); a background/scheduled sync also runs periodically."""
+    return plex_sync.sync(db)
+
+def _link_dict(link: PlexLink | None) -> dict:
+    return {
+        "linked": link is not None,
+        "sync_enabled": bool(link and link.sync_enabled),
+        "uses_admin": bool(link and link.uses_admin),
+        "initial_import_done": bool(link and link.initial_import_done),
+        "last_watch_sync_at": iso(link.last_watch_sync_at) if link else None,
+    }
+
+@router.get("/watch/link")
+def watch_link_status(
+    user: User = Depends(current_user), db: Session = Depends(get_db)
+) -> dict:
+    """This user's Plex watch-sync link status (drives the Settings toggle)."""
+    link = db.query(PlexLink).filter_by(user_id=user.id).first()
+    return _link_dict(link)
+
+class WatchLinkIn(BaseModel):
+    enabled: bool = False
+
+@router.post("/watch/link")
+def watch_link_set(
+    body: WatchLinkIn,
+    user: User = Depends(admin_user),
+    db: Session = Depends(get_db),
+) -> dict:
+    """Admin: turn two-way Plex watch-sync on/off for THIS (owner) account. The link uses the server
+    admin token, so no separate Plex login. Enabling it for the first time runs the one-time
+    "Plex is master" import immediately so the owner's existing Plex history shows up in Siftlode."""
+    if not sysconfig.get_bool(db, "plex_enabled"):
+        raise HTTPException(status_code=409, detail="The Plex module is disabled.")
+    enabled = body.enabled
+    link = db.query(PlexLink).filter_by(user_id=user.id).first()
+    if link is None:
+        link = PlexLink(user_id=user.id, uses_admin=True, linked_at=datetime.now(timezone.utc))
+        db.add(link)
+    link.sync_enabled = enabled
+    db.commit()
+
+    result: dict = {}
+    if enabled and not link.initial_import_done:
+        result = plex_watch.import_owner_watch_state(db, link)
+    return {**_link_dict(link), "import": result or None}
+
+@router.post("/watch/import")
+def watch_import_now(
+    user: User = Depends(admin_user), db: Session = Depends(get_db)
+) -> dict:
+    """Admin: (re-)run the "Plex is master" import for this account — Plex re-wins on the
+    intersection; Siftlode-only states are preserved. Idempotent."""
+    link = db.query(PlexLink).filter_by(user_id=user.id).first()
+    if link is None or not link.sync_enabled:
+        raise HTTPException(status_code=409, detail="Enable Plex watch-sync first.")
+    result = plex_watch.import_owner_watch_state(db, link)
+    return {**_link_dict(link), "import": result}

backend/app/routes/plex/playback.py

@@ -0,0 +1,450 @@
+"""Plex playback: image/subtitle proxies, HLS streaming, and watch-state writes."""
+
+import hashlib
+import os
+import tempfile
+from datetime import datetime, timezone
+import httpx
+from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request, Response
+from fastapi.responses import FileResponse
+from pydantic import BaseModel
+from sqlalchemy.orm import Session
+from app.auth import current_user, resolved_user_id
+from app.db import SessionLocal, get_db
+from app.models import (
+    PlexItem,
+    PlexSeason,
+    PlexShow,
+    PlexState,
+    User,
+)
+from app.plex import paths as plex_paths
+from app.plex import people as plex_people
+from app.plex import stream as plex_stream
+from app.plex import watch_sync as plex_watch
+from app.plex.client import PlexClient, PlexError, PlexNotConfigured
+from app.plex import imgcache, subtitles
+
+router = APIRouter(prefix="/api/plex", tags=["plex"])
+
+from app.routes.plex._shared import (
+    _item_or_404,
+)
+
+def _require_session(request: Request) -> int:
+    """Lightweight auth for the high-fan-out image proxies: validate the signed session cookie WITHOUT
+    a DB user-load, so these endpoints never hold a pool connection for the auth. Poster images are
+    low-sensitivity; a valid session is enough."""
+    uid, _ = resolved_user_id(request)
+    if not uid:
+        raise HTTPException(status_code=401, detail="Not authenticated")
+    return uid
+
+@router.get("/image/{rating_key}")
+def image(rating_key: str, request: Request, variant: str = "thumb") -> Response:
+    """Proxy a Plex poster/art image (keeps the admin token server-side; no image duplication). A
+    thin on-disk cache serves repeat views without re-hitting Plex.
+
+    ⚠️ This endpoint is called in BURSTS (a collections page fires 100+ concurrent poster requests),
+    so it must NOT hold a DB connection during the slow Plex fetch — doing so exhausted the pool
+    (QueuePool timeout, broken images on prod). Warm path (cache hit) touches NO DB; cold path opens a
+    brief session only to resolve the image key + read the Plex config, then RELEASES it before the
+    fetch. Auth is the signed-session check (no DB user-load)."""
+    _require_session(request)
+    v = "art" if variant == "art" else "thumb"
+    w, h = imgcache.SIZES[v]
+    # Size is part of the cache key: Plex resizes server-side (see PlexClient.image_bytes), so we cache
+    # the small variant. The width tag also means changing a target size just re-caches, never serves a
+    # stale differently-sized file (old full-size files under the un-tagged key are simply orphaned).
+    cache_key = f"item_{rating_key}_{v}_{w}"
+    hit = imgcache.read(cache_key)
+    if hit:
+        return imgcache.response(*hit)
+    # Cold miss: short-lived session for the key + Plex client config, released before the slow fetch.
+    try:
+        with SessionLocal() as db:
+            key = imgcache.key_for(db, str(rating_key), v)
+            if not key:
+                raise HTTPException(status_code=404, detail="No image")
+            plex = PlexClient(db)  # reads base URL + token now; the fetch below needs no DB
+    except PlexNotConfigured as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    try:
+        with plex:
+            data, ct = plex.image_bytes(key, w, h)
+    except PlexError as e:
+        raise HTTPException(status_code=502, detail=f"Plex image fetch failed: {e}")
+    imgcache.write(cache_key, data, ct)
+    return imgcache.response(data, ct)
+
+@router.get("/subtitle/{rating_key}/{sub_ord}")
+def subtitle(rating_key: str, sub_ord: int, request: Request, offset: float = 0.0) -> Response:
+    """Serve the ord-th subtitle track as WebVTT for the browser's <track> — so choosing a subtitle
+    overlays it directly (no HLS re-mux / session restart). Handles EXTERNAL sidecar subs (fetched
+    from Plex via the stream `key` — the old `-map 0:s` broke on these since they're not in the file)
+    AND embedded text subs (extracted from the local file). Image subs (PGS/VobSub) → 415. Disk-cached."""
+    _require_session(request)
+    # Cache the ABSOLUTE-time VTT once (per rk+ord); the per-session cue-shift is applied on the way
+    # out, so a resume/seek to any offset reuses the same cached track (no cache-per-offset explosion).
+    cache_key = f"sub_{rating_key}_{sub_ord}"
+    hit = subtitles.cache_read(cache_key)
+    if hit is not None:
+        return Response(subtitles.shift_vtt(hit, offset), media_type="text/vtt")
+    try:
+        with SessionLocal() as db:
+            it = _item_or_404(db, rating_key)
+            with PlexClient(db) as plex:
+                part = ((plex.metadata(it.rating_key) or {}).get("Media") or [{}])[0].get("Part") or [{}]
+                subs = [s for s in (part[0].get("Stream") or []) if s.get("streamType") == 3]
+                if sub_ord < 0 or sub_ord >= len(subs):
+                    raise HTTPException(status_code=404, detail="No such subtitle")
+                s = subs[sub_ord]
+                codec = (s.get("codec") or s.get("format") or "").lower()
+                if codec in subtitles.BITMAP_SUBS:
+                    raise HTTPException(status_code=415, detail="Image-based subtitle can't be shown as text")
+                if s.get("key"):  # external sidecar → fetch from Plex, convert to VTT
+                    raw = plex.raw_get(s["key"])
+                    if codec in ("srt", "subrip", ""):
+                        vtt = subtitles.srt_to_vtt(raw.decode("utf-8", "replace"))
+                    else:
+                        fd, tmp = tempfile.mkstemp(suffix="." + (codec or "sub"))
+                        try:
+                            os.write(fd, raw)
+                            os.close(fd)
+                            vtt = subtitles.ffmpeg_to_vtt(tmp, None)
+                        finally:
+                            os.unlink(tmp)
+                elif s.get("index") is not None:  # embedded → extract from the local file
+                    src = plex_paths.local_media_path(db, it.file_path)
+                    if src is None:
+                        raise HTTPException(status_code=404, detail="Media file not found on disk")
+                    emb_ord = sum(1 for x in subs[:sub_ord] if x.get("index") is not None)
+                    vtt = subtitles.ffmpeg_to_vtt(str(src), emb_ord)
+                else:
+                    raise HTTPException(status_code=404, detail="Subtitle not fetchable")
+    except PlexNotConfigured as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    except PlexError as e:
+        raise HTTPException(status_code=502, detail=f"Subtitle fetch failed: {e}")
+    subtitles.cache_write(cache_key, vtt)
+    return Response(subtitles.shift_vtt(vtt, offset), media_type="text/vtt")
+
+_PROGRESS_MIN_S = 5
+
+_FINISH_MARGIN_S = 30
+
+@router.get("/person-image")
+def person_image(u: str, request: Request) -> Response:
+    """Proxy a cast/crew photo from Plex's public metadata CDN (host-whitelisted; no token needed).
+    Signed-session auth (no DB) so it, like /image, never holds a pool connection during the fetch."""
+    _require_session(request)
+    if not u.startswith(plex_people.PERSON_IMG_HOST):
+        raise HTTPException(status_code=400, detail="Unsupported image host")
+    cache_key = "person_" + hashlib.sha1(u.encode()).hexdigest()
+    hit = imgcache.read(cache_key)
+    if hit:
+        return imgcache.response(*hit)
+    try:
+        with httpx.Client(timeout=10.0, follow_redirects=True) as c:
+            r = c.get(u)
+            r.raise_for_status()
+    except httpx.HTTPError:
+        raise HTTPException(status_code=502, detail="Image fetch failed")
+    ct = r.headers.get("content-type", "image/jpeg")
+    imgcache.write(cache_key, r.content, ct)
+    return imgcache.response(r.content, ct)
+
+def _push_watch(
+    background: BackgroundTasks,
+    db: Session,
+    user_id: int,
+    item_id: int,
+    rating_key: str,
+    action: str,
+    time_ms: int = 0,
+    duration_ms: int = 0,
+) -> None:
+    """Schedule a best-effort Siftlode→Plex watch push, but only for a user whose owner link has
+    push enabled — non-owner users never touch Plex. Guarding here (with the request session) avoids
+    spinning up a throwaway background session on every state change for the common no-link case."""
+    if plex_watch.link_for_push(db, user_id) is None:
+        return
+    # Capture the row's freshest timestamp NOW so the background push can tell if a newer local edit
+    # lands before it settles synced_to_plex (which would otherwise strand that newer value).
+    st = db.query(PlexState).filter_by(user_id=user_id, item_id=item_id).first()
+    background.add_task(
+        plex_watch.push_state_to_plex, user_id, item_id, rating_key, action, time_ms, duration_ms,
+        plex_watch._state_marker(st),
+    )
+
+class ItemProgressIn(BaseModel):
+    # float, truncated in the handler — the player may post a raw currentTime (a float); a plain
+    # `int` field would 422 it on the periodic progress checkpoint (the old `int(...)` truncated).
+    position_seconds: float = 0
+    duration_seconds: float = 0
+    final: bool = False
+
+@router.post("/item/{rating_key}/progress")
+def item_progress(
+    rating_key: str,
+    body: ItemProgressIn,
+    background: BackgroundTasks,
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> dict:
+    """Checkpoint the resume position while playing (near the end → mark watched + clear it).
+
+    `final` (sent on the pause/pagehide/unmount flush, not the periodic 10s checkpoint) is what
+    debounces the Plex resume push: we mirror the settled position to Plex only on a final flush, so
+    a single watch doesn't spray `/:/timeline` calls at the server every 10 seconds."""
+    it = _item_or_404(db, rating_key)
+    position = max(0, int(body.position_seconds))
+    duration = int(body.duration_seconds)
+    final = body.final
+    now = datetime.now(timezone.utc)
+    st = db.query(PlexState).filter_by(user_id=user.id, item_id=it.id).first()
+    was_watched = st is not None and st.status == "watched"
+
+    near_end = duration > 0 and position > duration - _FINISH_MARGIN_S
+    if near_end:
+        if st is None:
+            st = PlexState(user_id=user.id, item_id=it.id)
+            db.add(st)
+        st.status = "watched"
+        st.watched_at = now
+        st.position_seconds = 0
+        st.progress_updated_at = now
+        st.synced_to_plex = False
+        db.commit()
+        # Immediate on the watched transition; skip a repeat scrobble if it was already watched
+        # (a later checkpoint of the same finished item would otherwise bump Plex's viewCount again).
+        if not was_watched:
+            _push_watch(background, db, user.id, it.id, rating_key, "watched")
+        return {"status": "watched", "position_seconds": 0}
+
+    if position < _PROGRESS_MIN_S:
+        if st is not None and st.status == "new":
+            db.delete(st)
+        elif st is not None:
+            st.position_seconds = 0
+            st.progress_updated_at = now
+        db.commit()
+        return {"position_seconds": 0}
+
+    if st is None:
+        st = PlexState(user_id=user.id, item_id=it.id)
+        db.add(st)
+    st.position_seconds = position
+    st.progress_updated_at = now
+    if final:
+        st.synced_to_plex = False
+    db.commit()
+    if final:
+        _push_watch(
+            background, db, user.id, it.id, rating_key, "resume", position * 1000, duration * 1000
+        )
+    return {"position_seconds": position}
+
+class ItemStateIn(BaseModel):
+    status: str = ""
+
+@router.post("/item/{rating_key}/state")
+def item_state(
+    rating_key: str,
+    body: ItemStateIn,
+    background: BackgroundTasks,
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> dict:
+    """Set the per-user watch status (new|watched|hidden). watched→scrobble / →new→unscrobble push
+    to a linked Plex account immediately; `hidden` is a Siftlode-only concept with no Plex analogue,
+    so it never touches Plex."""
+    it = _item_or_404(db, rating_key)
+    status = body.status
+    if status not in ("new", "watched", "hidden"):
+        raise HTTPException(status_code=400, detail="Invalid status")
+    st = db.query(PlexState).filter_by(user_id=user.id, item_id=it.id).first()
+    prev_status = st.status if st is not None else "new"
+    if status == "new":
+        if st is not None:
+            db.delete(st)
+            db.commit()
+        if prev_status != "new":  # something → unwatched: mirror the un-mark to Plex
+            _push_watch(background, db, user.id, it.id, rating_key, "unwatched")
+        return {"status": "new"}
+    if st is None:
+        st = PlexState(user_id=user.id, item_id=it.id)
+        db.add(st)
+    st.status = status
+    if status == "watched":
+        st.watched_at = datetime.now(timezone.utc)
+        st.position_seconds = 0
+    st.synced_to_plex = False
+    db.commit()
+    if status == "watched" and prev_status != "watched":
+        _push_watch(background, db, user.id, it.id, rating_key, "watched")
+    return {"status": status}
+
+def _bulk_state(
+    background: BackgroundTasks, db: Session, user_id: int, eps: list[PlexItem], watched: bool
+) -> int:
+    """Mark every episode in `eps` watched (or unwatched) for a user, mirroring each change to a
+    linked Plex account in the background (best-effort). Returns how many rows actually changed."""
+    now = datetime.now(timezone.utc)
+    existing = {
+        s.item_id: s
+        for s in db.query(PlexState).filter(
+            PlexState.user_id == user_id, PlexState.item_id.in_([e.id for e in eps] or [0])
+        )
+    }
+    pushable = plex_watch.link_for_push(db, user_id) is not None
+    to_push: list[tuple[int, str, str]] = []
+    for e in eps:
+        st = existing.get(e.id)
+        prev = st.status if st is not None else "new"
+        if watched:
+            if st is None:
+                st = PlexState(user_id=user_id, item_id=e.id)
+                db.add(st)
+            st.status = "watched"
+            st.watched_at = now
+            st.position_seconds = 0
+            st.synced_to_plex = False
+            if prev != "watched":
+                to_push.append((e.id, e.rating_key, "watched"))
+        else:  # → new: drop any state (but keep the user's private "hidden" marks)
+            if st is not None and prev != "hidden":
+                db.delete(st)
+                if prev != "new":
+                    to_push.append((e.id, e.rating_key, "unwatched"))
+    db.commit()
+    if pushable and to_push:
+        # One coalesced background task for the whole batch (reuses a single session + Plex client),
+        # not one task per episode — a 200-ep show mark is 1 task, not 200.
+        background.add_task(plex_watch.push_bulk_state_to_plex, user_id, to_push)
+    return len(to_push)
+
+class WatchedIn(BaseModel):
+    watched: bool = False
+
+@router.post("/show/{rating_key}/state")
+def show_state(
+    rating_key: str,
+    body: WatchedIn,
+    background: BackgroundTasks,
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> dict:
+    """Mark a whole show watched/unwatched (all its episodes) for the current user + push to Plex."""
+    sh = db.query(PlexShow).filter_by(rating_key=str(rating_key)).first()
+    if sh is None:
+        raise HTTPException(status_code=404, detail="Unknown Plex show")
+    watched = body.watched
+    eps = db.query(PlexItem).filter_by(show_id=sh.id, kind="episode").all()
+    changed = _bulk_state(background, db, user.id, eps, watched)
+    return {"changed": changed, "watched": watched}
+
+@router.post("/season/{rating_key}/state")
+def season_state(
+    rating_key: str,
+    body: WatchedIn,
+    background: BackgroundTasks,
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> dict:
+    """Mark a whole season watched/unwatched (all its episodes) for the current user + push to Plex."""
+    se = db.query(PlexSeason).filter_by(rating_key=str(rating_key)).first()
+    if se is None:
+        raise HTTPException(status_code=404, detail="Unknown Plex season")
+    watched = body.watched
+    eps = db.query(PlexItem).filter_by(season_id=se.id, kind="episode").all()
+    changed = _bulk_state(background, db, user.id, eps, watched)
+    return {"changed": changed, "watched": watched}
+
+@router.post("/stream/{rating_key}/session")
+def stream_session(
+    rating_key: str,
+    start: float = 0.0,
+    audio: int | None = None,
+    aoff: float = 0.0,
+    multi: bool = False,
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> dict:
+    """Prepare playback from `start` seconds. Direct-playable files are served raw; remux files
+    (re)start an HLS session at the offset (seek-restart); transcode (HEVC/VP9) is P3. Selecting a
+    non-default audio track (`audio` = stream ordinal) forces the HLS path so the chosen audio can be
+    muxed in. Subtitles are independent WebVTT tracks (GET /subtitle) — no session restart."""
+    plex_stream.reap_idle()  # opportunistic cleanup of idle sessions on each new playback
+    it = _item_or_404(db, rating_key)
+    if it.playable == "transcode":
+        raise HTTPException(status_code=501, detail="This format needs transcoding (a later phase).")
+    # `multi` (item has ≥2 audio tracks) forces the HLS path so ALL audio tracks ship as renditions and
+    # the client can switch audio without a restart — even for otherwise direct-playable files.
+    if it.playable == "direct" and audio is None and not aoff and not multi:
+        return {"mode": "direct", "url": f"/api/plex/stream/{it.rating_key}/file", "start_s": 0.0}
+    if plex_paths.local_media_path(db, it.file_path) is None:
+        raise HTTPException(status_code=404, detail="Media file not found on disk")
+    s = plex_stream.start_session(db, it, start, audio, aoff, multi)
+    if s is None:
+        raise HTTPException(status_code=404, detail="Media file not found on disk")
+    # Report the REAL media start (the keyframe ffmpeg landed on, which hls.js zero-bases to), not the
+    # requested offset — the client's absolute clock and the subtitle cue-shift both key off this.
+    return {"mode": "hls", "start_s": s.media_start_s, "url": f"/api/plex/stream/{it.rating_key}/hls/{s.entry}"}
+
+@router.get("/stream/{rating_key}/file")
+def stream_file(
+    rating_key: str,
+    download: bool = False,
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> FileResponse:
+    """Serve the raw local media file with HTTP range support. Used both for direct-play (inline)
+    and for the player's "download original" button (`?download=1` → attachment). The download is
+    the untouched physical file — no re-encode/repackage."""
+    it = _item_or_404(db, rating_key)
+    src = plex_paths.local_media_path(db, it.file_path)
+    if src is None:
+        raise HTTPException(status_code=404, detail="Media file not found on disk")
+    if download:
+        # A clean, safe download name from the title + the real file extension.
+        ext = src.suffix or (f".{it.container}" if it.container else "")
+        base = "".join(c for c in (it.title or "video") if c.isalnum() or c in " -_.").strip() or "video"
+        return FileResponse(str(src), filename=f"{base}{ext}")
+    return FileResponse(str(src))  # Starlette honours the Range header (206)
+
+_HLS_CT = {".m3u8": "application/vnd.apple.mpegurl", ".ts": "video/mp2t", ".vtt": "text/vtt"}
+
+@router.get("/stream/{rating_key}/hls/{filename}")
+def stream_hls_file(
+    rating_key: str,
+    filename: str,
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> Response:
+    """Serve a file (playlist / segment / subtitle) from the current HLS session directory. The
+    playlist references its segments/subtitles by relative name, all resolving back here."""
+    # Filename must be a plain hls artifact — no path separators, only the expected extensions.
+    ext = "." + filename.rsplit(".", 1)[-1] if "." in filename else ""
+    if ext not in _HLS_CT or "/" in filename or "\\" in filename or ".." in filename:
+        raise HTTPException(status_code=404, detail="Not found")
+    s = plex_stream.current_session(str(rating_key))
+    if s is None:
+        # No session yet (e.g. reaped): the ENTRY playlist may auto-start one at 0 — master.m3u8 for a
+        # multi-audio item, index.m3u8 for a single-audio remux item.
+        it = _item_or_404(db, rating_key)
+        if filename == "master.m3u8":
+            s = plex_stream.start_session(db, it, 0.0, multi=True)
+        elif filename == "index.m3u8" and it.playable == "remux":
+            s = plex_stream.start_session(db, it, 0.0)
+        else:
+            raise HTTPException(status_code=404, detail="No active playback session")
+        if s is None:
+            raise HTTPException(status_code=404, detail="Media file not found on disk")
+    path = s.dir / filename
+    timeout = 30 if ext == ".m3u8" else 20
+    if not plex_stream.wait_for(path, timeout):
+        raise HTTPException(status_code=404, detail="Not available yet")
+    return Response(
+        content=path.read_bytes(), media_type=_HLS_CT[ext], headers={"Cache-Control": "no-store"}
+    )