video scan: harden library scope + pause ALL enrichers during any scan

1) Scan only the MAPPED libraries — never fall back to 'all'. The scan path used
   _sections/_views with the selected name, but an empty name returned ALL sections
   of that type — so a missing/unreadable selection silently scanned every library
   (how the 4K movie + 'YouTube' TV libraries leaked in as movies/shows). New
   _scan_sections / _scan_views return [] when a kind isn't mapped; available_libraries
   still lists all (for the Settings dropdown). Now an unmapped kind scans NOTHING.

2) A library scan (full/incremental/deep) now pauses EVERY enricher, including the
   YouTube date enricher — a separate singleton outside engine.workers that kept
   running through scans. pause_for_scan/resume_after_scan pause+resume it too, only
   if it wasn't already manually paused (never override the user).

kettui: source-scope tests (mapped-only / unmapped-scans-nothing / listing still
shows all) + engine pause tests (pauses+resumes YT / preserves manual pause).
123 scanner/source/enrichment tests + isolation guard green.

core/video/enrichment/engine.py

@@ -105,6 +105,19 @@ class VideoEnrichmentEngine:
             if not w.paused:
                 w.pause(persist=False)
                 self._scan_paused.add(service)
+        # The YouTube date enricher is a separate singleton (not in self.workers) —
+        # pause it too so a scan stops EVERY enricher, not just the matcher/backfill
+        # workers. Only if it wasn't already paused (never override a manual pause).
+        self._scan_paused_yt = False
+        try:
+            from core.video.youtube_enrichment import get_youtube_date_enricher
+            yt = get_youtube_date_enricher()
+            if yt and not getattr(yt, "_paused", False):
+                yt.pause()
+                self._scan_paused_yt = True
+                self._scan_paused.add("youtube")
+        except Exception:
+            logger.debug("video enrichment: could not pause YouTube date enricher for scan", exc_info=True)
         if self._scan_paused:
             logger.info("video enrichment: paused %s for library scan",
                         ", ".join(sorted(self._scan_paused)))
@@ -115,6 +128,15 @@ class VideoEnrichmentEngine:
             w = self.workers.get(service)
             if w:
                 w.resume(persist=False)
+        if getattr(self, "_scan_paused_yt", False):
+            try:
+                from core.video.youtube_enrichment import get_youtube_date_enricher
+                yt = get_youtube_date_enricher()
+                if yt:
+                    yt.resume()
+            except Exception:
+                logger.debug("video enrichment: could not resume YouTube date enricher", exc_info=True)
+            self._scan_paused_yt = False
         if getattr(self, "_scan_paused", None):
             logger.info("video enrichment: resumed %s after library scan",
                         ", ".join(sorted(self._scan_paused)))

core/video/sources.py

@@ -308,6 +308,12 @@ class PlexVideoSource:
             secs = [s for s in secs if s.title == name]
         return secs
 
