siftlode/backend/app/plex/stream.py

"""On-the-fly HLS remux for Plex playback (seek-restart session model).

Playback is from the LOCAL physical file. Browser-compatible files (playable="direct") are served
raw with HTTP range requests. Everything else that is h264 video (playable="remux") is remuxed on
the fly to HLS with **video stream-copy** (cheap, I/O-bound — no video re-encode) and audio to AAC
when needed. Non-h264 video (playable="transcode": HEVC/VP9/AV1/MPEG-2/XviD…) needs a full re-encode
to H.264: on the prod host that runs on the Intel iGPU via VAAPI (`h264_vaapi`, near-zero CPU), with
software libx264 as the automatic fallback where the render node is absent or the source codec isn't
hardware-decodable. Same HLS session/seek machinery either way — only the video encoder differs.

Seek model (Jellyfin-style): one ffmpeg session per item, started at a given offset via `-ss`
(fast keyframe seek). A seek beyond the generated region restarts the session at the new offset,
so seeking is responsive even on long movies without pre-generating the whole file. ffmpeg's own
HLS muxer does the segmentation (the only reliable way to cut a stream-copy at keyframes).

Sessions live in this (single) API process; a periodic reaper kills idle ones and frees the temp
segments. The remux is CPU-light (video copy + a tiny audio transcode), so it runs fine on the
CPU-only prod host; only P3's full transcode is CPU-heavy.
"""

import logging
import os
import re
import secrets
import shutil
import subprocess
import threading
import time
from collections.abc import Sequence
from pathlib import Path

from sqlalchemy.orm import Session as DbSession

from app import sysconfig
from app.config import settings
from app.downloads.edit import encode_thread_cap
from app.models import PlexItem
from app.plex import paths
from app.plex.sync import HW_DECODABLE_CODECS

log = logging.getLogger("siftlode.plex")

_HLS_ROOT = (
    Path(settings.plex_hls_dir)
    if settings.plex_hls_dir
    else Path(settings.download_root) / ".plex-hls"
)
_SEG_SECONDS = 6
# Cold-start warm-up. A freshly-spawned ffmpeg has produced only the first segment (or two) by the
# time we could report the session ready; hls.js parses the EVENT playlist, drains that tiny buffer,
# then STALLS a few seconds into playback waiting for the next segment the still-warming ffmpeg
# hasn't written yet. Very visible on auto-advance to the next episode (start_s==0, which previously
# returned with ZERO buffer). So before reporting ready we wait for the first few segments to exist:
# the client then starts with ~_WARMUP_SEGMENTS×_SEG_SECONDS of buffer and, since stream-copy runs
# faster than realtime, ffmpeg stays ahead from there and never stalls again. Best-effort and bounded
# — we stop at the first segment that isn't ready within the per-segment budget (a slow miss is no
# worse than the old zero-buffer behaviour).
_WARMUP_SEGMENTS = 3
_WARMUP_SEG_TIMEOUT_S = 8.0
# The HLS variant ffmpeg gives the VIDEO rendition of a multi-audio session. `-var_stream_map` names
# each audio rendition (`name:a0`, `a1`, …) but leaves the video one unnamed, so ffmpeg falls back to
# its index for that one. See `segment_name` for why this is a constant and not a comment.
_VIDEO_VARIANT = "0"
_SESSION_IDLE_S = 600  # reap a session with no access for this long
# A session directory is named `{rating_key}_{int(start_s)}_{tag}_{aoff:+.2f}` (see start_session);
# before multi-audio it was just `{rating_key}_{int(start_s)}`. `sweep_orphans` deletes ONLY names
# matching one of those two shapes: `_HLS_ROOT` is admin-configurable (PLEX_HLS_DIR), so it may
# legitimately point at a shared scratch path we do not own, and a blanket "delete every directory
# under the root" would take the neighbours with it.
#
# The KEY is anchored to digits (a Plex rating_key is a number) — that is what keeps foreign names
# out. A `.+` there let `snapshot_2024_backup_+1.00` through, and pairing `.+` with an optional tail
# degenerated all the way to "anything ending in _<digits>" (`backup_2024`, `release_10`). A
# non-numeric rating_key would merely leave its directory unswept, which is the safe direction to
# fail. Covered by test_plex_stream.py — this regex is the only thing standing between the sweep
# and someone else's data.
_SESSION_DIR_RE = re.compile(r"^\d+_\d+(_[^_]+_[+-]\d+\.\d{2})?$")
# When this process started. `sweep_orphans` only deletes directories older than this: everything
# newer belongs to a session THIS run created, and since `start_session` reuses a directory path
# for the same (rating_key, offset, tag), "stale name" is not the same as "stale directory".
_PROCESS_START = time.time()
# Slack on that comparison, because the two sides come from DIFFERENT clocks. A file's mtime is
# stamped from the kernel's COARSE clock (updated once per timer tick), while `time.time()` reads
# the fine-grained one — measured in this very container, a directory created 2ms AFTER
# `_PROCESS_START` got an mtime 2ms BEFORE it. Without slack the sweep would classify a session
# started in the first instants after boot as a leftover and delete it while ffmpeg writes into it
# — and "the first instants after boot" is exactly when the sweep runs. Erring on the keep side
# costs nothing: a genuine leftover from the last seconds before the restart simply survives until
# the next one.
_MTIME_GRACE_S = 2.0

_lock = threading.Lock()
_sessions: dict[str, "HlsSession"] = {}


