fix(e3): address /code-review high findings (5)

Round-1 review fixes on the E3 branch:

- prune partial-fetch (correctness): `_paginate` now reports whether it reached
  the server's reported `total`; `_prune_stale` prunes only on a COMPLETE
  non-empty listing. A truncated fetch (a mid-sequence empty/short page before
  total — a transient Plex hiccup) no longer deletes the rows it merely failed
  to list. New test: a listing that claims 5 but returns 1 prunes nothing.
- `_prune_stale` (robustness): `stats.get("pruned", 0) + 1` — no KeyError when a
  caller's stats dict omits the key.
- service transitions (conventions): pause/resume/redownload use the JobStatus
  StrEnum instead of loose "queued"/"paused"/"error" literals, matching cancel.
- ffmpeg progress writer (correctness): a stuck-percent heartbeat now runs on its
  OWN 1/interval budget, so it can't consume the progress-write slot and delay a
  real percent change by up to an interval.
- backend/app/README.md (conventions): reword the transaction-ownership rule to
  quote and extend CLAUDE.md's canonical ("handler/job-wrapper commits; helpers
  flush; only quota owns its session") instead of stating a contradicting
  "helpers MAY commit" carve-out.

ruff clean, full DB-lane 364 pass.

backend/app/README.md

@@ -31,15 +31,17 @@ pre-abstract for a single caller; don't leave shared logic duplicated once there
 
 ## Transaction ownership
 
-**A helper that mutates rows as part of a larger operation only `db.add()`s / stages its change; the
-CALLER commits, so the whole operation is one transaction.** `audit.record()` is the reference: it adds
-the audit row and the caller commits it together with the mutation it records, so an audit entry can
-never exist without its change (or vice-versa).
+This follows CLAUDE.md's canonical rule verbatim — **"the handler/job-wrapper commits; helpers flush
+(only `quota` owns its session)"** — so read it that way, not as an exception to it. The commit belongs
+to the **operation owner**; anything called *inside* another transaction only stages its change
+(`db.add` / `flush`) and lets the owner commit, so the whole operation is one transaction.
+`audit.record()` is the reference: it adds the audit row and the caller commits it together with the
+mutation it records, so an audit entry can never exist without its change (or vice-versa).
 
-- A self-contained transition that owns its whole unit of work MAY commit (e.g. `service.cancel/…`
-  each commit their one transition; the worker's short-lived `jobdb` helpers each commit their own
-  write). But a helper called *in the middle* of a caller's transaction must not commit — that splits
-  the caller's atomic operation in two.
+- The **operation owner** is usually the route handler. A `service.*` operation the handler delegates a
+  whole transition to (e.g. `service.cancel`) *is* that owner and commits its one transition — it isn't
+  a mid-operation helper. A helper that owns its OWN short-lived session (the worker's `jobdb` writes,
+  the same carve-out CLAUDE.md grants `quota`) commits that session. Everything else flushes.
 - `SessionLocal` is `expire_on_commit=False`: after a raw `UPDATE` + commit, an ORM object you already
   loaded is **stale**. Read the fresh value from the row (RETURNING, or a re-`SELECT`), not the ORM
   attribute — see `service.decr_ref_count`.

backend/app/downloads/pipeline.py

