fix(clip): a clip is held to its owner's storage cap, and its limits and failures say the server's truth

- The cut file is measured against what its owner has left, for a YouTube range (whose fetch never
  reads yt-dlp's size ceiling) and for a Plex clip (which fetches nothing), before the fetch and
  again before it is placed; a refusal leaves the asset pending for a Retry.
- A silent Plex file is cut silent; a file the probe cannot read says so.
- A second job on a failed or renamed clip gets the same YouTube diagnosis and the fetched title.
- The "too long" refusal names the server's cap; a contract test holds the client's copies of the
  clip limits to clip.py; the shipped player clients are pinned against the installed yt-dlp.
- One HDR transfer set and one rating_key lookup; job status words from JobStatus; the display-name
  guard sees subscript writes and the clip route; stale comments and types follow the clip job.

backend/app/downloads/clip.py

@@ -24,6 +24,7 @@ import re
 from pathlib import Path
 
 from app.downloads.edit import AAC_ARGS, EVEN_FRAME, INPUT_GUARD, X264_ARGS
+from app.plex.sync import HDR_TRANSFERS
 
 SOURCES = ("youtube", "plex")
 # Owner's cap: a clip is a moment, not a way to export a whole film to a public link.
@@ -39,8 +40,6 @@ _SIG_VERSION = 1
 _YOUTUBE_ID = re.compile(r"[A-Za-z0-9_-]{11}")
 _RATING_KEY = re.compile(r"[0-9]{1,32}")
 _MAX_AUDIO_TRACK = 63
-# The transfer characteristics of HDR video (PQ, HLG): an 8-bit SDR encode must tonemap them.
-_HDR_TRANSFERS = frozenset({"smpte2084", "arib-std-b67"})
 # HDR → SDR as measured on the owner's 4K HDR files (2026-10-10): linearise, tonemap (hable),
 # convert the primaries and transfer to BT.709 and back to 8-bit 4:2:0.
 _TONEMAP = (
@@ -59,10 +58,12 @@ _EDGE_SLACK = 1e-3
 
 
 class ClipError(Exception):
-    """A clip request that cannot be cut; `reason` is an i18n-mappable key."""
+    """A clip request that cannot be cut; `reason` is an i18n-mappable key, `params` the numbers its
+    message names (the cap itself, so the client never says a stale one)."""
 
-    def __init__(self, reason: str):
+    def __init__(self, reason: str, **params: float):
         self.reason = reason
+        self.params = params
         super().__init__(reason)
 
 
@@ -87,7 +88,7 @@ def normalize(
     if end - start < MIN_CLIP_S:
         raise ClipError("too_short")
     if end - start > MAX_CLIP_S:
-        raise ClipError("too_long")
+        raise ClipError("too_long", limit_s=MAX_CLIP_S)
     spec: dict = {"source": source, "ref": ref, "start_s": start, "end_s": end}
     if source == "plex" and audio_track is not None:
         if not 0 <= audio_track <= _MAX_AUDIO_TRACK:
@@ -151,7 +152,8 @@ def is_hdr(probe: dict) -> bool:
     """Whether the first video stream of an ffprobe `-show_streams` JSON carries an HDR transfer."""
     for s in probe.get("streams") or []:
         if s.get("codec_type") == "video":
-            return s.get("color_transfer") in _HDR_TRANSFERS
+            # PQ / HLG: an 8-bit SDR encode must tonemap them (the one set the Plex stream uses).
+            return s.get("color_transfer") in HDR_TRANSFERS
     return False
 
 

backend/app/downloads/formats.py

@@ -281,8 +281,9 @@ def build_ydl_opts(
     after the host-scope + private-asset gate in `downloads.credentials`. It is set AFTER the `extra`
     merge so a source adapter can never clobber it.
 
-    `max_bytes` (SEC-13) is the caller's remaining storage budget; yt-dlp refuses a format larger
-    than it rather than filling the disk and only then being caught by the quota accounting."""
+    `max_bytes` (SEC-13) is the caller's remaining storage budget; yt-dlp's HTTP downloader refuses
+    a file announced larger than it rather than filling the disk and only then being caught by the
+    quota accounting. Its ffmpeg downloader (a `download_ranges` fetch) does not read it."""
     s = normalize(spec)
     h = s["max_height"]
     opts: dict = {

backend/app/downloads/pipeline.py

@@ -7,7 +7,6 @@ from datetime import date, datetime, timedelta, timezone
 from pathlib import Path
 
 import yt_dlp
-from sqlalchemy import select
 from yt_dlp.utils import download_range_func
 
 from app import logs, sysconfig
@@ -30,6 +29,7 @@ from app.downloads.jobdb import (
 from app.downloads.stated import StatedFailure
 from app.downloads.status import JobStatus
 from app.models import DISPLAY_TEXT_MAX, DownloadJob, MediaAsset, PlexItem
+from app.plex import lookups as plex_lookups
 from app.plex import paths as plex_paths
 from app.notifications import create_notification
 from app.titles import normalize_title
@@ -363,7 +363,9 @@ def _extract_with_fallback(
     `credentials.resolve_cookiefile` (a clip: `resolve_clip_cookiefile`); passed to every attempt so
     the fallback stays authenticated.
 
-    `max_bytes` (SEC-13) is the owner's remaining storage budget, applied to every attempt.
+    `max_bytes` (SEC-13) is the owner's remaining storage budget, applied to every attempt. A
+    `section` fetch goes through yt-dlp's ffmpeg downloader, which does not read it: a clip holds
+    its own output to the cap (`_process_clip`).
 
     `section` = (start_s, end_s) fetches only that range (an ad-hoc clip, see downloads/clip): the
     range is stream-copied, and its first frame is the marked one (measured), so the clip's own
@@ -523,8 +525,8 @@ def _download(
             # SEC-13: what this user has left. `check_enqueue` only compared the footprint as it
             # stood when the job was QUEUED — a download's size is unknown until it finishes, so a
             # single job could previously blow through the cap by any amount. Handing the remainder
-            # to yt-dlp as `max_filesize` makes the format selector refuse an oversized file up
-            # front instead.
+            # to yt-dlp as `max_filesize` makes it refuse an oversized file (its HTTP downloader
+            # checks the announced size) instead.
             remaining_bytes = quota.remaining_bytes(cdb, user_id)
         # …and zero is a REFUSAL, not "no limit". Earlier jobs of this user may have filled the cap
         # while this one sat in the queue, and 0 is not an expressible yt-dlp `max_filesize`, so
@@ -927,8 +929,18 @@ def _process_edit(job_id: int, asset_id: int) -> None:
         shutil.rmtree(staging, ignore_errors=True)
 
 
+def _clip_room(user_id: int) -> int | None:
+    """What the clip's owner may still store (`quota.remaining_bytes`), refusing at zero — see
+    `_download`: zero is a refusal, not "no limit"."""
+    with SessionLocal() as db:
+        remaining = quota.remaining_bytes(db, user_id)
+    if remaining == 0:
+        raise _OwnerRefused(dlerrors.QUOTA_FULL_MESSAGE)
+    return remaining
+
+
 def _clip_source(
-    job_id: int, asset_id: int, spec: dict, staging: Path
+    job_id: int, asset_id: int, spec: dict, staging: Path, remaining: int | None
 ) -> tuple[Path, float, str | None]:
     """The file a clip is cut from, where in it the clip starts, and the source's title when only
     the fetch could tell it. A YouTube clip fetches ONLY its range into `staging` (so it starts at
@@ -939,15 +951,12 @@ def _clip_source(
         with SessionLocal() as db:
             job = db.get(DownloadJob, job_id)
             asset = db.get(MediaAsset, asset_id)
-            remaining = quota.remaining_bytes(db, job.user_id) if job else None
             # Signed in when the user has a YouTube cookie, as their downloads are: YouTube gates
             # single videos signed out ("not a bot"), and the clip is theirs alone.
             if job and asset:
                 cookiefile = credentials.resolve_clip_cookiefile(
                     db, job.user_id, url, asset, spec, staging
                 )
-        if remaining == 0:  # see `_download`: zero is a refusal, not "no limit"
-            raise _OwnerRefused(dlerrors.QUOTA_FULL_MESSAGE)
         _set_job(job_id, phase="downloading", progress=0)
         try:
             info = _extract_with_fallback(
@@ -976,9 +985,7 @@ def _clip_source(
         title = normalize_title(adapters.resolve_adapter("youtube", spec["ref"]).title(info))
         return _produced_file(info, staging), 0.0, title
     with SessionLocal() as db:
-        item = db.execute(
-            select(PlexItem).where(PlexItem.rating_key == spec["ref"])
-        ).scalar_one_or_none()
+        item = plex_lookups.by_rating_key(db, PlexItem, spec["ref"])
         src = plex_paths.local_media_path(db, item.file_path) if item else None
     if src is None:
         raise StatedFailure("The library file is not reachable from here.")
@@ -1013,10 +1020,18 @@ def _process_clip(job_id: int, asset_id: int) -> None:
         if staging.exists():
             shutil.rmtree(staging, ignore_errors=True)
         staging.mkdir(parents=True, exist_ok=True)
-        src, start, fetched_title = _clip_source(job_id, asset_id, spec, staging)
-        data = probe.probe_json(src) or {}
+        # Refused before the fetch and the encode when the owner's storage is already full.
+        remaining = _clip_room(user_id)
+        src, start, fetched_title = _clip_source(job_id, asset_id, spec, staging, remaining)
+        data = probe.probe_json(src)
+        if not data:
+            raise StatedFailure("The source file could not be read.")
         audio_track = spec.get("audio_track")
-        if audio_track is not None and audio_track >= clipmod.audio_track_count(data):
+        tracks = clipmod.audio_track_count(data)
+        if tracks == 0:
+            # A silent source: the player still names its track 0, and the clip is silent too.
+            audio_track = None
+        elif audio_track is not None and audio_track >= tracks:
             raise StatedFailure("The chosen audio track is not in the file.")
         crop = _clip_crop(spec, data)
         total_s = clipmod.duration(spec)
@@ -1041,6 +1056,12 @@ def _process_clip(job_id: int, asset_id: int) -> None:
             raise StatedFailure("ffmpeg produced no output file")
         # The phone-safe shape by construction; still held to the rule every ready file meets.
         out = _ensure_phone_safe(job_id, out, staging, "editing", round(total_s))
+        # Held to the cap by its real size: a range fetch passes yt-dlp's `max_filesize` to its
+        # ffmpeg downloader, which never reads it, and a Plex clip fetches nothing at all. Read
+        # again, since other jobs of this owner may have landed during the encode.
+        remaining = _clip_room(user_id)
+        if remaining is not None and out.stat().st_size > remaining:
+            raise _OwnerRefused(dlerrors.QUOTA_FULL_MESSAGE)
         ext = out.suffix.lstrip(".").lower() or "mp4"
         rel = storage.edit_rel_path(user_id, asset_id, ext)
         storage.place_file(out, settings.download_root, rel)

backend/app/downloads/quota.py

@@ -15,7 +15,8 @@ the footprint as it stood when the job was created and earlier jobs may have fil
 That is a per-job ceiling, not an aggregate ledger: concurrent jobs of one user each read the same
 remainder (in-flight bytes are not `ready`, so `footprint` cannot see them), so a wave of
 max_concurrent downloads can still land over the cap — the deliberate trade, since a download's size
-is unknown until it finishes. Edit jobs do not consult it at all (tracked as SEC-19).
+is unknown until it finishes. CLIP jobs consult it before their fetch or encode and again with
+their output's real size before it is placed. Edit jobs do not consult it at all (tracked as SEC-19).
 """
 
 from dataclasses import dataclass

backend/app/downloads/service.py

@@ -30,6 +30,7 @@ from app.models import (
     PlexShow,
     Video,
 )
+from app.plex import lookups as plex_lookups
 from app.plex import paths as plex_paths
 
 
@@ -193,7 +194,7 @@ def enqueue(
         profile_snapshot=spec,
         display_name=display_name
         or (asset.title[:DISPLAY_TEXT_MAX] if asset.status == "ready" and asset.title else None),
-        status="queued",
+        status=JobStatus.QUEUED,
         queue_pos=next_queue_pos(db),
     )
     db.add(job)
@@ -201,7 +202,7 @@ def enqueue(
 
     # Cache hit: the file already exists — no worker round-trip needed.
     if asset.status == "ready":
-        job.status = "done"
+        job.status = JobStatus.DONE
         job.progress = 100
         asset.last_access_at = datetime.now(timezone.utc)
 
@@ -272,14 +273,14 @@ def enqueue_edit(
         edit_spec=spec,
         profile_snapshot={},
         display_name=display_name or asset.title[:DISPLAY_TEXT_MAX],
-        status="queued",
+        status=JobStatus.QUEUED,
         queue_pos=next_queue_pos(db),
     )
     db.add(job)
     incr_ref_count(db, asset.id)
 
     if asset.status == "ready":  # identical clip already produced for this user → instant
-        job.status = "done"
+        job.status = JobStatus.DONE
         job.progress = 100
         asset.last_access_at = datetime.now(timezone.utc)
 
@@ -303,9 +304,7 @@ def _clip_source_facts(db: Session, spec: dict) -> dict:
             "thumbnail_url": video.thumbnail_url,
             "length_s": video.duration_seconds,
         }
-    item = db.execute(
-        select(PlexItem).where(PlexItem.rating_key == spec["ref"])
-    ).scalar_one_or_none()
+    item = plex_lookups.by_rating_key(db, PlexItem, spec["ref"])
     if item is None:
         raise clipmod.ClipError("not_found")
     if plex_paths.local_media_path(db, item.file_path) is None:
@@ -361,14 +360,14 @@ def enqueue_clip(
         edit_spec=spec,
         profile_snapshot={},
         display_name=display_name or asset.title[:DISPLAY_TEXT_MAX],
-        status="queued",
+        status=JobStatus.QUEUED,
         queue_pos=next_queue_pos(db),
     )
     db.add(job)
     incr_ref_count(db, asset.id)
 
     if asset.status == "ready":  # the same range was already cut for this user → instant
-        job.status = "done"
+        job.status = JobStatus.DONE
         job.progress = 100
         asset.last_access_at = datetime.now(timezone.utc)
 

backend/app/models.py

@@ -933,8 +933,9 @@ class DownloadJob(Base, TimestampMixin, UpdatedAtMixin):
     )
     source_kind: Mapped[str] = mapped_column(String(16))  # "youtube" | "url"
     source_ref: Mapped[str] = mapped_column(String(512))
-    # "download" (default), "edit" (a per-user trim/crop derivative of another job) or "upload" (a
-    # user's own file, converted by the worker when it is not phone-safe — see `downloads.upload`).
+    # "download" (default), "edit" (a per-user trim/crop derivative of another job), "upload" (a
+    # user's own file, converted by the worker when it is not phone-safe — see `downloads.upload`) or
+    # "clip" (a range cut straight from a YouTube video or a Plex item — see `downloads.clip`).
     job_kind: Mapped[str] = mapped_column(String(16), default="download", server_default="download")
     # For edit jobs: the parent download this clip is cut from (SET NULL if the parent is deleted).
     source_job_id: Mapped[int | None] = mapped_column(
@@ -943,7 +944,9 @@ class DownloadJob(Base, TimestampMixin, UpdatedAtMixin):
     source_asset_id: Mapped[int | None] = mapped_column(
         ForeignKey("media_assets.id", ondelete="SET NULL")
     )
-    # For edit jobs: the trim/crop recipe {trim?:{start_s,end_s}, crop?:{x,y,w,h}}.
+    # For edit jobs: the trim/crop recipe {trim?:{start_s,end_s}, crop?:{x,y,w,h}} in pixels. For clip
+    # jobs: the clip spec {source, ref, start_s, end_s, audio_track?, crop?:{x,y,w,h,aspect}}, its crop
+    # in fractions of the box it was drawn in (`downloads.clip.normalize`).
     edit_spec: Mapped[dict | None] = mapped_column(JSON)
     profile_id: Mapped[int | None] = mapped_column(
         ForeignKey("download_profiles.id", ondelete="SET NULL")

backend/app/plex/stream.py

@@ -35,7 +35,7 @@ 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
+from app.plex.sync import HDR_TRANSFERS, HW_DECODABLE_CODECS
 
 log = logs.get_logger(__name__)
 _HLS_ROOT = (
@@ -203,7 +203,8 @@ _HW_DECODABLE = HW_DECODABLE_CODECS
 # 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 single source is plex.sync (a clip's tonemap reads it too).
+_HDR_TRANSFERS = HDR_TRANSFERS
 
 
 # The three source classes that drive the transcode-pipeline choice (see _probe_video_kind).

backend/app/plex/sync.py

@@ -64,9 +64,11 @@ def classify_playable(vcodec: str | None, acodec: str | None, container: str | N
 # outside it (AV1, MPEG-4 ASP/XviD…) is software-only even on the iGPU (it can't decode it to feed its
 # own encoder).
 HW_DECODABLE_CODECS = frozenset({"h264", "hevc", "h265", "vp9", "vc1", "mpeg2video"})
-# Color-transfer values marking an HDR source (PQ / HLG). A 10-bit/HDR source breaks the VAAPI
-# scale=nv12 path on Gen 9.5 (see plex.stream) → software transcode, so it counts as NOT HW-capable.
-_HDR_TRANSFERS = frozenset({"smpte2084", "arib-std-b67"})
+# Color-transfer values marking an HDR source (PQ / HLG). THE single source of this set: plex.stream
+# (the tonemap choice) and downloads.clip (a clip's tonemap) import it here, so a transfer added for
+# one is tonemapped by all. A 10-bit/HDR source breaks the VAAPI scale=nv12 path on Gen 9.5 (see
+# plex.stream) → software transcode, so it counts as NOT HW-capable.
+HDR_TRANSFERS = frozenset({"smpte2084", "arib-std-b67"})
 
 
 def transcode_hw_capable(
@@ -107,7 +109,7 @@ def _video_stream_props(meta: dict) -> tuple[int | None, bool]:
                 # whole library would drown the sync's own lines.
                 except (TypeError, ValueError):
                     depth = None
-                hdr = str(s.get("colorTrc") or "").lower() in _HDR_TRANSFERS
+                hdr = str(s.get("colorTrc") or "").lower() in HDR_TRANSFERS
                 return depth, hdr
     return None, False
 

backend/app/routes/downloads.py

@@ -610,10 +610,12 @@ def enqueue_clip(
         spec = clipmod.normalize(
             body.source, body.ref, body.start_s, body.end_s, body.audio_track, crop
         )
-        job = service.enqueue_clip(db, user.id, spec, display_text(body.display_name))
+        display_name = display_text(body.display_name)
+        job = service.enqueue_clip(db, user.id, spec, display_name=display_name)
     except clipmod.ClipError as e:
         status = 404 if e.reason == "not_found" else 422
-        raise HTTPException(status_code=status, detail={"code": "clip", "reason": e.reason}) from e
+        detail = {"code": "clip", "reason": e.reason, **e.params}
+        raise HTTPException(status_code=status, detail=detail) from e
     except quota.QuotaExceeded as e:
         raise HTTPException(status_code=422, detail=_quota_detail(e)) from e
     return _serialize_job(db, job)

backend/app/worker.py

@@ -29,6 +29,7 @@ from datetime import datetime, timezone
 from functools import partial
 
 from sqlalchemy import text
+from sqlalchemy.orm import Session
 
 from app import logs
 from app import logsink
@@ -44,7 +45,7 @@ from app.downloads.jobdb import (
     _reset_asset_pending,
     _staging_dir,
 )
-from app.downloads.status import STOPPED_BY_ROUTE, JobStatus
+from app.downloads.status import STOPPED_BY_ROUTE, JobKind, JobStatus
 from app.models import DISPLAY_TEXT_MAX, DownloadJob, MediaAsset
 
 log = logs.get_logger(__name__)
@@ -437,7 +438,7 @@ def _run_supervised(job_id: int, asset_id: int) -> None:
             if status is None or status in STOPPED_BY_ROUTE:
                 reason = "cancelled"  # a route stopped it (or the row vanished)
                 break
-            if status == "running":
+            if status == JobStatus.RUNNING:
                 if stale is not None and stale > stall_s:
                     reason = "stalled"
                     break
@@ -448,7 +449,7 @@ def _run_supervised(job_id: int, asset_id: int) -> None:
             # teardown. A still-running job over the cap is failed ("timeout"); a terminal one (already
             # done/error) just gets its lingering child killed, its DB status left intact ("abandon").
             if time.monotonic() - started > wall_s:
-                reason = "timeout" if status == "running" else "abandon"
+                reason = "timeout" if status == JobStatus.RUNNING else "abandon"
                 break
             # else: a terminal status under the cap (done/error, or a re-queued contended asset) means
             # the child is exiting on its own — keep polling; NEVER kill a finishing child (that could
@@ -479,7 +480,7 @@ def _run_supervised(job_id: int, asset_id: int) -> None:
             _reset_asset_pending(asset_id)
             if proc.returncode != _CHILD_OK:
                 status, _ = _job_liveness(job_id)
-                if status == "running":
+                if status == JobStatus.RUNNING:
                     _cleanup_staging(asset_id)
                     _fail_running(
                         job_id,
@@ -560,15 +561,12 @@ def _process_job(job_id: int) -> None:
         asset_id = asset.id if asset else None
         asset_status = asset.status if asset else None
         asset_error = asset.error if asset else None
-        if asset_status == "error":
+        if asset is not None and asset_status == "error":
             # A job that meets an already-failed asset gets that failure's explanation — not a bare
             # "Download failed". Only the text is on the asset, so it is diagnosed again, for this
             # link, with whether a cookie goes with it asked of the same gate the download child
             # used (`credentials.usable_cookie`) — as it stands now, so the advice fits this job's Retry.
-            url = service.source_url(job.source_kind, job.source_ref)
-            cookie = credentials.usable_cookie(
-                db, job.user_id, url, asset, formats.normalize(job.profile_snapshot)
-            )
+            url, cookie = _failed_link_and_cookie(db, job, asset)
             diagnosis = dlerrors.diagnose(asset_error, url=url, cookie_attached=cookie is not None)
             del cookie  # only whether there is one matters here
 
@@ -605,13 +603,34 @@ def _process_job(job_id: int) -> None:
     _run_supervised(job_id, asset_id)
 
 
+def _failed_link_and_cookie(
+    db: Session, job: DownloadJob, asset: MediaAsset
+) -> tuple[str | None, str | None]:
+    """The link a job's failed asset was fetched from, and the cookie that fetch would go with now.
+    A clip's link is its source's (a `clip` ref names the user's range, no page), asked of the
+    clip's own cookie gate, as its fetch was (`_clip_source`); a Plex clip fetches nothing."""
+    if job.job_kind == JobKind.CLIP:
+        spec = dict(job.edit_spec or {})
+        if spec.get("source") != "youtube":
+            return None, None
+        url = service.source_url("youtube", spec["ref"])
+        return url, credentials.usable_clip_cookie(db, job.user_id, url, asset, spec)
+    url = service.source_url(job.source_kind, job.source_ref)
+    spec = formats.normalize(job.profile_snapshot)
+    return url, credentials.usable_cookie(db, job.user_id, url, asset, spec)
+
+
 def _finish_from_cache(job_id: int, asset_id: int) -> None:
     with SessionLocal() as db:
         job = db.get(DownloadJob, job_id)
         asset = db.get(MediaAsset, asset_id)
         if job and asset:
             fields = {"status": JobStatus.DONE, "progress": 100, "phase": None}
-            if not job.display_name and asset.title:
+            # A clip queued under the placeholder takes the title its sibling's fetch found.
+            unnamed = not job.display_name or (
+                job.job_kind == JobKind.CLIP and job.display_name == service.clip_title(None, None)
+            )
+            if unnamed and asset.title:
                 # display_name is VARCHAR(255); a title can be longer (e.g. a recipe in the title).
                 fields["display_name"] = asset.title[:DISPLAY_TEXT_MAX]
             asset.last_access_at = datetime.now(timezone.utc)