class HlsSession:
    def __init__(
        self,
        key: str,
        directory: Path,
        proc: subprocess.Popen,
        start_s: float,
        variants: "Sequence[str]",
        transcode: bool = False,
        hw: bool = False,
        tonemap: bool = False,
    ):
        # Unguessable per-session id, carried in the URLs the client is handed. Sessions are keyed
        # by rating_key alone, so a seek-restart or a SECOND viewer of the same item replaces the
        # registered one — and the old viewer's next request for `seg_12.ts`, resolved by rating_key,
        # was answered 200 with the NEW session's twelfth segment, produced from a different offset
        # under `-copyts`. Wrong bytes, not an error, so the client's reclaim never fired on it
        # (deep-review round 5). A request naming a session that is no longer registered is a 404,
        # which the player recovers from by restarting at the playhead.
        self.id = secrets.token_urlsafe(9)
        self.key = key
        self.dir = directory
        self.proc = proc
        self.transcode = transcode  # video re-encoded (non-h264 source), not stream-copied
        self.hw = hw  # transcode ran on the VAAPI hardware encoder (else libx264 software)
        self.tonemap = tonemap  # HW HDR/10-bit path: OpenCL tonemap between VAAPI decode and encode
        self.start_s = start_s  # the REQUESTED seek offset
        # The REAL media start: with `-ss X -c:v copy` ffmpeg can only cut at the nearest earlier
        # keyframe K<=X, and hls.js zero-bases the timeline to K (so video.currentTime=0 is K, not X).
        # We measure K from the first segment's absolute PTS (kept absolute by `-copyts`) and report
        # THAT as the session start, so the client's absolute clock (start + currentTime) and the
        # subtitle cue-shift both key off the true content position — no drift. Provisional = X until
        # the first segment is probed.
        self.media_start_s = start_s
        # The rendition names this session's ffmpeg was told to write — () for a single-audio
        # session. Held on the session because it is what `artifact_may_arrive` needs to tell a name
        # this session CAN produce from one belonging to the other argv branch (see it).
        self.variants = tuple(variants)
        # The playlist filename hls.js should load: the master of a multi-audio session, else the
        # single index. Derived, not passed, so it cannot disagree with the argv.
        self.entry = _MASTER_PL if self.variants else _INDEX_PL
        self.last_access = time.time()


def _seek_opts(start_s: float) -> list[str]:
    # -noaccurate_seek: with `-ss` before -i, video (stream-copy) backs up to the keyframe K<=X,
    # but the re-encoded audio is normally trimmed to the exact X — so audio starts (X-K) AFTER
    # video → seconds of silence at the start of every seek/audio-switch. noaccurate_seek makes
    # audio ALSO start at K, aligned with the video (K is our real clock anchor anyway).
    return ["-noaccurate_seek", "-ss", f"{start_s:.3f}"] if start_s > 0 else []


_HLS_TAIL = [
    "-f",
    "hls",
    "-hls_time",
    str(_SEG_SECONDS),
    "-hls_list_size",
    "0",
    # EVENT (not VOD): ffmpeg writes/appends the playlist AS segments complete, so playback can
    # start immediately from this session's offset. VOD only writes the playlist at the end. The
    # full seekbar comes from our known duration + the seek-restart model, not from the playlist.
    "-hls_playlist_type",
    "event",
    "-hls_segment_type",
    "mpegts",
    "-hls_flags",
    "independent_segments+temp_file",
]


# --- 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. 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.
#   * libx264 software — the automatic fallback where `/dev/dri` is absent (dev box, or prod before the
#     iGPU enablement) or the source codec isn't hardware-decodable (AV1/XviD on Gen 9.5). CPU-bound, so
#     it reserves a core via `-threads` (see encode_thread_cap).
# Everything else — the `-copyts`/`_HLS_TAIL`/segment/seek-restart model and the AAC audio path — is
# unchanged, so hls.js, subtitles, multi-audio and seeking keep working exactly as for remux.
_VAAPI_DEVICE = "/dev/dri/renderD128"
# Where an OpenCL ICD registers its driver. The Intel NEO runtime (intel-opencl-icd) drops an .icd file
# here; its presence is the precondition for the HW HDR→SDR tonemap graph (see _opencl_present).
_OPENCL_ICD_DIR = Path("/etc/OpenCL/vendors")
# Codecs the prod iGPU (UHD 630 / Quick Sync Gen 9.5) can hardware-DECODE — the precondition for the
# full-GPU VAAPI pipeline. A source outside it (AV1, MPEG-4 ASP/XviD…) falls back to software even with
# the render node present, because the GPU can't decode it to feed its own encoder. THE single source is
# plex.sync, so the runtime encoder choice here and the catalog `transcode_hw` facet can't disagree by
# codec; aliased to the old name to keep the call sites below unchanged.
_HW_DECODABLE = HW_DECODABLE_CODECS
# Color-transfer values that mark a true-HDR source (PQ / HLG). A HW-decodable HDR source can't take the
# plain scale_vaapi=format=nv12 path (Gen 9.5 VPP refuses it) — it needs the OpenCL tonemap graph
# (_venc_opts tonemap branch) when the OpenCL runtime is present, else software libx264. A 10-bit source
# WITHOUT one of these transfers is SDR, not HDR, and must not be tone-mapped. See _probe_video_kind.
_HDR_TRANSFERS = {"smpte2084", "arib-std-b67"}


# The three source classes that drive the transcode-pipeline choice (see _probe_video_kind).
_VK_HWSAFE = "hwsafe"  # 8-bit SDR — the plain scale_vaapi=format=nv12 GPU path
_VK_HDR = "hdr"  # PQ/HLG HDR — needs the OpenCL tonemap graph (or software if no runtime)
_VK_SDR10 = "sdr10"  # 10-bit *SDR* — a plain 10→8-bit downconvert, NEVER the HDR tone curve


