youtube fulfillment: drain the whole wishlist, cap concurrency not total

ditch the per-run batch cap (a 200-video backlog would've taken weeks). now the
'Download YouTube Wishlist' automation queues the ENTIRE wishlist as 'queued'
rows, starts up to max_concurrent (default 3) right away, and each finished
download starts the next (one-out-one-in in the worker) so it all drains in a
controlled stream. the knob is 'max simultaneous downloads', not a total cap.

mirrors the music download worker's lesson (cap concurrency + space starts to
avoid yt-dlp 429s) but stays isolated on the video side:
- youtube_download: _pace() staggers fetch starts (3s); start_next_queued()
  claims+spawns the next; run_youtube_download chains on finish.
- db.count_active_youtube_downloads() + claim_next_youtube_queued() (atomic,
  race-safe). block field batch_size -> max_concurrent.

all seam-tested (pure select + pump); 680-test sweep green.

core/automation/blocks.py

@@ -273,9 +273,9 @@ ACTIONS: list[dict] = [
          {"key": "backfill_count", "type": "number", "label": "Always keep the last N videos from each channel", "default": 10, "min": 0}
      ]},
     {"type": "video_process_youtube_wishlist", "label": "Download YouTube Wishlist", "icon": "download", "scope": "video",
-     "description": "Grab wished YouTube videos: pushes a batch into the download queue (yt-dlp), organised as a Plex 'TV by date' show (channel/year/date). A completed download leaves the wishlist. Needs the YouTube library folder set on Settings → Downloads. Pair with a schedule to drain continuously.", "available": True,
+     "description": "Download wished YouTube videos (yt-dlp), organised as a Plex 'TV by date' show (channel/year/date). Queues the WHOLE wishlist — the setting only limits how many download at the same time; each finished one starts the next, so it all drains. A completed download leaves the wishlist. Needs the YouTube library folder set on Settings → Downloads.", "available": True,
      "config_fields": [
-         {"key": "batch_size", "type": "number", "label": "Max downloads to queue per run", "default": 3, "min": 0}
+         {"key": "max_concurrent", "type": "number", "label": "Max simultaneous downloads", "default": 3, "min": 1}
      ]},
     # Video twins of the music maintenance actions. Distinct action_type (the
     # system seeder keys on action_type, so a shared key would collide with the

core/automation/handlers/video_process_youtube_wishlist.py

@@ -1,20 +1,23 @@
 """Automation handler: ``video_process_youtube_wishlist`` action.
 
 The drain side of the YouTube fulfillment lane. The watchlist-channels scan keeps the
-wishlist fed with new uploads; THIS takes wished YouTube videos and pushes them into the
+wishlist fed; THIS queues wished videos for download and keeps a few flowing at a time.
-shared ``video_downloads`` queue, spawning the yt-dlp worker per video. The worker
-(``core.video.youtube_download``) downloads → organises (channel/year/date) → archives to
-history → removes the video from the wishlist, so a completed grab leaves the wishlist and
-won't be re-queued.
 
-Polite by default: only a small BATCH is enqueued per run (a big first-time backlog drains
+There is NO cap on how much of the wishlist gets processed — it queues the WHOLE thing.
-over several scheduled runs rather than spawning hundreds of yt-dlp processes at once), and
+The only limit is how many download SIMULTANEOUSLY (``max_concurrent``, default 3): every
-videos already in flight (an active ``source='youtube'`` download) are skipped so re-runs
+wished video becomes a ``queued`` row in the shared ``video_downloads`` table, the handler
+starts up to the limit, and each finished download starts the next (one-out-one-in, in the
+worker) so the entire queue drains in a controlled stream. This mirrors the music side's
+download worker — a concurrency cap plus a small inter-download delay (handled in the
+worker) to avoid yt-dlp 429s — but stays on the isolated video side.
+
+The worker (``core.video.youtube_download``) downloads → organises (channel/year/date) →
+archives to history → removes the video from the wishlist, so a completed grab leaves the
+wishlist and won't be re-queued; videos already queued or downloading are skipped so re-runs
 never double-grab.
 
 Shared automation side (may import ``core.video`` / ``api.video``); owns its own progress.
-All I/O is injected as seams, so the selection + batching is a pure unit-testable function;
+All I/O is injected as seams, so selection + the pump are pure unit-testable functions.
-production lazily binds the real DB + the worker spawn.
 """
 
 from __future__ import annotations
