fix(sync): an empty answer the remote's own count contradicts is not believed, one level down

E32 review round 1 (B1, B2, B3, B8).

- Plex watch reconcile: a section the server SAYS is empty (totalSize 0, nothing listed) is a
  complete listing. It read as "cut short" forever, so one empty library stopped un-watches from
  ever syncing. A listing with no total at all stays incomplete (the client now returns None for
  it instead of 0), and a whole scan that listed nothing still clears nothing. Tests now cover
  completeness across two sections, both orders.
- Plex collections: no members from a collection whose listing counts some (childCount) keeps
  the membership we hold, in the full sync and in the targeted resync after a write.
- YouTube playlists: no items for a playlist whose listing counts some (itemCount) keeps the
  mirror in the bulk pull, and is refused as retryable in "reset to YouTube" and in the push
  plan/merge, where it read as "YouTube deleted them all". The rule sits beside vanished().
- A playlist deleted on YouTube answered "reconnect your account"; it now says it is gone
  (errors.youtube.gone, HU+EN).
- The pure 3-way merge moved to sync/playlist_merge.py; playlists.py is back under 800 lines and
  off the size list.

backend/app/sync/playlists.py

@@ -1,6 +1,7 @@
 """YouTube playlist sync, both directions. Read: mirror each user's own YouTube playlists
 into local `source='youtube'` playlists (kept fresh by the scheduler). Write-back: two-way sync
-local edits with YouTube (`plan_merge` / `merge_playlist`), behind the optional write scope.
+local edits with YouTube (`plan_merge` / `merge_playlist`), behind the optional write scope; the
+pure 3-way merge they run is `sync/playlist_merge.py`.
 
 YouTube's special Watch Later / History playlists are not exposed by the Data API and are
 never synced. Videos in a playlist that aren't in our shared catalog yet are fetched and
@@ -10,7 +11,7 @@ import hashlib
 import json
 from collections.abc import Callable, Iterator
 from contextlib import AbstractContextManager, contextmanager
-from dataclasses import dataclass
+
 from datetime import datetime, timezone
 from typing import TypeVar
 
@@ -21,8 +22,9 @@ from sqlalchemy.pool import NullPool
 from app import logs, progress, quota
 from app.auth import has_read_scope
 from app.models import Channel, Playlist, PlaylistItem, User, Video
+from app.sync.playlist_merge import merge_plan
 from app.sync.runner import token_users
-from app.sync.subscriptions import vanished
+from app.sync.subscriptions import blank_listing, refuse_blank_listing, vanished
 from app.sync.videos import apply_video_details, parse_dt
 from app.youtube.client import YouTubeClient, YouTubeError
 
@@ -376,6 +378,9 @@ def pull_user_playlists(db: Session, user: User) -> dict:
                     # The next scheduled run refreshes it.
                     continue
                 ids = list(yt.iter_my_playlist_video_ids(ytid))
+                if blank_listing(ids, p.get("item_count")):
+                    log.warning("YouTube listed no items for non-empty playlist %s; kept", pl.id)
+                    continue
                 # Ingest first — it commits — so the row lock below is held from the dirty check
                 # to the commit that replaces the items.
                 _ensure_videos(db, yt, ids)
@@ -407,90 +412,6 @@ def _local_video_ids(db: Session, pl: Playlist) -> list[str]:
     )
 
 
-def _reorder_moves(current: list[str], desired: list[str]) -> int:
-    """Count the playlistItems.update moves an insertion-sort would need to turn
-    `current` order into `desired` order (both already the same set of video ids).
-    Items already in place are skipped; this matches what apply_push actually executes."""
-    model = list(current)
-    moves = 0
-    for i, want in enumerate(desired):
-        if i < len(model) and model[i] == want:
-            continue
-        j = model.index(want, i)
-        model.pop(j)
-        model.insert(i, want)
-        moves += 1
-    return moves
-
-
-@dataclass
-class MergePlan:
-    """The 3-way merge of a linked playlist: what each side gains and loses so neither side
-    loses an item, plus the merged order. `final` is the id set both sides end up with."""
-
-    final: list[str]  # merged order (local order, YouTube-only adds appended)
-    add_to_yt: list[str]  # local-adds to push out to YouTube
-    add_to_local: list[str]  # YouTube-adds to pull into Siftlode
-    remove_from_yt: list[str]  # items deleted locally that are still on YouTube
-    remove_from_local: list[str]  # items deleted on YouTube that are still local
-    reorders: int  # estimated YouTube playlistItems.update moves to reach `final`
-    baseline_known: bool
-
-
-def merge_plan(
-    baseline: list[str] | None,
-    local: list[str],
-    youtube: list[str],
-    unavailable: set[str] | None = None,
-) -> MergePlan:
-    """Pure 3-way merge. Classify each id against the last-synced `baseline`:
-    - in local but not youtube, not deleted → add to YouTube (a local-add)
-    - in youtube but not local, not deleted → add to Siftlode (a YouTube-add)
-    - in baseline but gone from local → a genuine local delete → propagate to YouTube
-    - in baseline but gone from youtube → a genuine YouTube delete → propagate to Siftlode
-    With NO baseline (first-ever merge) we can't tell an add from a delete, so we UNION both
-    sides (never delete) — the lossless default. Merged order: local order first, YouTube-only
-    additions appended in their YouTube order.
-    `unavailable` = known un-ingestable-but-live YouTube ids: dropped from the YouTube side up
-    front so they're never re-classified as a YouTube-add (they can't ingest) nor as a local
-    delete (they were never local) — otherwise such a video would make the playlist never
-    converge. They stay in place on YouTube, untouched by this merge."""
-    un = set(unavailable or ())
-    L = list(dict.fromkeys(local))
-    Y = [v for v in dict.fromkeys(youtube) if v not in un]
-    Lset, Yset = set(L), set(Y)
-    if baseline is None:
-        add_to_yt = [v for v in L if v not in Yset]
-        add_to_local = [v for v in Y if v not in Lset]
-        remove_from_yt: list[str] = []
-        remove_from_local: list[str] = []
-        final = L + [v for v in Y if v not in Lset]
-    else:
-        B = set(baseline)
-        deleted_local = B - Lset  # was synced, removed on our side
-        deleted_yt = B - Yset  # was synced, removed on YouTube's side
-        gone = deleted_local | deleted_yt
-        remove_from_yt = [v for v in Y if v in deleted_local]
-        remove_from_local = [v for v in L if v in deleted_yt]
-        add_to_yt = [v for v in L if v not in Yset and v not in gone]
-        add_to_local = [v for v in Y if v not in Lset and v not in gone]
-        final = [v for v in L if v not in gone] + [v for v in Y if v not in Lset and v not in gone]
-    final_set = set(final)
-    # On YouTube, after membership reconciliation the order is current-kept + inserts appended;
-    # the reorder pass then moves items into `final`.
-    after_membership = [v for v in Y if v in final_set] + add_to_yt
-    reorders = _reorder_moves(after_membership, final)
-    return MergePlan(
-        final=final,
-        add_to_yt=add_to_yt,
-        add_to_local=add_to_local,
-        remove_from_yt=remove_from_yt,
-        remove_from_local=remove_from_local,
-        reorders=reorders,
-        baseline_known=baseline is not None,
-    )
-
-
 def plan_merge(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
     """Compute (without writing) what a two-way sync of this playlist would do: create vs
     update, how many items each side gains/loses, reorders, and the quota estimate (YouTube
@@ -518,6 +439,7 @@ def plan_merge(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
             snippet = yt.get_playlist_snippet(pl.yt_playlist_id)
     rename = bool(snippet) and (snippet.get("title") or "") != pl.name
     yt_ids = [it["video_id"] for it in current]
+    refuse_blank_listing(yt_ids, (snippet or {}).get("item_count"))
     mp = merge_plan(parse_baseline(pl), local, yt_ids, parse_unavailable(pl))
     ops = len(mp.add_to_yt) + len(mp.remove_from_yt) + mp.reorders + (1 if rename else 0)
     return {
@@ -664,6 +586,8 @@ def merge_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
             current = list(yt.iter_playlist_items_with_ids(pl.yt_playlist_id))
         item_id_by_vid = {it["video_id"]: it["item_id"] for it in current}
         yt_ids = [it["video_id"] for it in current]
+        # A merge would read "YouTube lists none of them" as "YouTube deleted them all".
+        refuse_blank_listing(yt_ids, (snippet or {}).get("item_count"))
         # Re-attempt ingestion of ids previously marked un-ingestable that are still on YouTube: a
         # video that has since become available (e.g. private→public) should re-enter the merge, not
         # stay skipped forever. Those that still won't ingest remain filtered (`effective_unavailable`).
@@ -802,8 +726,11 @@ def repull_playlist(db: Session, user: User, pl: Playlist) -> dict:
     with YouTubeClient(db, user) as yt:
         snippet = yt.get_playlist_snippet(pl.yt_playlist_id)
         if snippet is None:
-            raise YouTubeError("Playlist no longer exists on YouTube")
+            raise YouTubeError(
+                "Playlist no longer exists on YouTube", reason="playlistNotFound", path="playlists"
+            )
         ids = list(yt.iter_my_playlist_video_ids(pl.yt_playlist_id))
+        refuse_blank_listing(ids, snippet.get("item_count"))
         _ensure_videos(db, yt, ids)
         _lock_or_gone(db, pl)
         pl.name = snippet.get("title") or pl.name