+    def _scan_sections(self, kind: str, name):
+        """Sections to SCAN for a kind. UNLIKE _sections, an empty name means
+        'this kind isn't mapped' → scan NOTHING (never fall back to all sections).
+        Prevents a missing selection from silently pulling every library."""
+        return self._sections(kind, name) if name else []
+
     def available_libraries(self) -> dict:
         return {
             "movies": [{"title": s.title} for s in self._sections("movie")],
@@ -316,14 +322,14 @@ class PlexVideoSource:
 
     def counts(self, incremental=False) -> dict:
         """Cheap item totals (no full fetch) for the progress bar."""
-        m = sum(int(getattr(s, "totalSize", 0) or 0) for s in self._sections("movie", self._movies_lib))
+        m = sum(int(getattr(s, "totalSize", 0) or 0) for s in self._scan_sections("movie", self._movies_lib))
-        sh = sum(int(getattr(s, "totalSize", 0) or 0) for s in self._sections("show", self._tv_lib))
+        sh = sum(int(getattr(s, "totalSize", 0) or 0) for s in self._scan_sections("show", self._tv_lib))
         if incremental:
             m, sh = min(m, 100), min(sh, 50)
         return {"movies": m, "shows": sh}
 
     def iter_movies(self, incremental=False):
-        for section in self._sections("movie", self._movies_lib):
+        for section in self._scan_sections("movie", self._movies_lib):
             items = section.search(sort="addedAt:desc", maxresults=100) if incremental else section.all()
             for m in items:
                 try:
@@ -332,7 +338,7 @@ class PlexVideoSource:
                     logger.exception("Plex: skipping movie %s", getattr(m, "title", "?"))
 
     def iter_shows(self, incremental=False):
-        for section in self._sections("show", self._tv_lib):
+        for section in self._scan_sections("show", self._tv_lib):
             items = section.search(sort="addedAt:desc", maxresults=50) if incremental else section.all()
             for sh in items:
                 try:
@@ -345,7 +351,7 @@ class PlexVideoSource:
         get indexed. (plexapi ``section.update()`` triggers the library scan.)"""
         n = 0
         for kind, name in (("movie", self._movies_lib), ("show", self._tv_lib)):
-            for s in self._sections(kind, name):
+            for s in self._scan_sections(kind, name):
                 try:
                     s.update()
                     n += 1
@@ -508,6 +514,12 @@ class JellyfinVideoSource:
             views = [v for v in views if v.get("Name") == name]
         return views
 
+    def _scan_views(self, collection_type: str, name):
+        """Views to SCAN. An empty name means this kind isn't mapped → scan NOTHING
+        (never fall back to all views), so a missing selection can't pull every
+        library. (available_libraries still lists all via _views.)"""
+        return self._views(collection_type, name) if name else []
+
     def available_libraries(self) -> dict:
         return {
             "movies": [{"title": v.get("Name")} for v in self._views("movies")],
@@ -522,7 +534,7 @@ class JellyfinVideoSource:
         if not base:
             return {"ok": False, "sections": 0}
         headers = {"X-Emby-Token": self._c.api_key or ""}
-        views = list(self._views("movies", self._movies_lib)) + list(self._views("tvshows", self._tv_lib))
+        views = list(self._scan_views("movies", self._movies_lib)) + list(self._scan_views("tvshows", self._tv_lib))
         n = 0
         for v in views:
             vid = v.get("Id")
@@ -543,8 +555,8 @@ class JellyfinVideoSource:
                 "ParentId": view["Id"], "IncludeItemTypes": itype,
                 "Recursive": "true", "Limit": "0"}) or {}
             return int(resp.get("TotalRecordCount", 0) or 0)
-        m = sum(total(v, "Movie") for v in self._views("movies", self._movies_lib))
+        m = sum(total(v, "Movie") for v in self._scan_views("movies", self._movies_lib))
-        sh = sum(total(v, "Series") for v in self._views("tvshows", self._tv_lib))
+        sh = sum(total(v, "Series") for v in self._scan_views("tvshows", self._tv_lib))
         if incremental:
             m, sh = min(m, 100), min(sh, 50)
         return {"movies": m, "shows": sh}
@@ -589,7 +601,7 @@ class JellyfinVideoSource:
 
     def iter_movies(self, incremental=False):
         path = f"/Users/{self.uid}/Items"
-        for view in self._views("movies", self._movies_lib):
+        for view in self._scan_views("movies", self._movies_lib):
             params = {"ParentId": view["Id"], "IncludeItemTypes": "Movie",
                       "Recursive": "true", "Fields": _JF_MOVIE_FIELDS}
             if incremental:
@@ -631,7 +643,7 @@ class JellyfinVideoSource:
 
     def iter_shows(self, incremental=False):
         path = f"/Users/{self.uid}/Items"
-        for view in self._views("tvshows", self._tv_lib):
+        for view in self._scan_views("tvshows", self._tv_lib):
             params = {"ParentId": view["Id"], "IncludeItemTypes": "Series",
                       "Recursive": "true", "Fields": _JF_SHOW_FIELDS}
             if incremental:

tests/test_video_scan_scope.py

@@ -0,0 +1,103 @@
+"""Video scan SCOPE + worker-pause coupling.
+
+1. The scan must read ONLY the user-mapped Movies/TV libraries — and when a kind
+   isn't mapped, scan NOTHING (never silently fall back to all libraries, which is
+   how YouTube/4K libraries leaked in).
+2. A library scan (any mode) pauses EVERY enricher — including the YouTube date
+   enricher, which is a separate singleton outside engine.workers.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from core.video.sources import PlexVideoSource
+from database.video_database import VideoDatabase
+from core.video.enrichment.engine import VideoEnrichmentEngine
+import core.video.youtube_enrichment as yt_mod
+
+
+# ── 1. scan scope ──────────────────────────────────────────────────────────
+
+class _Sec:
+    def __init__(self, type_, title):
+        self.type = type_
+        self.title = title
+
+
+class _Lib:
+    def __init__(self, secs):
+        self._secs = secs
+
+    def sections(self):
+        return self._secs
+
+
+class _Server:
+    def __init__(self, secs):
+        self.library = _Lib(secs)
+
+
+_SECTIONS = [_Sec("movie", "Movies"), _Sec("movie", "4K Movies"),
+             _Sec("show", "TV Shows"), _Sec("show", "YouTube")]
+
+
+def test_scan_uses_only_the_mapped_libraries():
+    src = PlexVideoSource(_Server(_SECTIONS), movies_lib="Movies", tv_lib="TV Shows")
+    assert [s.title for s in src._scan_sections("movie", src._movies_lib)] == ["Movies"]
+    assert [s.title for s in src._scan_sections("show", src._tv_lib)] == ["TV Shows"]
+    # the 4K + YouTube libraries are NOT scanned…
+    titles = [s.title for s in src._scan_sections("movie", src._movies_lib)] + \
+             [s.title for s in src._scan_sections("show", src._tv_lib)]
+    assert "4K Movies" not in titles and "YouTube" not in titles
+    # …but available_libraries still LISTS them all (for the Settings dropdown).
+    avail = src.available_libraries()
+    assert {x["title"] for x in avail["movies"]} == {"Movies", "4K Movies"}
+    assert {x["title"] for x in avail["tv"]} == {"TV Shows", "YouTube"}
+
+
+def test_unmapped_kind_scans_nothing_not_everything():
+    # The hardening: a missing selection must scan NOTHING, never fall back to all.
+    src = PlexVideoSource(_Server(_SECTIONS), movies_lib=None, tv_lib=None)
+    assert src._scan_sections("movie", src._movies_lib) == []
+    assert src._scan_sections("show", src._tv_lib) == []
+    # _sections (used for LISTING) still returns all — only the SCAN path is gated.
+    assert len(src._sections("movie")) == 2
+
+
+# ── 2. scan pauses every enricher (incl. the YouTube date singleton) ───────
+
+class _FakeYT:
+    def __init__(self, paused=False):
+        self._paused = paused
+
+    def pause(self):
+        self._paused = True
+
+    def resume(self):
+        self._paused = False
+
+
+@pytest.fixture()
+def engine(tmp_path):
+    db = VideoDatabase(database_path=str(tmp_path / "video_library.db"))
+    return VideoEnrichmentEngine(db, clients={})
+
+
+def test_scan_pauses_and_resumes_youtube_date_enricher(engine, monkeypatch):
+    fake = _FakeYT()
+    monkeypatch.setattr(yt_mod, "get_youtube_date_enricher", lambda: fake)
+    engine.pause_for_scan()
+    assert fake._paused is True
+    assert "youtube" in engine._scan_paused
+    engine.resume_after_scan()
+    assert fake._paused is False
+
+
+def test_scan_never_resumes_a_manually_paused_youtube_enricher(engine, monkeypatch):
+    fake = _FakeYT(paused=True)   # the user paused it themselves
+    monkeypatch.setattr(yt_mod, "get_youtube_date_enricher", lambda: fake)
+    engine.pause_for_scan()
+    assert "youtube" not in engine._scan_paused   # we didn't touch it
+    engine.resume_after_scan()
+    assert fake._paused is True                    # still paused after the scan