fix(plex): a listing cut short no longer clears watch state or Hungarian text

The daily watch reconcile reads "absent from Plex's scan" as "Plex un-watched it" and deletes the
row — but it walked the sections without asking _paginate whether the walk reached the server's
total, so a Plex hiccup mid-walk (an empty page early) un-watched every title it failed to reach
until the next complete run. The scan now reports completeness; an incomplete one still pulls in
what it saw, re-pushes dirty rows, and clears nothing (logged as plex.watch.reconcile_incomplete).

The HU overlay had the same shape: clear every row's HU text, then refill from the listing. It now
reads the listing first and clears only after a complete one. The overlay moved to plex/overlay.py
(plex/sync.py drops under the 800-line limit and off the size ratchet's list).

Tests: plex/paths (23% -> 100%: prefix boundary, longest prefix, dead mount), the reconcile's
clear/keep/re-push rules, the cut-short scan and listing (each seen red with the guard removed).

backend/app/plex/overlay.py

@@ -0,0 +1,150 @@
+"""The HU overlay: a Hungarian Plex section written onto its English twin's rows (E30), split out of
+`plex/sync.py` (E32 S2), which runs it once per paired section during the full sync.
+
+The pairing is admin-configured (`plex_overlay_map`, ``huKey:enKey,...``). An overlay section creates
+no browsable rows of its own; it only fills the `*_hu` columns of the EN rows it matches by guid.
+`plex/sync.py` and this module import each other as MODULES (attribute access at call time), which is
+what lets either be imported first.
+"""
+
+from collections.abc import Iterable
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app import logs, sysconfig
+from app.models import PlexItem, PlexLibrary, PlexShow
+from app.plex import sync as plex_sync
+from app.plex.client import PlexClient
+from app.plex.genres import hu_genres
+
+log = logs.get_logger(__name__)
+
+
+def overlay_map(db: Session) -> dict[str, str]:
+    """The configured HU-overlay pairing: {huSectionKey: enSectionKey} from `plex_overlay_map`
+    ('huKey:enKey,...'). Each HU section is synced as a non-browsable overlay onto its EN twin.
+    Malformed pairs are skipped."""
+    raw = sysconfig.get_str(db, "plex_overlay_map")
+    out: dict[str, str] = {}
+    for pair in raw.split(","):
+        hu, sep, en = pair.partition(":")
+        hu, en = hu.strip(), en.strip()
+        if sep and hu and en:
+            out[hu] = en
+    return out
+
+
+def sync_overlay(
+    db: Session, plex: PlexClient, lib: PlexLibrary, en_lib_id: int, stats: dict
+) -> None:
+    """Write an overlay section's Hungarian metadata onto its EN twin's rows, matched by guid. Creates
+    NO browsable rows — an overlay library is never listed on its own. Only title/summary/tagline/genres
+    differ by language; the language-neutral people_text feeds both search vectors. A show section's
+    episodes are overlaid too (E30): the HU section lists every episode with its Hungarian title and
+    summary, and the episode rows live in the EN show library.
+
+    This is a full RECONCILE: after a COMPLETE HU listing every guid-bearing EN row's HU fields are
+    cleared, then the matches are repopulated — so an EN row whose HU twin vanished (or was re-guid'd)
+    falls back to EN instead of keeping stale Hungarian metadata. A listing cut short only adds. HU
+    items with no EN match are skipped."""
+    if lib.kind == "movie":
+        passes = [
+            (
+                plex_sync._TYPE_MOVIE,
+                select(PlexItem).where(
+                    PlexItem.library_id == en_lib_id,
+                    PlexItem.kind == "movie",
+                    PlexItem.guid.is_not(None),
+                ),
+            )
+        ]
+    else:
+        passes = [
+            (
+                plex_sync._TYPE_SHOW,
+                select(PlexShow).where(
+                    PlexShow.library_id == en_lib_id, PlexShow.guid.is_not(None)
+                ),
+            ),
+            (
+                plex_sync._TYPE_EPISODE,
+                select(PlexItem).where(
+                    PlexItem.library_id == en_lib_id,
+                    PlexItem.kind == "episode",
+                    PlexItem.guid.is_not(None),
+                ),
+            ),
+        ]
+    for item_type, rows_q in passes:
+        _overlay_pass(db, plex, lib, item_type, db.execute(rows_q).scalars(), stats)
+    db.flush()
+
+
+def _overlay_pass(
+    db: Session,
+    plex: PlexClient,
+    lib: PlexLibrary,
+    item_type: int,
+    rows: Iterable[PlexItem] | Iterable[PlexShow],
+    stats: dict,
+) -> None:
+    """One kind of the overlay (movies, shows or episodes) — see `sync_overlay`."""
+    # guid -> the EN rows carrying it (a list: two editions of one film can share a guid; both overlay).
+    by_guid: dict[str, list[PlexItem | PlexShow]] = {}
+    for r in rows:
+        if r.guid:  # the passes select guid-bearing rows only
+            by_guid.setdefault(r.guid, []).append(r)
+    # The HU listing is read first and applied after, so the clear can wait for its verdict.
+    hu_by_guid, complete = _read_hu_listing(plex, lib, item_type, by_guid)
+    episodes = item_type == plex_sync._TYPE_EPISODE
+    if complete:
+        _clear_hu(by_guid, episodes)
+    else:
+        stats["overlay_incomplete"] = stats.get("overlay_incomplete", 0) + 1
+        log.warning(
+            "the HU listing ended before the server's total; existing HU text is kept",
+            extra=logs.event("plex.sync.overlay_incomplete", library=lib.plex_key),
+        )
+    for g, meta in hu_by_guid.items():
+        targets = by_guid[g]
+        title_hu = (meta.get("title") or "").strip() or None
+        summary_hu = meta.get("summary")
+        tagline_hu = (meta.get("tagline") or "").strip() or None
+        genres_hu = hu_genres(plex_sync._tags(meta, "Genre"))
+        for row in targets:
+            row.title_hu = title_hu
+            row.summary_hu = summary_hu
+            if not episodes:
+                row.tagline_hu = tagline_hu
+                row.genres_hu = genres_hu or row.genres_hu
+        stats["overlaid"] = stats.get("overlaid", 0) + 1
+
+
+def _read_hu_listing(
+    plex: PlexClient, lib: PlexLibrary, item_type: int, by_guid: dict
+) -> tuple[dict[str, dict], bool]:
+    """The HU section's entries for the guids the EN side carries, and whether the walk reached the
+    server's total. The first HU entry for a guid wins; a dup HU item can't overwrite or double-count."""
+    progress: dict = {}
+    hu_by_guid: dict[str, dict] = {}
+    for meta in plex_sync._paginate(plex, lib.plex_key, item_type, progress):
+        g = plex_sync._canonical_guid(meta)
+        if g and g in by_guid and g not in hu_by_guid:
+            hu_by_guid[g] = meta
+    return hu_by_guid, bool(progress.get("complete"))
+
+
+def _clear_hu(by_guid: dict, episodes: bool) -> None:
+    """Reconcile step: clear HU fields on all guid-bearing rows; the matches are re-set after. Only
+    rows that actually held HU data become dirty (SQLAlchemy skips a no-net-change UPDATE), so this is
+    cheap. The genres are the exception: every row gets Hungarian ones, from its HU twin or, lacking
+    one, from its own English list — otherwise the HU genre list mixed "Drama" (no twin) with "Dráma".
+    Only after a COMPLETE listing: one a Plex hiccup cut short would clear the HU text of every title
+    it failed to reach, until the next full sync (E32 S2, the watch reconcile's twin)."""
+    for group in by_guid.values():
+        for row in group:
+            row.title_hu = row.summary_hu = None
+            if not episodes:
+                row.tagline_hu = None
+                row.genres_hu = hu_genres(row.genres)

backend/app/plex/sync.py

@@ -15,7 +15,6 @@ dead `enabled` column) is a separate follow-up. Cast + intro/credit markers are
 item-detail/player path (P2), not here.
 """
 
-from collections.abc import Iterable
 from datetime import datetime, timezone
 
 from sqlalchemy import func, select
@@ -23,7 +22,7 @@ from sqlalchemy.orm import Session
 
 from app import logs, sysconfig
 from app.models import PlexCollection, PlexItem, PlexLibrary, PlexSeason, PlexShow
-from app.plex import detail, people
+from app.plex import detail, overlay, people
 from app.plex.client import PlexClient, PlexError, PlexNotConfigured
 from app.plex.genres import hu_genres
 from app.plex.lookups import by_rating_key
@@ -194,20 +193,6 @@ def _enabled_section_keys(db: Session) -> set[str] | None:
     return keys or None
 
 
-def _overlay_map(db: Session) -> dict[str, str]:
-    """The configured HU-overlay pairing: {huSectionKey: enSectionKey} from `plex_overlay_map`
-    ('huKey:enKey,...'). Each HU section is synced as a non-browsable overlay onto its EN twin.
-    Malformed pairs are skipped."""
-    raw = sysconfig.get_str(db, "plex_overlay_map")
-    out: dict[str, str] = {}
-    for pair in raw.split(","):
-        hu, sep, en = pair.partition(":")
-        hu, en = hu.strip(), en.strip()
-        if sep and hu and en:
-            out[hu] = en
-    return out
-
-
 def sync(db: Session) -> dict:
     """Full reconcile of the enabled movie/show sections. Returns a small stats dict."""
     if not sysconfig.get_bool(db, "plex_enabled"):
@@ -223,7 +208,7 @@ def sync(db: Session) -> dict:
         "pruned": 0,
     }
     wanted = _enabled_section_keys(db)
-    overlay_map = _overlay_map(db)  # {huKey: enKey}
+    overlay_map = overlay.overlay_map(db)  # {huKey: enKey}
     overlay_keys = set(overlay_map)
     try:
         with PlexClient(db) as plex:
@@ -267,7 +252,7 @@ def sync(db: Session) -> dict:
                 lib.metadata_lang = "hu"
                 lib.overlay_of = en_lib.id
                 db.flush()
-                _sync_overlay(db, plex, lib, en_lib.id, stats)
+                overlay.sync_overlay(db, plex, lib, en_lib.id, stats)
                 db.commit()
             # Rich-fetch a bounded batch of not-yet-enriched titles' FULL cast, so the whole library
             # becomes filterable over successive runs (the cheap listing above only stored the top ~3).
@@ -661,98 +646,6 @@ def _sync_shows(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) ->
     _prune_stale(db, shows, shows_seen, stats, complete=shows_progress.get("complete", False))
 
 
-def _sync_overlay(
-    db: Session, plex: PlexClient, lib: PlexLibrary, en_lib_id: int, stats: dict
-) -> None:
-    """Write an overlay section's Hungarian metadata onto its EN twin's rows, matched by guid. Creates
-    NO browsable rows — an overlay library is never listed on its own. Only title/summary/tagline/genres
-    differ by language; the language-neutral people_text feeds both search vectors. A show section's
-    episodes are overlaid too (E30): the HU section lists every episode with its Hungarian title and
-    summary, and the episode rows live in the EN show library.
-
-    This is a full RECONCILE: every guid-bearing EN row's HU fields are cleared first, then the HU
-    section repopulates the matches — so an EN row whose HU twin vanished (or was re-guid'd) falls back
-    to EN instead of keeping stale Hungarian metadata. HU items with no EN match are skipped."""
-    if lib.kind == "movie":
-        passes = [
-            (
-                _TYPE_MOVIE,
-                select(PlexItem).where(
-                    PlexItem.library_id == en_lib_id,
-                    PlexItem.kind == "movie",
-                    PlexItem.guid.is_not(None),
-                ),
-            )
-        ]
-    else:
-        passes = [
-            (
-                _TYPE_SHOW,
-                select(PlexShow).where(
-                    PlexShow.library_id == en_lib_id, PlexShow.guid.is_not(None)
-                ),
-            ),
-            (
-                _TYPE_EPISODE,
-                select(PlexItem).where(
-                    PlexItem.library_id == en_lib_id,
-                    PlexItem.kind == "episode",
-                    PlexItem.guid.is_not(None),
-                ),
-            ),
-        ]
-    for item_type, rows_q in passes:
-        _overlay_pass(db, plex, lib, item_type, db.execute(rows_q).scalars(), stats)
-    db.flush()
-
-
-def _overlay_pass(
-    db: Session,
-    plex: PlexClient,
-    lib: PlexLibrary,
-    item_type: int,
-    rows: Iterable[PlexItem] | Iterable[PlexShow],
-    stats: dict,
-) -> None:
-    """One kind of the overlay (movies, shows or episodes) — see `_sync_overlay`."""
-    # guid -> the EN rows carrying it (a list: two editions of one film can share a guid; both overlay).
-    by_guid: dict[str, list[PlexItem | PlexShow]] = {}
-    for r in rows:
-        if r.guid:  # the passes select guid-bearing rows only
-            by_guid.setdefault(r.guid, []).append(r)
-    # Reconcile step: clear HU fields on all guid-bearing rows. Only rows that actually held HU data
-    # become dirty (SQLAlchemy skips a no-net-change UPDATE), so this is cheap; matches are re-set below.
-    # The genres are the exception: every row gets Hungarian ones, from its HU twin or, lacking one, from
-    # its own English list — otherwise the HU genre list mixed "Drama" (no twin) with "Dráma".
-    episodes = item_type == _TYPE_EPISODE
-    for group in by_guid.values():
-        for row in group:
-            row.title_hu = row.summary_hu = None
-            if not episodes:
-                row.tagline_hu = None
-                row.genres_hu = hu_genres(row.genres)
-    seen: set[str] = (
-        set()
-    )  # first HU entry for a guid wins; a dup HU item can't overwrite or double-count
-    for meta in _paginate(plex, lib.plex_key, item_type):
-        g = _canonical_guid(meta)
-        targets = by_guid.get(g) if g else None
-        if not targets or g in seen:
-            continue
-        seen.add(g)
-        title_hu = (meta.get("title") or "").strip() or None
-        summary_hu = meta.get("summary")
-        tagline_hu = (meta.get("tagline") or "").strip() or None
-        genres_hu = hu_genres(_tags(meta, "Genre"))
-        for row in targets:
-            row.title_hu = title_hu
-            row.summary_hu = summary_hu
-            if not episodes:
-                row.tagline_hu = tagline_hu
-                row.genres_hu = genres_hu or row.genres_hu
-        stats["overlaid"] = stats.get("overlaid", 0) + 1
-
-
 def _upsert_collection(
     db: Session, existing: PlexCollection | None, rk: str, lib_id: int, meta: dict
 ) -> PlexCollection:
@@ -815,6 +708,15 @@ def _sync_collections(db: Session, plex: PlexClient, lib: PlexLibrary, stats: di
             select(PlexCollection).where(PlexCollection.library_id == lib.id)
         ).scalars()
     }
+    if not cols and existing:
+        # The catalog's own rule (see the module doc): an empty listing while we hold rows is a Plex
+        # glitch, not a library emptied of collections. Reconciling on it dropped every collection
+        # (its user-set `editable` flag with it) and every member's keys (E32 S2).
+        log.warning(
+            "Plex listed no collections for a library that has some; keeping them this run",
+            extra=logs.event("plex.sync.collections_empty", library=lib.plex_key),
+        )
+        return
     seen: set[str] = set()
     membership: dict[str, list[str]] = {}  # member rating_key -> [collection rating_key, ...]
     for meta in cols:

backend/app/plex/watch_sync.py

@@ -72,13 +72,17 @@ def _rk_to_id(db: Session) -> dict[str, int]:
 
 def _scan_plex_states(
     db: Session, plex: PlexClient
-) -> tuple[dict[int, tuple[str, int, datetime | None, datetime | None]], int]:
+) -> tuple[dict[int, tuple[str, int, datetime | None, datetime | None]], int, bool]:
     """Scan the enabled movie/show sections and return ``{item_id: (status, pos, watched_at,
-    prog_at)}`` for every mirrored leaf Plex holds a watch signal for, plus the scanned-leaf count.
+    prog_at)}`` for every mirrored leaf Plex holds a watch signal for, the scanned-leaf count, and
-    Shared by the one-time import and the full reconcile (both need Plex's whole current picture)."""
+    whether EVERY section's walk reached the server's total. Shared by the one-time import and the
+    full reconcile (both need Plex's whole current picture). The last one matters to the reconcile
+    alone: it reads "absent from the scan" as "Plex un-watched it", which a walk cut short by a
+    Plex hiccup does not mean (see `_paginate`)."""
     item_id_by_rk = _rk_to_id(db)
     wanted = _enabled_section_keys(db)
     scanned = 0
+    complete = True
     out: dict[int, tuple[str, int, datetime | None, datetime | None]] = {}
     for s in plex.sections():
         stype = s.get("type")
@@ -88,7 +92,8 @@ def _scan_plex_states(
         if wanted is not None and key not in wanted:
             continue
         item_type = _TYPE_MOVIE if stype == "movie" else _TYPE_EPISODE
-        for meta in _paginate(plex, key, item_type):
+        progress: dict = {}
+        for meta in _paginate(plex, key, item_type, progress):
             scanned += 1
             item_id = item_id_by_rk.get(str(meta.get("ratingKey") or ""))
             if item_id is None:
@@ -98,7 +103,8 @@ def _scan_plex_states(
             )
             if res is not None:
                 out[item_id] = res
-    return out, scanned
+        complete = complete and bool(progress.get("complete"))
+    return out, scanned, complete
 
 
 def import_owner_watch_state(db: Session, link: PlexLink) -> dict:
@@ -111,7 +117,8 @@ def import_owner_watch_state(db: Session, link: PlexLink) -> dict:
     stats = {"watched": 0, "in_progress": 0, "scanned": 0}
     try:
         with PlexClient(db) as plex:
-            scanned_states, stats["scanned"] = _scan_plex_states(db, plex)
+            # A partial scan is still safe here: the import only ever ADDS what Plex reports.
+            scanned_states, stats["scanned"], _complete = _scan_plex_states(db, plex)
             # One row per item (keyed by item_id → inherently de-duped) then bulk-UPSERT. An upsert is
             # idempotent AND concurrency-safe: a repeated/overlapping enable (the toggle firing twice
             # while the multi-second import runs) can't raise a duplicate-key error — it re-updates.
@@ -590,6 +597,20 @@ def run_plex_watch_sync(db: Session) -> dict:
     return total
 
 
+def _clear_unwatched(db: Session, user_id: int, plex_states: dict) -> int:
+    """Clear the user's rows Plex no longer has a record for — the mirrored ones (Plex un-watched
+    them); hidden and dirty rows stay (a dirty one is re-pushed by `_repush_dirty`). Called only
+    after a COMPLETE scan: a walk a Plex hiccup cut short leaves out rows it merely failed to reach,
+    and clearing those un-watched them here until the next complete run (E32 S2)."""
+    n = 0
+    for st in db.execute(select(PlexState).where(PlexState.user_id == user_id)).scalars().all():
+        if st.item_id in plex_states or st.status == "hidden" or not st.synced_to_plex:
+            continue
+        db.delete(st)
+        n += 1
+    return n
+
+
 def run_plex_watch_reconcile(db: Session) -> dict:
     """Scheduler job (default daily) — the authoritative full reconcile. Rescans every section and,
     per item, uses `synced_to_plex` to settle what the incremental feed can't:
@@ -608,7 +629,7 @@ def run_plex_watch_reconcile(db: Session) -> dict:
         try:
             with PlexClient(db) as plex:
                 _resolve_account_id(db, plex, link)
-                plex_states, scanned = _scan_plex_states(db, plex)
+                plex_states, scanned, complete = _scan_plex_states(db, plex)
                 total["scanned"] += scanned
 
                 for item_id, (status, pos, watched_at, prog_at) in plex_states.items():
@@ -616,18 +637,14 @@ def run_plex_watch_reconcile(db: Session) -> dict:
                     if ch:
                         total["applied"] += 1
 
-                # Siftlode rows Plex no longer has a record for: clear the ones that were mirrored
+                if complete:
-                # (Plex un-watched them); leave hidden + dirty (dirty gets re-pushed just below).
+                    total["cleared"] += _clear_unwatched(db, link.user_id, plex_states)
-                for st in (
+                else:
-                    db.execute(select(PlexState).where(PlexState.user_id == link.user_id))
+                    total["incomplete"] = total.get("incomplete", 0) + 1
-                    .scalars()
+                    log.warning(
-                    .all()
+                        "the Plex scan ended before the server's total; nothing is cleared this run",
-                ):
+                        extra=logs.event("plex.watch.reconcile_incomplete", user_id=link.user_id),
-                    if st.item_id in plex_states or st.status == "hidden":
+                    )
-                        continue
-                    if st.synced_to_plex:
-                        db.delete(st)
-                        total["cleared"] += 1
 
                 total["repushed"] += _repush_dirty(db, plex, link)
                 link.last_watch_sync_at = datetime.now(timezone.utc)

backend/tests/test_plex_overlay.py

@@ -16,18 +16,21 @@ from app import sysconfig
 from app.db import engine
 from app.models import PlexItem, PlexLibrary, PlexShow
 from app.plex.genres import hu_genre, hu_genres
-from app.plex.sync import _apply_item, _canonical_guid, _sync_overlay
+from app.plex.overlay import sync_overlay
+from app.plex.sync import _apply_item, _canonical_guid
 from app.plex.query import content_libs
 
 
 class _StubPlex:
-    """Minimal PlexClient stand-in: _sync_overlay only calls section_items via _paginate."""
+    """Minimal PlexClient stand-in: sync_overlay only calls section_items via _paginate. `total`
+    over the item count is a listing cut short (the server said more than it handed over)."""
 
-    def __init__(self, items: list[dict]):
+    def __init__(self, items: list[dict], total: int | None = None):
         self._items = items
+        self._total = len(items) if total is None else total
 
     def section_items(self, key, item_type, start, size):
-        return self._items[start : start + size], len(self._items)
+        return self._items[start : start + size], self._total
 
 
 def _hu_movie(rk: str, title: str, summary: str, genres: list[str], guid: str | None) -> dict:
@@ -71,7 +74,7 @@ def test_sync_overlay_writes_hu_onto_matched_en_rows(db):
         ]
     )
     stats = {"overlaid": 0}