def _probe_video_kind(src: Path) -> str:
    """Classify the source for the transcode pipeline — one of `_VK_HWSAFE` / `_VK_HDR` / `_VK_SDR10`:

      * `_VK_HWSAFE` — 8-bit SDR: safe for the plain VAAPI path (`scale_vaapi=format=nv12` + h264_vaapi).
      * `_VK_HDR`    — PQ/HLG HDR (any depth): the plain scale path breaks on Gen 9.5 VPP, so it needs the
        OpenCL tonemap graph (`tonemap_opencl=hable`) when the runtime is present, else software libx264.
      * `_VK_SDR10`  — 10-bit but *SDR* (bt709 etc.): NOT HDR. Applying the HDR hable tone curve here would
        visibly alter an SDR picture (crushed highlights/midtones) where only a 10→8-bit downconvert was
        needed, so it must NOT take the tonemap graph — it goes to software libx264, which CPU-decodes and
        downconverts to 8-bit yuv420p exactly. (Kept off the plain scale_vaapi 10-bit path too: that path
        is unverified on Gen 9.5 VPP, and software is the known-correct downconvert.)

    Ground truth from ffprobe (bit depth via pix_fmt / bits_per_raw_sample, HDR via color_transfer); the
    Plex mirror doesn't store these. Returns `_VK_HWSAFE` on any probe failure: the runtime HW→SW fallback
    in start_session backstops a HW session that still produces nothing, so a probe miss only costs the
    rare wasted first attempt, never a stranded item."""
    try:
        out = subprocess.run(
            [
                "ffprobe",
                "-v",
                "error",
                "-select_streams",
                "v:0",
                "-show_entries",
                "stream=pix_fmt,bits_per_raw_sample,color_transfer",
                "-of",
                "default=noprint_wrappers=1",
                str(src),
            ],
            capture_output=True,
            text=True,
            timeout=10,
        )
    except (subprocess.SubprocessError, OSError):
        return _VK_HWSAFE
    pix_fmt = bits = transfer = ""
    for line in out.stdout.splitlines():
        k, _, v = line.strip().partition("=")
        v = v.strip().lower()
        if k == "pix_fmt":
            pix_fmt = v
        elif k == "bits_per_raw_sample":
            bits = v
        elif k == "color_transfer":
            transfer = v
    if transfer in _HDR_TRANSFERS:
        return _VK_HDR  # PQ/HLG — needs a real HDR→SDR tonemap, not a plain downconvert
    # >8-bit but not HDR = 10-bit SDR. bits_per_raw_sample OR the pix_fmt bit-depth suffix
    # (yuv420p10le, p010le, yuv444p12le…) signals >8-bit even when the other is absent.
    if (bits.isdigit() and int(bits) > 8) or re.search(r"(?:10|12|16)(?:le|be)$", pix_fmt):
        return _VK_SDR10
    return _VK_HWSAFE


def _opencl_present() -> bool:
    """Whether an OpenCL runtime (Intel NEO / intel-opencl-icd) is installed — the precondition for the
    HW HDR→SDR tonemap graph. Gen 9.5 VPP can't tonemap natively (tonemap_vaapi fails), so the working
    HW path for a 10-bit/HDR source is VAAPI-decode → OpenCL tonemap → VAAPI-encode, which needs an ICD
    registered here. Absent on the dev box and until the image bakes NEO in → those sources take
    software libx264. Its own function so it's monkeypatchable in tests, mirroring _render_node_present."""
    try:
        return any(_OPENCL_ICD_DIR.glob("*.icd"))
    except OSError:
        return False


def _hwaccel_in_opts(hw: bool, tonemap: bool = False) -> list[str]:
    """VAAPI decode options, placed before `-i` so the source is decoded (and kept) on the GPU.
    `-hwaccel_output_format vaapi` keeps frames as GPU surfaces all the way to `h264_vaapi` — no
    round-trip to system memory. Empty for the software path.

    The tonemap variant additionally derives an OpenCL device FROM the VAAPI one (`opencl=ocl@va`) so
    tonemap_opencl can map the decoded surfaces zero-copy (see _venc_opts); it needs named devices, so it
    uses `-init_hw_device` + `-hwaccel_device` rather than the plain path's implicit `-vaapi_device`."""
    if not hw:
        return []
    if tonemap:
        return [
            "-init_hw_device",
            f"vaapi=va:{_VAAPI_DEVICE}",
            "-init_hw_device",
            "opencl=ocl@va",
            "-hwaccel",
            "vaapi",
            "-hwaccel_output_format",
            "vaapi",
            "-hwaccel_device",
            "va",
        ]
    return ["-hwaccel", "vaapi", "-hwaccel_output_format", "vaapi", "-vaapi_device", _VAAPI_DEVICE]


# Audio is ALWAYS re-encoded on the HLS path: the source is typically AC3/DTS 5.1, which no browser
# plays, so it is downmixed to stereo AAC. `-ac 2` uses ffmpeg's standard downmix matrix; nothing is
# normalised, which is why a film mastered for a cinema arrives with whispered dialogue and deafening
# action (bug hunt 2026-09, item 14).
_AUDIO_ENC = ["-c:a", "aac", "-ac", "2", "-b:a", "192k"]
# Optional loudness normalisation (admin knob `plex_audio_loudnorm`, default OFF). EBU R128 single-pass
# `loudnorm` — two-pass is impossible here, since a seek-restart session must start emitting segments
# immediately and cannot analyse the file first.
#
# The downmix is FORCED FIRST (`aformat=channel_layouts=stereo`) rather than left to `-ac 2`: measuring
# 5.1 and then folding it down would normalise something other than what the viewer hears. `loudnorm`
# runs its own resampler at 192 kHz internally, so the chain resamples back to 48 kHz for the AAC
# encoder. Measured on a real 5.1 AC3 film (Westworld S1E6, 60 s from 15:00): I -27.6 → -16.6 LUFS,
# LRA 13.3 → 9.8 LU — the quiet/loud gap narrows, and everything gets ~11 dB louder, so expect to turn
# the volume down after switching this on. The filter's 3 s lookahead does NOT desync A/V: measured
# under this pipeline's own `-copyts`, the audio start PTS is identical with and without it.
_LOUDNORM_AF = "aformat=channel_layouts=stereo,loudnorm=I=-16:TP=-1.5:LRA=11,aresample=48000"


def _aenc_opts(loudnorm: bool) -> list[str]:
    """Audio encoder args for one output, with the normalisation filter when the admin enabled it."""
    return [*_AUDIO_ENC, *(["-af", _LOUDNORM_AF] if loudnorm else [])]


