Add DownloadSourcePlugin contract + registry

`core/download_plugins/` defines the canonical interface every
download source must satisfy and the registry that holds them.
Single source of truth replacing the orchestrator's hardcoded
`[self.soulseek, self.youtube, ...]` lists scattered across 6+
dispatch sites.

Pure additive — no consumers wired through the registry yet.

core/download_plugins/__init__.py

@@ -0,0 +1,25 @@
+"""Download source plugin contract + registry.
+
+This package defines the canonical interface every download source
+(Soulseek, YouTube, Tidal, Qobuz, HiFi, Deezer, Lidarr, SoundCloud,
+and future additions like Usenet) must satisfy. The orchestrator
+dispatches through this contract instead of hardcoded
+`if self.youtube ... elif self.tidal ...` chains.
+
+This is the foundation step of a multi-commit refactor. Subsequent
+commits extract shared logic (background download worker, search
+query normalization, post-processing context building) into the
+contract so adding a new source becomes a one-class plugin instead
+of a 700+ LOC copy-paste loop.
+
+See `core/download_plugins/base.py` for the protocol contract and
+`core/download_plugins/registry.py` for the dispatch entry point.
+"""
+
+from core.download_plugins.base import DownloadSourcePlugin
+from core.download_plugins.registry import DownloadPluginRegistry
+
+__all__ = [
+    "DownloadSourcePlugin",
+    "DownloadPluginRegistry",
+]

core/download_plugins/base.py

@@ -0,0 +1,128 @@
+"""Canonical contract every download source plugin must satisfy.
+
+`DownloadSourcePlugin` is a structural Protocol — any class that
+implements these methods with matching signatures is automatically
+treated as a download source. No inheritance required, no manual
+registration required beyond the registry entry.
+
+The protocol is intentionally narrow — only the methods the
+orchestrator dispatches generically across all sources. Source-
+specific extras (Soulseek's slskd internals, Lidarr's album-only
+flow, etc.) stay on the underlying client and are accessed through
+the registry's typed accessor.
+
+This file is the FOUNDATION step. Existing client classes
+(SoulseekClient, YouTubeClient, TidalDownloadClient, etc.) already
+conform structurally — they grew the same shape independently
+because every consumer site needed the same calls. This file just
+makes the implicit contract explicit so:
+
+- Type checkers can flag drift if a new source forgets a method.
+- The orchestrator can iterate plugins generically instead of
+  hardcoding `[self.soulseek, self.youtube, ...]` everywhere.
+- Future PRs can move shared logic INTO the contract (a base
+  class with default implementations) without changing the
+  signature surface every consumer already depends on.
+"""
+
+from __future__ import annotations
+
+from typing import List, Optional, Protocol, Tuple, runtime_checkable
+
+# Soulseek client owns the canonical TrackResult / AlbumResult / DownloadStatus
+# dataclasses — every other source already imports from there. Keep it that way
+# until a future PR lifts those into a neutral location.
+from core.soulseek_client import AlbumResult, DownloadStatus, TrackResult
+
+
+@runtime_checkable
+class DownloadSourcePlugin(Protocol):
+    """Structural contract for a download source.
+
+    `runtime_checkable` lets `isinstance(client, DownloadSourcePlugin)`
+    work for the conformance test, but it ONLY checks method names —
+    not signatures or async-ness. The conformance test in
+    ``tests/test_download_plugin_conformance.py`` does the deeper
+    signature check.
+    """
+
+    # ------------------------------------------------------------------
+    # Configuration / lifecycle
+    # ------------------------------------------------------------------
+
+    def is_configured(self) -> bool:
+        """Return True iff this source has the credentials / settings
+        it needs to function. Used by the orchestrator to skip
+        unconfigured sources during hybrid fallback."""
+        ...
+
+    async def check_connection(self) -> bool:
+        """Probe the source's API / endpoint. Return True if the
+        source is reachable. May make a live network call."""
+        ...
+
+    # ------------------------------------------------------------------
+    # Search
+    # ------------------------------------------------------------------
+
+    async def search(
+        self,
+        query: str,
+        timeout: Optional[int] = None,
+        progress_callback=None,
+    ) -> Tuple[List[TrackResult], List[AlbumResult]]:
+        """Search the source for tracks (and albums where supported).
+
+        Returns a tuple of (track_results, album_results). Either
+        list may be empty. Sources that don't expose album-level
+        search return ``[]`` as the second element.
+        """
+        ...
+
+    # ------------------------------------------------------------------
+    # Download
+    # ------------------------------------------------------------------
+
+    async def download(
+        self,
+        username: str,
+        filename: str,
+        file_size: int = 0,
+    ) -> Optional[str]:
+        """Kick off a download. Returns a download_id string the
+        caller can poll via ``get_download_status``. Returns ``None``
+        if the source can't / won't handle this download.
+
+        ``username`` is the source-name string for streaming sources
+        (e.g. ``'youtube'``, ``'tidal'``) and the actual slskd peer
+        username for Soulseek. ``filename`` is source-specific —
+        Soulseek file path, YouTube ``video_id||title``, Tidal /
+        Qobuz ``track_id||display``, etc.
+        """
+        ...
+
+    async def get_all_downloads(self) -> List[DownloadStatus]:
+        """Return live status of all downloads currently tracked by
+        this source. The orchestrator concatenates results from
+        every plugin to build the global download list."""
+        ...
+
+    async def get_download_status(self, download_id: str) -> Optional[DownloadStatus]:
+        """Return status for a single download or ``None`` if this
+        source doesn't know about that download_id."""
+        ...
+
+    async def cancel_download(
+        self,
+        download_id: str,
+        username: Optional[str] = None,
+        remove: bool = False,
+    ) -> bool:
+        """Cancel an active download. ``remove=True`` also drops
+        the row from the source's active-downloads tracking."""
+        ...
+
+    async def clear_all_completed_downloads(self) -> bool:
+        """Drop completed downloads from active tracking. Sources
+        that don't keep completed history return True with no-op."""
+        ...

