Lyra/docs/superpowers/plans/2026-07-10-lyra-youtube-adapter.md

Lyra YouTube Adapter Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Ship Lyra's first REAL source adapter — YouTube via yt-dlp — implementing the existing SourceAdapter contract, and wire it into registry.build_adapters() so the running worker actually acquires music from YouTube. The adapter's logic is unit-tested offline against a fake yt-dlp client; the real network path is exercised only by an opt-in live test the user runs.

Architecture: YouTubeAdapter depends on a small injectable YtClient seam (two methods: search_album, download). Tests inject a FakeYtClient returning recorded-shape dicts — no network, no credentials, deterministic. The real YtDlpClient wraps yt_dlp.YoutubeDL and is constructed only inside registry.build_adapters(). Adding YouTube changes nothing in the pipeline/ranker/confidence — they already operate purely on Candidate/SourceAdapter.

Tech Stack: Python 3.12 worker; yt-dlp (new dependency); ffmpeg in the worker image (for FFmpegExtractAudio). Tests: pytest with a fake client (offline) + an opt-in live test gated by LYRA_LIVE_TESTS.

Global Constraints

• Python 3.12 in the worker container; local test runs may use newer 3.x — keep deps version-flexible.
• New dependency allowed: yt-dlp (this is the first plan that adds real source tooling). No other new deps.
• The SourceAdapter protocol, Candidate/Quality/DownloadResult/MBTarget types, the ranker, confidence, and pipeline are UNCHANGED. The adapter only implements the protocol.
• Candidate.source_tier is the single source of truth for ranking (set by the adapter / stamped by the pipeline). YouTube: name="youtube", tier=2, quality is lossy (Quality(fmt="OPUS", lossless=False, bitrate_kbps=160)).
• The real YtDlpClient is NOT unit-tested offline (it hits the network); it is covered by an OPT-IN live test skipped unless env LYRA_LIVE_TESTS is set. Never make normal CI/test runs depend on network.
• registry.build_adapters() returns only REAL, usable adapters. After this plan that is [YouTubeAdapter(YtDlpClient())]; Qobuz/Soulseek are added by later plans. The fake adapters remain in adapters/fakes.py for pipeline tests only.
• Every task ends with a commit.

Shared interfaces (defined by tasks below; listed for consistency)

# worker/lyra_worker/adapters/youtube.py (Task 1)
class YtClient(Protocol):
    def search_album(self, artist: str, album: str) -> list[dict]: ...   # each: {source_ref, title, artist, track_count}
    def download(self, source_ref: str, dest: str, on_progress: Callable[[float], None]) -> dict: ...  # {track_count, path}; raises on failure

class YouTubeAdapter:   # implements SourceAdapter
    name = "youtube"; tier = 2
    def __init__(self, client: YtClient): ...
    def health(self) -> bool: ...
    def search(self, target: MBTarget) -> list[Candidate]: ...
    def download(self, candidate: Candidate, dest: str, on_progress) -> DownloadResult: ...

# worker/lyra_worker/adapters/_ytdlp.py (Task 2)
class YtDlpClient:   # real YtClient wrapping yt_dlp.YoutubeDL
    def search_album(self, artist, album) -> list[dict]: ...
    def download(self, source_ref, dest, on_progress) -> dict: ...

---

Task 1: YouTubeAdapter + client seam (offline, fake client)

Files:

• Create: worker/lyra_worker/adapters/youtube.py
• Test: worker/tests/test_youtube_adapter.py

Interfaces:

• Consumes: MBTarget, Candidate, Quality, DownloadResult (types.py); SourceAdapter (adapters/base.py).

• Produces: the YtClient Protocol and YouTubeAdapter (see Shared interfaces). search maps each client result dict to a Candidate(source="youtube", source_ref, matched_artist=result["artist"], matched_album=result["title"], quality=lossy OPUS 160, track_count, source_tier=2). download calls client.download; on exception returns DownloadResult(ok=False, error=str(e)), else DownloadResult(ok=True, path, track_count).

• Step 1: Write the failing test

worker/tests/test_youtube_adapter.py:

from lyra_worker.adapters.youtube import YouTubeAdapter
from lyra_worker.types import Candidate, MBTarget, Quality

TARGET = MBTarget(artist="Radiohead", album="In Rainbows", track_count=10)


class FakeYtClient:
    def __init__(self, results=None, fail=False):
        self._results = results if results is not None else [
            {"source_ref": "yt:playlist:abc", "title": "In Rainbows", "artist": "Radiohead", "track_count": 10}
        ]
        self._fail = fail
        self.downloaded = []

    def search_album(self, artist, album):
        return self._results

    def download(self, source_ref, dest, on_progress):
        if self._fail:
            raise RuntimeError("network boom")
        on_progress(1.0)
        self.downloaded.append((source_ref, dest))
        return {"track_count": 10, "path": dest}