def _venc_opts(
    transcode: bool,
    hw: bool,
    tonemap: bool = False,
    preset: str = "veryfast",
    crf: int = 20,
    max_height: int = 0,
) -> list[str]:
    """The video-encoder args that replace `-c:v copy` for a transcode session (returns the copy args
    when not transcoding). A re-encode must MINT its own keyframes at the segment boundaries — the copy
    path leaned on the source's keyframes, but a re-encoded stream has none the HLS muxer can cut on
    unless we force them (`-force_key_frames` every `_SEG_SECONDS`), else segments aren't independent and
    seeking breaks. `preset`/`crf`/`max_height` (software only) come from admin sysconfig."""
    if not transcode:
        return ["-c:v", "copy"]
    force_kf = ["-force_key_frames", f"expr:gte(t,n_forced*{_SEG_SECONDS})"]
    if hw and tonemap:
        # HDR/10-bit HW path: Gen 9.5 VPP can't tonemap (scale_vaapi=format=nv12 fails), so the decoded
        # VAAPI surfaces are mapped to OpenCL (hwmap=derive_device=opencl), tone-mapped HDR→SDR to 8-bit
        # nv12 there (tonemap_opencl, hable curve, bt709 out), then mapped back to VAAPI (reverse=1) for
        # h264_vaapi. `-filter_hw_device ocl` tells the filter graph which device to run the OpenCL
        # filters on. Full GPU pipeline — ~1 CPU core, several×realtime on the UHD 630 (measured).
        return [
            "-filter_hw_device",
            "ocl",
            "-vf",
            "hwmap=derive_device=opencl,"
            "tonemap_opencl=tonemap=hable:transfer=bt709:matrix=bt709:primaries=bt709:format=nv12,"
            "hwmap=derive_device=vaapi:reverse=1",
            "-c:v",
            "h264_vaapi",
            *force_kf,
        ]
    if hw:
        # scale_vaapi with format=nv12 downconverts a 10-bit (p010) HEVC source to 8-bit nv12 so
        # h264_vaapi (8-bit High) can encode it; for an 8-bit source it's a cheap format pass-through.
        # The GPU encode is cheap, so it runs at source resolution (max_height is a software-only knob).
        return ["-vf", "scale_vaapi=format=nv12", "-c:v", "h264_vaapi", *force_kf]
    # Software fallback: same browser-safe recipe as downloads/edit.py (libx264). yuv420p forces 8-bit
    # High (not High10) so any browser decodes it; -threads reserves a core so the CPU-bound encode can't
    # starve request handling / playback on the shared box.
    args: list[str] = []
    if max_height and max_height > 0:
        # Downscale-ONLY to the height cap (min(ih,H)), keeping aspect with an even width (-2), so a tall
        # source doesn't make the software re-encode ruinously expensive. The comma in min() is escaped
        # (\,) so ffmpeg's filter parser reads one filter, not two. Never upscales a smaller source.
        args += ["-vf", f"scale=-2:min(ih\\,{max_height})"]
    args += [
        "-c:v",
        "libx264",
        "-preset",
        preset,
        "-crf",
        str(crf),
        "-pix_fmt",
        "yuv420p",
        "-threads",
        str(encode_thread_cap()),
        *force_kf,
    ]
    return args


def _render_node_present() -> bool:
    """Whether the VAAPI render node is available — i.e. `/dev/dri` was passed into the container.
    Present on prod after the iGPU enablement; absent on the CPU-only dev box and in the localdev
    Docker. Its own function so the selection is monkeypatchable in tests and one place for S1c's knob."""
    return Path(_VAAPI_DEVICE).exists()


def _select_encode(db: DbSession, item: PlexItem, transcode: bool, src: Path) -> tuple[bool, bool]:
    """Pick the transcode video pipeline for this session — returns `(hw, tonemap)`:
      * `(False, False)` → software libx264 (the fallback: no render node, non-decodable source, or a
        10-bit/HDR source with no OpenCL runtime to tonemap on the GPU).
      * `(True, False)`  → full-GPU VAAPI (`scale_vaapi=format=nv12` + `h264_vaapi`) — an 8-bit SDR source.
      * `(True, True)`   → VAAPI decode → OpenCL tonemap → VAAPI encode — a true-HDR (PQ/HLG) source, when
        the OpenCL runtime is present (Gen 9.5 VPP can't tonemap natively).

    A 10-bit *SDR* source is deliberately NOT a `(True, True)` case: it needs a plain 10→8-bit downconvert,
    not the HDR tone curve (which would visibly darken/flatten an SDR picture), so it drops to software.

    Honours the admin `plex_transcode_hw` policy: "off" → always software; "on" → HW wherever the render
    node exists (trusting the admin, skipping the decodability gate — the runtime HW→SW fallback still
    backstops a session that produces nothing — but still routing a true-HDR source through the tonemap
    graph so it doesn't hit the scale_vaapi wall); "auto" (default) → HW only when the render node is
    present AND the iGPU can hardware-decode the source, with tonemap chosen only for a true-HDR (PQ/HLG)
    source that has an OpenCL runtime, while a 10-bit SDR source drops to software."""
    if not transcode:
        return (False, False)
    mode = (sysconfig.get_str(db, "plex_transcode_hw") or "auto").lower()
    if mode == "off" or not _render_node_present():
        return (False, False)
    if mode == "on":
        # Trust the admin: HW wherever the render node exists, skipping the decodability gate (the runtime
        # backstop catches a miss). Only a true-HDR (PQ/HLG) source needs the tonemap graph — a 10-bit SDR
        # source takes the plain scale path here, NOT the HDR tone curve. Probe only when OpenCL exists
        # (short-circuit), so "on" stays probe-free where tonemap can't apply anyway.
        tonemap = _opencl_present() and _probe_video_kind(src) == _VK_HDR
        return (True, tonemap)
    # auto
    if (item.codec_video or "").lower() not in _HW_DECODABLE:
        return (False, False)
    kind = _probe_video_kind(src)
    if kind == _VK_HWSAFE:
        return (True, False)  # 8-bit SDR → full-GPU scale path
    if kind == _VK_HDR:
        # PQ/HLG: HW OpenCL tonemap when the runtime is present, else software libx264.
        return (True, True) if _opencl_present() else (False, False)
    # _VK_SDR10 — 10-bit SDR: a plain 10→8-bit downconvert, done exactly on the CPU. Never the HDR tone
    # curve (it would alter an SDR picture), and off the unverified scale_vaapi 10-bit path — software.
    return (False, False)