@@ -193,22 +193,27 @@ def _cancel_poll(job_id: int, interval_s: float = _CANCEL_POLL_INTERVAL_S, *, no
 
 
 def _ffmpeg_progress_writer(job_id: int, phase: str, *, now=time.monotonic):
-    """`on_progress` for run_ffmpeg: persist at most one row write per interval, and only when the
-    whole-percent value actually changed (the bar shows integers)."""
-    allow = _rate_limiter(_DB_WRITE_INTERVAL_S, now=now)
+    """`on_progress` for run_ffmpeg: persist a progress row write at most once per interval and only
+    when the whole-percent value changed (the bar shows integers), plus a separate once-per-interval
+    heartbeat while the percent is stuck (so a slow re-encode isn't mistaken for a stall)."""
+    # Two independent 1/interval budgets so a heartbeat never consumes the progress-write slot — a
+    # percent change right after a heartbeat must still write immediately, not wait out the interval.
+    allow_write = _rate_limiter(_DB_WRITE_INTERVAL_S, now=now)
+    allow_beat = _rate_limiter(_DB_WRITE_INTERVAL_S, now=now)
     state = {"pct": -1}
 
     def write(pct: float) -> None:
-        if not allow():
-            return
         whole = int(pct)
         if whole == state["pct"]:
             # Same integer percent, but ffmpeg IS still working — a slow re-encode (AV1/VP9 -> H.264
             # or a crop on the CPU-only host) can sit on one percent for minutes. Heartbeat the row's
-            # updated_at so the stall watchdog doesn't kill a live job as "stalled" (B-sub-4), without
-            # rewriting the unchanged progress bar.
+            # updated_at (on its OWN budget) so the stall watchdog doesn't kill a live job as "stalled"
+            # (B-sub-4), without rewriting the unchanged bar or delaying the next real write.
+            if allow_beat():
                 _touch_job(job_id)
             return
+        if not allow_write():
+            return
         state["pct"] = whole
         _set_job(job_id, progress=whole, phase=phase)
 

backend/app/downloads/service.py

@@ -318,8 +318,8 @@ def release_asset(db: Session, job: DownloadJob) -> None:
 def pause(db: Session, job: DownloadJob) -> None:
     """Pause a queued/running job (a running worker aborts cooperatively via the cancel poll); a no-op
     in any other state."""
-    if job.status in ("queued", "running"):
-        job.status = "paused"
+    if job.status in (JobStatus.QUEUED, JobStatus.RUNNING):
+        job.status = JobStatus.PAUSED
         db.commit()
 
 
@@ -327,16 +327,16 @@ def resume(db: Session, job: DownloadJob) -> None:
     """Requeue a paused/errored job for a fresh attempt, clearing its error. If the SHARED asset itself
     failed, reset it to pending so the worker actually re-downloads — otherwise the requeued job would
     instantly re-inherit the asset's stale error (the worker short-circuits an errored asset)."""
-    if job.status in ("paused", "error"):
+    if job.status in (JobStatus.PAUSED, JobStatus.ERROR):
         job.progress = 0
         job.error = None
         job.error_code = None
         if job.asset_id:
             asset = db.get(MediaAsset, job.asset_id)
-            if asset and asset.status == "error":
+            if asset and asset.status == "error":  # MediaAsset.status (no enum), not a JobStatus
                 asset.status = "pending"
                 asset.error = None
-        job.status = "queued"
+        job.status = JobStatus.QUEUED
         db.commit()
 
 
@@ -370,7 +370,7 @@ def redownload(db: Session, job: DownloadJob) -> None:
     release_asset(db, job)
     incr_ref_count(db, asset.id)
     job.asset_id = asset.id
-    job.status = "queued"
+    job.status = JobStatus.QUEUED
     job.progress = 0
     job.phase = None
     job.error = None

backend/app/plex/sync.py

@@ -284,8 +284,14 @@ def _upsert_library(db: Session, key: str, title: str, kind: str) -> PlexLibrary
     return lib
 
 
-def _paginate(plex: PlexClient, key: str, item_type: int):
+def _paginate(plex: PlexClient, key: str, item_type: int, progress: dict | None = None):
+    """Yield a section's items page by page. If `progress` is given, set `progress["complete"]` True
+    ONLY when the walk reached the server's reported `total` — a truncated fetch (an empty/short page
+    mid-sequence, e.g. a transient Plex hiccup, before `total` is reached) leaves it False so the
+    caller can skip pruning. Pruning on a partial listing would delete rows the listing merely failed
+    to mention (the metadata is re-syncable, but that still cascades watch-state)."""
     start = 0
+    total = None
     while True:
         items, total = plex.section_items(key, item_type, start, _PAGE)
         if not items:
@@ -297,6 +303,10 @@ def _paginate(plex: PlexClient, key: str, item_type: int):
             break
         if len(items) < _PAGE:
             break
+    if progress is not None:
+        # A truthy `total` we reached = a trustworthy complete listing. `total` unknown/zero-with-rows
+        # (a Plex that doesn't report totalSize) is treated as NOT complete, so prune stays conservative.
+        progress["complete"] = bool(total) and start >= total
 
 
 # The section listing exposes at most the top ~20 cast (Plex's own cap); a stored cast longer than this
@@ -356,18 +366,21 @@ def _apply_item(row: PlexItem, lib_id: int, kind: str, meta: dict) -> None:
     row.playable = classify_playable(mf.get("codec_video"), mf.get("codec_audio"), mf.get("container"))
 
 
-def _prune_stale(db: Session, existing: dict, seen: set[str], stats: dict) -> None:
+def _prune_stale(
+    db: Session, existing: dict, seen: set[str], stats: dict, *, complete: bool
+) -> None:
     """Delete stored rows whose rating_key is no longer in the freshly-listed `seen` set — the titles
-    that disappeared from Plex (their watch-state / playlist links cascade via the FKs). GUARDED: an
-    empty `seen` (a section that listed zero items — a real empty library, or a transient Plex glitch
-    we can't tell apart) prunes NOTHING, so a hiccup can never wipe a whole library. Callers pass the
-    rating_key -> row map they already built for the upsert."""
-    if not seen:
+    that disappeared from Plex (their watch-state / playlist links cascade via the FKs). GUARDED on a
+    COMPLETE, non-empty listing: a truncated fetch (`complete` False — a mid-sequence empty/short page
+    before the server's total) or a zero-item listing prunes NOTHING, so a transient Plex hiccup can
+    never wipe a library over rows it merely failed to list. Callers pass the rating_key -> row map
+    they already built for the upsert."""
+    if not complete or not seen:
         return
     for rk, row in existing.items():
         if rk not in seen:
             db.delete(row)
-            stats["pruned"] += 1
+            stats["pruned"] = stats.get("pruned", 0) + 1
     db.flush()
 
 
@@ -379,7 +392,8 @@ def _sync_movies(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -
         ).scalars()
     }
     seen: set[str] = set()
-    for meta in _paginate(plex, lib.plex_key, _TYPE_MOVIE):
+    progress: dict = {}
+    for meta in _paginate(plex, lib.plex_key, _TYPE_MOVIE, progress):
         rk = str(meta.get("ratingKey") or "")
         if not rk:
             continue
@@ -390,7 +404,7 @@ def _sync_movies(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -
         seen.add(rk)
         stats["movies"] += 1
     db.flush()
-    _prune_stale(db, existing, seen, stats)
+    _prune_stale(db, existing, seen, stats, complete=progress.get("complete", False))
 
 
 def _sync_shows(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -> None:
@@ -402,7 +416,8 @@ def _sync_shows(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) ->
         ).scalars()
     }
     shows_seen: set[str] = set()
-    for meta in _paginate(plex, lib.plex_key, _TYPE_SHOW):
+    shows_progress: dict = {}
+    for meta in _paginate(plex, lib.plex_key, _TYPE_SHOW, shows_progress):
         rk = str(meta.get("ratingKey") or "")
         if not rk:
             continue
@@ -433,7 +448,8 @@ def _sync_shows(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) ->
         ).scalars()
     }
     seasons_seen: set[str] = set()
-    for meta in _paginate(plex, lib.plex_key, _TYPE_SEASON):
+    seasons_progress: dict = {}
+    for meta in _paginate(plex, lib.plex_key, _TYPE_SEASON, seasons_progress):
         rk = str(meta.get("ratingKey") or "")
         sid = show_id.get(str(meta.get("parentRatingKey") or ""))
         if not rk or sid is None:
@@ -459,7 +475,8 @@ def _sync_shows(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) ->
         ).scalars()
     }
     episodes_seen: set[str] = set()
-    for meta in _paginate(plex, lib.plex_key, _TYPE_EPISODE):
+    episodes_progress: dict = {}
+    for meta in _paginate(plex, lib.plex_key, _TYPE_EPISODE, episodes_progress):
         rk = str(meta.get("ratingKey") or "")
         if not rk:
             continue
@@ -478,9 +495,9 @@ def _sync_shows(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) ->
     # Prune leaf-first (episodes -> seasons -> shows) so a stale row is deleted directly, before its
     # parent's cascade would reach it — avoids deleting the same ORM object twice. Each level is
     # independently guarded on a non-empty listing by _prune_stale.
-    _prune_stale(db, episodes, episodes_seen, stats)
-    _prune_stale(db, seasons, seasons_seen, stats)
-    _prune_stale(db, shows, shows_seen, stats)
+    _prune_stale(db, episodes, episodes_seen, stats, complete=episodes_progress.get("complete", False))
+    _prune_stale(db, seasons, seasons_seen, stats, complete=seasons_progress.get("complete", False))
+    _prune_stale(db, shows, shows_seen, stats, complete=shows_progress.get("complete", False))
 
 
 def _sync_overlay(

backend/tests/test_plex_query_conversion.py

@@ -149,7 +149,9 @@ class TestRollup:
 
 
 def _fake_paginate(by_type):
-    def _p(_plex, _key, typ):
+    def _p(_plex, _key, typ, progress=None):
+        if progress is not None:
+            progress["complete"] = True  # the fake returns the whole list — a complete listing
         return list(by_type.get(typ, []))
     return _p
 

backend/tests/test_plex_sync_prune.py

@@ -37,6 +37,21 @@ class _StubPlex:
         return items[start : start + size], len(items)
 
 
+class _TruncatedPlex:
+    """Simulates a truncated listing (a transient Plex hiccup): it reports a large `total` but returns
+    only one short page, so pagination stops BEFORE reaching total. Prune must treat this as an
+    incomplete fetch and delete nothing — the un-listed rows are present, just not mentioned."""
+
+    def __init__(self, items: list[dict], claimed_total: int):
+        self._items = items
+        self._total = claimed_total
+
+    def section_items(self, key, item_type, start, size):
+        if start == 0:
+            return list(self._items), self._total  # one short page, total says there's more
+        return [], self._total
+
+
 def _seed_movies(db):
     lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
     db.add(lib)
@@ -88,6 +103,22 @@ def test_empty_listing_prunes_nothing(db):
     assert stats["pruned"] == 0
 
 
+def test_truncated_listing_prunes_nothing(db):
+    """A partial fetch (Plex reports 5 items but the listing returns only 1 then stops) must NOT prune
+    the 4 it failed to mention — the completeness guard, not just the empty-listing guard."""
+    lib, m1_id, _m2_id, _user_id = _seed_movies(db)
+    # Only m1 is listed, but total claims 5 → the walk never reaches total → incomplete → no prune.
+    stub = _TruncatedPlex([{"ratingKey": "m1", "title": "Kept Movie"}], claimed_total=5)
+    stats = {"movies": 0, "pruned": 0}
+
+    _sync_movies(db, stub, lib, stats)
+    db.commit()
+
+    live = {r.rating_key for r in db.execute(select(PlexItem)).scalars()}
+    assert live == {"m1", "m2"}, "a truncated listing must not prune the un-listed rows"
+    assert stats["pruned"] == 0
+
+
 def test_show_prune_removes_stale_show_season_and_episode(db):
     lib = PlexLibrary(plex_key="2", title="Shows", kind="show", enabled=True)
     db.add(lib)