def test_name_and_tier():
    a = YouTubeAdapter(FakeYtClient())
    assert a.name == "youtube"
    assert a.tier == 2
    assert a.health() is True


def test_search_maps_results_to_candidates():
    cands = YouTubeAdapter(FakeYtClient()).search(TARGET)
    assert len(cands) == 1
    c = cands[0]
    assert c.source == "youtube"
    assert c.source_ref == "yt:playlist:abc"
    assert c.matched_artist == "Radiohead"
    assert c.matched_album == "In Rainbows"
    assert c.track_count == 10
    assert c.source_tier == 2
    assert c.quality.lossless is False


def test_search_empty_when_no_results():
    assert YouTubeAdapter(FakeYtClient(results=[])).search(TARGET) == []


def test_download_success():
    client = FakeYtClient()
    cand = YouTubeAdapter(client).search(TARGET)[0]
    seen = []
    result = YouTubeAdapter(client).download(cand, "/tmp/x", seen.append)
    assert result.ok is True
    assert result.track_count == 10
    assert seen and seen[-1] == 1.0


def test_download_failure_returns_not_ok():
    client = FakeYtClient(fail=True)
    cand = YouTubeAdapter(FakeYtClient()).search(TARGET)[0]
    result = YouTubeAdapter(client).download(cand, "/tmp/x", lambda _p: None)
    assert result.ok is False
    assert result.error

• Step 2: Run the test to verify it fails

Run:

cd worker
worker/.venv/bin/python -m pytest tests/test_youtube_adapter.py -v

Expected: FAIL — ModuleNotFoundError: No module named 'lyra_worker.adapters.youtube'.

• Step 3: Implement the adapter

worker/lyra_worker/adapters/youtube.py:

from typing import Callable, Protocol

from lyra_worker.types import Candidate, DownloadResult, MBTarget, Quality

_YT_QUALITY = Quality(fmt="OPUS", lossless=False, bitrate_kbps=160)


class YtClient(Protocol):
    def search_album(self, artist: str, album: str) -> list[dict]:
        """Return album matches: dicts with keys source_ref, title, artist, track_count."""
        ...

    def download(self, source_ref: str, dest: str, on_progress: Callable[[float], None]) -> dict:
        """Download into dest; return {track_count, path}. Raise on failure."""
        ...


class YouTubeAdapter:
    name = "youtube"
    tier = 2

    def __init__(self, client: YtClient):
        self._client = client

    def health(self) -> bool:
        return True

    def search(self, target: MBTarget) -> list[Candidate]:
        candidates: list[Candidate] = []
        for r in self._client.search_album(target.artist, target.album):
            candidates.append(
                Candidate(
                    source="youtube",
                    source_ref=r["source_ref"],
                    matched_artist=r.get("artist", ""),
                    matched_album=r.get("title", ""),
                    quality=_YT_QUALITY,
                    track_count=r.get("track_count", 1) or 1,
                    source_tier=self.tier,
                )
            )
        return candidates

    def download(
        self, candidate: Candidate, dest: str, on_progress: Callable[[float], None]
    ) -> DownloadResult:
        try:
            info = self._client.download(candidate.source_ref, dest, on_progress)
        except Exception as e:  # network/extraction failures -> fall-through in the pipeline
            return DownloadResult(ok=False, error=str(e))
        return DownloadResult(
            ok=True,
            path=info.get("path", dest),
            track_count=info.get("track_count", candidate.track_count),
        )

• Step 4: Run the test to verify it passes

Run:

cd worker
worker/.venv/bin/python -m pytest tests/test_youtube_adapter.py -v

Expected: PASS — all five tests green.

• Step 5: Commit

git add worker/lyra_worker/adapters/youtube.py worker/tests/test_youtube_adapter.py
git commit -m "feat: add YouTubeAdapter with injectable client seam"

---

Task 2: Real YtDlpClient + yt-dlp dependency + ffmpeg + opt-in live test

Files:

• Create: worker/lyra_worker/adapters/_ytdlp.py
• Create: worker/tests/test_youtube_live.py
• Modify: worker/requirements.txt
• Modify: worker/Dockerfile

Interfaces:

• Consumes: yt_dlp.

• Produces: YtDlpClient implementing the YtClient seam using yt_dlp.YoutubeDL. search_album runs a ytsearch5: query and maps entries; download fetches bestaudio into dest with a progress hook and FFmpegExtractAudio.

• Step 1: Add the dependency and ffmpeg

Append to worker/requirements.txt:

yt-dlp>=2024.1

Install into the venv: worker/.venv/bin/pip install -r worker/requirements.txt

In worker/Dockerfile, add ffmpeg to the image (yt-dlp needs it for audio extraction). Change the first lines from:

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

to:

FROM python:3.12-slim
WORKDIR /app
RUN apt-get update && apt-get install -y --no-install-recommends ffmpeg && rm -rf /var/lib/apt/lists/*
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

• Step 2: Implement the real client

worker/lyra_worker/adapters/_ytdlp.py:

import os
from typing import Callable

import yt_dlp


class YtDlpClient:
    """Real YtClient backed by yt_dlp. Not unit-tested offline; see test_youtube_live.py."""

    def search_album(self, artist: str, album: str) -> list[dict]:
        query = f"ytsearch5:{artist} {album} full album"
        opts = {
            "quiet": True,
            "no_warnings": True,
            "extract_flat": "in_playlist",
            "skip_download": True,
        }
        with yt_dlp.YoutubeDL(opts) as ydl:
            info = ydl.extract_info(query, download=False)
        results: list[dict] = []
        for e in (info.get("entries") or []):
            if not e:
                continue
            results.append(
                {
                    "source_ref": e.get("url") or e.get("webpage_url") or e.get("id"),
                    "title": e.get("title") or "",
                    "artist": e.get("uploader") or e.get("channel") or "",
                    "track_count": e.get("playlist_count") or 1,
                }
            )
        return results

    def download(self, source_ref: str, dest: str, on_progress: Callable[[float], None]) -> dict:
        os.makedirs(dest, exist_ok=True)

        def hook(d: dict) -> None:
            if d.get("status") == "downloading":
                total = d.get("total_bytes") or d.get("total_bytes_estimate")
                if total:
                    on_progress(min(1.0, d.get("downloaded_bytes", 0) / total))
            elif d.get("status") == "finished":
                on_progress(1.0)

        opts = {
            "quiet": True,
            "no_warnings": True,
            "format": "bestaudio/best",
            "outtmpl": os.path.join(dest, "%(playlist_index)s-%(title)s.%(ext)s"),
            "progress_hooks": [hook],
            "postprocessors": [{"key": "FFmpegExtractAudio", "preferredcodec": "opus"}],
        }
        with yt_dlp.YoutubeDL(opts) as ydl:
            info = ydl.extract_info(source_ref, download=True)
        entries = info.get("entries") if info.get("entries") is not None else [info]
        return {"track_count": len([e for e in entries if e]), "path": dest}

• Step 3: Write the opt-in live test

worker/tests/test_youtube_live.py:

import os

import pytest

from lyra_worker.adapters._ytdlp import YtDlpClient

pytestmark = pytest.mark.skipif(
    not os.environ.get("LYRA_LIVE_TESTS"),
    reason="live network test; set LYRA_LIVE_TESTS=1 to run",
)


def test_live_search_returns_results():
    results = YtDlpClient().search_album("Radiohead", "In Rainbows")
    assert len(results) >= 1
    assert all(r["source_ref"] for r in results)

• Step 4: Verify (offline) the module imports and the live test is skipped

Run:

cd worker
worker/.venv/bin/python -c "from lyra_worker.adapters._ytdlp import YtDlpClient; print('import ok')"
worker/.venv/bin/python -m pytest tests/test_youtube_live.py -v

Expected: prints import ok; the live test is reported skipped (LYRA_LIVE_TESTS not set). Do NOT set LYRA_LIVE_TESTS here — the live test is for the user to run on their network.

• Step 5: Commit

git add worker/lyra_worker/adapters/_ytdlp.py worker/tests/test_youtube_live.py worker/requirements.txt worker/Dockerfile
git commit -m "feat: add real yt-dlp client and opt-in live test; ffmpeg in worker image"

---

Task 3: Wire the real YouTube adapter into the registry

Files:

• Modify: worker/lyra_worker/registry.py
• Modify: worker/tests/test_registry.py

Interfaces:

• Consumes: YouTubeAdapter (Task 1), YtDlpClient (Task 2).

• Produces: build_adapters() returns [YouTubeAdapter(YtDlpClient())] — the real, usable adapters. (Qobuz/Soulseek added by later plans.)

• Step 1: Update the registry test (write the new expectation first)

Replace worker/tests/test_registry.py with:

from lyra_worker.adapters.youtube import YouTubeAdapter
from lyra_worker.registry import build_adapters


def test_registry_returns_real_youtube_adapter():
    adapters = build_adapters()
    names = [a.name for a in adapters]
    assert "youtube" in names
    yt = next(a for a in adapters if a.name == "youtube")
    assert isinstance(yt, YouTubeAdapter)
    assert yt.tier == 2
    assert yt.health() is True


def test_registry_adapter_names_are_unique():
    names = [a.name for a in build_adapters()]
    assert len(names) == len(set(names))

• Step 2: Run the test to verify it fails

Run:

cd worker
worker/.venv/bin/python -m pytest tests/test_registry.py -v

Expected: FAIL — the old build_adapters returns the three fakes, so isinstance(yt, YouTubeAdapter) fails (the youtube entry is a FakeYouTube).

• Step 3: Update the registry

Replace worker/lyra_worker/registry.py with:

from lyra_worker.adapters._ytdlp import YtDlpClient
from lyra_worker.adapters.base import SourceAdapter
from lyra_worker.adapters.youtube import YouTubeAdapter


def build_adapters() -> list[SourceAdapter]:
    """The enabled real source adapters, best-tier first.

    Only real, usable adapters are returned. Plan 3c/3d add Qobuz (streamrip)
    and Soulseek (slskd), each gated on its config being present via health().
    """
    return [YouTubeAdapter(YtDlpClient())]

• Step 4: Run the test to verify it passes

Run:

cd worker
worker/.venv/bin/python -m pytest tests/test_registry.py -v

Expected: PASS — both tests green.

• Step 5: Commit

git add worker/lyra_worker/registry.py worker/tests/test_registry.py
git commit -m "feat: wire real YouTube adapter into the registry"

---

Task 4: Pipeline integration with the real adapter class

Files:

• Test: worker/tests/test_youtube_pipeline.py

Interfaces:

• Consumes: run_pipeline (pipeline.py), YouTubeAdapter (Task 1), the conn fixture + insert_request.

• Produces: proof that the REAL YouTubeAdapter class (driven by a fake yt-dlp client, so offline) flows through the full pipeline to imported, writing a LibraryItem and a chosen Candidate with source="youtube".

• Step 1: Write the failing test

worker/tests/test_youtube_pipeline.py:

from lyra_worker.adapters.youtube import YouTubeAdapter
from lyra_worker.claim import claim_next
from lyra_worker.pipeline import run_pipeline
from tests.conftest import insert_request
from tests.test_youtube_adapter import FakeYtClient


def test_youtube_adapter_flows_through_pipeline(conn):
    job_id = insert_request(conn, artist="Radiohead", album="In Rainbows")
    claim_next(conn)
    run_pipeline(conn, job_id, [YouTubeAdapter(FakeYtClient())], dest_root="/tmp/lib")

    with conn.cursor() as cur:
        cur.execute('SELECT state, "currentStage" FROM "Job" WHERE id = %s', (job_id,))
        assert cur.fetchone() == ("imported", "import")
        cur.execute('SELECT source FROM "Candidate" WHERE "jobId" = %s AND chosen = true', (job_id,))
        assert cur.fetchone()[0] == "youtube"
        cur.execute('SELECT source FROM "LibraryItem" WHERE artist = %s', ("Radiohead",))
        assert cur.fetchone()[0] == "youtube"

• Step 2: Run the test to verify it fails, then passes

Run:

cd worker
export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra
worker/.venv/bin/python -m pytest tests/test_youtube_pipeline.py -v

Expected: This test should PASS immediately — it exercises already-implemented code (the adapter from Task 1 + the existing pipeline) with no new production code. If it FAILS, investigate a real integration mismatch (e.g. the FakeYtClient's default result must score above the confidence gate against "Radiohead"/"In Rainbows" — it does, since title/artist match exactly). This task is a characterization/integration test; no implementation step is expected.

• Step 3: Run the full worker suite

Run:

cd worker
export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra
worker/.venv/bin/python -m pytest -v

Expected: PASS — every suite green (youtube adapter, youtube pipeline, registry, config, crypto, pipeline, ranker, etc.), the live test skipped, output pristine.

• Step 4: Commit

git add worker/tests/test_youtube_pipeline.py
git commit -m "test: verify real YouTubeAdapter flows through the pipeline"

---

Notes for the next plans

registry.build_adapters() now returns the real YouTube adapter. Plan 3c adds QobuzAdapter (streamrip) and Plan 3d adds SoulseekAdapter (slskd REST client + an slskd service in docker-compose), each reading credentials via get_config(conn) and included in build_adapters() only when configured (checked through health()). The YtClient/injected-client pattern established here (real client separate from the adapter, fake client for offline tests, opt-in live test) is the template for both. Plan 4 then adds real MusicBrainz resolution in intake (filling MBTarget.track_count/total_duration_s, which will make the confidence gate and YouTube's track_count reporting meaningfully stricter) and real tagging/file-move in tag/import.