def _want_hw(db: DbSession, item: PlexItem, transcode: bool, src: Path) -> bool:
    """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]


# CPU-nice increment for the SOFTWARE transcode child. The Plex ffmpeg is a Popen in the long-lived API
# process (which must NOT itself be niced), so — unlike the download worker's _deprioritize (run once in
# its own main()) — we lower priority per-child via preexec_fn in the fork. The GPU path needs none of
# this. Reuses the download child's nice default so all co-located re-encoders share one handicap.
_SW_TRANSCODE_NICE = settings.download_child_nice


def _sw_child_preexec() -> None:
    """Runs in the forked child (POSIX) before exec: renice this software-transcode ffmpeg down so it
    can't starve the co-located API / Postgres / playback. Best-effort — a failed renice must not fail
    playback."""
    try:
        os.nice(_SW_TRANSCODE_NICE)
    except OSError:
        pass


def _probe_audio_langs(src: Path) -> list[str]:
    """One entry per audio stream (its language tag, or "" if untagged). len() = audio-track count.
    Used to build the multi-rendition var_stream_map. Empty on probe failure → single-audio path."""
    try:
        out = subprocess.run(
            [
                "ffprobe",
                "-v",
                "error",
                "-select_streams",
                "a",
                "-show_entries",
                "stream_tags=language",
                "-of",
                "csv=p=0",
                str(src),
            ],
            capture_output=True,
            text=True,
            timeout=10,
        )
        return [ln.strip() for ln in out.stdout.splitlines()]
    except (subprocess.SubprocessError, OSError):
        return []


def _ffmpeg_cmd(
    src: Path,
    start_s: float,
    out_dir: Path,
    audio_ord: int | None,
    aoff: float = 0.0,
    audio_langs: list[str] | None = None,
    transcode: bool = False,
    hw: bool = False,
    tonemap: bool = False,
    preset: str = "veryfast",
    crf: int = 20,
    max_height: int = 0,
    loudnorm: bool = False,
) -> list[str]:
    # For a transcode session the video is re-encoded (VAAPI or libx264) instead of stream-copied; the
    # hwaccel decode options go before -i, the encoder args replace `-c:v copy` below. Everything else is
    # identical to remux. `tonemap` is the HDR/10-bit HW variant (OpenCL tonemap between VAAPI decode and
    # encode) — it needs matching decode-side (named devices) and encode-side (the OpenCL filter graph) opts.
    venc = _venc_opts(transcode, hw, tonemap, preset, crf, max_height)
    args = ["ffmpeg", "-nostdin", "-loglevel", "error", *_hwaccel_in_opts(hw, tonemap)]
    # Keep ORIGINAL timestamps on the output so the first segment carries the true absolute PTS of the
    # keyframe ffmpeg actually seeked to. hls.js still zero-bases playback to that PTS; we read it back
    # (see _probe_first_pts) to learn the real content offset. Without -copyts the segment PTS is
    # rewritten toward 0 and the true offset is unknowable, which caused the ~seconds subtitle lead.
    args += _seek_opts(start_s) + ["-copyts", "-i", str(src)]
    ai = 0  # input index the audio maps come from
    if aoff:
        # A/V-sync offset: read the file a SECOND time for audio only, with `-itsoffset` shifting the
        # audio PTS by ±aoff relative to the (input-0) video. Maps audio from input 1 then.
        args += ["-itsoffset", f"{aoff:.3f}"] + _seek_opts(start_s) + ["-copyts", "-i", str(src)]
        ai = 1
    # Video always stream-copied; audio → AAC. Subtitles are NOT muxed here — they're separate WebVTT
    # tracks (GET /subtitle) the browser overlays, so choosing a subtitle doesn't restart this session.
    if audio_langs:
        # MULTI-RENDITION: map EVERY audio track as an alternate HLS rendition in one group, so hls.js
        # switches audio CLIENT-SIDE on the same timeline (no session restart, no drift) — the premium
        # audio-switch path. Produces master.m3u8 + one stream_%v.m3u8 and seg_%v_*.ts per variant,
        # where %v is the video rendition's index ("0") and each audio rendition's `name:` (a0, a1,
        # …). Those names are spelled by `segment_name` / `variant_playlist_name` below, which the
        # guard and the warm-up read back — see them for the measurement behind the ordering.
        args += ["-map", "0:v:0"]
        for i in range(len(audio_langs)):
            args += ["-map", f"{ai}:a:{i}?"]
        # One `-af` covers EVERY audio output of this file, so each rendition is normalised.
        args += ["-sn", *venc, *_aenc_opts(loudnorm)]
        var = ["v:0,agroup:aud"]
        for i, lang in enumerate(audio_langs):
            entry = f"a:{i},agroup:aud,name:a{i}"
            if lang:
                entry += f",language:{lang}"
            if i == 0:
                entry += ",default:yes"
            var.append(entry)
        # The name patterns come from the vocabulary functions, with ffmpeg's own placeholders passed
        # through them — so the producer and the parser cannot spell the layout differently (round 7).
        args += _HLS_TAIL + [
            "-master_pl_name",
            _MASTER_PL,
            "-var_stream_map",
            " ".join(var),
            "-hls_segment_filename",
            str(out_dir / segment_name("%d", "%v")),
            str(out_dir / variant_playlist_name("%v")),
        ]
    else:
        ao = audio_ord if audio_ord is not None else 0
        args += ["-map", "0:v:0", "-map", f"{ai}:a:{ao}?", "-sn"]
        args += [*venc, *_aenc_opts(loudnorm)]
        args += _HLS_TAIL + [
            "-hls_segment_filename",
            str(out_dir / segment_name("%d")),
            str(out_dir / _INDEX_PL),
        ]
    return args


def _probe_first_pts(seg: Path) -> float | None:
    """Read the first video packet's absolute PTS from a finished HLS segment (`-copyts` kept it
    absolute). This is the true content position of the segment's first frame — the value hls.js
    zero-bases the timeline to. Returns None on any failure (caller falls back to the requested offset)."""
    try:
        out = subprocess.run(
            [
                "ffprobe",
                "-v",
                "error",
                "-select_streams",
                "v:0",
                "-read_intervals",
                "%+#1",
                "-show_entries",
                "packet=pts_time",
                "-of",
                "csv=p=0",
                str(seg),
            ],
            capture_output=True,
            text=True,
            timeout=10,
        )
        for line in out.stdout.splitlines():
            tok = line.strip().strip(",")
            if tok:
                return float(tok)
    except (subprocess.SubprocessError, ValueError, OSError):
        pass
    return None


def _kill(s: HlsSession) -> None:
    try:
        s.proc.terminate()
        try:
            s.proc.wait(timeout=3)
        except subprocess.TimeoutExpired:
            s.proc.kill()
    except Exception:
        pass
    shutil.rmtree(s.dir, ignore_errors=True)


def _enforce_cap(cap: int) -> None:
    # Caller holds _lock. Only TRANSCODE sessions count toward plex_max_transcodes — remux
    # (stream-copy) and direct-serve are cheap and deliberately uncapped (see config.py:
    # plex_max_transcodes, and the user-facing config hint). Drop the least-recently-accessed
    # transcode sessions over the cap, leaving any concurrent remux streams untouched.
    tx = [(key, s) for key, s in _sessions.items() if s.transcode]
    if len(tx) <= cap:
        return
    for key, s in sorted(tx, key=lambda kv: kv[1].last_access)[: len(tx) - cap]:
        _kill(s)
        _sessions.pop(key, None)


def start_session(
    db: DbSession,
    item: PlexItem,
    start_s: float,
    audio_ord: int | None = None,
    aoff: float = 0.0,
    multi: bool = False,
    transcode: bool = False,
) -> HlsSession | None:
    """(Re)start the HLS session for an item at the given offset. `transcode` (item is non-h264,
    playable="transcode") re-encodes the video to H.264 (VAAPI or libx264) instead of stream-copying it;
    otherwise the video is copied (remux). `multi` (item has >1 audio track) maps every audio track as an
    HLS rendition so the client switches audio without a restart; otherwise a single audio track
    (`audio_ord`) is muxed. Subtitles are separate WebVTT tracks (not muxed here). Returns None if the
    local file can't be read."""
    src = paths.local_media_path(db, item.file_path)
    if src is None:
        return None
    # Read once, here: the knob has to be identical for the directory name and the ffmpeg argv, and a
    # session respawn (the HW->SW fallback) must not pick up a mid-playback config change.
    loudnorm = sysconfig.get_bool(db, "plex_audio_loudnorm")
    key = item.rating_key
    start_s = max(0.0, float(start_s))
    audio_langs = _probe_audio_langs(src) if multi else []
    is_multi = len(audio_langs) > 1  # only worth a master playlist when there really are ≥2 tracks
    # The rendition names the argv below will produce — one source for the argv, the warm-up's
    # segment names, the entry playlist and the artifact guard.
    variants = session_variants(audio_langs if is_multi else None)
    # Loudness normalisation changes the AUDIO THE SEGMENTS CONTAIN, so it belongs in the directory
    # name beside the other parameters that do (offset, audio track, A/V offset): a session left over
    # from before the admin flipped the knob must not be mistaken for one that honours it. Appended to
    # the tag with a DASH — `_SESSION_DIR_RE` treats the tag as one underscore-free field, and that
    # regex is the only thing keeping the orphan sweep off a shared scratch path (see its comment).
    tag = ("multi" if is_multi else str(audio_ord)) + ("-ln" if loudnorm else "")
    directory = _HLS_ROOT / f"{key}_{int(start_s)}_{tag}_{aoff:+.2f}"

    def _seg(i: int) -> Path:
        return directory / segment_name(i, _VIDEO_VARIANT if variants else None)

    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
        if transcode and not hw:
            preset = sysconfig.get_str(db, "plex_transcode_preset") or "veryfast"
            crf = sysconfig.get_int(db, "plex_transcode_crf")
            max_height = sysconfig.get_int(db, "plex_transcode_max_height")
        # Nice the SOFTWARE transcode child (not the HW path, and not remux) so a CPU-bound re-encode
        # can't starve the co-located API/playback. preexec_fn runs in the forked child (POSIX); on a
        # 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)
            shutil.rmtree(directory, ignore_errors=True)
            directory.mkdir(parents=True, exist_ok=True)
            proc = subprocess.Popen(
                _ffmpeg_cmd(
                    src,
                    start_s,
                    directory,
                    audio_ord,
                    aoff,
                    audio_langs if is_multi else None,
                    transcode=transcode,
                    hw=hw,
                    tonemap=tonemap,
                    preset=preset,
                    crf=crf,
                    max_height=max_height,
                    loudnorm=loudnorm,
                ),
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                preexec_fn=preexec,
            )
            s = HlsSession(
                key,
                directory,
                proc,
                start_s,
                variants,
                transcode=transcode,
                hw=hw,
                tonemap=tonemap,
            )
            _sessions[key] = s
            # Admin-configurable concurrency cap (Configuration → Plex → Max concurrent transcodes);
            # at least 1 so playback can't be capped to zero.
            _enforce_cap(max(1, sysconfig.get_int(db, "plex_max_transcodes")))
        return s

    s = _spawn(*_select_encode(db, item, transcode, src))
    # Wait for the first video segment outside the lock (this blocks on ffmpeg, and we must not hold up
    # other sessions). Pass the proc so a HW session that dies at VAAPI init fails fast (not after 25s).
    ready = wait_for(_seg(0), timeout=25.0, proc=s.proc)
    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. 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
        )
        respawned = _spawn(False, False, replace_only=s)
        if respawned is None:
            # Superseded/culled mid-fallback — yield whoever is registered now (a newer seek-restart), or
            # our own now-dead session if the key was cap-culled. Never None here: the POST /session route
            # turns a None return into a "Media file not found on disk" 404 — dishonest for a file that
            # exists. A (dead) session instead degrades to a retryable 404 ("No active playback session" /
            # "Not available yet") the client re-requests.
            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,
        # so the client's absolute clock + subtitle shift key off the true keyframe, not the requested
        # offset. On probe failure we keep the provisional requested offset (old behaviour, small drift).
        if start_s > 0:
            k = _probe_first_pts(_seg(0))
            if k is not None and k >= 0:
                s.media_start_s = k
        # Warm up the buffer (see _WARMUP_SEGMENTS) so the client doesn't outrun the cold ffmpeg. Stop
        # at the first segment that isn't ready in time — ffmpeg produces them faster than realtime once
        # warm, so a miss means something slow and we proceed rather than block the request further.
        for i in range(1, _WARMUP_SEGMENTS):
            if not wait_for(_seg(i), timeout=_WARMUP_SEG_TIMEOUT_S, proc=s.proc):
                break
    log.info(
        "plex hls session start key=%s req=%.1f real=%.3f multi=%s audio=%s transcode=%s hw=%s tonemap=%s",
        key,
        start_s,
        s.media_start_s,
        is_multi,
        audio_ord,
        transcode,
        s.hw,
        s.tonemap,
    )
    return s