core/download_plugins/registry.py

@@ -0,0 +1,192 @@
+"""Plugin registry — single source of truth for which download
+sources exist, what their canonical names are, and which client
+class implements each.
+
+Replaces the orchestrator's hardcoded ``[self.soulseek,
+self.youtube, self.tidal, ...]`` lists and ``source_map`` dicts
+that historically had to be touched in 6+ places to add a source.
+With the registry:
+
+- One ``register()`` call adds a source to every dispatch path.
+- Iteration helpers replace hand-maintained lists.
+- The orchestrator stays oblivious to source-specific quirks.
+- Adding Usenet (planned) becomes a one-line registry entry plus
+  the new client class — no orchestrator changes.
+
+This is the foundation step. Subsequent commits move shared logic
+(thread workers, search query normalization, post-processing
+context building) out of the orchestrator and the per-source
+clients into helpers the registry exposes.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Callable, Dict, Iterator, List, Optional, Tuple
+
+from utils.logging_config import get_logger
+
+from core.download_plugins.base import DownloadSourcePlugin
+
+# Eager client imports — keep the import-order behavior the orchestrator
+# had before this refactor. Some integration tests inject mock modules
+# into ``sys.modules`` at collection time (see
+# ``tests/test_tidal_search_shortening.py``); making the registry's
+# factories lazy-import would cause tidalapi-dependent code to bind to
+# whichever ``tidalapi`` object happens to be in ``sys.modules`` at the
+# moment ``DownloadOrchestrator()`` is constructed — which is later
+# than the legacy module-top imports here. Importing everything at
+# registry-load time pins the bindings the same way the legacy
+# orchestrator did.
+from core.deezer_download_client import DeezerDownloadClient
+from core.hifi_client import HiFiClient
+from core.lidarr_download_client import LidarrDownloadClient
+from core.qobuz_client import QobuzClient
+from core.soulseek_client import SoulseekClient
+from core.soundcloud_client import SoundcloudClient
+from core.tidal_download_client import TidalDownloadClient
+from core.youtube_client import YouTubeClient
+
+logger = get_logger("download_plugins.registry")
+
+
+@dataclass(frozen=True)
+class PluginSpec:
+    """Static descriptor for a download source. The ``factory`` is
+    a zero-arg callable that builds the client instance — kept as a
+    callable rather than a class so each source can do its own
+    setup (e.g. SoulseekClient calls ``_setup_client`` after init,
+    Deezer reads ARL from config). ``aliases`` lets the registry
+    accept multiple historical names (e.g. ``deezer_dl`` is the
+    legacy alias for ``deezer``)."""
+
+    name: str
+    factory: Callable[[], DownloadSourcePlugin]
+    display_name: str
+    aliases: Tuple[str, ...] = field(default_factory=tuple)
+
+
+class DownloadPluginRegistry:
+    """Holds the live plugin instances + name → instance lookup.
+
+    Construction is two-phase:
+    1. Specs are registered (cheap — just stores callable refs).
+    2. ``initialize()`` calls each factory once and stores the
+       resulting client. Failures are caught and logged so one
+       broken source doesn't take down the orchestrator (mirrors
+       the existing ``_safe_init`` behavior).
+
+    Iteration helpers (``all_plugins``, ``configured_plugins``)
+    replace the hand-maintained lists scattered across the
+    orchestrator's ``get_all_downloads``, ``cancel_all_downloads``,
+    etc. so adding a source touches the registry alone.
+    """
+
+    def __init__(self) -> None:
+        self._specs: Dict[str, PluginSpec] = {}
+        self._instances: Dict[str, Optional[DownloadSourcePlugin]] = {}
+        self._init_failures: List[str] = []
+
+    def register(self, spec: PluginSpec) -> None:
+        """Register a plugin spec under its canonical name + each alias.
+        Aliases all resolve to the same instance after ``initialize``."""
+        if spec.name in self._specs:
+            raise ValueError(f"Plugin already registered: {spec.name}")
+        self._specs[spec.name] = spec
+
+    def initialize(self) -> None:
+        """Build every registered plugin's instance. Failures captured
+        in ``init_failures`` and the slot is set to None so the
+        orchestrator can skip unavailable sources without crashing."""
+        for spec in self._specs.values():
+            try:
+                instance = spec.factory()
+                self._instances[spec.name] = instance
+            except Exception as exc:
+                logger.error("%s download client failed to initialize: %s", spec.display_name, exc)
+                self._init_failures.append(spec.display_name)
+                self._instances[spec.name] = None
+
+    @property
+    def init_failures(self) -> List[str]:
+        return list(self._init_failures)
+
+    def get(self, name: str) -> Optional[DownloadSourcePlugin]:
+        """Resolve a name (or alias) to its plugin instance, or
+        None if the source failed to initialize / isn't registered."""
+        if not name:
+            return None
+        # Direct hit
+        if name in self._instances:
+            return self._instances[name]
+        # Alias lookup
+        for spec in self._specs.values():
+            if name in spec.aliases:
+                return self._instances.get(spec.name)
+        return None
+
+    def get_spec(self, name: str) -> Optional[PluginSpec]:
+        if name in self._specs:
+            return self._specs[name]
+        for spec in self._specs.values():
+            if name in spec.aliases:
+                return spec
+        return None
+
+    def display_name(self, name: str) -> str:
+        spec = self.get_spec(name)
+        return spec.display_name if spec else name
+
+    def names(self) -> List[str]:
+        """Canonical names of every registered source (regardless of
+        whether it initialized successfully)."""
+        return list(self._specs.keys())
+
+    def all_plugins(self) -> Iterator[Tuple[str, DownloadSourcePlugin]]:
+        """Yield (name, plugin) for every successfully-initialized
+        plugin. Replaces the orchestrator's hand-maintained client
+        lists in get_all_downloads / cancel_all_downloads / etc."""
+        for name, instance in self._instances.items():
+            if instance is not None:
+                yield name, instance
+
+    def configured_plugins(self) -> Iterator[Tuple[str, DownloadSourcePlugin]]:
+        """Yield (name, plugin) for every initialized AND configured
+        plugin. Useful for hybrid mode and any operation that should
+        skip sources the user hasn't set up."""
+        for name, instance in self.all_plugins():
+            try:
+                if instance.is_configured():
+                    yield name, instance
+            except Exception:
+                continue
+
+
+def build_default_registry() -> DownloadPluginRegistry:
+    """Construct the registry with SoulSync's eight built-in download
+    sources. Called once during orchestrator init.
+
+    Adding a new source (Usenet, etc.) means adding one ``register``
+    call here — no orchestrator changes required.
+
+    The factory itself is just the class constructor — clients are
+    imported eagerly at the top of this module so they bind to the
+    real third-party libs (tidalapi, etc.) at import time, not at
+    factory-call time. See the import-block comment above for why.
+    """
+    registry = DownloadPluginRegistry()
+
+    registry.register(PluginSpec(name='soulseek',  factory=SoulseekClient,         display_name='Soulseek'))
+    registry.register(PluginSpec(name='youtube',   factory=YouTubeClient,          display_name='YouTube'))
+    registry.register(PluginSpec(name='tidal',     factory=TidalDownloadClient,    display_name='Tidal'))
+    registry.register(PluginSpec(name='qobuz',     factory=QobuzClient,            display_name='Qobuz'))
+    registry.register(PluginSpec(name='hifi',      factory=HiFiClient,             display_name='HiFi'))
+    # 'deezer_dl' is the legacy name used in config + per-source dispatch
+    # strings (e.g. orchestrator's ``source_map``). Canonical name is
+    # ``deezer`` so future-facing code reads naturally.
+    registry.register(PluginSpec(name='deezer',    factory=DeezerDownloadClient,   display_name='Deezer',
+                                 aliases=('deezer_dl',)))
+    registry.register(PluginSpec(name='lidarr',    factory=LidarrDownloadClient,   display_name='Lidarr'))
+    registry.register(PluginSpec(name='soundcloud',factory=SoundcloudClient,       display_name='SoundCloud'))
+
+    return registry