-    _sync_overlay(db, stub, hu_lib, en_lib.id, stats)
+    sync_overlay(db, stub, hu_lib, en_lib.id, stats)
     db.commit()
 
     db.refresh(matched)
@@ -185,13 +188,51 @@ def test_overlay_reconcile_clears_stale_hu_when_twin_vanishes(db):
     )
     db.add(row)
     db.commit()
-    # The HU section no longer lists this film (empty overlay).
+    # The HU section, read in full, no longer lists this film (only another one).
-    _sync_overlay(db, _StubPlex([]), hu_lib, en_lib.id, {"overlaid": 0})
+    other = _hu_movie("h9", "Más", "más", [], "tmdb://99")
+    sync_overlay(db, _StubPlex([other]), hu_lib, en_lib.id, {"overlaid": 0})
     db.commit()
     db.refresh(row)
     assert row.title_hu is None and row.summary_hu is None and row.genres_hu is None
 
 
+def test_a_listing_cut_short_keeps_the_hungarian_text_it_did_not_reach(db):
+    """E32 S2: the reconcile clears HU text only after a COMPLETE listing. A walk a Plex hiccup cut
+    short (an empty page before the server's total), or an empty section, says nothing about the
+    titles it did not reach — clearing them dropped their Hungarian text until the next full sync."""
+    en_lib = PlexLibrary(plex_key="4", title="Movies", kind="movie")
+    hu_lib = PlexLibrary(plex_key="9", title="Filmek", kind="movie", metadata_lang="hu")
+    db.add_all([en_lib, hu_lib])
+    db.flush()
+    hu_lib.overlay_of = en_lib.id
+    kept = PlexItem(
+        rating_key="m1",
+        library_id=en_lib.id,
+        kind="movie",
+        title="Heat",
+        guid="tmdb:1",
+        title_hu="Szemtől szemben",
+        summary_hu="régi",
+        genres_hu=["Akció"],
+    )
+    fresh = PlexItem(rating_key="m2", library_id=en_lib.id, kind="movie", title="Up", guid="tmdb:2")
+    db.add_all([kept, fresh])
+    db.commit()
+    stats: dict = {}
+    reached = _hu_movie("h2", "Fel", "fel", [], "tmdb://2")
+    sync_overlay(db, _StubPlex([reached], total=2), hu_lib, en_lib.id, stats)
+    db.commit()
+    db.refresh(kept)
+    db.refresh(fresh)
+    assert kept.title_hu == "Szemtől szemben" and kept.summary_hu == "régi"
+    assert fresh.title_hu == "Fel"  # what the listing did reach is still applied
+    assert stats["overlay_incomplete"] == 1
+    sync_overlay(db, _StubPlex([]), hu_lib, en_lib.id, {})  # an empty section: the same answer
+    db.commit()
+    db.refresh(kept)
+    assert kept.title_hu == "Szemtől szemben"
+
+
 def test_apply_item_keeps_prior_guid_when_listing_lacks_guid():
     row = PlexItem(rating_key="m1", kind="movie", guid="tmdb:1")
     _apply_item(row, lib_id=7, kind="movie", meta={"title": "Heat"})  # no Guid array