def current_session(key: str) -> HlsSession | None:
    with _lock:
        s = _sessions.get(key)
        if s is not None:
            s.last_access = time.time()
        return s


# The names ffmpeg writes into a session directory, in ONE place — and "one place" means the argv
# BUILDS its patterns from these same functions (`segment_name("%d", "%v")`), not that a comment
# says so. Getting this wrong is invisible in both directions: a name we failed to anticipate used
# to fall back to "wait and see" (round 6's HIGH), and now that the rule is positive it falls back
# to a 404 instead — so the producer and the parser have to be literally the same spelling.
#
# MEASURED (ffmpeg 7.1.5 in the API container — the binary that actually writes these files — and
# re-checked on 8.1.1): `-hls_segment_filename seg_%v_%d.ts` substitutes `%v` with the VARIANT and
# `%d` with the SEQUENCE — variant FIRST. A dual-audio session writes seg_0_0.ts… (video, the
# unnamed variant) plus seg_a0_0.ts… / seg_a1_0.ts… (the `name:aN` audio renditions), and the
# playlists beside them are master.m3u8 + stream_0/a0/a1.m3u8. Single-audio uses `seg_%d.ts` +
# index.m3u8, with no variant field at all.
#
# The two argv branches are DISJOINT vocabularies, which is why the guard below is keyed to the
# SESSION and not to HLS in general: a single-audio session can never write `master.m3u8`,
# `stream_*.m3u8` or any `seg_<variant>_<n>.ts`, and a multi-audio one can never write `index.m3u8`
# or `seg_<n>.ts`. Accepting the union meant a hopeless name of the other shape was waited out for
# the full playlist/segment timeout — the threadpool parking this guard exists to end (round 7).
_MASTER_PL = "master.m3u8"
_INDEX_PL = "index.m3u8"
_SEG_NAME_RE = re.compile(r"^seg_(?:([0-9a-z]{1,4})_)?(\d{1,9})\.ts$")


