diff --git a/core/video/youtube_download.py b/core/video/youtube_download.py
new file mode 100644
index 00000000..4438b1cc
--- /dev/null
+++ b/core/video/youtube_download.py
@@ -0,0 +1,224 @@
+"""YouTube download worker — the fulfillment lane for wished YouTube videos.
+
+Model B (Boulder): YouTube grabs flow through the SAME ``video_downloads`` queue +
+history as movies/TV, so the Downloads page, live progress, and History modal all work
+for YouTube for free. But the mechanism is different — there's no slskd transfer to poll;
+yt-dlp fetches the stream directly. So this worker owns a YouTube download end to end:
+
+    pick stream (quality profile → yt-dlp format) → download into the library, organised
+    as a Plex "TV by date" show (channel/Season YEAR/channel - DATE - title) → mark the
+    row completed + archive it to history → remove the video from the wishlist.
+
+The slskd ``download_monitor`` simply SKIPS ``source='youtube'`` rows (they have no
+transfer to match), so this lane never disturbs the movie/TV pipeline.
+
+The orchestration (``process_youtube_download``) is PURE — the actual yt-dlp run and all
+DB writes are injected seams — so the lifecycle (dest planning, completion → archive +
+unwish, failure → archive, no unwish) is unit-tested without a network or a DB. Production
+(``run_youtube_download``) lazily binds the real calls and runs it on a worker thread.
+
+Isolated: imports only sibling ``core.video`` modules; nothing from the music side.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+from typing import Any, Callable, Dict, Optional
+
+from core.video import organization, youtube_quality
+from core.video.youtube_quality import format_selection
+
+logger = logging.getLogger(__name__)
+
+try:
+    import yt_dlp
+except Exception:   # noqa: BLE001 - optional at import; absence handled at call time
+    yt_dlp = None
+
+
+def youtube_fields_from_download(dl: Dict[str, Any]) -> Dict[str, Any]:
+    """Organising fields for a YouTube download row. The channel/video-title/date that
+    ``render_path('youtube', …)`` needs ride in ``search_ctx`` (a generic JSON column);
+    fall back to the row's own columns when absent."""
+    ctx = dl.get("search_ctx")
+    if isinstance(ctx, str):
+        try:
+            ctx = json.loads(ctx)
+        except (ValueError, TypeError):
+            ctx = {}
+    if not isinstance(ctx, dict):
+        ctx = {}
+    return {
+        "channel": ctx.get("channel") or dl.get("title"),
+        "title": ctx.get("video_title") or dl.get("title"),
+        "published_at": ctx.get("published_at") or dl.get("year"),
+        "youtube_id": dl.get("media_id"),
+    }
+
+
+def plan_destination(dl: Dict[str, Any], settings: Dict[str, Any], container: str) -> Dict[str, str]:
+    """Where this video lands in the library: ``{dir, filename, path}`` under the youtube
+    root (``target_dir``), organised by the youtube template. Pure."""
+    ext = "." + str(container or "mp4").lstrip(".")
+    return organization.render_path("youtube", dl.get("target_dir"),
+                                    youtube_fields_from_download(dl), settings, ext)
+
+
+def ydl_download_opts(profile: Any, dest_dir: str, dest_stem: str,
+                      *, progress_hook: Optional[Callable] = None, cookie_opts: Optional[dict] = None) -> dict:
+    """The yt-dlp options dict for one download: format selection from the quality profile,
+    a fixed output path (dir + stem + yt-dlp's own ext), polite defaults. Pure."""
+    sel = format_selection(profile)
+    opts = {
+        "quiet": True,
+        "no_warnings": True,
+        "noplaylist": True,
+        "retries": 3,
+        "format": sel["format"],
+        "format_sort": sel["format_sort"],
+        "merge_output_format": sel["merge_output_format"],
+        "paths": {"home": str(dest_dir or "")},
+        "outtmpl": dest_stem + ".%(ext)s",
+    }
+    if cookie_opts:
+        opts.update(cookie_opts)
+    if progress_hook:
+        opts["progress_hooks"] = [progress_hook]
+    return opts
+
+
+def _stem_and_container(dest: Dict[str, str], container: str) -> tuple:
+    """(filename stem, final ext) — strip the ext render_path put on the filename so
+    yt-dlp can own the extension during the merge."""
+    fn = dest.get("filename") or "download"
+    cont = str(container or "mp4").lstrip(".")
+    stem = fn[:-(len(cont) + 1)] if fn.lower().endswith("." + cont.lower()) else os.path.splitext(fn)[0]
+    return stem or "download", cont
+
+
+def download_one(video_id: Any, dest_dir: str, dest_stem: str, profile: Any, container: str,
+                 *, ydl_factory=None, progress_hook=None, cookie_opts=None) -> Dict[str, Any]:
+    """Run yt-dlp for ONE video into ``dest_dir/dest_stem.ext``. Returns
+    ``{ok, dest_path|None, error|None}``. The yt-dlp class is injectable for tests."""
+    vid = str(video_id or "").strip()
+    if not vid:
+        return {"ok": False, "dest_path": None, "error": "No video id"}
+    factory = ydl_factory or (yt_dlp.YoutubeDL if yt_dlp else None)
+    if factory is None:
+        return {"ok": False, "dest_path": None, "error": "yt-dlp unavailable"}
+    opts = ydl_download_opts(profile, dest_dir, dest_stem,
+                             progress_hook=progress_hook, cookie_opts=cookie_opts)
+    url = vid if vid.startswith("http") else "https://www.youtube.com/watch?v=" + vid
+    try:
+        with factory(opts) as ydl:
+            ydl.download([url])
+    except Exception as e:   # noqa: BLE001 - any yt-dlp failure → a failed download, not a crash
+        logger.info("youtube download failed for %s: %s", vid, e)
+        return {"ok": False, "dest_path": None, "error": str(e)}
+    dest_path = os.path.join(str(dest_dir or ""), dest_stem + "." + str(container or "mp4").lstrip("."))
+    return {"ok": True, "dest_path": dest_path, "error": None}
+
+
+def process_youtube_download(
+    dl: Dict[str, Any],
+    *,
+    profile: Any,
+    settings: Dict[str, Any],
+    download: Callable = download_one,
+    update_row: Callable[..., Any],
+    archive: Callable[[Dict[str, Any], Dict[str, Any]], Any],
+    clear_wishlist: Callable[[Any], Any],
+    progress_hook: Optional[Callable] = None,
+    cookie_opts: Optional[dict] = None,
+    now: Optional[Callable[[], str]] = None,
+) -> Dict[str, Any]:
+    """Fulfil one queued YouTube download. PURE — all I/O injected.
+
+    On success: row → completed (with dest_path), snapshot to history, and remove the video
+    from the wishlist (it's in the library now; history is the permanent record). On
+    failure: row → failed + a history snapshot; the wishlist row is LEFT so a later scan/run
+    can retry. Returns a small result dict."""
+    now = now or (lambda: "")
+    settings = settings if isinstance(settings, dict) else {}
+    container = format_selection(profile)["merge_output_format"]
+    dest = plan_destination(dl, settings, container)
+    stem, cont = _stem_and_container(dest, container)
+
+    # Record where it's going up front (Downloads page shows the organised name).
+    update_row(dl.get("id"), status="downloading", progress=0,
+               target_dir=dest.get("dir"), filename=dest.get("filename"))
+
+    res = download(dl.get("media_id"), dest.get("dir"), stem, profile, cont,
+                   progress_hook=progress_hook, cookie_opts=cookie_opts)
+
+    if res.get("ok"):
+        dest_path = res.get("dest_path") or dest.get("path")
+        completed = now()
+        update_row(dl.get("id"), status="completed", progress=100,
+                   dest_path=dest_path, completed_at=completed)
+        archive(dl, {"status": "completed", "dest_path": dest_path, "completed_at": completed})
+        try:
+            clear_wishlist(dl.get("media_id"))
+        except Exception:   # noqa: BLE001 - unwish is best-effort; the file is already in place
+            logger.exception("youtube download %s: unwish failed", dl.get("id"))
+        return {"status": "completed", "dest_path": dest_path}
+
+    err = res.get("error") or "Download failed"
+    completed = now()
+    update_row(dl.get("id"), status="failed", error=err, completed_at=completed)
+    archive(dl, {"status": "failed", "error": err, "completed_at": completed})
+    return {"status": "failed", "error": err}
+
+
+# ── 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."""
+    db = db_provider()
+    dl = db.get_video_download(dl_id)
+    if not dl:
+        return
+    profile = youtube_quality.load(db)
+    settings = organization.load(db)
+    from datetime import datetime, timezone
+
+    def _now():
+        return datetime.now(timezone.utc).isoformat(timespec="seconds")
+
+    def _progress(d):
+        # yt-dlp progress hook → row progress %. Best-effort; never raises into yt-dlp.
+        try:
+            if d.get("status") == "downloading":
+                total = d.get("total_bytes") or d.get("total_bytes_estimate") or 0
+                got = d.get("downloaded_bytes") or 0
+                if total:
+                    db.update_video_download(dl_id, progress=int(got * 100 / total))
+        except Exception:   # noqa: BLE001, S110 - a progress glitch must not abort the download
+            pass
+
+    def _archive(row, upd):
+        try:
+            db.record_download_history({**row, **upd})
+        except Exception:
+            logger.exception("youtube download %s: history snapshot failed", dl_id)
+
+    cookie_opts = None
+    try:
+        from core.video.youtube import _cookie_opts
+        cookie_opts = _cookie_opts()
+    except Exception:   # noqa: BLE001 - cookies are optional
+        cookie_opts = None
+
+    process_youtube_download(
+        dl, profile=profile, settings=settings,
+        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)
+
+
+__all__ = [
+    "youtube_fields_from_download", "plan_destination", "ydl_download_opts",
+    "download_one", "process_youtube_download", "run_youtube_download",
+]
diff --git a/tests/test_youtube_download.py b/tests/test_youtube_download.py
new file mode 100644
index 00000000..7c47a9aa
--- /dev/null
+++ b/tests/test_youtube_download.py
@@ -0,0 +1,153 @@
+"""YouTube download worker — the fulfillment lane for wished YouTube videos. Pure
+orchestration (dest planning, yt-dlp opts, completion → archive + unwish, failure →
+archive only) tested with the yt-dlp run + all DB writes injected.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+
+from core.video import youtube_download as ytd
+from core.video.youtube_quality import default_profile
+
+
+# ── organising fields from the queue row ──────────────────────────────────────
+def test_fields_prefer_search_ctx_then_fall_back():
+    dl = {"title": "Some Channel", "year": "2024-01-01", "media_id": "vid1",
+          "search_ctx": json.dumps({"channel": "Veritasium", "video_title": "Electricity",
+                                     "published_at": "2024-03-15"})}
+    f = ytd.youtube_fields_from_download(dl)
+    assert f == {"channel": "Veritasium", "title": "Electricity",
+                 "published_at": "2024-03-15", "youtube_id": "vid1"}
+
+
+def test_fields_fall_back_to_row_when_ctx_absent_or_garbage():
+    dl = {"title": "Chan", "year": "2024-02-02", "media_id": "v2", "search_ctx": "{bad"}
+    f = ytd.youtube_fields_from_download(dl)
+    assert f["channel"] == "Chan" and f["title"] == "Chan"
+    assert f["published_at"] == "2024-02-02" and f["youtube_id"] == "v2"
+
+
+# ── destination planning ──────────────────────────────────────────────────────
+def test_plan_destination_uses_the_youtube_template():
+    dl = {"target_dir": "/yt", "media_id": "v1",
+          "search_ctx": json.dumps({"channel": "Veritasium", "video_title": "How It Works",
+                                    "published_at": "2024-03-15"})}
+    dest = ytd.plan_destination(dl, {}, "mp4")
+    assert dest["path"] == os.path.join("/yt", "Veritasium", "Season 2024",
+                                        "Veritasium - 2024-03-15 - How It Works.mp4")
+
+
+# ── yt-dlp opts ───────────────────────────────────────────────────────────────
+def test_ydl_opts_carry_format_selection_and_fixed_output():
+    opts = ytd.ydl_download_opts(default_profile(), "/yt/dir", "Chan - 2024-03-15 - Title")
+    assert opts["format"] == "bv*[height<=1080]+ba/b[height<=1080]/bv*+ba/b"
+    assert opts["merge_output_format"] == "mp4"
+    assert opts["paths"] == {"home": "/yt/dir"}
+    assert opts["outtmpl"] == "Chan - 2024-03-15 - Title.%(ext)s"
+    assert opts["noplaylist"] is True
+
+
+# ── download_one with an injected yt-dlp ───────────────────────────────────────
+class _FakeYDL:
+    def __init__(self, opts):
+        self.opts = opts
+        _FakeYDL.last = self
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *a):
+        return False
+
+    def download(self, urls):
+        self.urls = urls
+
+
+class _BoomYDL(_FakeYDL):
+    def download(self, urls):
+        raise RuntimeError("403 blocked")
+
+
+def test_download_one_success_returns_built_dest_path():
+    res = ytd.download_one("vid1", "/yt/Chan/Season 2024", "Chan - 2024-03-15 - T",
+                           default_profile(), "mp4", ydl_factory=_FakeYDL)
+    assert res["ok"] is True
+    assert res["dest_path"] == os.path.join("/yt/Chan/Season 2024", "Chan - 2024-03-15 - T.mp4")
+    assert _FakeYDL.last.urls == ["https://www.youtube.com/watch?v=vid1"]
+
+
+def test_download_one_failure_is_captured_not_raised():
+    res = ytd.download_one("vid1", "/yt", "stem", default_profile(), "mp4", ydl_factory=_BoomYDL)
+    assert res["ok"] is False and "403 blocked" in res["error"]
+
+
+def test_download_one_no_factory_is_unavailable():
+    res = ytd.download_one("vid1", "/yt", "stem", default_profile(), "mp4", ydl_factory=None)
+    # yt_dlp may or may not be importable in the test env; either way no real run happens
+    if res["ok"] is False:
+        assert res["error"]
+
+
+# ── the orchestration: completion vs failure ──────────────────────────────────
+def _recorder():
+    calls = {"rows": [], "archive": [], "unwish": []}
+
+    def update_row(dl_id, **kw):
+        calls["rows"].append((dl_id, kw))
+
+    def archive(row, upd):
+        calls["archive"].append(upd)
+
+    def clear_wishlist(vid):
+        calls["unwish"].append(vid)
+
+    return calls, update_row, archive, clear_wishlist
+
+
+def _dl():
+    return {"id": 7, "media_id": "vid1", "target_dir": "/yt", "title": "Chan", "year": "2024-03-15",
+            "search_ctx": json.dumps({"channel": "Chan", "video_title": "T", "published_at": "2024-03-15"})}
+
+
+def test_process_completion_archives_and_unwishes():
+    calls, update_row, archive, clear = _recorder()
+    res = ytd.process_youtube_download(
+        _dl(), profile=default_profile(), settings={},
+        download=lambda *a, **k: {"ok": True, "dest_path": "/yt/Chan/Season 2024/Chan - 2024-03-15 - T.mp4"},
+        update_row=update_row, archive=archive, clear_wishlist=clear, now=lambda: "2026-06-25T00:00:00+00:00")
+    assert res["status"] == "completed"
+    # row marked completed, history snapshot 'completed', and the video unwished
+    statuses = [kw.get("status") for _, kw in calls["rows"]]
+    assert "downloading" in statuses and statuses[-1] == "completed"
+    assert calls["archive"][-1]["status"] == "completed"
+    assert calls["unwish"] == ["vid1"]
+
+
+def test_process_failure_archives_but_keeps_the_wish():
+    calls, update_row, archive, clear = _recorder()
+    res = ytd.process_youtube_download(
+        _dl(), profile=default_profile(), settings={},
+        download=lambda *a, **k: {"ok": False, "error": "yt-dlp said no"},
+        update_row=update_row, archive=archive, clear_wishlist=clear, now=lambda: "t")
+    assert res["status"] == "failed" and "yt-dlp said no" in res["error"]
+    assert calls["rows"][-1][1]["status"] == "failed"
+    assert calls["archive"][-1]["status"] == "failed"
+    assert calls["unwish"] == []                         # wish kept so it can retry later
+
+
+def test_process_passes_the_organised_dir_to_the_downloader():
+    seen = {}
+
+    def fake_download(video_id, dest_dir, stem, profile, container, **kw):
+        seen.update(video_id=video_id, dest_dir=dest_dir, stem=stem, container=container)
+        return {"ok": True, "dest_path": "/x"}
+
+    calls, update_row, archive, clear = _recorder()
+    ytd.process_youtube_download(_dl(), profile=default_profile(), settings={},
+                                 download=fake_download, update_row=update_row,
+                                 archive=archive, clear_wishlist=clear, now=lambda: "t")
+    assert seen["video_id"] == "vid1"
+    assert seen["dest_dir"] == os.path.join("/yt", "Chan", "Season 2024")
+    assert seen["stem"] == "Chan - 2024-03-15 - T" and seen["container"] == "mp4"