@@ -225,7 +266,7 @@ def test_overlay_applies_to_all_editions_sharing_a_guid(db):
         ]
     )
     stats = {"overlaid": 0}
-    _sync_overlay(db, stub, hu_lib, en_lib.id, stats)
+    sync_overlay(db, stub, hu_lib, en_lib.id, stats)
     db.commit()
     db.refresh(ed1)
     db.refresh(ed2)
@@ -291,7 +332,7 @@ def test_a_show_overlay_writes_hungarian_episode_titles_and_summaries(db):
             ],
         }
     )
-    _sync_overlay(db, stub, hu_lib, en_lib.id, {})
+    sync_overlay(db, stub, hu_lib, en_lib.id, {})
     db.commit()
     db.refresh(ep)
     db.refresh(show)
@@ -328,7 +369,7 @@ def test_the_overlay_gives_every_row_one_hungarian_word_per_genre(db):
         "h1", "Szemtől szemben", "HU.", ["Action", "Bűnügyi", "Drama", "Dráma"], "tmdb://1"
     )
     meta["tagline"] = "Magyar mottó."
-    _sync_overlay(db, _StubPlex([meta]), hu_lib, en_lib.id, {})
+    sync_overlay(db, _StubPlex([meta]), hu_lib, en_lib.id, {})
     db.commit()
     db.refresh(twin)
     db.refresh(lone)
