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/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).
-    _sync_overlay(db, _StubPlex([]), hu_lib, en_lib.id, {"overlaid": 0})
+    # The HU section, read in full, no longer lists this film (only another one).
+    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,