fix(review): resolve deep-review findings across quickfix batch #2

- Button: `loading` always disables (|| not ??) — stops a double-submit
  when a call-site passes an explicit boolean `disabled` (VideoEditor).
- playlists: track un-ingestable-but-live YouTube ids in the sync baseline
  so a linked playlist containing a private/deleted video converges instead
  of re-classifying it as a pending pull forever; re-attempt ingestion each
  sync so a restored (private->public) video heals, and prune ids that
  leave YouTube.
- SettingsPanel: download-auth section renders an honest error/loading
  state (StateMessage + retry) instead of a bare empty section on a failed
  credentials fetch; HU+EN loadError string.
- tests: reorder-order, partial-failure baseline hold, un-ingestable
  converge/heal/prune, Button loading-disables, download-auth error/retry.

backend/app/sync/playlists.py

@@ -49,22 +49,49 @@ def _dump_ids(ids: list[str]) -> str:
     return json.dumps(ids)
 
 
-def parse_baseline(pl: Playlist) -> list[str] | None:
-    """The last-synced ordered id set, or None when no baseline has been recorded yet (a fresh
-    export, or a link created before S7). None makes the first merge union losslessly."""
+def _load_baseline(pl: Playlist) -> tuple[list[str] | None, set[str]]:
+    """Parse `synced_video_ids` into (ordered id set, un-ingestable id set). Two on-disk forms:
+    a plain JSON list (legacy / no un-ingestable ids) or `{"ids": [...], "unavailable": [...]}`.
+    `unavailable` = YouTube-only videos that exist on YouTube but the Data API won't return
+    (private/deleted/region-locked), so they never became local rows — tracked so the merge can
+    skip them instead of forever re-classifying them as a YouTube-add (they'd never converge)."""
     if not pl.synced_video_ids:
-        return None
+        return None, set()
     try:
         val = json.loads(pl.synced_video_ids)
     except (ValueError, TypeError):
-        return None
-    return [str(v) for v in val] if isinstance(val, list) else None
+        return None, set()
+    if isinstance(val, list):
+        return [str(v) for v in val], set()
+    if isinstance(val, dict):
+        raw_ids = val.get("ids")
+        ids = [str(v) for v in raw_ids] if isinstance(raw_ids, list) else None
+        raw_un = val.get("unavailable")
+        un = {str(v) for v in raw_un} if isinstance(raw_un, list) else set()
+        return ids, un
+    return None, set()
 
 
-def _set_baseline(pl: Playlist, ids: list[str]) -> None:
-    """Record `ids` as both the hash (drives `dirty`) and the raw ordered set (drives merge)."""
+def parse_baseline(pl: Playlist) -> list[str] | None:
+    """The last-synced ordered id set, or None when no baseline has been recorded yet (a fresh
+    export, or a link created before S7). None makes the first merge union losslessly."""
+    return _load_baseline(pl)[0]
+
+
+def parse_unavailable(pl: Playlist) -> set[str]:
+    """The last-synced set of un-ingestable-but-live YouTube ids (see `_load_baseline`)."""
+    return _load_baseline(pl)[1]
+
+
+def _set_baseline(pl: Playlist, ids: list[str], unavailable: list[str] | None = None) -> None:
+    """Record `ids` as both the hash (drives `dirty`, over the local ids only) and the raw ordered
+    set (drives merge). `unavailable` = un-ingestable-but-live YouTube ids to carry forward so the
+    next merge skips them; stored only when non-empty (keeps the common case a plain list)."""
     pl.synced_fingerprint = fingerprint(pl.name, ids)
-    pl.synced_video_ids = _dump_ids(ids)
+    if unavailable:
+        pl.synced_video_ids = json.dumps({"ids": ids, "unavailable": list(unavailable)})
+    else:
+        pl.synced_video_ids = _dump_ids(ids)
     pl.dirty = False
 
 
@@ -262,7 +289,12 @@ class MergePlan:
     baseline_known: bool
 
 