def session_variants(audio_langs: "Sequence[str | None] | None") -> tuple[str, ...]:
    """The `-var_stream_map` rendition names a session will write, or () for a single-audio one.

    The video rendition is left unnamed in the map, so ffmpeg falls back to its index
    (`_VIDEO_VARIANT`); each audio rendition is named `aN` in track order."""
    if not audio_langs:
        return ()
    return (_VIDEO_VARIANT, *(f"a{i}" for i in range(len(audio_langs))))


def segment_name(seq: "int | str", variant: "str | None" = None) -> str:
    """The file ffmpeg writes for the `seq`-th segment of `variant`.

    `variant=None` is a single-audio session, whose argv carries no `%v` field at all. `seq` and
    `variant` are typed loosely because `_ffmpeg_cmd` builds its own `-hls_segment_filename` pattern
    through this function, passing ffmpeg's `"%d"` / `"%v"` placeholders — that is what keeps the
    producer and the parser from drifting, rather than a comment asking the next editor not to."""
    return f"seg_{variant}_{seq}.ts" if variant else f"seg_{seq}.ts"


def variant_playlist_name(variant: str) -> str:
    """The sub-playlist ffmpeg writes for one rendition of a multi-audio session."""
    return f"stream_{variant}.m3u8"


def playlist_names(variants: "Sequence[str]") -> frozenset[str]:
    """Every playlist name this session's argv can write — the master plus one per rendition, or
    just `index.m3u8` for a single-audio session."""
    if not variants:
        return frozenset({_INDEX_PL})
    return frozenset({_MASTER_PL, *(variant_playlist_name(v) for v in variants)})


def artifact_may_arrive(s: "HlsSession", filename: str) -> bool:
    """Whether waiting for `filename` can possibly end in a file OF THIS SESSION.

    Positive rule, not a blocklist: a session produces segments and playlists and NOTHING else, so
    any other name is refused at once. ffmpeg also produces segments strictly in order, so a segment
    whose PREDECESSOR does not exist yet is not the next one — it is beyond the generated region,
    and no amount of waiting will conjure it. Playlists are the one thing that must still be waited
    for: they are rewritten continuously and have no ordering to read.

    Keyed to the SESSION's own vocabulary, not to HLS's: the two argv branches are disjoint, so a
    name belonging to the other shape is as hopeless here as a random string. Reading those as "an
    HLS name, so wait and see" left the whole class reachable — measured through the real route
    against a live single-audio session, `master.m3u8` and `stream_q7z.m3u8` were 404s after 30.0 s
    and `seg_a0_0.ts` / `seg_zzzz_0.ts` after 20.1 s each (deep-review round 7).

    Saying no at once matters because the route that waits is a SYNC one: it holds a thread off the
    ~40-slot pool, and a pooled DB connection with it, for the whole timeout. An ordinary seek past
    the produced region was enough to park one for 20 s, and forty concurrent requests for a name
    that will never exist parked the whole API (deep-review round 5). Round 5's guard read only
    single-audio segment names, so both halves of that were still fully reachable: every multi-audio
    name fell through, and so did any `.m3u8`/`.vtt` name at all."""
    if filename in playlist_names(s.variants):
        return True
    m = _SEG_NAME_RE.match(filename)
    if m is None:
        return False
    variant, seq = m.group(1), int(m.group(2))
    # The variant field must be one THIS session writes: present and known for a multi-audio
    # session, absent for a single-audio one.
    if variant not in (s.variants or (None,)):
        return False
    if seq == 0:
        return True
    try:
        return (s.dir / segment_name(seq - 1, variant)).exists()
    except OSError:
        return False