@@ -24,21 +27,23 @@ from typing import Any, Callable, Dict, Iterable, List, Optional
 from core.automation.deps import AutomationDeps
 
 
-def select_to_enqueue(wanted: List[Dict[str, Any]], active_ids: Iterable, batch_size: int) -> List[Dict[str, Any]]:
+def videos_to_enqueue(wanted: List[Dict[str, Any]], already_ids: Iterable) -> List[Dict[str, Any]]:
-    """Which wished videos to enqueue now: skip ones already in flight, cap at the batch
+    """Wished videos not already queued or downloading. NO cap — the whole backlog is
-    size (0 = no cap). Pure."""
+    queued; concurrency is bounded at start time, not here. Pure."""
-    active = {str(x) for x in (active_ids or ()) if x}
+    already = {str(x) for x in (already_ids or ()) if x}
     out: List[Dict[str, Any]] = []
     for v in wanted or []:
         vid = v.get("video_id")
-        if not vid or str(vid) in active:
+        if vid and str(vid) not in already:
-            continue
+            out.append(v)
-        out.append(v)
-        if batch_size and len(out) >= batch_size:
-            break
     return out
 
 
+def slots_free(running: int, max_concurrent: int) -> int:
+    """How many new downloads may start now given how many are already fetching. Pure."""
+    return max(0, int(max_concurrent) - max(0, int(running)))
+
+
 # ── production seams ──────────────────────────────────────────────────────────
 def _default_youtube_root() -> str:
     from api.video import get_video_db
@@ -56,27 +61,34 @@ def _default_active_ids() -> List[Any]:
             if d.get("source") == "youtube" and d.get("media_id")]
 
 
+def _default_running_count() -> int:
+    from api.video import get_video_db
+    return get_video_db().count_active_youtube_downloads()
+
+
 def _default_enqueue(video: Dict[str, Any], root: str) -> Any:
-    """Create the download row + spawn the yt-dlp worker thread. Returns the row id."""
+    """Create a QUEUED download row (no thread spawned here — the pump starts it). Returns
+    the row id."""
     import json
-    import threading
     from api.video import get_video_db
     from core.video.sources import resolve_video_server
