Post-download scan: wait until Plex's scan queue is actually idle, not a fixed 2min

A fixed debounce can't fit a big library — 8500 movies + 4500 shows scan sequentially
through Plex's queue and can take 10-20 min, so the old 120s wait read the DB before
Plex finished and fresh downloads showed up late. Now Stage 1 (video_scan_server)
fires the rescan then POLLS the server until its scan queue goes idle, then emits the
done event.

- sources: PlexVideoSource.is_scanning (section.refreshing + activity feed, scoped by
  media_type) and JellyfinVideoSource.is_scanning (scheduled-task state), plus
  video_server_scan_in_progress() returning True/False/None.
- handler: pure wait_for_server_scan(scan_status, sleep, …) — grace, then poll every
  interval until idle or a generous cap; falls back to the fixed wait only when the
  server can't report status (None). debounce_seconds is now that fallback; new
  max_wait_minutes caps the poll.

Seam tests for the poll logic (idle/poll/fallback/cap/lost-status), the handler wiring,
and Plex scan-status detection.

core/automation/blocks.py

@@ -228,14 +228,15 @@ ACTIONS: list[dict] = [
     # Post-download chain actions (two stages, like music's scan_library +
     # start_database_update). Stage 1 nudges the server; stage 2 reads it in.
     {"type": "video_scan_server", "label": "Scan Video Server", "icon": "refresh", "scope": "video",
-     "description": "Tell the media server to rescan your video sections, then fire 'Video Library Scan Done'", "available": True,
+     "description": "Tell the media server to rescan your video sections, wait until it actually finishes indexing, then fire 'Video Library Scan Done'", "available": True,
      "config_fields": [
          {"key": "media_type", "type": "select", "label": "Library",
           "options": [{"value": "all", "label": "Movies + TV"},
                       {"value": "movie", "label": "Movies only"},
                       {"value": "show", "label": "TV only"}],
           "default": "all"},
-         {"key": "debounce_seconds", "type": "number", "label": "Wait for indexing (sec)", "default": 120, "min": 10}
+         {"key": "max_wait_minutes", "type": "number", "label": "Max wait for scan (min)", "default": 60, "min": 1},
+         {"key": "debounce_seconds", "type": "number", "label": "Fallback wait if status unknown (sec)", "default": 120, "min": 10}
      ]},
     {"type": "video_update_database", "label": "Update Video Database", "icon": "database", "scope": "video",
      "description": "Read newly-indexed media from the server into SoulSync (incremental)", "available": True,

core/automation/handlers/video_scan_library.py

@@ -174,31 +174,81 @@ def auto_video_scan_library(
 
 # ── post-download chain (video twin of music's scan_library → start_database_update) ──
 # Split into two stages so the calendar/library reflect a download promptly, mirroring
-# the music side: tell the server to rescan, then (after it has had time to index) read
-# the new state into video.db. Stage 1 emits 'video_library_scan_completed' on a debounce
-# (like music's web_scan_manager time-based completion); stage 2 listens for that.
+# the music side: tell the server to rescan, WAIT until its scan queue is actually idle
+# (a big library can take 10-20 min — a fixed wait would read too early), then fire
+# 'video_library_scan_completed'; stage 2 listens for that and reads the fresh state.
+
+def _default_scan_status(media_type: str = "all"):
+    """Production wiring: True/False if the active server is/ isn't scanning, or None
+    when it can't report (so the caller falls back to a fixed wait)."""
+    from core.video.sources import video_server_scan_in_progress
+    return video_server_scan_in_progress(media_type)
+
+
+def wait_for_server_scan(scan_status, sleep, *, grace_seconds: int = 15,
+                         interval_seconds: int = 10, cap_seconds: int = 3600,
+                         fallback_seconds: int = 120) -> int:
+    """Block until the media server's scan queue goes idle, then return ~seconds waited.
+
+    ``scan_status()`` returns True (scanning), False (idle) or None (can't tell). After
+    a short grace — so the just-triggered scan has time to register as running — poll
+    every ``interval_seconds`` until idle or ``cap_seconds`` (a generous backstop for a
+    huge library). If the server can't report its state, fall back to a fixed
+    ``fallback_seconds`` wait — exactly the old behaviour. ``sleep`` is injected and
+    elapsed is counted from the sleeps, so tests never actually block."""
+    grace = max(0, grace_seconds)
+    if grace:
+        sleep(grace)
+    waited = grace
+    try:
+        status = scan_status()
+    except Exception:   # noqa: BLE001 - status is best-effort
+        status = None
+    if status is None:
+        extra = max(0, fallback_seconds - waited)
+        if extra:
+            sleep(extra)
+        return waited + extra
+    while status is True and waited < cap_seconds:
+        sleep(interval_seconds)
+        waited += interval_seconds
+        try:
+            status = scan_status()
+        except Exception:   # noqa: BLE001
+            status = None
+        if status is None:   # lost the ability to tell — stop waiting, don't hang
+            break
+    return waited
+
 
 def auto_video_scan_server(
     config: Dict[str, Any],
     deps: AutomationDeps,
     *,
-    server_refresh: Optional[Callable[[], Dict[str, Any]]] = None,
+    server_refresh: Optional[Callable[..., Dict[str, Any]]] = None,
     sleep: Optional[Callable[[float], None]] = None,
     emit: Optional[Callable[[str, Dict[str, Any]], None]] = None,
+    scan_status: Optional[Callable[..., Any]] = None,
 ) -> Dict[str, Any]:
-    """Stage 1: tell the media server to rescan the video sections, wait a debounce
-    for it to index, then fire 'video_library_scan_completed' so the DB-update twin
-    reads the fresh state. (Video twin of music's scan_library.)"""
+    """Stage 1: tell the media server to rescan the video sections, WAIT until its scan
+    queue is idle (polling the server; falls back to a fixed wait if it can't report),
+    then fire 'video_library_scan_completed' so the DB-update twin reads the fresh state.
+    (Video twin of music's scan_library.)"""
     server_refresh = server_refresh or _default_server_refresh
     sleep = sleep or time.sleep
     emit = emit or deps.engine.emit
+    scan_status = scan_status or _default_scan_status
     automation_id = config.get('_automation_id')
     media_type = config.get('media_type') or 'all'
     lib_label = {'movie': 'Movie', 'show': 'TV'}.get(media_type, 'video')
     try:
-        debounce = int(config.get('debounce_seconds') or 120)
+        fallback = int(config.get('debounce_seconds') or 120)
     except (TypeError, ValueError):
-        debounce = 120
+        fallback = 120
+    try:
+        cap = int(config.get('max_wait_minutes') or 60) * 60
+    except (TypeError, ValueError):
+        cap = 3600
     try:
         deps.update_progress(automation_id, phase=f'Asking media server to rescan {lib_label}…', progress=20,
                              log_line=f'Triggering server-side {lib_label} scan', log_type='info')
@@ -207,13 +257,15 @@ def auto_video_scan_server(
             deps.update_progress(automation_id,
                                  log_line='Server scan trigger unavailable: ' + str(refresh.get('error') or 'unknown'),
                                  log_type='warning')
-        deps.update_progress(automation_id, phase=f'Waiting for the server to index ({debounce}s)…', progress=55)
-        sleep(debounce)
+        deps.update_progress(automation_id, phase='Waiting for the server to finish indexing…', progress=55)
+        waited = wait_for_server_scan(lambda: scan_status(media_type), sleep,
+                                      fallback_seconds=fallback, cap_seconds=cap)
         # Carry the scope so stage 2 (Update DB) reads only the library we rescanned.
         emit('video_library_scan_completed',
              {'server': refresh.get('server') or '', 'media_type': media_type})
         deps.update_progress(automation_id, status='finished', progress=100, phase='Complete',
-                             log_line='Server scan done — updating the database', log_type='success')
+                             log_line=f'Server scan finished (~{waited}s) — updating the database',
+                             log_type='success')
         return {'status': 'completed', '_manages_own_progress': True}
     except Exception as e:  # noqa: BLE001
         deps.update_progress(automation_id, status='error', phase='Error', log_line=str(e), log_type='error')

core/video/sources.py

@@ -296,6 +296,21 @@ def refresh_video_server_sections(media_type="all"):
         return {"ok": False, "error": str(e)}
 
 
+def video_server_scan_in_progress(media_type="all"):
+    """True if the active video server is mid-scan for the given library (or either,
+    for 'all'); False if idle; None if it can't be determined — no server, or an
+    adapter that can't report scan state. Callers fall back to a fixed wait on None."""
+    media_type = normalize_media_type(media_type)
+    src = get_active_video_source()
+    if src is None or not hasattr(src, "is_scanning"):
+        return None
+    try:
+        return bool(src.is_scanning(media_type))
+    except Exception:
+        logger.debug("video sources: scan-status check failed", exc_info=True)
+        return None
+
+
 def list_video_libraries():
     """Discover the active server's video libraries for the mapping UI:
     {'server', 'movies': [{'title'}], 'tv': [{'title'}]} or None."""
@@ -376,6 +391,26 @@ class PlexVideoSource:
                     logger.exception("Plex: refresh failed for section %s", getattr(s, "title", "?"))
         return {"ok": n > 0, "sections": n}
 
+    def is_scanning(self, media_type="all") -> bool:
+        """True if any SELECTED video section (scoped by media_type) is currently
+        being scanned by Plex. Checks the per-section refreshing flag, then the
+        server activity feed (real-time) — mirrors the music PlexClient check."""
+        sections = []
+        for kind, name in (("movie", self._movies_lib), ("show", self._tv_lib)):
+            if media_type != "all" and media_type != kind:
+                continue
+            sections.extend(self._scan_sections(kind, name))
+        for s in sections:
+            if getattr(s, "refreshing", False):
+                return True
+        titles = {(getattr(s, "title", "") or "").lower() for s in sections}
+        for act in self._server.activities():
+            if getattr(act, "type", "") in ("library.scan", "library.refresh"):
+                at = (getattr(act, "title", "") or "").lower()
+                if any(t and t in at for t in titles):
+                    return True
+        return False
+
     @staticmethod
     def _part_file(obj):
         try:
@@ -571,6 +606,26 @@ class JellyfinVideoSource:
                 logger.exception("Jellyfin: refresh failed for view %s", vid)
         return {"ok": n > 0, "sections": n}
 
+    def is_scanning(self, media_type="all") -> bool:
+        """True if Jellyfin's library-scan scheduled task is running. Jellyfin's
+        scan isn't per-library, so media_type is ignored (any scan counts)."""
+        import requests
+        base = (self._c.base_url or "").rstrip("/")
+        if not base:
+            return False
+        try:
+            r = requests.get(base + "/ScheduledTasks",
+                             headers={"X-Emby-Token": self._c.api_key or ""}, timeout=10)
+            tasks = r.json() if r.ok else []
+        except Exception:
+            return False
+        for task in tasks or []:
+            name = (task.get("Name") or "").lower()
+            if (("scan" in name or "refresh" in name or "library" in name)
+                    and task.get("State") in ("Running", "Cancelling")):
+                return True
+        return False
+
     def counts(self, incremental=False) -> dict:
         def total(view, itype):
             resp = self._req(f"/Users/{self.uid}/Items", {

tests/automation/test_handler_video_scan_library.py

@@ -229,29 +229,74 @@ class TestHandlerNeverRaises:
 
 # ── post-download chain: video_scan_server (stage 1) ───────────────────────
 from core.automation.handlers.video_scan_library import (  # noqa: E402
-    auto_video_scan_server, auto_video_update_database)
+    auto_video_scan_server, auto_video_update_database, wait_for_server_scan)
+
+
+class TestWaitForServerScan:
+    """Poll the server until its scan queue is idle; fixed wait only as a fallback."""
+
+    def test_returns_quickly_when_already_idle(self):
+        slept = []
+        waited = wait_for_server_scan(lambda: False, slept.append, grace_seconds=15)
+        assert waited == 15 and slept == [15]                 # just the grace, then idle
+
+    def test_polls_until_the_scan_finishes(self):
+        # scanning for three polls, then idle — handles a 10-20 min big-library scan
+        seq = iter([True, True, True, False])
+        slept = []
+        waited = wait_for_server_scan(lambda: next(seq), slept.append,
+                                      grace_seconds=15, interval_seconds=10)
+        assert waited == 15 + 30                               # grace + 3 polls
+        assert slept == [15, 10, 10, 10]
+
+    def test_falls_back_to_fixed_wait_when_status_unknown(self):
+        slept = []
+        waited = wait_for_server_scan(lambda: None, slept.append,
+                                      grace_seconds=15, fallback_seconds=120)
+        assert waited == 120 and slept == [15, 105]           # grace + (fallback - grace)
+
+    def test_stops_at_the_cap_on_a_runaway_scan(self):
+        slept = []
+        waited = wait_for_server_scan(lambda: True, slept.append,
+                                      grace_seconds=0, interval_seconds=10, cap_seconds=50)
+        assert waited == 50                                   # never hangs forever
+
+    def test_stops_if_status_becomes_unknown_mid_poll(self):
+        seq = iter([True, None])
+        slept = []
+        wait_for_server_scan(lambda: next(seq), slept.append, grace_seconds=0, interval_seconds=10)
+        assert slept == [10]                                  # one poll, then bail — no hang
 
 
 class TestScanServerStage:
-    def test_refreshes_waits_then_emits_scan_done(self):
+    def test_waits_for_idle_then_emits_scan_done(self):
         deps = _RecordingDeps()
         events = []
-        slept = []
+        # scanning once, then idle
+        seq = iter([True, False])
         r = auto_video_scan_server(
-            {'_automation_id': 'a', 'debounce_seconds': 90}, deps,
-            server_refresh=_refresh_ok(), sleep=lambda s: slept.append(s),
+            {'_automation_id': 'a'}, deps,
+            server_refresh=_refresh_ok(), sleep=lambda s: None,
+            scan_status=lambda mt: next(seq),
             emit=lambda ev, data: events.append((ev, data)))
         assert r['status'] == 'completed'
-        assert slept == [90]                                  # waited the debounce
         assert events == [('video_library_scan_completed', {'server': '', 'media_type': 'all'})]
 
-    def test_default_debounce_and_server_unavailable_still_emits(self):
+    def test_falls_back_to_fixed_wait_when_server_cant_report(self):
+        slept = []
+        auto_video_scan_server(
+            {'_automation_id': 'a', 'debounce_seconds': 90}, _RecordingDeps(),
+            server_refresh=_refresh_ok(), sleep=slept.append,
+            scan_status=lambda mt: None, emit=lambda ev, data: None)
+        assert sum(slept) == 90                               # honoured the fallback wait
+
+    def test_server_unavailable_still_emits(self):
         deps = _RecordingDeps()
         events = []
         auto_video_scan_server(
             {'_automation_id': 'a'}, deps,
             server_refresh=lambda media_type=None: {'ok': False, 'error': 'no server'},
-            sleep=lambda s: None, emit=lambda ev, data: events.append(ev))
+            sleep=lambda s: None, scan_status=lambda mt: False, emit=lambda ev, data: events.append(ev))
         assert events == ['video_library_scan_completed']     # fires even if refresh failed
         assert 'warning' in deps.log_types()
 
@@ -260,22 +305,24 @@ class TestScanServerStage:
         r = auto_video_scan_server(
             {'_automation_id': 'a'}, deps,
             server_refresh=lambda media_type=None: (_ for _ in ()).throw(RuntimeError('boom')),
-            sleep=lambda s: None, emit=lambda *a: None)
+            sleep=lambda s: None, scan_status=lambda mt: False, emit=lambda *a: None)
         assert r['status'] == 'error' and r['error'] == 'boom'
 
-    def test_scopes_refresh_and_carries_media_type_on_the_event(self):
+    def test_scopes_refresh_and_status_and_carries_media_type_on_the_event(self):
         seen = {}
         events = []
 
         def _refresh(media_type=None):
-            seen['mt'] = media_type
+            seen['refresh_mt'] = media_type
             return {'ok': True}
 
         auto_video_scan_server(
             {'_automation_id': 'a', 'media_type': 'show'}, _RecordingDeps(),
             server_refresh=_refresh, sleep=lambda s: None,
+            scan_status=lambda mt: seen.setdefault('status_mt', mt) and False,
             emit=lambda ev, data: events.append((ev, data)))
-        assert seen['mt'] == 'show'                                   # only TV sections nudged
+        assert seen['refresh_mt'] == 'show'                           # only TV sections nudged
+        assert seen['status_mt'] == 'show'                            # polled TV scan status
         assert events[0][1]['media_type'] == 'show'                   # stage 2 inherits the scope
 
 

tests/test_video_scan_scope.py

@@ -126,3 +126,42 @@ def test_refresh_sections_scopes_by_media_type():
 
     res = src.refresh_sections("all")
     assert secs[0].updated and secs[1].updated and res["sections"] == 2   # both
+
+
+# ── scan-status detection (poll-until-idle uses this) ───────────────────────
+
+def test_plex_is_scanning_reads_section_refreshing_flag_scoped():
+    class Sec:
+        def __init__(self, type_, title, refreshing=False):
+            self.type, self.title, self.refreshing = type_, title, refreshing
+
+    class Srv:
+        def __init__(self, secs): self.library = _Lib(secs)
+        def activities(self): return []
+
+    # TV section refreshing, movie idle
+    secs = [Sec("movie", "Movies", False), Sec("show", "TV Shows", True)]
+    src = PlexVideoSource(Srv(secs), movies_lib="Movies", tv_lib="TV Shows")
+    assert src.is_scanning("show") is True        # TV is mid-scan
+    assert src.is_scanning("movie") is False       # movie idle → not scanning for that scope
+    assert src.is_scanning("all") is True          # either counts
+
+
+def test_plex_is_scanning_falls_back_to_activity_feed():
+    class Sec:
+        def __init__(self, type_, title): self.type, self.title, self.refreshing = type_, title, False
+    class Act:
+        def __init__(self, type_, title): self.type, self.title = type_, title
+    class Srv:
+        def __init__(self, secs, acts): self.library, self._acts = _Lib(secs), acts
+        def activities(self): return self._acts
+    secs = [Sec("movie", "Movies")]
+    src = PlexVideoSource(Srv(secs, [Act("library.refresh", "Scanning Movies…")]),
+                          movies_lib="Movies", tv_lib="TV Shows")
+    assert src.is_scanning("movie") is True        # no refreshing flag, but the feed shows a scan
+
+
+def test_scan_status_helper_is_none_when_no_server(monkeypatch):
+    import core.video.sources as srcmod
+    monkeypatch.setattr(srcmod, "get_active_video_source", lambda: None)
+    assert srcmod.video_server_scan_in_progress("all") is None   # caller falls back to fixed wait