-def merge_plan(baseline: list[str] | None, local: list[str], youtube: list[str]) -> MergePlan:
+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)
@@ -270,9 +302,14 @@ def merge_plan(baseline: list[str] | None, local: list[str], youtube: list[str])
     - 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."""
+    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 = list(dict.fromkeys(youtube))
+    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]
@@ -333,7 +370,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]
-    mp = merge_plan(parse_baseline(pl), local, yt_ids)
+    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 {
         "action": "update",
@@ -460,7 +497,20 @@ 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]
-        mp = merge_plan(parse_baseline(pl), local, yt_ids)
+        # 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`).
+        prev_unavailable = parse_unavailable(pl)
+        still_listed = [v for v in yt_ids if v in prev_unavailable]
+        if still_listed:
+            _ensure_videos(db, yt, still_listed)
+            healed = set(
+                db.execute(select(Video.id).where(Video.id.in_(still_listed))).scalars().all()
+            )
+            effective_unavailable = prev_unavailable - healed
+        else:
+            effective_unavailable = prev_unavailable
+        mp = merge_plan(parse_baseline(pl), local, yt_ids, effective_unavailable)
 
         # --- Siftlode side (free, direct): pull YouTube-adds in, drop YouTube-deletes, land the
         # merged order authoritatively. Do this first so local reflects the merge even if a later
@@ -529,7 +579,15 @@ def merge_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
             model.pop(j)
             model.insert(i, want)
         if not failures:
-            _set_baseline(pl, final_local)
+            # Carry forward un-ingestable-but-live YouTube ids so the next merge skips them
+            # (see merge_plan): this run's failed pulls, plus prior ones still on YouTube, minus
+            # any that have since left YouTube. Recomputed each success → self-healing.
+            yt_id_set = set(yt_ids)
+            unavailable = sorted(
+                (effective_unavailable | {v for v in mp.add_to_local if v not in present})
+                & yt_id_set
+            )
+            _set_baseline(pl, final_local, unavailable)
         db.commit()
     return {
         "created": created,

backend/tests/test_playlist_merge.py

@@ -12,7 +12,14 @@ import pytest
 
 from app.models import Channel, Playlist, PlaylistItem, Video
 from app.sync import playlists as pl_mod
-from app.sync.playlists import merge_plan, merge_playlist, parse_baseline
+from app.sync.playlists import (
+    merge_plan,
+    merge_playlist,
+    parse_baseline,
+    parse_unavailable,
+    plan_merge,
+)
+from app.youtube.client import YouTubeError
 
 # --- pure merge classification (no DB) --------------------------------------------------------
 
@@ -70,10 +77,22 @@ class FakeYouTube:
     and the write ops that mutate it. Every catalog video is pre-seeded so _ensure_videos never
     needs get_videos."""
 
-    def __init__(self, *, title: str, items: list[str]):
+    def __init__(
+        self,
+        *,
+        title: str,
+        items: list[str],
+        unavailable: set[str] | None = None,
+        fail_add: set[str] | None = None,
+    ):
         self.title = title
         self._seq = 0
         self.items = [(self._new_id(), v) for v in items]
+        # Ids the Data API won't return (private/deleted) — get_videos omits them, so _ensure_videos
+        # can't ingest them. They still live in the playlist (`items`) like a real un-ingestable video.
+        self.unavailable = set(unavailable or ())
+        # Ids whose add_playlist_item raises, to exercise the partial-failure baseline-hold path.
+        self.fail_add = set(fail_add or ())
 
     def _new_id(self) -> str:
         self._seq += 1
@@ -95,10 +114,28 @@ class FakeYouTube:
     def iter_playlist_items_with_ids(self, pid):
         return [{"item_id": iid, "video_id": v} for iid, v in self.items]
 
+    def get_videos(self, ids):
+        # Mirror the Data API: a minimal ingestable record for each id, EXCEPT the un-ingestable
+        # ones (private/deleted), which are simply absent from the response.
+        return [
+            {
+                "id": v,
+                "snippet": {
+                    "channelId": "UCtest",
+                    "channelTitle": "Test",
+                    "publishedAt": "2020-01-01T00:00:00Z",
+                },
+            }
+            for v in ids
+            if v not in self.unavailable
+        ]
+
     def update_playlist_title(self, pid, title):
         self.title = title
 
     def add_playlist_item(self, pid, vid, position=None):
+        if vid in self.fail_add:
+            raise YouTubeError("simulated add failure")
         iid = self._new_id()
         self.items.append((iid, vid))
         return iid
@@ -219,3 +256,105 @@ def test_create_path_appends_and_records_baseline(db, make_user, _fake_yt):
     assert r["created"] == 1 and r["inserted"] == 2
     assert pl.yt_playlist_id  # created and linked
     assert parse_baseline(pl) == ["a", "b"]