def wait_for(path: Path, timeout: float = 20.0, proc: subprocess.Popen | None = None) -> bool:
    """Wait until a session file (playlist / segment) exists and is non-empty. Segments are
    produced ~faster than realtime, so a segment just ahead of playback appears quickly; a segment
    far beyond the generated region won't (the frontend restarts the session at a seek instead).

    With `proc` given, also stop as soon as that ffmpeg exits: a VAAPI session that fails to init
    (10-bit/HDR VAProfile refusal) dies within ~1s producing nothing, and without this we'd block the
    whole `timeout` before the HW→SW fallback could kick in — turning the fix into a stall."""
    end = time.time() + timeout
    while time.time() < end:
        try:
            if path.exists() and path.stat().st_size > 0:
                return True
        except OSError:
            pass
        if proc is not None and proc.poll() is not None:
            # the producer exited — no more segments are coming; one last check then give up.
            try:
                return path.exists() and path.stat().st_size > 0
            except OSError:
                return False
        time.sleep(0.15)
    return path.exists()


def sweep_orphans(older_than: float | None = None) -> int:
    """Delete OUR segment directories that no live session owns.

    Sessions live only in THIS process, so at startup every session directory under `_HLS_ROOT` is
    a leftover from a crashed or killed run — and nothing else ever reaps them (the download GC
    deliberately skips `.plex-hls` as a system tree), so a few restarts mid-playback leave
    gigabytes of `.ts` segments behind forever.

    Only entries whose name matches `_SESSION_DIR_RE` are touched: the root is admin-configurable,
    so anything else under it belongs to someone else and is left alone. Blocking (rmtree) — call
    it off the event loop.

    TWO guards against deleting a LIVE session, because this now runs CONCURRENTLY with request
    handling (the caller stopped awaiting it so startup isn't held up):
    * `older_than` (default: this process's start, minus `_MTIME_GRACE_S` of cross-clock slack) —
      a directory touched since we booted was made by THIS run, so it is by definition not a
      leftover. This is the guard that matters, because `start_session` REUSES the same path for
      the same (key, offset, tag): a stale directory can become a live one at any moment.
    * the live-session check and the rmtree happen together under `_lock`, so a session can't be
      registered in the gap between them. The lock is held for one directory at a time — long
      enough to be atomic, short enough that a boot-time sweep of a big backlog doesn't stall
      playback."""
    cutoff = (_PROCESS_START - _MTIME_GRACE_S) if older_than is None else older_than
    dropped = 0
    try:
        entries = list(_HLS_ROOT.iterdir())
    except OSError:  # the root doesn't exist yet — nothing to sweep
        return 0
    for p in entries:
        if not _SESSION_DIR_RE.match(p.name):
            continue
        if not p.is_dir():
            continue
        try:
            if p.stat().st_mtime >= cutoff:
                continue  # this run made (or reused) it — never ours to delete
        except OSError:  # vanished under us (a concurrent restart of that session)
            continue
        with _lock:
            if any(s.dir == p for s in _sessions.values()):
                continue
            shutil.rmtree(p, ignore_errors=True)
        dropped += 1
    return dropped


# How long a session whose ffmpeg has FINISHED may go unfetched before a new session request reaps
# it (see reap_idle).
_FINISHED_IDLE_S = 30


def reap_idle(finished_idle_s: float = _FINISHED_IDLE_S) -> int:
    """Kill sessions nobody has said they still want for `_SESSION_IDLE_S` — or, once their ffmpeg
    has finished, for `finished_idle_s`.

    A finished session is also what a PAUSED viewer has: a remux runs far faster than realtime, so
    ffmpeg is done a minute or two in and writes ENDLIST; hls.js then stops asking for the playlist,
    and with its forward buffer full a paused player fetches nothing at all. With the 30 s rule that
    killed a one-minute pause, and playback died ~30 s after resuming (deep-review round 4).

    What makes the short rule safe again is that fetching is no longer the only signal: a player
    parked MID-ITEM checkpoints its position every 10 s, and that route calls `touch_session`, so a
    viewer who is merely paused says so far more often than this rule asks (round 5). The heartbeat
    stops once the item is finished or before a position is known (`usePlexProgress.flush`) — both
    states where reaping is correct, and where the player's 404 → reclaim recovers a viewer who
    scrubs back in anyway. Do not restate this as "every 10 s whether or not it is playing": round 6
    found that claim in two comments and not in the code. Round 4's blanket
    `reap_abandoned` bought the same safety by keeping every finished session for ten minutes — each
    one a whole segmented copy of the film on a volume shared with the download center."""
    now = time.time()
    dropped = 0
    with _lock:
        for key, s in list(_sessions.items()):
            done = s.proc.poll() is not None
            idle = now - s.last_access
            if idle > _SESSION_IDLE_S or (done and idle > finished_idle_s):
                _kill(s)
                _sessions.pop(key, None)
                dropped += 1
    return dropped


def touch_session(key: str) -> None:
    """Say a 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
    viewer's session alive was to keep EVERY abandoned one just as long."""
    with _lock:
        s = _sessions.get(key)
        if s is not None:
            s.last_access = time.time()