@@ -353,7 +394,7 @@ def test_a_twin_without_genres_keeps_the_rows_own_hungarian_list(db):
     )
     db.add(row)
     db.commit()
-    _sync_overlay(
+    sync_overlay(
         db,
         _StubPlex([_hu_movie("h1", "Szemtől szemben", "HU.", [], "tmdb://1")]),
         hu_lib,

backend/tests/test_plex_paths.py

@@ -0,0 +1,82 @@
+"""plex/paths — the Plex-side path → local mount translation (E32 S2: 23% covered, and a wrong answer
+here reads as "media file not found" for a whole library at once, with nothing else failing).
+
+Pure logic: the admin map is read through sysconfig, stubbed per test.
+"""
+
+from pathlib import Path
+
+import pytest
+
+from app.plex import paths
+
+
+@pytest.fixture
+def path_map(monkeypatch):
+    def use(raw: str) -> None:
+        monkeypatch.setattr(paths.sysconfig, "get_str", lambda db, key: raw)
+
+    return use
+
+
+def test_the_map_parses_both_separators_drops_junk_and_puts_the_longest_prefix_first():
+    raw = " /data=/mnt/a ,\n/data/movies = /mnt/m\nnot-a-pair,=/x,/y=, /data/tv=/mnt/tv "
+    assert paths.parse_map(raw) == [
+        ("/data/movies", "/mnt/m"),
+        ("/data/tv", "/mnt/tv"),
+        ("/data", "/mnt/a"),
+    ]
+    assert paths.parse_map("") == []
+
+
+def test_the_longest_matching_prefix_wins(path_map):
+    path_map("/data=/mnt/a,/data/movies=/mnt/m")
+    assert paths.map_path(None, "/data/movies/Foo.mkv") == str(Path("/mnt/m") / "Foo.mkv")
+    assert paths.map_path(None, "/data/tv/Bar.mkv") == str(Path("/mnt/a") / "tv/Bar.mkv")
+
+
+def test_a_prefix_matches_only_at_a_path_boundary(path_map):
+    # "/data/movies2" shares the characters of "/data/movies" but is another folder.
+    path_map("/data/movies=/mnt/m")
+    assert paths.map_path(None, "/data/movies2/Foo.mkv") == "/data/movies2/Foo.mkv"
+
+
+def test_the_prefix_itself_and_a_trailing_slash_in_the_map(path_map):
+    path_map("/data/movies/=/mnt/m")
+    assert paths.map_path(None, "/data/movies/Foo.mkv") == str(Path("/mnt/m") / "Foo.mkv")
+    path_map("/data/movies=/mnt/m")
+    assert paths.map_path(None, "/data/movies") == "/mnt/m"
+
+
+def test_no_map_leaves_the_path_as_plex_sees_it(path_map):
+    path_map("")
+    assert paths.map_path(None, "/data/movies/Foo.mkv") == "/data/movies/Foo.mkv"
+
+
+def test_the_local_file_must_exist(path_map, tmp_path):
+    (tmp_path / "movies").mkdir()
+    (tmp_path / "movies" / "Foo.mkv").write_bytes(b"x")
+    path_map(f"/data={tmp_path}")
+    assert (
+        paths.local_media_path(None, "/data/movies/Foo.mkv")
+        == (tmp_path / "movies" / "Foo.mkv").resolve()
+    )
+    assert paths.local_media_path(None, "/data/movies/Missing.mkv") is None
+    assert paths.local_media_path(None, "/data/movies") is None  # a folder is not a file
+    assert paths.local_media_path(None, None) is None
+    assert paths.local_media_path(None, "") is None
+
+
+def test_a_path_that_cannot_be_resolved_is_logged_and_answered_none(path_map, monkeypatch, caplog):
+    path_map("")
+
+    def broken(self, strict=False):
+        raise OSError("dead mount")
+
+    monkeypatch.setattr(paths.Path, "resolve", broken)
+    assert paths.local_media_path(None, "/data/movies/Foo.mkv") is None
+    rec = next(
+        r for r in caplog.records if getattr(r, "event", None) == "plex.paths.resolve_failed"
+    )
+    # The title stays out of the log line: the path carries it.
+    assert "Foo" not in rec.getMessage() and rec.attrs == {"reason": "OSError"}

backend/tests/test_plex_watch_reconcile.py

@@ -0,0 +1,167 @@
+"""The daily full watch reconcile (`run_plex_watch_reconcile`) — the one pass that DELETES watch state:
+a mirrored row Plex no longer has a record for is read as "Plex un-watched it" and cleared. E32 S2
+found it trusting a scan a Plex hiccup had cut short; these hold what it may and may not clear.
+"""
+
+import pytest
+from sqlalchemy import select
+
+from app import sysconfig
+from app.models import PlexItem, PlexLibrary, PlexLink, PlexState, User
+from app.plex import watch_sync
+
+
+class _FakePlex:
+    """A PlexClient stand-in: one movie section whose listing reports `total` (more than it hands
+    over = a walk cut short), and a record of every write the reconcile sends back to Plex."""
+
+    def __init__(self, leaves: list[dict], total: int | None = None):
+        self.leaves = leaves
+        self.total = len(leaves) if total is None else total
+        self.calls: list[tuple] = []
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *exc):
+        return False
+
+    def sections(self):
+        return [{"type": "movie", "key": "1"}]
+
+    def section_items(self, key, item_type, start, size):
+        return self.leaves[start : start + size], self.total
+
+    def scrobble(self, rk):
+        self.calls.append(("scrobble", rk))
+
+    def unscrobble(self, rk):
+        self.calls.append(("unscrobble", rk))
+
+    def set_timeline(self, rk, time_ms, duration_ms):
+        self.calls.append(("timeline", rk, time_ms))
+
+
+@pytest.fixture
+def world(db):
+    """Five mirrored movies and the owner's link; each test sets the local states it needs."""
+    sysconfig.set_value(db, "plex_enabled", True)
+    lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
+    user = User(email="reconcile@example.com")
+    db.add_all([lib, user])
+    db.flush()
+    items = {}
+    for rk in ("m1", "m2", "m3", "m4", "m5"):
+        items[rk] = PlexItem(
+            rating_key=rk, library_id=lib.id, kind="movie", title=rk, duration_s=6000
+        )
+        db.add(items[rk])
+    db.add(
+        PlexLink(
+            user_id=user.id,
+            uses_admin=True,
+            sync_enabled=True,
+            initial_import_done=True,
+            plex_account_id=1,
+        )
+    )
+    db.flush()
+
+    def state(rk, **kw):
+        db.add(PlexState(user_id=user.id, item_id=items[rk].id, **kw))
+
+    def rows():
+        db.expire_all()
+        by_id = {i.id: rk for rk, i in items.items()}
+        return {
+            by_id[s.item_id]: (s.status, s.position_seconds, s.synced_to_plex)
+            for s in db.execute(select(PlexState).where(PlexState.user_id == user.id)).scalars()
+        }
+
+    return state, rows
+
+
+def _run(db, monkeypatch, plex):
+    monkeypatch.setattr(watch_sync, "PlexClient", lambda _db: plex)
+    return watch_sync.run_plex_watch_reconcile(db)
+
+
+def _local_states(state):
+    state("m1", status="watched", synced_to_plex=True)  # mirrored; Plex has no record now
+    state("m2", status="watched", synced_to_plex=False)  # local change that never reached Plex
+    state("m3", status="hidden", synced_to_plex=True)  # Siftlode-only, never cleared
+    state("m5", status="new", position_seconds=300, synced_to_plex=True)  # Plex says otherwise
+
+
+def test_a_complete_scan_clears_what_plex_unwatched_and_keeps_the_rest(db, monkeypatch, world):
+    state, rows = world
+    _local_states(state)
+    db.commit()
+    plex = _FakePlex(
+        [
+            {"ratingKey": "m3"},  # listed, no watch signal
+            {"ratingKey": "m4", "viewCount": 1, "lastViewedAt": 1_700_000_000},
+            {"ratingKey": "m5", "viewOffset": 120_000, "lastViewedAt": 1_700_000_000},
+        ]
+    )
+    out = _run(db, monkeypatch, plex)
+    assert rows() == {
+        "m2": ("watched", 0, True),  # re-pushed, then settled
+        "m3": ("hidden", 0, True),
+        "m4": ("watched", 0, True),  # pulled in
+        "m5": ("new", 120, True),  # Plex's position wins
+    }
+    assert ("scrobble", "m2") in plex.calls
+    assert out["cleared"] == 1 and out["repushed"] == 1 and "incomplete" not in out
+
+
+def test_a_scan_cut_short_clears_nothing(db, monkeypatch, world):
+    state, rows = world
+    _local_states(state)
+    db.commit()
+    # The server said 5, the walk got 3: m1 may simply not have been reached.
+    plex = _FakePlex(
+        [
+            {"ratingKey": "m3"},
+            {"ratingKey": "m4", "viewCount": 1, "lastViewedAt": 1_700_000_000},
+            {"ratingKey": "m5", "viewOffset": 120_000, "lastViewedAt": 1_700_000_000},
+        ],
+        total=5,
+    )
+    out = _run(db, monkeypatch, plex)
+    after = rows()
+    assert after["m1"] == ("watched", 0, True)  # NOT un-watched on a partial picture
+    assert after["m4"] == ("watched", 0, True) and after["m5"] == ("new", 120, True)  # pulls stand
+    assert after["m2"] == ("watched", 0, True)  # the dirty row is still pushed up
+    assert out["cleared"] == 0 and out["incomplete"] == 1
+
+
+def test_an_empty_listing_is_not_a_library_of_unwatched_titles(db, monkeypatch, world):
+    state, rows = world
+    state("m1", status="watched", synced_to_plex=True)
+    db.commit()
+    out = _run(db, monkeypatch, _FakePlex([]))
+    assert rows() == {"m1": ("watched", 0, True)}
+    assert out["cleared"] == 0
+
+
+def test_a_dirty_resume_is_re_pushed_as_a_timeline_and_a_dirty_reset_as_an_unscrobble(
+    db, monkeypatch, world
+):
+    state, rows = world
+    state("m1", status="new", position_seconds=300, synced_to_plex=False)
+    state("m2", status="new", position_seconds=2, synced_to_plex=False)  # under the 5 s floor
+    db.commit()
+    plex = _FakePlex([{"ratingKey": rk} for rk in ("m1", "m2", "m3", "m4", "m5")])
+    _run(db, monkeypatch, plex)
+    assert ("timeline", "m1", 300_000) in plex.calls
+    assert ("unscrobble", "m2") in plex.calls
+    assert {rk: synced for rk, (_s, _p, synced) in rows().items()} == {"m1": True, "m2": True}
+
+
+def test_watch_fields_map_to_a_state_only_when_plex_holds_a_signal():
+    to = watch_sync._plex_watch_to_state
+    assert to(None, None, None) is None
+    assert to(0, 4_000, None) is None  # a stray few seconds is not "in progress"
+    assert to(0, 90_000, 1_700_000_000)[:2] == ("new", 90)
+    assert to("2", 90_000, 1_700_000_000)[:2] == ("watched", 0)  # watched wins over an offset

backend/tests/test_source_size.py

@@ -32,7 +32,6 @@ BIG_MODULES: dict[str, int] = {
     "app/models.py": 1607,
     "app/plex/stream.py": 1128,
     "app/plex/subtitles.py": 1092,
-    "app/plex/sync.py": 841,
     "app/routes/admin.py": 864,
     "app/routes/auth.py": 1353,
     "app/routes/downloads.py": 1237,