+
+
+def test_reorder_reaches_the_merged_order_on_youtube(db, make_user, _fake_yt):
+    # Same membership on both sides, but the user reordered locally to c,a,b. The apply-side
+    # insertion-sort must drive YouTube to that exact order — asserted as an ordered list, not a
+    # set, so an off-by-one or a dropped reorder pass can't pass unnoticed.
+    user = make_user()
+    _seed_videos(db, ["a", "b", "c"])
+    pl = _linked_playlist(db, user, local_ids=["c", "a", "b"], baseline=["a", "b", "c"])
+    fake = _fake_yt(FakeYouTube(title="Mix", items=["a", "b", "c"]))
+
+    r = merge_playlist(db, user, pl)
+
+    assert r["failures"] == []
+    assert fake.video_ids == ["c", "a", "b"], "YouTube order matches the merged (local) order"
+    assert r["reordered"] >= 1
+    assert parse_baseline(pl) == ["c", "a", "b"]
+
+
+def test_partial_youtube_failure_holds_the_baseline_for_retry(db, make_user, _fake_yt):
+    # A local-add whose YouTube insert fails must NOT advance the baseline — otherwise the retry
+    # sync would read the un-pushed add as a genuine YouTube-side delete and silently drop it.
+    user = make_user()
+    _seed_videos(db, ["a", "b", "c"])
+    pl = _linked_playlist(db, user, local_ids=["a", "b", "c"], baseline=["b", "c"])
+    fake = _fake_yt(FakeYouTube(title="Mix", items=["b", "c"], fail_add={"a"}))
+
+    r = merge_playlist(db, user, pl)
+
+    assert r["failures"] == ["a"], "the failed local-add is reported"
+    assert "a" not in fake.video_ids, "the add never landed on YouTube"
+    # Baseline is UNCHANGED so the next sync re-classifies 'a' as a local-add and re-pushes it,
+    # rather than treating it as a YouTube delete and removing it locally.
+    assert parse_baseline(pl) == ["b", "c"]
+    assert "a" in set(_local_ids(db, pl)), "'a' stays local for the retry"
+
+
+def test_unavailable_youtube_video_converges(db, make_user, _fake_yt):
+    # A YouTube-only video that can't be ingested (private/deleted) must not make the playlist
+    # loop forever re-classifying it as a pending pull. It is recorded as un-ingestable and skipped
+    # by the next merge; the playlist reports "up to date".
+    user = make_user()
+    _seed_videos(db, ["a", "b"])  # note: 'x' is NOT in the catalog and can't be ingested
+    pl = _linked_playlist(db, user, local_ids=["a", "b"], baseline=["a", "b"])
+    fake = _fake_yt(FakeYouTube(title="Mix", items=["a", "b", "x"], unavailable={"x"}))
+
+    r1 = merge_playlist(db, user, pl)
+
+    assert r1["failures"] == []
+    assert r1["pulled"] == 0, "the un-ingestable video did not become a local row"
+    assert set(_local_ids(db, pl)) == {"a", "b"}
+    assert parse_unavailable(pl) == {"x"}, "'x' recorded as un-ingestable"
+    assert "x" in fake.video_ids, "'x' left untouched on YouTube"
+
+    # Second sync: the preview reports nothing pending (plan_merge filters 'x' via the stored
+    # unavailable set). The sync itself re-attempts ingestion of 'x' (the heal path), but since it
+    # still won't ingest, 'x' stays recorded as unavailable — no perpetual pending pull.
+    plan = plan_merge(db, user, pl)
+    assert plan["add_local"] == 0, "no perpetual pending pull — the playlist is up to date"
+    r2 = merge_playlist(db, user, pl)
+    assert r2["pulled"] == 0 and r2["failures"] == []
+    assert parse_unavailable(pl) == {"x"}
+    assert "x" in fake.video_ids
+
+
+def test_unavailable_video_that_becomes_available_is_pulled_in(db, make_user, _fake_yt):
+    # A video marked un-ingestable (e.g. private) that later becomes available (public) must
+    # re-enter the merge on a normal sync — not stay skipped forever.
+    user = make_user()
+    _seed_videos(db, ["a", "b"])
+    pl = _linked_playlist(db, user, local_ids=["a", "b"], baseline=["a", "b"])
+    fake = _fake_yt(FakeYouTube(title="Mix", items=["a", "b", "x"], unavailable={"x"}))
+
+    merge_playlist(db, user, pl)
+    assert parse_unavailable(pl) == {"x"}, "recorded un-ingestable on the first sync"
+
+    # 'x' becomes available; the next sync re-attempts ingestion and pulls it in.
+    fake.unavailable.clear()
+    r2 = merge_playlist(db, user, pl)
+
+    assert r2["failures"] == []
+    assert "x" in set(_local_ids(db, pl)), "the now-available video is pulled into Siftlode"
+    assert parse_unavailable(pl) == set(), "'x' is no longer marked un-ingestable"
+
+
+def test_unavailable_id_that_leaves_youtube_is_pruned(db, make_user, _fake_yt):
+    # A recorded un-ingestable id that is later removed from the YouTube playlist must be pruned
+    # from the persisted `unavailable` set (the `& yt_id_set` self-heal), not carried forever.
+    user = make_user()
+    _seed_videos(db, ["a", "b"])
+    pl = _linked_playlist(db, user, local_ids=["a", "b"], baseline=["a", "b"])
+    fake = _fake_yt(FakeYouTube(title="Mix", items=["a", "b", "x"], unavailable={"x"}))
+
+    merge_playlist(db, user, pl)
+    assert parse_unavailable(pl) == {"x"}, "recorded un-ingestable on the first sync"
+
+    # 'x' is removed from the YouTube playlist entirely; the next sync must drop it.
+    fake.items = [row for row in fake.items if row[1] != "x"]
+    r2 = merge_playlist(db, user, pl)
+
+    assert r2["failures"] == []
+    assert parse_unavailable(pl) == set(), "'x' pruned once it left YouTube"