-    from core.video.youtube_download import run_youtube_download
+    return get_video_db().add_video_download({
-    db = get_video_db()
-    dl_id = db.add_video_download({
         "kind": "youtube", "source": "youtube", "media_source": "youtube",
         "title": video.get("video_title") or video.get("channel_title"),
-        "media_id": video.get("video_id"), "target_dir": root, "status": "downloading",
+        "media_id": video.get("video_id"), "target_dir": root, "status": "queued",
         "year": video.get("published_at"), "poster_url": video.get("thumbnail_url"),
         "search_ctx": json.dumps({"channel": video.get("channel_title"),
                                   "video_title": video.get("video_title"),
                                   "published_at": video.get("published_at"),
                                   "server_source": resolve_video_server()}),
     })
-    threading.Thread(target=run_youtube_download, args=(dl_id, get_video_db),
+
-                     daemon=True, name="yt-dl-%s" % dl_id).start()
+
-    return dl_id
+def _default_start_next() -> Any:
+    """Claim + start the next queued YouTube download (or None). The worker chains the rest."""
+    from api.video import get_video_db
+    from core.video.youtube_download import start_next_queued
+    return start_next_queued(get_video_db)
 
 
 def auto_video_process_youtube_wishlist(
@@ -86,17 +98,21 @@ def auto_video_process_youtube_wishlist(
     youtube_root: Optional[Callable[[], str]] = None,
     fetch_wanted: Optional[Callable[[], List[Dict[str, Any]]]] = None,
     active_ids: Optional[Callable[[], Iterable]] = None,
+    running_count: Optional[Callable[[], int]] = None,
     enqueue: Optional[Callable[[Dict[str, Any], str], Any]] = None,
+    start_next: Optional[Callable[[], Any]] = None,
 ) -> Dict[str, Any]:
-    """Enqueue a batch of wished YouTube videos for download.
+    """Queue the whole YouTube wishlist for download and start up to ``max_concurrent`` now.
 
-    Returns ``{'status': 'completed', 'queued': int, 'remaining': int, ...}``."""
+    Returns ``{'status': 'completed', 'queued': int, 'started': int, 'running': int, ...}``."""
     youtube_root = youtube_root or _default_youtube_root
     fetch_wanted = fetch_wanted or _default_fetch_wanted
     active_ids = active_ids or _default_active_ids
+    running_count = running_count or _default_running_count
     enqueue = enqueue or _default_enqueue
+    start_next = start_next or _default_start_next
     automation_id = config.get('_automation_id')
-    batch_size = max(0, int(config.get('batch_size', 3) or 0))
+    max_concurrent = max(1, int(config.get('max_concurrent', 3) or 3))
 
     try:
         root = youtube_root()
@@ -106,39 +122,44 @@ def auto_video_process_youtube_wishlist(
                                  log_line=msg, log_type='error')
             return {'status': 'error', 'error': msg, '_manages_own_progress': True}
 
-        deps.update_progress(automation_id, phase='Checking the YouTube wishlist…', progress=10,
+        deps.update_progress(automation_id, phase='Checking the YouTube wishlist…', progress=15,
-                             log_line='Looking for new videos to download', log_type='info')
+                             log_line='Queueing new videos for download', log_type='info')
         wanted = fetch_wanted() or []
-        active = list(active_ids() or [])
+        already = list(active_ids() or [])
-        picks = select_to_enqueue(wanted, active, batch_size)
+        new = videos_to_enqueue(wanted, already)
-        if not picks:
-            note = ('All %d wished video(s) are already downloading' % len(active)) if active \
-                else 'No wished YouTube videos to download'
-            deps.update_progress(automation_id, status='finished', progress=100, phase='Complete',
-                                 log_line=note, log_type='info')
-            return {'status': 'completed', 'queued': 0, 'remaining': max(0, len(wanted) - len(active)),
-                    '_manages_own_progress': True}
 
         queued = 0
-        for i, v in enumerate(picks):
+        for v in new:
-            deps.update_progress(automation_id, phase='Queueing downloads…',
-                                 progress=20 + int(70 * i / max(len(picks), 1)),
-                                 log_line="Queued '%s'" % (v.get('video_title') or v.get('video_id')),
-                                 log_type='info')
             try:
                 if enqueue(v, root) is not None:
                     queued += 1
-            except Exception:   # noqa: BLE001 - one bad enqueue shouldn't stop the batch
+            except Exception:   # noqa: BLE001 - one bad enqueue shouldn't stop the rest
                 deps.update_progress(automation_id, log_type='warning',
                                      log_line="Couldn't queue '%s'" % (v.get('video_title') or v.get('video_id')))
 
-        remaining = max(0, len(wanted) - len(active) - queued)
+        # Fill the concurrency slots now; each finished download starts the next, so the
-        done = 'Queued %d YouTube download(s)' % queued
+        # whole queue drains on its own from here.
-        if remaining:
+        deps.update_progress(automation_id, phase='Starting downloads…', progress=70,
-            done += ' · %d more waiting (drains over the next runs)' % remaining
+                             log_line='Queued %d new video(s)' % queued, log_type='info')
+        started = 0
+        for _ in range(slots_free(running_count() or 0, max_concurrent)):
+            if start_next() is None:
+                break
+            started += 1
+
+        running = (running_count() or 0)
+        if queued or started:
+            done = 'Queued %d new · %d downloading now (the rest drain automatically)' % (queued, running)
+            log_type = 'success'
+        elif running:
+            done = '%d already downloading; nothing new to queue' % running
+            log_type = 'info'
+        else:
+            done = 'No wished YouTube videos to download'
+            log_type = 'info'
         deps.update_progress(automation_id, status='finished', progress=100, phase='Complete',
-                             log_line=done, log_type='success')
+                             log_line=done, log_type=log_type)
-        return {'status': 'completed', 'queued': queued, 'remaining': remaining,
+        return {'status': 'completed', 'queued': queued, 'started': started, 'running': running,
                 '_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/youtube_download.py

@@ -25,6 +25,8 @@ from __future__ import annotations
 import json
 import logging
 import os
+import threading
+import time
 from typing import Any, Callable, Dict, Optional
 
 from core.video import organization, youtube_quality
@@ -172,13 +174,52 @@ def process_youtube_download(
     return {"status": "failed", "error": err}
 
 
+# ── concurrency + pacing (the music side's lesson: cap concurrency AND space starts) ──
+# yt-dlp 429s if hammered, so fetch STARTS are spaced ≥ this far apart across all workers.
+_DELAY_SECONDS = 3.0
+_pace_lock = threading.Lock()
+_last_start = [0.0]
+
+
+def _pace(delay: float) -> None:
+    """Block until at least ``delay`` seconds after the previous fetch start, then reserve
+    this start slot. Reserving under the lock (sleeping outside it) staggers concurrent
+    workers without serialising them past the delay."""
+    if delay <= 0:
+        return
+    with _pace_lock:
+        now = time.monotonic()
+        start_at = max(now, _last_start[0] + delay) if _last_start[0] else now
+        _last_start[0] = start_at
+    wait = start_at - time.monotonic()
+    if wait > 0:
+        time.sleep(wait)
+
+
+def _spawn_worker(dl_id: Any, db_provider: Callable) -> None:
+    threading.Thread(target=run_youtube_download, args=(dl_id, db_provider),
+                     daemon=True, name="yt-dl-%s" % dl_id).start()
+
+
+def start_next_queued(db_provider: Callable) -> Any:
+    """Claim the next queued YouTube download and start it. Returns its id, or None if the
+    queue is empty. The wishlist pump uses this to fill slots; each finished worker calls it
+    once (one-out-one-in) so the queue drains continuously at the established concurrency."""
+    row = db_provider().claim_next_youtube_queued()
+    if not row:
+        return None
+    _spawn_worker(row["id"], db_provider)
+    return row["id"]
+
+
 # ── production wiring ─────────────────────────────────────────────────────────
 def run_youtube_download(dl_id: Any, db_provider: Callable) -> None:
     """Production entry: fetch the row, bind real seams, fulfil it. Called on a worker
-    thread by the enqueue path; swallows nothing silently into the DB."""
+    thread by the pump; on finish it starts the next queued download (one-out-one-in)."""
     db = db_provider()
     dl = db.get_video_download(dl_id)
     if not dl:
+        start_next_queued(db_provider)        # keep the queue moving even on a stale id
         return
     profile = youtube_quality.load(db)
     settings = organization.load(db)
@@ -211,14 +252,19 @@ def run_youtube_download(dl_id: Any, db_provider: Callable) -> None:
     except Exception:   # noqa: BLE001 - cookies are optional
         cookie_opts = None
 
-    process_youtube_download(
+    try:
-        dl, profile=profile, settings=settings,
+        _pace(_DELAY_SECONDS)                  # space fetch starts to avoid yt-dlp 429s
-        update_row=db.update_video_download, archive=_archive,
+        process_youtube_download(
-        clear_wishlist=lambda vid: db.remove_youtube_from_wishlist("video", vid),
+            dl, profile=profile, settings=settings,
-        progress_hook=_progress, cookie_opts=cookie_opts, now=_now)
+            update_row=db.update_video_download, archive=_archive,
+            clear_wishlist=lambda vid: db.remove_youtube_from_wishlist("video", vid),
+            progress_hook=_progress, cookie_opts=cookie_opts, now=_now)
+    finally:
+        start_next_queued(db_provider)         # one out, one in — drain the queue
 
 
 __all__ = [
     "youtube_fields_from_download", "plan_destination", "ydl_download_opts",
     "download_one", "process_youtube_download", "run_youtube_download",
+    "start_next_queued",
 ]

database/video_database.py

@@ -1428,6 +1428,37 @@ class VideoDatabase:
         finally:
             conn.close()
 
+    def count_active_youtube_downloads(self) -> int:
+        """How many YouTube downloads are ACTIVELY fetching right now (not queued) —
+        the concurrency gauge the wishlist pump checks before starting more."""
+        conn = self._get_connection()
+        try:
+            r = conn.execute("SELECT COUNT(*) AS n FROM video_downloads "
+                             "WHERE source='youtube' AND status='downloading'").fetchone()
+            return int(r["n"]) if r else 0
+        finally:
+            conn.close()
+
+    def claim_next_youtube_queued(self) -> dict | None:
+        """Atomically take the oldest QUEUED YouTube download and flip it to
+        'downloading', returning the row (or None if none are queued). The
+        ``WHERE status='queued'`` guard makes the claim race-safe — two workers finishing
+        at once can't grab the same row."""
+        conn = self._get_connection()
+        try:
+            row = conn.execute(
+                "SELECT * FROM video_downloads WHERE source='youtube' AND status='queued' "
+                "ORDER BY id LIMIT 1").fetchone()
+            if not row:
+                return None
+            cur = conn.execute(
+                "UPDATE video_downloads SET status='downloading', updated_at=datetime('now') "
+                "WHERE id=? AND status='queued'", (row["id"],))
+            conn.commit()
+            return dict(row) if cur.rowcount else None   # rowcount 0 = lost the race
+        finally:
+            conn.close()
+
     def media_tmdb_id(self, kind: str, media_id) -> tuple:
         """(tmdb_id, imdb_id) for a library movie/show row — used to resolve sidecar /
         subtitle metadata for an owned re-grab (whose media_id is the library id, not a

tests/test_video_process_youtube_wishlist.py

@@ -1,6 +1,6 @@
-"""Drain side of the YouTube fulfillment lane: enqueue wished videos into the download
+"""Drain side of the YouTube fulfillment lane: queue the WHOLE wishlist (no total cap),
-queue in polite batches, skipping in-flight ones. Pure selection + handler with all I/O
+start up to a concurrency limit, and let finished downloads start the next. Pure selection +
-injected, plus the DB query that feeds it.
+pump with all I/O injected, plus the DB queue methods that back it.
 """
 
 from __future__ import annotations
@@ -9,7 +9,8 @@ import pytest
 
 from core.automation.handlers.video_process_youtube_wishlist import (
     auto_video_process_youtube_wishlist,
-    select_to_enqueue,
+    slots_free,
+    videos_to_enqueue,
 )
 
 
@@ -26,74 +27,108 @@ def _v(vid, title="T", date="2024-01-01"):
             "thumbnail_url": "/t.jpg", "published_at": date}
 
 
-# ── pure batching ─────────────────────────────────────────────────────────────
+# ── pure ──────────────────────────────────────────────────────────────────────
-def test_select_skips_inflight_and_caps_batch():
+def test_videos_to_enqueue_skips_already_queued_no_cap():
     wanted = [_v("a"), _v("b"), _v("c"), _v("d")]
-    picks = select_to_enqueue(wanted, active_ids=["b"], batch_size=2)
+    out = videos_to_enqueue(wanted, already_ids=["b"])
-    assert [p["video_id"] for p in picks] == ["a", "c"]   # b skipped (in flight), capped at 2
+    assert [v["video_id"] for v in out] == ["a", "c", "d"]   # b in flight; rest ALL kept
 
 
-def test_select_zero_batch_means_no_cap():
+def test_videos_to_enqueue_drops_idless():
-    wanted = [_v("a"), _v("b"), _v("c")]
+    assert videos_to_enqueue([{"video_title": "no id"}], []) == []
-    assert len(select_to_enqueue(wanted, [], 0)) == 3
 
 
-def test_select_drops_idless():
+def test_slots_free():
-    assert select_to_enqueue([{"video_title": "no id"}], [], 5) == []
+    assert slots_free(running=0, max_concurrent=3) == 3
+    assert slots_free(running=2, max_concurrent=3) == 1
+    assert slots_free(running=5, max_concurrent=3) == 0     # over the limit → no new starts
 
 
-# ── handler ───────────────────────────────────────────────────────────────────
+# ── handler: queue everything, start up to the limit ──────────────────────────
-def _run(wanted, *, active=None, root="/yt", batch=3):
+def _run(wanted, *, active=None, running=0, root="/yt", max_concurrent=3, start_results=None):
-    enq = []
+    enq, starts = [], {"n": 0}
 
     def enqueue(video, r):
         enq.append((video["video_id"], r))
         return len(enq)
 
+    # start_next returns an id until the (simulated) queue is exhausted
+    seq = list(start_results) if start_results is not None else [1] * 999
+
+    def start_next():
+        if starts["n"] < len(seq) and seq[starts["n"]] is not None:
+            starts["n"] += 1
+            return seq[starts["n"] - 1]
+        return None
+
     deps = _Deps()
     res = auto_video_process_youtube_wishlist(
-        {"_automation_id": "a", "batch_size": batch}, deps,
+        {"_automation_id": "a", "max_concurrent": max_concurrent}, deps,
         youtube_root=lambda: root, fetch_wanted=lambda: wanted,
-        active_ids=lambda: list(active or []), enqueue=enqueue)
+        active_ids=lambda: list(active or []), running_count=lambda: running,
-    return res, enq, deps
+        enqueue=enqueue, start_next=start_next)
+    return res, enq, starts["n"], deps
 
 
-def test_enqueues_a_batch_and_reports_remaining():
+def test_queues_entire_wishlist_and_starts_up_to_the_limit():
-    wanted = [_v("a"), _v("b"), _v("c"), _v("d"), _v("e")]
+    wanted = [_v(c) for c in "abcdefgh"]          # 8 wished
-    res, enq, _ = _run(wanted, batch=2)
+    res, enq, started, _ = _run(wanted, running=0, max_concurrent=3)
-    assert res["status"] == "completed" and res["queued"] == 2
+    assert res["status"] == "completed"
-    assert [vid for vid, _ in enq] == ["a", "b"]
+    assert res["queued"] == 8                       # the WHOLE wishlist is queued (no cap)
-    assert res["remaining"] == 3                          # 5 wanted - 2 queued
+    assert [vid for vid, _ in enq] == list("abcdefgh")
-    assert enq[0][1] == "/yt"                             # root passed through
+    assert started == 3 and res["started"] == 3     # only 3 start now; rest drain via workers
+    assert enq[0][1] == "/yt"
 
 
-def test_inflight_videos_are_not_requeued():
+def test_does_not_exceed_concurrency_when_some_already_running():
-    wanted = [_v("a"), _v("b")]
+    wanted = [_v(c) for c in "abcde"]
-    res, enq, _ = _run(wanted, active=["a", "b"], batch=5)
+    res, enq, started, _ = _run(wanted, running=2, max_concurrent=3)
-    assert enq == [] and res["queued"] == 0
+    assert res["queued"] == 5                        # still queues everything
+    assert started == 1                              # only 1 free slot (3 - 2 already running)
+
+
+def test_full_pipeline_starts_nothing_new():
+    wanted = [_v("a")]
+    res, enq, started, _ = _run(wanted, active=["a"], running=3, max_concurrent=3)
+    assert enq == []                                 # 'a' already in flight → not re-queued
+    assert started == 0
+
+
+def test_stops_starting_when_queue_drains_early():
+    # only 2 things can actually start even though 3 slots are free
+    res, enq, started, _ = _run([_v("a"), _v("b")], running=0, max_concurrent=3,
+                                start_results=[10, 11, None])
+    assert started == 2
 
 
 def test_missing_youtube_folder_is_an_error():
-    res, enq, deps = _run([_v("a")], root="")
+    res, enq, started, deps = _run([_v("a")], root="")
     assert res["status"] == "error" and "library folder" in res["error"]
     assert enq == [] and any(p.get("status") == "error" for p in deps.progress)
 
 
-def test_nothing_wanted_is_a_clean_noop():
+def test_nothing_wanted_and_empty_queue_is_a_clean_noop():
-    res, enq, _ = _run([])
+    res, enq, started, _ = _run([], start_results=[None])     # nothing wanted, queue empty
-    assert res["status"] == "completed" and res["queued"] == 0 and enq == []
+    assert res["status"] == "completed" and res["queued"] == 0 and enq == [] and started == 0
 
 
-def test_one_bad_enqueue_does_not_stop_the_batch():
+def test_drains_leftover_queue_even_with_nothing_new_wanted():
+    # a prior run queued items; this run adds nothing new but still fills the slots
+    res, enq, started, _ = _run([], running=0, max_concurrent=3, start_results=[1, 2, 3])
+    assert res["queued"] == 0 and enq == [] and started == 3
+
+
+def test_one_bad_enqueue_does_not_stop_the_rest():
     def enqueue(video, r):
         if video["video_id"] == "a":
-            raise RuntimeError("queue full")
+            raise RuntimeError("disk full")
         return 1
 
     res = auto_video_process_youtube_wishlist(
-        {"_automation_id": "x", "batch_size": 5}, _Deps(),
+        {"_automation_id": "x", "max_concurrent": 5}, _Deps(),
         youtube_root=lambda: "/yt", fetch_wanted=lambda: [_v("a"), _v("b")],
-        active_ids=lambda: [], enqueue=enqueue)
+        active_ids=lambda: [], running_count=lambda: 0, enqueue=enqueue,
+        start_next=lambda: None)
     assert res["status"] == "completed" and res["queued"] == 1   # b still queued
 
 
@@ -105,7 +140,7 @@ def test_top_level_error_is_caught():
     assert res["status"] == "error" and "db down" in res["error"]
 
 
-# ── the DB query that feeds it ────────────────────────────────────────────────
+# ── the DB queue methods ──────────────────────────────────────────────────────
 from database.video_database import VideoDatabase  # noqa: E402
 
 
@@ -124,12 +159,26 @@ def test_youtube_wishlist_to_download_shape(db):
     top = rows[0]
     assert top["channel_id"] == "UC1" and top["channel_title"] == "Cool Channel"
     assert top["video_title"] == "Second" and top["published_at"] == "2024-05-01"
-    assert top["thumbnail_url"] == "/2.jpg"
 
 
-def test_youtube_wishlist_to_download_excludes_movies(db):
+def test_count_and_claim_queue(db):
-    db.add_movie_to_wishlist(99, "A Movie")
+    a = db.add_video_download({"kind": "youtube", "source": "youtube", "media_id": "v1",
-    db.add_videos_to_wishlist({"youtube_id": "UC1", "title": "Ch"},
+                               "title": "A", "status": "queued"})
-                              [{"youtube_id": "v1", "title": "Vid", "published_at": "2024-01-01"}])
+    db.add_video_download({"kind": "youtube", "source": "youtube", "media_id": "v2",
-    rows = db.youtube_wishlist_to_download()
+                           "title": "B", "status": "queued"})
-    assert [r["video_id"] for r in rows] == ["v1"]           # the movie isn't a youtube video
+    assert db.count_active_youtube_downloads() == 0           # nothing fetching yet
+
+    claimed = db.claim_next_youtube_queued()
+    assert claimed["id"] == a and claimed["media_id"] == "v1"  # oldest first
+    assert db.count_active_youtube_downloads() == 1           # now one is 'downloading'
+
+    db.claim_next_youtube_queued()
+    assert db.count_active_youtube_downloads() == 2
+    assert db.claim_next_youtube_queued() is None             # queue empty
+
+
+def test_claim_ignores_non_youtube_and_terminal(db):
+    db.add_video_download({"kind": "movie", "source": "soulseek", "media_id": "m1",
+                           "title": "Movie", "status": "queued"})
+    assert db.claim_next_youtube_queued() is None             # soulseek queue isn't ours
+    assert db.count_active_youtube_downloads() == 0