fix(plex): the watch reconcile clears only what it walked, and doubts a mirrored 0

A library left out of the allowlist is never scanned, so its watch states were cleared on
every reconcile; clearing is now limited to the sections walked. A section we mirror items
from that says it holds 0 is not believed complete. The page size is no longer read as the
section total when totalSize is missing (measured: the server always sends totalSize on a
paged request, 0 for an empty one).

backend/app/plex/client.py

@@ -94,7 +94,9 @@ class PlexClient:
                 "X-Plex-Container-Size": size,
             },
         )
-        total = mc.get("totalSize", mc.get("size"))
+        # `totalSize` only: `size` is THIS page's count, and read as the total it made a single
+        # page look like the whole section — "complete" to a prune or a reconcile (E32 review).
+        total = mc.get("totalSize")
         return mc.get("Metadata", []) or [], None if total is None else int(total)
 
     def children(self, rating_key: str) -> list[dict]:

backend/app/plex/watch_sync.py

@@ -29,7 +29,7 @@ from sqlalchemy.orm import Session
 
 from app import logs, sysconfig
 from app.db import SessionLocal
-from app.models import PlexItem, PlexLink, PlexState
+from app.models import PlexItem, PlexLibrary, PlexLink, PlexState
 from app.plex.client import PlexClient, PlexError, PlexNotConfigured
 from app.plex.sync import (
     _TYPE_EPISODE,
@@ -72,17 +72,25 @@ 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, bool]:
+) -> tuple[dict[int, tuple[str, int, datetime | None, datetime | None]], int, bool, set[str]]:
     """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, the scanned-leaf count, and
-    whether EVERY section's walk reached the server's total (or the server said it holds none).
+    prog_at)}`` for every mirrored leaf Plex holds a watch signal for, the scanned-leaf count,
+    whether EVERY section's walk reached the server's total (or the server said it holds none), and
+    the keys of the sections walked — the only ones whose absences mean anything.
     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)
+    # The sections we mirror something from: a total of 0 from one of those contradicts the mirror.
+    mirrored = set(
+        db.execute(
+            select(PlexLibrary.plex_key).join(PlexItem, PlexItem.library_id == PlexLibrary.id)
+        ).scalars()
+    )
     scanned = 0
     complete = True
+    walked: set[str] = set()
     out: dict[int, tuple[str, int, datetime | None, datetime | None]] = {}
     for s in plex.sections():
         stype = s.get("type")
@@ -92,6 +100,7 @@ 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
+        walked.add(key)
         progress: dict = {}
         before = scanned
         for meta in _paginate(plex, key, item_type, progress):
@@ -106,9 +115,11 @@ def _scan_plex_states(
                 out[item_id] = res
         # A section the server SAYS is empty (total 0, nothing listed) is complete too: without this
         # one genuinely empty library kept every run "incomplete" and un-watches never synced again.
-        said_empty = progress.get("total") == 0 and scanned == before
+        # Believed only for a section we mirror nothing from: one whose items we hold answering 0 is
+        # a Plex that lost the library for a moment, not one that un-watched all of it (E32 review).
+        said_empty = progress.get("total") == 0 and scanned == before and key not in mirrored
         complete = complete and (bool(progress.get("complete")) or said_empty)
-    return out, scanned, complete
+    return out, scanned, complete, walked
 
 
 def import_owner_watch_state(db: Session, link: PlexLink) -> dict:
@@ -122,7 +133,7 @@ def import_owner_watch_state(db: Session, link: PlexLink) -> dict:
     try:
         with PlexClient(db) as 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)
+            scanned_states, stats["scanned"], _complete, _walked = _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.
@@ -601,13 +612,21 @@ def run_plex_watch_sync(db: Session) -> dict:
     return total
 
 
-def _clear_unwatched(db: Session, user_id: int, plex_states: dict) -> int:
+def _clear_unwatched(db: Session, user_id: int, plex_states: dict, walked: set[str]) -> 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)."""
+    and clearing those un-watched them here until the next complete run (E32 S2). Only rows of the
+    sections the scan WALKED: an item of a library the allowlist leaves out is absent from every
+    scan, and was cleared on every run (E32 review)."""
     n = 0
-    for st in db.execute(select(PlexState).where(PlexState.user_id == user_id)).scalars().all():
+    rows = db.execute(
+        select(PlexState)
+        .join(PlexItem, PlexItem.id == PlexState.item_id)
+        .join(PlexLibrary, PlexLibrary.id == PlexItem.library_id)
+        .where(PlexState.user_id == user_id, PlexLibrary.plex_key.in_(walked))
+    )
+    for st in rows.scalars().all():
         if st.item_id in plex_states or st.status == "hidden" or not st.synced_to_plex:
             continue
         db.delete(st)
@@ -633,7 +652,7 @@ def run_plex_watch_reconcile(db: Session) -> dict:
         try:
             with PlexClient(db) as plex:
                 _resolve_account_id(db, plex, link)
-                plex_states, scanned, complete = _scan_plex_states(db, plex)
+                plex_states, scanned, complete, walked = _scan_plex_states(db, plex)
                 total["scanned"] += scanned
 
                 for item_id, (status, pos, watched_at, prog_at) in plex_states.items():
@@ -644,7 +663,7 @@ def run_plex_watch_reconcile(db: Session) -> dict:
                 # A whole scan that listed nothing at all is not "Plex un-watched everything" either:
                 # every section can say 0 only when the server lost its libraries, never as news.
                 if complete and scanned:
-                    total["cleared"] += _clear_unwatched(db, link.user_id, plex_states)
+                    total["cleared"] += _clear_unwatched(db, link.user_id, plex_states, walked)
                 else:
                     total["incomplete"] = total.get("incomplete", 0) + 1
                     log.warning(

backend/tests/test_plex_client.py

@@ -79,11 +79,16 @@ def test_a_section_page_asks_for_guids_and_reports_the_server_total(plex):
     assert params["includeGuids"] == "1" and params["type"] == "1"
     assert params["X-Plex-Container-Start"] == "200" and params["X-Plex-Container-Size"] == "50"
 
-    plex.routes["/library/sections/7/all"] = {"Metadata": [], "size": 0}
+    # Measured on the real server (E32 review): a paged request carries `totalSize` even when the
+    # page is empty, past the end, or the section holds nothing — `size: 0, totalSize: 0`.
+    plex.routes["/library/sections/7/all"] = {"Metadata": [], "size": 0, "totalSize": 0}
     assert plex.section_items("7", 2) == ([], 0)
     # No total at all is not "0": the caller must not read it as a complete, empty section.
     plex.routes["/library/sections/7/all"] = {"Metadata": []}
     assert plex.section_items("7", 2) == ([], None)
+    # …and `size` is THIS page's count, never the section's: one full page is not the whole section.
+    plex.routes["/library/sections/7/all"] = {"Metadata": [{"ratingKey": "1"}], "size": 1}
+    assert plex.section_items("7", 2) == ([{"ratingKey": "1"}], None)
 
 
 def test_related_flattens_the_hubs_and_degrades_to_empty(plex):

backend/tests/test_plex_watch_reconcile.py

@@ -157,6 +157,51 @@ def test_a_genuinely_empty_second_section_still_lets_the_reconcile_clear(db, mon
     assert out["cleared"] == 1 and "incomplete" not in out
 
 
+def test_a_section_we_mirror_that_says_0_is_not_believed(db, monkeypatch, world):
+    # A total of 0 from the very section our items come from is a Plex that lost the library for a
+    # moment, not one that un-watched all of it (E32 review round 2): the shows section below says
+    # 0 and lists nothing — while we hold an episode from it.
+    state, rows = world
+    _local_states(state)
+    shows = PlexLibrary(plex_key="2", title="Shows", kind="show", enabled=True)
+    db.add(shows)
+    db.flush()
+    db.add(PlexItem(rating_key="e1", library_id=shows.id, kind="episode", title="e1"))
+    db.commit()
+    out = _run(db, monkeypatch, _FakePlex(_LISTED, more=[("show", [], 0)]))
+    assert rows()["m1"] == ("watched", 0, True)
+    assert out["cleared"] == 0 and out["incomplete"] == 1
+
+
+def test_a_library_the_allowlist_leaves_out_keeps_its_watch_states(db, monkeypatch, world):
+    # Its items are absent from every scan because the scan never walks it — not because Plex
+    # un-watched them. It cleared them on every run (E32 review round 2).
+    state, _rows = world
+    other = PlexLibrary(plex_key="2", title="Shows", kind="show", enabled=True)
+    db.add(other)
+    db.flush()
+    ep = PlexItem(rating_key="e1", library_id=other.id, kind="episode", title="e1")
+    db.add(ep)
+    db.flush()
+    user_id = db.execute(select(PlexLink.user_id)).scalar_one()
+    db.add(PlexState(user_id=user_id, item_id=ep.id, status="watched", synced_to_plex=True))
+    _local_states(state)
+    sysconfig.set_value(db, "plex_libraries", "1")
+    db.commit()
+    out = _run(db, monkeypatch, _FakePlex(_LISTED, more=[("show", [], 0)]))
+    db.expire_all()
+    left = {
+        rk
+        for rk in db.execute(
+            select(PlexItem.rating_key).join(PlexState, PlexState.item_id == PlexItem.id)
+        ).scalars()
+    }
+    assert "m1" not in left  # the walked section still clears
+    kept = db.execute(select(PlexState).where(PlexState.item_id == ep.id)).scalar_one_or_none()
+    assert kept is not None and kept.status == "watched"
+    assert out["cleared"] == 1
+
+
 def test_a_cut_short_section_is_not_rescued_by_a_complete_one_after_it(db, monkeypatch, world):
     # Completeness is every section's, not the last one's: movies cut short (5 said, 3 handed),
     # then a complete shows listing — m1 may simply not have been reached.