diff --git a/shelfmark/release_sources/direct_download/__init__.py b/shelfmark/release_sources/direct_download/__init__.py
new file mode 100644
index 00000000..b9d20353
--- /dev/null
+++ b/shelfmark/release_sources/direct_download/__init__.py
@@ -0,0 +1,20 @@
+"""Direct Download release source and public entry points.
+
+Importing the source and handler classes registers them with Shelfmark.
+"""
+
+from shelfmark.release_sources.direct_download.annas_archive import search_books
+from shelfmark.release_sources.direct_download.common import DirectDownloadUnavailableError
+from shelfmark.release_sources.direct_download.handler import DirectDownloadHandler
+from shelfmark.release_sources.direct_download.source import DirectDownloadSource
+
+__all__ = [
+    "DirectDownloadUnavailableError",
+    "DirectDownloadHandler",
+    "DirectDownloadSource",
+    "SearchUnavailableError",
+    "search_books",
+]
+
+# Compatibility alias for integrations that imported the old module-level name.
+SearchUnavailableError = DirectDownloadUnavailableError
diff --git a/shelfmark/release_sources/direct_download.py b/shelfmark/release_sources/direct_download/annas_archive.py
similarity index 79%
rename from shelfmark/release_sources/direct_download.py
rename to shelfmark/release_sources/direct_download/annas_archive.py
index abbb5d73..c3855c23 100644
--- a/shelfmark/release_sources/direct_download.py
+++ b/shelfmark/release_sources/direct_download/annas_archive.py
@@ -1,9 +1,8 @@
-"""Direct download source - Anna's Archive/Libgen with fallback cascade."""
+"""Anna's Archive search, metadata parsing, and MD5 mirror download cascade."""
 
 import itertools
 import json
 import re
-import threading
 import time
 import unicodedata
 from contextlib import contextmanager
@@ -11,7 +10,7 @@ from contextvars import ContextVar
 from dataclasses import replace
 from http import HTTPStatus
 from pathlib import Path
-from typing import TYPE_CHECKING, ClassVar, NoReturn, TypedDict
+from typing import TYPE_CHECKING, NoReturn, TypedDict
 from urllib.parse import quote, urlparse
 
 import requests
@@ -19,30 +18,33 @@ from bs4 import BeautifulSoup, Tag
 from bs4.element import NavigableString
 
 from shelfmark.bypass.challenge import MAX_CHALLENGE_HTML_CHARS, challenge_marker
-from shelfmark.config.env import DEBUG_SKIP_SOURCES, TMP_DIR
+from shelfmark.config.env import DEBUG_SKIP_SOURCES
 from shelfmark.core import search_deadline
 from shelfmark.core.config import config
-from shelfmark.core.languages import language_alias_map
 from shelfmark.core.logger import setup_logger
-from shelfmark.core.models import DownloadTask, SearchFilters, build_filename
-from shelfmark.core.utils import CONTENT_TYPES, get_aa_content_type_dir
-from shelfmark.core.utils import is_audiobook as check_audiobook
+from shelfmark.core.models import SearchFilters
+from shelfmark.core.utils import CONTENT_TYPES
 from shelfmark.download import http as downloader
 from shelfmark.download import network
-from shelfmark.release_sources import (
-    BrowseRecord,
-    ColumnAlign,
-    ColumnColorHint,
-    ColumnRenderType,
-    ColumnSchema,
-    DownloadHandler,
-    Release,
-    ReleaseColumnConfig,
-    ReleaseProtocol,
-    ReleaseSource,
-    SourceUnavailableError,
-    register_handler,
-    register_source,
+from shelfmark.release_sources import BrowseRecord
+from shelfmark.release_sources.direct_download.common import (
+    MIN_VALID_FILE_SIZE as _MIN_VALID_FILE_SIZE,
+)
+from shelfmark.release_sources.direct_download.common import (
+    DirectDownloadUnavailableError,
+    ParsedSearchResult,
+    get_attr,
+    get_supported_formats,
+    html_response_text,
+    language_alias_to_code,
+    normalize_language_token,
+    normalize_requested_languages,
+    normalize_size,
+    parse_search_items,
+    parse_search_page,
+)
+from shelfmark.release_sources.direct_download.common import (
+    book_matches_requested_languages as _book_matches_requested_languages,
 )
 
 if TYPE_CHECKING:
@@ -67,18 +69,6 @@ def _raise_runtime_error(message: str) -> NoReturn:
     raise RuntimeError(message)
 
 
-def _coerce_str_list(value: object) -> list[str]:
-    """Return only string items from a config value."""
-    if not isinstance(value, list | tuple):
-        return []
-    return [item for item in value if isinstance(item, str)]
-
-
-def _get_supported_formats() -> list[str]:
-    """Return configured supported formats as a clean string list."""
-    return _coerce_str_list(config.SUPPORTED_FORMATS)
-
-
 def _parse_source_priority_entries(
     value: object,
     *,
@@ -108,13 +98,6 @@ def _parse_source_priority_entries(
     return entries
 
 
-def _html_response_text(response: str | tuple[str, str]) -> str:
-    """Extract the HTML body from downloader responses."""
-    if isinstance(response, tuple):
-        return response[0]
-    return response
-
-
 def _html_response_url(response: str | tuple[str, str]) -> str | None:
     """The URL that actually answered, when the downloader was asked to report it.
 
@@ -125,22 +108,6 @@ def _html_response_url(response: str | tuple[str, str]) -> str | None:
     return None
 
 
-def _attr_to_str(value: object) -> str | None:
-    """Convert a BeautifulSoup attribute value to a plain string."""
-    if isinstance(value, str):
-        return value
-    if isinstance(value, list):
-        for item in value:
-            if isinstance(item, str):
-                return item
-    return None
-
-
-def _get_attr(tag: Tag, attr: str) -> str | None:
-    """Safely fetch a tag attribute as a string."""
-    return _attr_to_str(tag.get(attr))
-
-
 def _first_stripped_text(tag: Tag | None) -> str | None:
     """Return the first non-empty stripped string from a tag."""
     if tag is None:
@@ -212,9 +179,9 @@ _DOWNLOAD_SOURCES = [
 ]
 
 _SOURCE_FAILURE_THRESHOLD = 4
-_MIN_VALID_FILE_SIZE = 10 * 1024
 _AA_COUNTDOWN_MAX_SECONDS = 300
 
+
 # --- Distant-path language detection ---
 
 _DISTANT_PATH_EXTENSIONS = (
@@ -252,8 +219,6 @@ _LANGUAGE_CODE_TOKEN_PATTERN = re.compile(
     r"(?:^|[\s_./\\\-\[(])([A-Za-z]{2,3})(?=$|[\s_./\\\-)\]])"
 )
 _LANGUAGE_NAME_TOKEN_PATTERN = re.compile(r"[a-z]{4,}(?:-[a-z0-9]+)?")
-_LANGUAGE_ALIAS_TO_CODE: dict[str, str] | None = None
-_LANGUAGE_ALIAS_LOCK = threading.Lock()
 _LANGUAGE_PLACEHOLDERS = frozenset({"", "-", "--", "unknown", "unk", "n/a", "na"})
 # Short codes that appear in common words — require bracket/key context to accept
 _AMBIGUOUS_SHORT_LANGUAGE_CODES = frozenset({"de", "en", "it", "la", "no", "or", "is", "in"})
@@ -269,34 +234,11 @@ def _is_language_from_path_enabled() -> bool:
     return bool(config.get("DIRECT_DOWNLOAD_LANGUAGE_FROM_PATH", False))
 
 
-def _normalize_language_token(value: str) -> str:
-    normalized = value.strip().lower()
-    for dash in ("‑", "–", "—", "−"):
-        normalized = normalized.replace(dash, "-")
-    return normalized
-
-
 def _fold_text(value: str) -> str:
     normalized = unicodedata.normalize("NFKD", value)
     return "".join(c for c in normalized if not unicodedata.combining(c)).lower()
 
 
-def _language_alias_to_code() -> dict[str, str]:
-    """Alias to code map, delegating to the shared language data."""
-    global _LANGUAGE_ALIAS_TO_CODE
-    cached = _LANGUAGE_ALIAS_TO_CODE
-    if cached is not None:
-        return cached
-
-    with _LANGUAGE_ALIAS_LOCK:
-        cached = _LANGUAGE_ALIAS_TO_CODE
-        if cached is not None:
-            return cached
-
-        _LANGUAGE_ALIAS_TO_CODE = language_alias_map()
-        return _LANGUAGE_ALIAS_TO_CODE
-
-
 def _extract_distant_path(row: Tag, *, enabled: bool) -> str | None:
     """Extract the Windows-style file path from an AA search result row."""
     if not enabled:
@@ -342,7 +284,7 @@ def _detect_language_from_distant_path(path: str | None) -> str | None:
     if not path:
         return None
 
-    aliases = _language_alias_to_code()
+    aliases = language_alias_to_code()
     if not aliases:
         return None
 
@@ -350,12 +292,12 @@ def _detect_language_from_distant_path(path: str | None) -> str | None:
     strong_candidates: list[str] = []
 
     for code in _BRACKETED_LANGUAGE_CODE_PATTERN.findall(path):
-        normalized = _normalize_language_token(code)
+        normalized = normalize_language_token(code)
         if normalized in aliases:
             strong_candidates.append(aliases[normalized])
 
     for code in _KEYED_LANGUAGE_CODE_PATTERN.findall(path):
-        normalized = _normalize_language_token(code)
+        normalized = normalize_language_token(code)
         if normalized in aliases:
             strong_candidates.append(aliases[normalized])
 
@@ -364,7 +306,7 @@ def _detect_language_from_distant_path(path: str | None) -> str | None:
         return non_ambiguous[0]
 
     for token in _LANGUAGE_NAME_TOKEN_PATTERN.findall(folded_path):
-        normalized = _normalize_language_token(token)
+        normalized = normalize_language_token(token)
         if normalized in aliases:
             candidate = aliases[normalized]
             if candidate not in _AMBIGUOUS_SHORT_LANGUAGE_CODES:
@@ -374,7 +316,7 @@ def _detect_language_from_distant_path(path: str | None) -> str | None:
         return strong_candidates[0]
 
     for code in _LANGUAGE_CODE_TOKEN_PATTERN.findall(path):
-        normalized = _normalize_language_token(code)
+        normalized = normalize_language_token(code)
         if normalized in _AMBIGUOUS_SHORT_LANGUAGE_CODES:
             continue
         if normalized in aliases:
@@ -386,38 +328,7 @@ def _detect_language_from_distant_path(path: str | None) -> str | None:
 def _is_missing_or_placeholder_language(language: str | None) -> bool:
     if language is None:
         return True
-    return _normalize_language_token(language) in _LANGUAGE_PLACEHOLDERS
-
-
-def _normalize_requested_languages(languages: list[str] | None) -> set[str]:
-    if not languages:
-        return set()
-    aliases = _language_alias_to_code()
-    normalized: set[str] = set()
-    for value in languages:
-        token = _normalize_language_token(str(value))
-        if not token or token == "all":  # noqa: S105 - "all" is a language sentinel
-            continue
-        normalized.add(aliases.get(token, token))
-    return normalized
-
-
-def _book_matches_requested_languages(book_language: str | None, requested: set[str]) -> bool:
-    """Return True when a book's language matches the requested filter.
-
-    Books with unknown/missing language always pass — the server-side &lang= filter
-    already narrowed the result set, so dropping unlabelled rows hides valid results.
-    """
-    if not requested:
-        return True
-    if not book_language:
-        return True
-    aliases = _language_alias_to_code()
-    normalized_book = aliases.get(
-        _normalize_language_token(book_language),
-        _normalize_language_token(book_language),
-    )
-    return normalized_book in requested
+    return normalize_language_token(language) in _LANGUAGE_PLACEHOLDERS
 
 
 def _is_configured_zlib_link(url: str) -> bool:
@@ -516,7 +427,7 @@ def _is_source_enabled(source_id: str) -> bool:
     return False
 
 
-def _get_direct_download_unavailable_reason() -> str | None:
+def get_unavailable_reason() -> str | None:
     """Return a user-facing reason when Direct Download cannot be used."""
     from shelfmark.core import mirrors
 
@@ -534,23 +445,14 @@ def _get_direct_download_unavailable_reason() -> str | None:
     return None
 
 
-def _ensure_direct_download_available() -> None:
+def ensure_available() -> None:
     """Raise a source-unavailable error when Direct Download is disabled or unconfigured."""
-    reason = _get_direct_download_unavailable_reason()
+    reason = get_unavailable_reason()
     if reason:
         raise SearchUnavailableError(reason)
 
 
-_SIZE_UNIT_PATTERN = re.compile(r"(kb|mb|gb|tb)", re.IGNORECASE)
-
-
-def _normalize_size(size_str: str) -> str:
-    """Normalize size string by uppercasing units (e.g., '5.2 mb' -> '5.2 MB')."""
-    return _SIZE_UNIT_PATTERN.sub(lambda m: m.group(1).upper(), size_str.strip())
-
-
-class SearchUnavailableError(SourceUnavailableError):
-    """Raised when Anna's Archive cannot be reached via any mirror/DNS."""
+SearchUnavailableError = DirectDownloadUnavailableError
 
 
 # Markers that prove a 200 really came from Anna's Archive, and markers that mean we
@@ -599,7 +501,7 @@ _search_page_cache: ContextVar[dict[str, tuple[str, Tag | None]] | None] = Conte
 
 
 @contextmanager
-def _search_page_reuse() -> Iterator[None]:
+def search_page_reuse() -> Iterator[None]:
     """Fetch each distinct AA search URL at most once per search.
 
     One search asks AA for the same URL more than once. The language-filter retry in
@@ -717,7 +619,7 @@ def _fetch_search_table_uncached(
             allow_bypasser_fallback=True,
             include_response_url=True,
         )
-        html = _html_response_text(response)
+        html = html_response_text(response)
         # Checked on the body, not on `response`: with include_response_url the give-up
         # shape is the tuple ("", url), and a tuple is truthy.
         if not html:
@@ -808,7 +710,7 @@ def search_books(query: str, filters: SearchFilters) -> list[BrowseRecord]:
     filters_query = ""
 
     path_language_enabled = _is_language_from_path_enabled()
-    requested_langs = _normalize_requested_languages(filters.lang)
+    requested_langs = normalize_requested_languages(filters.lang)
 
     # When path-language inference is on and a language is requested, skip the
     # server-side &lang= filter: lgli files often have no AA language metadata
@@ -826,7 +728,7 @@ def search_books(query: str, filters: SearchFilters) -> list[BrowseRecord]:
         for value in filters.content:
             filters_query += f"&content={quote(value)}"
 
-    formats_to_use = filters.format or _get_supported_formats()
+    formats_to_use = filters.format or get_supported_formats()
 
     index = 1
     for filter_type, filter_values in vars(filters).items():
@@ -860,16 +762,18 @@ def search_books(query: str, filters: SearchFilters) -> list[BrowseRecord]:
         msg = f"Expected results table tag, got {type(tbody).__name__}"
         raise TypeError(msg)
 
-    books = []
-    for line_tr in tbody.find_all("tr"):
-        book = _parse_search_result_row(line_tr)
-        if book:
-            books.append(book)
+    books = parse_search_page(
+        tbody,
+        filters,
+        provider_id="annas_archive",
+        item_selector="tr",
+        extract_item=_extract_aa_search_result,
+    )
 
     if path_language_enabled and requested_langs:
         books = [b for b in books if _book_matches_requested_languages(b.language, requested_langs)]
 
-    supported_formats = _get_supported_formats()
+    supported_formats = get_supported_formats()
 
     books.sort(
         key=lambda x: (
@@ -905,13 +809,13 @@ def get_book_info(book_id: str, *, fetch_download_count: bool = True) -> BrowseR
         )
         raise SearchUnavailableError(f"Unable to reach download source. {detail}")
 
-    soup = BeautifulSoup(_html_response_text(html), "html.parser")
+    soup = BeautifulSoup(html_response_text(html), "html.parser")
 
     return _parse_book_info_page(soup, book_id, fetch_download_count=fetch_download_count)
 
 
-def _parse_search_result_row(row: Tag) -> BrowseRecord | None:
-    """Parse a single search result row into a browse record."""
+def _extract_aa_search_result(row: Tag) -> ParsedSearchResult | None:
+    """Extract Anna's Archive table fields for the shared parser."""
     try:
         if row.text.strip().lower().startswith("your ad here"):
             return None
@@ -921,7 +825,7 @@ def _parse_search_result_row(row: Tag) -> BrowseRecord | None:
         if len(cells) < 11 or not anchors:
             return None
 
-        record_id = (_get_attr(anchors[0], "href") or "").split("/")[-1]
+        record_id = (get_attr(anchors[0], "href") or "").split("/")[-1]
         if not record_id:
             return None
 
@@ -929,7 +833,7 @@ def _parse_search_result_row(row: Tag) -> BrowseRecord | None:
         distant_path = _extract_distant_path(row, enabled=path_language_enabled)
 
         preview_img = cells[0].find("img")
-        preview = _get_attr(preview_img, "src") if isinstance(preview_img, Tag) else None
+        preview = get_attr(preview_img, "src") if isinstance(preview_img, Tag) else None
 
         title_span = cells[1].find("span")
         if isinstance(title_span, Tag):
@@ -963,25 +867,37 @@ def _parse_search_result_row(row: Tag) -> BrowseRecord | None:
             detected = _detect_language_from_distant_path(distant_path)
             language = detected or "unknown"
 
-        return BrowseRecord(
-            id=record_id,
+        return ParsedSearchResult(
+            key=record_id,
+            record_id=record_id,
             title=title,
-            source="direct_download",
+            formats=(file_format.lower(),),
             preview=preview,
             author=author,
             publisher=publisher,
             year=year,
             language=language,
             content=content.lower() if content else None,
-            format=file_format.lower() if file_format else None,
             size=size,
             download_path=distant_path,
+            source_url=f"{network.get_aa_base_url()}/md5/{record_id}",
         )
     except (AttributeError, IndexError, KeyError, TypeError) as e:
         logger.error_trace(f"Error parsing search result row: {e}")
         return None
 
 
+def _parse_search_result_row(row: Tag) -> BrowseRecord | None:
+    """Compatibility wrapper for parsing one Anna's Archive result row."""
+    records = parse_search_items(
+        [row],
+        None,
+        provider_id="annas_archive",
+        extract_item=_extract_aa_search_result,
+    )
+    return records[0] if records else None
+
+
 def _parse_book_info_page(
     soup: BeautifulSoup,
     book_id: str,
@@ -999,7 +915,7 @@ def _parse_book_info_page(
 
     node = data.select_one("div:nth-of-type(1) > img")
     if isinstance(node, Tag):
-        preview = _get_attr(node, "src") or ""
+        preview = get_attr(node, "src") or ""
 
     main_inner = next(
         (tag for tag in soup.find_all("div", {"class": "main-inner"}) if isinstance(tag, Tag)),
@@ -1023,7 +939,7 @@ def _parse_book_info_page(
     for anchor in soup.find_all("a"):
         try:
             text = anchor.text.strip().lower()
-            href = _get_attr(anchor, "href")
+            href = get_attr(anchor, "href")
             if not href:
                 continue
 
@@ -1075,7 +991,7 @@ def _parse_book_info_page(
     file_format = ""
     size = ""
     content = ""
-    supported_formats = _get_supported_formats()
+    supported_formats = get_supported_formats()
 
     for _details in all_details:
         _details = _details.split(" · ")
@@ -1084,7 +1000,7 @@ def _parse_book_info_page(
             if file_format == "" and stripped_lower in supported_formats:
                 file_format = f.strip().lower()
             if size == "" and any(u in f.strip().lower() for u in ("mb", "kb", "gb")):
-                size = _normalize_size(f)
+                size = normalize_size(f)
             if content == "":
                 for ct in CONTENT_TYPES:
                     if ct in f.strip().lower():
@@ -1096,7 +1012,7 @@ def _parse_book_info_page(
                 if file_format == "" and stripped and " " not in stripped:
                     file_format = stripped
                 if size == "" and "." in stripped:
-                    size = _normalize_size(f)
+                    size = normalize_size(f)
 
     book_title = (_find_in_divs(divs, "🔍") or [""])[0].strip("🔍").strip()
 
@@ -1134,7 +1050,7 @@ def _parse_book_info_page(
                 summary_url, selector=network.AAMirrorSelector(), allow_bypasser_fallback=False
             )
             if summary_response:
-                summary_data = json.loads(_html_response_text(summary_response))
+                summary_data = json.loads(html_response_text(summary_response))
                 if "downloads_total" in summary_data:
                     info["Downloads"] = [str(summary_data["downloads_total"])]
         except (
@@ -1456,11 +1372,11 @@ def _get_download_urls_from_welib(
         logger.warning("Welib page empty for %s", book_id)
         return []
 
-    soup = BeautifulSoup(_html_response_text(html), "html.parser")
+    soup = BeautifulSoup(html_response_text(html), "html.parser")
     links = [
         downloader.get_absolute_url(url, href)
         for a in soup.find_all("a", href=True)
-        if (href := _get_attr(a, "href")) and "/slow_download/" in href
+        if (href := get_attr(a, "href")) and "/slow_download/" in href
     ]
     return list(dict.fromkeys(links))  # Dedupe while preserving order
 
@@ -1524,7 +1440,7 @@ def _extract_libgen_download_url(link: str, cancel_flag: Event | None = None) ->
         return download_url
 
 
-def _download_book(
+def download_book(
     book_info: BrowseRecord,
     book_path: Path,
     progress_callback: Callable[[float], None] | None = None,
@@ -1650,7 +1566,7 @@ def _get_download_url(
         page = downloader.html_get_page(
             link, selector=sel, cancel_flag=cancel_flag, status_callback=status_callback
         )
-        page_data = json.loads(_html_response_text(page))
+        page_data = json.loads(html_response_text(page))
         download_url = page_data.get("download_url", "")
         return (
             downloader.get_absolute_url(link, download_url) if isinstance(download_url, str) else ""
@@ -1665,7 +1581,7 @@ def _get_download_url(
     if not html:
         return ""
 
-    soup = BeautifulSoup(_html_response_text(html), "html.parser")
+    soup = BeautifulSoup(html_response_text(html), "html.parser")
     url = ""
 
     # Z-Library
@@ -1678,9 +1594,9 @@ def _get_download_url(
                 link, selector=sel, cancel_flag=cancel_flag, status_callback=status_callback
             )
             if html:
-                soup = BeautifulSoup(_html_response_text(html), "html.parser")
+                soup = BeautifulSoup(html_response_text(html), "html.parser")
                 dl = soup.find("a", href=True, class_="addDownloadedBook")
-        url = (_get_attr(dl, "href") or "") if isinstance(dl, Tag) else ""
+        url = (get_attr(dl, "href") or "") if isinstance(dl, Tag) else ""
 
     # AA slow download / partner servers
     elif "/slow_download/" in link:
@@ -1693,7 +1609,7 @@ def _get_download_url(
             soup, "Download"
         )
         if get_btn:
-            url = _get_attr(get_btn, "href") or ""
+            url = get_attr(get_btn, "href") or ""
         else:
             logger.warning("Unknown source type, couldn't find download link: %s", link)
             url = ""
@@ -1727,11 +1643,11 @@ def _extract_slow_download_url(
         soup, "Download now", contains=True
     )
     if dl_link:
-        return _get_attr(dl_link, "href") or ""
+        return get_attr(dl_link, "href") or ""
 
     for a_tag in soup.find_all("a", href=True):
         if a_tag.has_attr("download"):
-            href = _get_attr(a_tag, "href")
+            href = get_attr(a_tag, "href")
             if not href:
                 continue
             if href.startswith("http") and "/slow_download/" not in href:
@@ -1762,7 +1678,7 @@ def _extract_slow_download_url(
         parent = copy_text.parent
         next_link = parent.find_next("a", href=True)
         if isinstance(next_link, Tag):
-            next_href = _get_attr(next_link, "href")
+            next_href = get_attr(next_link, "href")
             if next_href:
                 return next_href
         code_elem = parent.find_next("code")
@@ -1825,7 +1741,7 @@ def _extract_slow_download_url(
         )
         if not html:
             return ""
-        new_soup = BeautifulSoup(_html_response_text(html), "html.parser")
+        new_soup = BeautifulSoup(html_response_text(html), "html.parser")
         return _extract_slow_download_url(
             new_soup,
             link,
@@ -1915,119 +1831,48 @@ def _parse_countdown_seconds_from_element(element: Tag) -> int | None:
     return None
 
 
-def _browse_record_to_release(record: BrowseRecord) -> Release:
-    """Convert a browse record to a Release object.
+class AnnasArchiveProvider:
+    """Anna's Archive provider, including its specialized MD5 mirror cascade."""
 
-    This bridges the direct source's browse data to the generic release model.
-    """
-    return Release(
-        source=record.source,
-        source_id=record.id,
-        title=record.title,
-        format=record.format,
-        language=record.language,  # Top-level language for filtering
-        size=record.size,
-        download_url=record.download_urls[0] if record.download_urls else None,
-        info_url=f"{network.get_aa_base_url()}/md5/{record.id}",
-        protocol=ReleaseProtocol.HTTP,
-        indexer="Direct Download",
-        content_type=record.content,  # Preserve content type from source
-        extra={
-            "author": record.author,
-            "publisher": record.publisher,
-            "year": record.year,
-            "language": record.language,
-            "preview": record.preview,
-            "description": record.description,
-            "download_urls": record.download_urls,
-            "info": record.info,
-        },
-    )
-
-
-@register_source("direct_download")
-class DirectDownloadSource(ReleaseSource):
-    """Direct download source - searches web sources for books.
-
-    This wraps the search_books() functionality to provide releases
-    via the plugin interface.
-    """
-
-    name = "direct_download"
-    display_name = "Direct Download"
-    supported_content_types: ClassVar[list[str]] = ["ebook"]  # Direct downloads only support ebooks
+    id = "annas_archive"
+    display_name = "Anna's Archive"
 
     def __init__(self) -> None:
-        """Initialize per-instance search state for direct downloads."""
-        # Tracks which search method was used in the last search() call
-        # "isbn" = ISBN search returned results, "title_author" = title+author was used
-        self._last_search_type: str = "title_author"
+        self._last_search_type = "title_author"
 
     @property
     def last_search_type(self) -> str:
-        """Returns the search type used in the last search() call."""
         return self._last_search_type
 
-    def get_column_config(self) -> ReleaseColumnConfig:
-        """Column configuration for Direct Download source.
+    def is_enabled(self) -> bool:
+        from shelfmark.core import mirrors
 
-        Shows language, format, and size badges for each release.
-        Language is hidden on mobile; format and size are shown.
-        """
-        return ReleaseColumnConfig(
-            columns=[
-                ColumnSchema(
-                    key="extra.language",
-                    label="Language",
-                    render_type=ColumnRenderType.BADGE,
-                    align=ColumnAlign.CENTER,
-                    width="60px",
-                    hide_mobile=False,  # Language shown on mobile
-                    color_hint=ColumnColorHint(type="map", value="language"),
-                    uppercase=True,
-                ),
-                ColumnSchema(
-                    key="format",
-                    label="Format",
-                    render_type=ColumnRenderType.BADGE,
-                    align=ColumnAlign.CENTER,
-                    width="80px",
-                    hide_mobile=False,  # Format shown on mobile
-                    color_hint=ColumnColorHint(type="map", value="format"),
-                    uppercase=True,
-                ),
-                ColumnSchema(
-                    key="size",
-                    label="Size",
-                    render_type=ColumnRenderType.SIZE,
-                    align=ColumnAlign.CENTER,
-                    width="80px",
-                    hide_mobile=False,  # Size shown on mobile
-                ),
-            ],
-            grid_template="minmax(0,2fr) 60px 80px 80px",
-            supported_filters=["format", "language"],  # AA has reliable language metadata
+        return mirrors.has_aa_mirror_configuration()
+
+    def handles(self, url: str) -> bool:
+        from shelfmark.core import mirrors
+
+        hostname = (urlparse(url).hostname or "").lower().rstrip(".")
+        if not hostname:
+            return False
+        return any(
+            hostname == (urlparse(base_url).hostname or "").lower().rstrip(".")
+            for base_url in mirrors.get_aa_mirrors()
         )
 
-    def get_record(
-        self,
-        record_id: str,
-        *,
-        fetch_download_count: bool = True,
-    ) -> BrowseRecord | None:
-        """Resolve a direct-download record for direct-mode info/download flows."""
-        _ensure_direct_download_available()
+    def get_record(self, record_id: str, *, fetch_download_count: bool = True) -> BrowseRecord:
+        ensure_available()
         return get_book_info(record_id, fetch_download_count=fetch_download_count)
 
-    def search_results_are_releases(self) -> bool:
-        """Direct search results already represent concrete downloadable releases."""
-        return True
-
-    def get_destination_override(self, task: DownloadTask) -> Path | None:
-        """Apply Anna's Archive content-type routing when configured."""
-        if check_audiobook(task.content_type):
-            return None
-        return get_aa_content_type_dir(task.content_type)
+    def download(
+        self,
+        book_info: BrowseRecord,
+        book_path: Path,
+        progress_callback: Callable[[float], None] | None,
+        cancel_flag: Event | None,
+        status_callback: Callable[[str, str | None], None] | None,
+    ) -> str | None:
+        return download_book(book_info, book_path, progress_callback, cancel_flag, status_callback)
 
     def _search_books_with_language_fallback(
         self,
@@ -2055,14 +1900,15 @@ class DirectDownloadSource(ReleaseSource):
         *,
         expand_search: bool = False,
         content_type: str = "ebook",
-    ) -> list[Release]:
-        """Search for releases using the book's metadata.
-
-        The whole fan-out runs under one page cache, so a URL built twice by different
-        passes is fetched once. See `_search_page_reuse`.
-        """
-        with _search_page_reuse():
-            return self._search(book, plan, expand_search=expand_search, content_type=content_type)
+    ) -> list[BrowseRecord]:
+        """Search for releases using the book's metadata with request-local page reuse."""
+        with search_page_reuse():
+            return self._search(
+                book,
+                plan,
+                expand_search=expand_search,
+                content_type=content_type,
+            )
 
     def _search(
         self,
@@ -2071,8 +1917,8 @@ class DirectDownloadSource(ReleaseSource):
         *,
         expand_search: bool = False,
         content_type: str = "ebook",
-    ) -> list[Release]:
-        """Search for releases using the book's metadata.
+    ) -> list[BrowseRecord]:
+        """Run Anna's Archive's ISBN-first and localized-title search strategy.
 
         Priority: ISBN search first (most precise), then title+author fallback.
         For non-English languages, uses localized titles from book.titles_by_language.
@@ -2085,7 +1931,7 @@ class DirectDownloadSource(ReleaseSource):
             content_type: Ignored - Direct download uses format filtering instead
 
         """
-        _ensure_direct_download_available()
+        ensure_available()
         lang_filter = plan.languages
 
         # Reset search type tracking
@@ -2102,7 +1948,7 @@ class DirectDownloadSource(ReleaseSource):
                 query, filters, search_label="manual"
             )
             self._last_search_type = "manual" if query else "title_author"
-            return [_browse_record_to_release(record) for record in results]
+            return results
 
         # ISBN search first (unless expand_search requested)
         if plan.manual_query:
@@ -2119,7 +1965,7 @@ class DirectDownloadSource(ReleaseSource):
                     if results:
                         logger.info("Found %s releases via ISBN", len(results))
                         self._last_search_type = "isbn"
-                        return [_browse_record_to_release(record) for record in results]
+                        return results
                     logger.debug("No ISBN results, falling back to title+author")
                 except SearchUnavailableError:
                     raise
@@ -2184,155 +2030,4 @@ class DirectDownloadSource(ReleaseSource):
                 except Exception:
                     logger.exception("Search error")
 
-        return [_browse_record_to_release(record) for record in all_results]
-
-    def is_available(self) -> bool:
-        """Check if Direct Download has been explicitly enabled and configured."""
-        return _get_direct_download_unavailable_reason() is None
-
-
-@register_handler("direct_download")
-class DirectDownloadHandler(DownloadHandler):
-    """Handler for direct HTTP downloads from Anna's Archive, Libgen, etc.
-
-    Receives a DownloadTask with task_id (AA MD5 hash) and cascades through
-    sources in priority order. The AA page is only fetched if AA slow sources
-    are enabled in the user's source priority configuration.
-    """
-
-    def download(
-        self,
-        task: DownloadTask,
-        cancel_flag: Event,
-        progress_callback: Callable[[float], None],
-        status_callback: Callable[[str, str | None], None],
-    ) -> str | None:
-        """Execute a direct HTTP download.
-
-        Uses task.task_id (AA MD5 hash) to cascade through sources in priority
-        order. The AA page is only fetched if AA slow sources are enabled.
-
-        Args:
-            task: Download task with task_id (AA MD5 hash)
-            cancel_flag: Event to check for cancellation
-            progress_callback: Called with progress percentage (0-100)
-            status_callback: Called with (status, message) for status updates
-
-        Returns:
-            Path to downloaded file if successful, None otherwise
-
-        """
-        try:
-            # Check for cancellation before starting
-            if cancel_flag.is_set():
-                logger.info("Download cancelled before starting: %s", task.task_id)
-                status_callback("cancelled", "Cancelled")
-                return None
-
-            # Create browse record from task data - NO AA page fetch here
-            # AA page is fetched lazily by _fetch_aa_page_urls only when
-            # we actually reach an AA slow source in the priority order
-            book_info = BrowseRecord(
-                id=task.task_id,
-                title=task.title,
-                source="direct_download",
-                author=task.author,
-                year=task.year,
-                format=task.format,
-                size=task.size,
-                preview=task.preview,
-            )
-
-            return self._execute_download(
-                book_info, cancel_flag, progress_callback, status_callback
-            )
-
-        except Exception as e:
-            if cancel_flag.is_set():
-                logger.info("Download cancelled during error handling: %s", task.task_id)
-                status_callback("cancelled", "Cancelled")
-            else:
-                logger.exception("Error downloading book")
-                status_callback("error", str(e))
-            return None
-
-    def _execute_download(
-        self,
-        book_info: BrowseRecord,
-        cancel_flag: Event,
-        progress_callback: Callable[[float], None],
-        status_callback: Callable[[str, str | None], None],
-    ) -> str | None:
-        """Execute the direct-download flow with a fetched browse record.
-
-        This contains the core download logic: cascade through sources,
-        handle bypass, move to final location.
-        """
-        try:
-            logger.debug("Starting download: %s", book_info.title)
-
-            # Prepare paths - use descriptive staging filename, orchestrator will rename
-            # based on FILE_ORGANIZATION setting
-            file_org = config.get("FILE_ORGANIZATION", "rename")
-            if file_org == "none":
-                book_name = f"{book_info.id}.{book_info.format or 'bin'}"
-            else:
-                book_name = build_filename(
-                    book_info.title,
-                    book_info.author,
-                    book_info.year,
-                    book_info.format,
-                )
-            book_path = TMP_DIR / book_name
-
-            # Check cancellation before download
-            if cancel_flag.is_set():
-                logger.info("Download cancelled before download call: %s", book_info.id)
-                status_callback("cancelled", "Cancelled")
-                return None
-
-            # Execute download via _download_book (handles cascade and bypass)
-            status_callback("resolving", "Finding download source")
-            success_url = _download_book(
-                book_info, book_path, progress_callback, cancel_flag, status_callback
-            )
-
-            # Check for cancellation after download
-            if cancel_flag.is_set():
-                logger.info("Download cancelled during download: %s", book_info.id)
-                if book_path.exists():
-                    book_path.unlink()
-                status_callback("cancelled", "Cancelled")
-                return None
-
-            if not success_url:
-                if network.dns_interference_detected():
-                    status_callback(
-                        "error",
-                        "All sources failed - your network/ISP appears to be blocking "
-                        "Anna's Archive. Enable DNS-over-HTTPS in settings.",
-                    )
-                else:
-                    status_callback("error", "All download sources failed")
-                return None
-
-            # Return temp path - orchestrator handles post-processing (archive extraction, ingest)
-            return str(book_path)
-
-        except Exception:
-            if cancel_flag.is_set():
-                logger.info("Download cancelled during error handling: %s", book_info.id)
-                status_callback("cancelled", "Cancelled")
-            else:
-                logger.exception("Error downloading book")
-            return None
-
-    def cancel(self, task_id: str) -> bool:
-        """Cancel an in-progress download.
-
-        Cancellation is handled via the cancel_flag passed to download().
-        This method exists for the interface but actual cancellation
-        happens through the Event flag mechanism.
-        """
-        # Cancellation is handled by the orchestrator via cancel_flag
-        return False
+        return all_results
diff --git a/shelfmark/release_sources/direct_download/common.py b/shelfmark/release_sources/direct_download/common.py
new file mode 100644
index 00000000..e1348941
--- /dev/null
+++ b/shelfmark/release_sources/direct_download/common.py
@@ -0,0 +1,264 @@
+"""Shared contracts and result normalization for Direct Download websites."""
+
+import hashlib
+import re
+import threading
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Protocol, runtime_checkable
+
+from bs4 import BeautifulSoup, Tag
+
+from shelfmark.core.config import config
+from shelfmark.core.languages import language_alias_map
+from shelfmark.release_sources import BrowseRecord, SourceUnavailableError
+
+if TYPE_CHECKING:
+    from collections.abc import Callable, Iterable
+    from pathlib import Path
+    from threading import Event
+
+    from shelfmark.core.models import SearchFilters
+    from shelfmark.core.search_plan import ReleaseSearchPlan
+    from shelfmark.metadata_providers import BookMetadata
+
+_LANGUAGE_ALIAS_TO_CODE: dict[str, str] | None = None
+_LANGUAGE_ALIAS_LOCK = threading.Lock()
+_SIZE_UNIT_PATTERN = re.compile(r"(kb|mb|gb|tb)", re.IGNORECASE)
+MIN_VALID_FILE_SIZE = 10 * 1024
+
+
+class DirectDownloadUnavailableError(SourceUnavailableError):
+    """Raised when the composite Direct Download source cannot be reached."""
+
+
+def coerce_str_list(value: object) -> list[str]:
+    """Return only string items from a config value."""
+    if not isinstance(value, list | tuple):
+        return []
+    return [item for item in value if isinstance(item, str)]
+
+
+def get_supported_formats() -> list[str]:
+    """Return configured supported formats as a clean string list."""
+    return coerce_str_list(config.SUPPORTED_FORMATS)
+
+
+def html_response_text(response: str | tuple[str, str]) -> str:
+    """Extract the HTML body from downloader responses."""
+    if isinstance(response, tuple):
+        return response[0]
+    return response
+
+
+def attr_to_str(value: object) -> str | None:
+    """Convert a BeautifulSoup attribute value to a plain string."""
+    if isinstance(value, str):
+        return value
+    if isinstance(value, list):
+        for item in value:
+            if isinstance(item, str):
+                return item
+    return None
+
+
+def get_attr(tag: Tag, attr: str) -> str | None:
+    """Safely fetch a tag attribute as a string."""
+    return attr_to_str(tag.get(attr))
+
+
+@dataclass(frozen=True)
+class ParsedSearchResult:
+    """Provider-neutral fields extracted from one search-result element."""
+
+    key: str
+    title: str
+    formats: tuple[str, ...]
+    record_id: str | None = None
+    author: str | None = None
+    publisher: str | None = None
+    year: str | None = None
+    language: str | None = None
+    content: str | None = None
+    size: str | None = None
+    preview: str | None = None
+    source_url: str | None = None
+    download_path: str | None = None
+
+
+class DirectDownloadProvider(Protocol):
+    """Provider lifecycle used by the composite Direct Download source."""
+
+    id: str
+    display_name: str
+
+    def is_enabled(self) -> bool: ...
+
+    def handles(self, url: str) -> bool: ...
+
+    def search(
+        self,
+        book: BookMetadata,
+        plan: ReleaseSearchPlan,
+        *,
+        expand_search: bool = False,
+        content_type: str = "ebook",
+    ) -> list[BrowseRecord]: ...
+
+    def download(
+        self,
+        book_info: BrowseRecord,
+        book_path: Path,
+        progress_callback: Callable[[float], None] | None,
+        cancel_flag: Event | None,
+        status_callback: Callable[[str, str | None], None] | None,
+    ) -> str | None: ...
+
+
+@runtime_checkable
+class RecordLookupProvider(Protocol):
+    """Optional capability for providers that can reopen source-native records."""
+
+    def get_record(
+        self, record_id: str, *, fetch_download_count: bool = True
+    ) -> BrowseRecord | None: ...
+
+
+def normalize_language_token(value: str) -> str:
+    normalized = value.strip().lower()
+    for dash in ("‑", "–", "—", "−"):
+        normalized = normalized.replace(dash, "-")
+    return normalized
+
+
+def language_alias_to_code() -> dict[str, str]:
+    """Alias to code map, delegating to the shared language data."""
+    global _LANGUAGE_ALIAS_TO_CODE
+    cached = _LANGUAGE_ALIAS_TO_CODE
+    if cached is not None:
+        return cached
+
+    with _LANGUAGE_ALIAS_LOCK:
+        cached = _LANGUAGE_ALIAS_TO_CODE
+        if cached is not None:
+            return cached
+
+        _LANGUAGE_ALIAS_TO_CODE = language_alias_map()
+        return _LANGUAGE_ALIAS_TO_CODE
+
+
+def normalize_requested_languages(languages: list[str] | None) -> set[str]:
+    if not languages:
+        return set()
+    aliases = language_alias_to_code()
+    normalized: set[str] = set()
+    for value in languages:
+        token = normalize_language_token(str(value))
+        if not token or token == "all":  # noqa: S105 - "all" is a language sentinel
+            continue
+        normalized.add(aliases.get(token, token))
+    return normalized
+
+
+def book_matches_requested_languages(book_language: str | None, requested: set[str]) -> bool:
+    """Return True when a book's language matches the requested filter.
+
+    Books with unknown/missing language always pass — the server-side &lang= filter
+    already narrowed the result set, so dropping unlabelled rows hides valid results.
+    """
+    if not requested:
+        return True
+    if not book_language:
+        return True
+    aliases = language_alias_to_code()
+    normalized_book = aliases.get(
+        normalize_language_token(book_language),
+        normalize_language_token(book_language),
+    )
+    return normalized_book in requested
+
+
+def normalize_size(size_str: str) -> str:
+    """Normalize size string by uppercasing units (e.g., '5.2 mb' -> '5.2 MB')."""
+    return _SIZE_UNIT_PATTERN.sub(lambda m: m.group(1).upper(), size_str.strip())
+
+
+def parse_search_items(
+    items: Iterable[Tag],
+    filters: SearchFilters | None,
+    *,
+    provider_id: str,
+    extract_item: Callable[[Tag], ParsedSearchResult | None],
+) -> list[BrowseRecord]:
+    """Normalize provider-specific HTML elements into Direct Download records.
+
+    Providers only describe how fields are extracted from their DOM. Language and
+    format filtering, stable IDs, and BrowseRecord construction stay shared.
+    """
+    requested_languages = normalize_requested_languages(filters.lang) if filters else set()
+    requested_formats = (
+        {value.casefold() for value in (filters.format or get_supported_formats())}
+        if filters
+        else set()
+    )
+    records: list[BrowseRecord] = []
+
+    for item in items:
+        parsed = extract_item(item)
+        if parsed is None:
+            continue
+
+        normalized_language = normalize_language_token(parsed.language) if parsed.language else ""
+        language = language_alias_to_code().get(normalized_language, normalized_language) or None
+        if not book_matches_requested_languages(language, requested_languages):
+            continue
+
+        formats = parsed.formats or ("",)
+        for book_format in formats:
+            normalized_format = book_format.casefold()
+            if (
+                normalized_format
+                and requested_formats
+                and normalized_format not in requested_formats
+            ):
+                continue
+            record_id = parsed.record_id
+            if not record_id or len(formats) > 1:
+                source_key = f"{parsed.key}#{normalized_format}"
+                digest = hashlib.blake2b(source_key.encode(), digest_size=16).hexdigest()
+                record_id = f"{provider_id}:{digest}"
+            records.append(
+                BrowseRecord(
+                    id=record_id,
+                    title=parsed.title,
+                    source="direct_download",
+                    author=parsed.author,
+                    publisher=parsed.publisher,
+                    year=parsed.year,
+                    language=language,
+                    format=normalized_format or None,
+                    size=parsed.size,
+                    preview=parsed.preview,
+                    content=parsed.content,
+                    source_url=parsed.source_url,
+                    download_path=parsed.download_path,
+                )
+            )
+    return records
+
+
+def parse_search_page(
+    page: str | BeautifulSoup | Tag,
+    filters: SearchFilters | None,
+    *,
+    provider_id: str,
+    item_selector: str,
+    extract_item: Callable[[Tag], ParsedSearchResult | None],
+) -> list[BrowseRecord]:
+    """Parse a result page using provider-specific selectors and extraction."""
+    root = BeautifulSoup(page, "html.parser") if isinstance(page, str) else page
+    return parse_search_items(
+        (item for item in root.select(item_selector) if isinstance(item, Tag)),
+        filters,
+        provider_id=provider_id,
+        extract_item=extract_item,
+    )
diff --git a/shelfmark/release_sources/direct_download/handler.py b/shelfmark/release_sources/direct_download/handler.py
new file mode 100644
index 00000000..59d69834
--- /dev/null
+++ b/shelfmark/release_sources/direct_download/handler.py
@@ -0,0 +1,177 @@
+"""Direct Download routing, staging, and cancellation."""
+
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+from shelfmark.config.env import TMP_DIR
+from shelfmark.core.config import config
+from shelfmark.core.logger import setup_logger
+from shelfmark.core.models import DownloadTask, build_filename
+from shelfmark.download import network
+from shelfmark.release_sources import (
+    BrowseRecord,
+    DownloadHandler,
+    register_handler,
+)
+from shelfmark.release_sources.direct_download import registry
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+    from pathlib import Path
+    from threading import Event
+
+
+logger = setup_logger(__name__)
+
+
+def _download_book(
+    book_info: BrowseRecord,
+    book_path: Path,
+    progress_callback: Callable[[float], None] | None = None,
+    cancel_flag: Event | None = None,
+    status_callback: Callable[[str, str | None], None] | None = None,
+) -> str | None:
+    """Route a website record or an Anna's Archive MD5 to its download flow."""
+    provider = registry.provider_for_record(book_info)
+    if provider is None:
+        msg = f"No Direct Download provider owns record {book_info.id!r}"
+        raise RuntimeError(msg)
+    return provider.download(book_info, book_path, progress_callback, cancel_flag, status_callback)
+
+
+@register_handler("direct_download")
+class DirectDownloadHandler(DownloadHandler):
+    """Route and stage downloads from registered Direct Download providers."""
+
+    def download(
+        self,
+        task: DownloadTask,
+        cancel_flag: Event,
+        progress_callback: Callable[[float], None],
+        status_callback: Callable[[str, str | None], None],
+    ) -> str | None:
+        """Execute a provider-owned direct HTTP download.
+
+        Args:
+            task: Download task with a provider-owned source ID
+            cancel_flag: Event to check for cancellation
+            progress_callback: Called with progress percentage (0-100)
+            status_callback: Called with (status, message) for status updates
+
+        Returns:
+            Path to downloaded file if successful, None otherwise
+
+        """
+        try:
+            # Check for cancellation before starting
+            if cancel_flag.is_set():
+                logger.info("Download cancelled before starting: %s", task.task_id)
+                status_callback("cancelled", "Cancelled")
+                return None
+
+            # Reconstruct the provider-owned record without resolving it again.
+            book_info = BrowseRecord(
+                id=task.task_id,
+                title=task.title,
+                source="direct_download",
+                author=task.author,
+                year=task.year,
+                format=task.format,
+                size=task.size,
+                preview=task.preview,
+                source_url=task.source_url,
+            )
+
+            return self._execute_download(
+                book_info, cancel_flag, progress_callback, status_callback
+            )
+
+        except Exception as e:
+            if cancel_flag.is_set():
+                logger.info("Download cancelled during error handling: %s", task.task_id)
+                status_callback("cancelled", "Cancelled")
+            else:
+                logger.exception("Error downloading book")
+                status_callback("error", str(e))
+            return None
+
+    def _execute_download(
+        self,
+        book_info: BrowseRecord,
+        cancel_flag: Event,
+        progress_callback: Callable[[float], None],
+        status_callback: Callable[[str, str | None], None],
+    ) -> str | None:
+        """Execute the direct-download flow with a fetched browse record.
+
+        This contains the core download logic: cascade through sources,
+        handle bypass, move to final location.
+        """
+        try:
+            logger.debug("Starting download: %s", book_info.title)
+
+            # Prepare paths - use descriptive staging filename, orchestrator will rename
+            # based on FILE_ORGANIZATION setting
+            file_org = config.get("FILE_ORGANIZATION", "rename")
+            if file_org == "none":
+                book_name = f"{book_info.id}.{book_info.format or 'bin'}"
+            else:
+                book_name = build_filename(
+                    book_info.title,
+                    book_info.author,
+                    book_info.year,
+                    book_info.format,
+                )
+            book_path = TMP_DIR / book_name
+
+            # Check cancellation before download
+            if cancel_flag.is_set():
+                logger.info("Download cancelled before download call: %s", book_info.id)
+                status_callback("cancelled", "Cancelled")
+                return None
+
+            # Execute download via _download_book (handles cascade and bypass)
+            status_callback("resolving", "Finding download source")
+            success_url = _download_book(
+                book_info, book_path, progress_callback, cancel_flag, status_callback
+            )
+
+            # Check for cancellation after download
+            if cancel_flag.is_set():
+                logger.info("Download cancelled during download: %s", book_info.id)
+                if book_path.exists():
+                    book_path.unlink()
+                status_callback("cancelled", "Cancelled")
+                return None
+
+            if not success_url:
+                if network.dns_interference_detected():
+                    status_callback(
+                        "error",
+                        "All sources failed - your network/ISP appears to be blocking "
+                        "Anna's Archive. Enable DNS-over-HTTPS in settings.",
+                    )
+                else:
+                    status_callback("error", "All download sources failed")
+                return None
+
+            # Return temp path - orchestrator handles post-processing (archive extraction, ingest)
+            return str(book_path)
+
+        except Exception:
+            if cancel_flag.is_set():
+                logger.info("Download cancelled during error handling: %s", book_info.id)
+                status_callback("cancelled", "Cancelled")
+            else:
+                logger.exception("Error downloading book")
+            return None
+
+    def cancel(self, task_id: str) -> bool:
+        """Cancel an in-progress download.
+
+        Cancellation is handled via the cancel_flag passed to download().
+        This method exists for the interface but actual cancellation
+        happens through the Event flag mechanism.
+        """
+        # Cancellation is handled by the orchestrator via cancel_flag
+        return False
diff --git a/shelfmark/release_sources/direct_download/registry.py b/shelfmark/release_sources/direct_download/registry.py
new file mode 100644
index 00000000..bb8e422f
--- /dev/null
+++ b/shelfmark/release_sources/direct_download/registry.py
@@ -0,0 +1,93 @@
+"""Provider composition and dispatch for the Direct Download release source."""
+
+import re
+from typing import TYPE_CHECKING
+
+from shelfmark.core.config import config
+from shelfmark.release_sources.direct_download.annas_archive import AnnasArchiveProvider
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+    from shelfmark.release_sources import BrowseRecord
+    from shelfmark.release_sources.direct_download.common import DirectDownloadProvider
+
+
+PROVIDER_TYPES = (AnnasArchiveProvider,)
+_AA_MD5_PATTERN = re.compile(r"^[0-9a-f]{32}$", re.IGNORECASE)
+
+
+def create_providers() -> tuple[DirectDownloadProvider, ...]:
+    """Create request-local providers so mutable search state is not shared."""
+    return tuple(provider_type() for provider_type in PROVIDER_TYPES)
+
+
+def enabled_providers(
+    providers: Sequence[DirectDownloadProvider] | None = None,
+) -> tuple[DirectDownloadProvider, ...]:
+    if not config.get("DIRECT_DOWNLOAD_ENABLED", False):
+        return ()
+    candidates = providers if providers is not None else create_providers()
+    return tuple(provider for provider in candidates if provider.is_enabled())
+
+
+def get_unavailable_reason(
+    providers: Sequence[DirectDownloadProvider] | None = None,
+) -> str | None:
+    if not config.get("DIRECT_DOWNLOAD_ENABLED", False):
+        return "Direct Download is disabled. Enable the source in Settings."
+    if not enabled_providers(providers):
+        return (
+            "Direct Download is not configured. Enable and configure at least one "
+            "download provider in Settings."
+        )
+    return None
+
+
+def provider_by_id(
+    provider_id: str | None,
+    providers: Sequence[DirectDownloadProvider] | None = None,
+) -> DirectDownloadProvider | None:
+    if not provider_id:
+        return None
+    candidates = providers if providers is not None else create_providers()
+    return next((provider for provider in candidates if provider.id == provider_id), None)
+
+
+def provider_for_record(
+    record: BrowseRecord,
+    providers: Sequence[DirectDownloadProvider] | None = None,
+) -> DirectDownloadProvider | None:
+    """Resolve a record explicitly, retaining safe compatibility with legacy tasks."""
+    candidates = providers if providers is not None else create_providers()
+    prefix, separator, _remainder = record.id.partition(":")
+    if separator:
+        provider = provider_by_id(prefix, candidates)
+        if provider is not None:
+            return provider
+
+    if record.source_url:
+        provider = next(
+            (provider for provider in candidates if provider.handles(record.source_url)),
+            None,
+        )
+        if provider is not None:
+            return provider
+
+    # Anna's Archive records historically carried only their raw MD5. Preserve those
+    # persisted tasks without treating arbitrary unknown URLs as Anna's Archive.
+    if _AA_MD5_PATTERN.fullmatch(record.id):
+        return provider_by_id("annas_archive", candidates)
+    return None
+
+
+def provider_for_record_id(
+    record_id: str,
+    providers: Sequence[DirectDownloadProvider] | None = None,
+) -> DirectDownloadProvider | None:
+    candidates = providers if providers is not None else create_providers()
+    prefix, separator, _remainder = record_id.partition(":")
+    if separator:
+        return provider_by_id(prefix, candidates)
+    # Record lookup predates provider-qualified IDs, so unqualified IDs are AA IDs.
+    return provider_by_id("annas_archive", candidates)
diff --git a/shelfmark/release_sources/direct_download/source.py b/shelfmark/release_sources/direct_download/source.py
new file mode 100644
index 00000000..9a23bfdb
--- /dev/null
+++ b/shelfmark/release_sources/direct_download/source.py
@@ -0,0 +1,204 @@
+"""Direct Download search and release-source integration."""
+
+from pathlib import Path
+from typing import TYPE_CHECKING, ClassVar
+
+import requests
+
+from shelfmark.core.logger import setup_logger
+from shelfmark.core.utils import get_aa_content_type_dir
+from shelfmark.core.utils import is_audiobook as check_audiobook
+from shelfmark.release_sources import (
+    BrowseRecord,
+    ColumnAlign,
+    ColumnColorHint,
+    ColumnRenderType,
+    ColumnSchema,
+    Release,
+    ReleaseColumnConfig,
+    ReleaseProtocol,
+    ReleaseSource,
+    SourceUnavailableError,
+    register_source,
+)
+from shelfmark.release_sources.direct_download import registry
+from shelfmark.release_sources.direct_download.common import (
+    DirectDownloadUnavailableError,
+    RecordLookupProvider,
+)
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    from shelfmark.core.models import DownloadTask
+    from shelfmark.core.search_plan import ReleaseSearchPlan
+    from shelfmark.metadata_providers import BookMetadata
+
+logger = setup_logger(__name__)
+
+
+def _browse_record_to_release(record: BrowseRecord) -> Release:
+    """Convert a browse record to a Release object.
+
+    This bridges the direct source's browse data to the generic release model.
+    """
+    provider = registry.provider_for_record(record)
+    provider_id = provider.id if provider is not None else None
+    return Release(
+        source=record.source,
+        source_id=record.id,
+        title=record.title,
+        format=record.format,
+        language=record.language,  # Top-level language for filtering
+        size=record.size,
+        download_url=record.source_url
+        or (record.download_urls[0] if record.download_urls else None),
+        info_url=record.source_url,
+        protocol=ReleaseProtocol.HTTP,
+        indexer="Direct Download",
+        content_type=record.content,  # Preserve content type from source
+        extra={
+            "author": record.author,
+            "publisher": record.publisher,
+            "year": record.year,
+            "language": record.language,
+            "preview": record.preview,
+            "description": record.description,
+            "download_urls": record.download_urls,
+            "info": record.info,
+            "direct_download_provider": provider_id,
+            # Kept for older frontends and persisted request payloads.
+            "web_provider": provider_id if provider_id != "annas_archive" else None,
+        },
+    )
+
+
+@register_source("direct_download")
+class DirectDownloadSource(ReleaseSource):
+    """Direct download source - searches web sources for books.
+
+    This wraps the search_books() functionality to provide releases
+    via the plugin interface.
+    """
+
+    name = "direct_download"
+    display_name = "Direct Download"
+    supported_content_types: ClassVar[list[str]] = ["ebook"]  # Direct downloads only support ebooks
+
+    def __init__(self) -> None:
+        """Initialize per-instance search state for direct downloads."""
+        self._providers = registry.create_providers()
+
+    @property
+    def last_search_type(self) -> str:
+        """Returns the search type used in the last search() call."""
+        provider = registry.provider_by_id("annas_archive", self._providers)
+        return str(getattr(provider, "last_search_type", "title_author"))
+
+    def get_column_config(self) -> ReleaseColumnConfig:
+        """Column configuration for Direct Download source.
+
+        Shows language, format, and size badges for each release.
+        Language is hidden on mobile; format and size are shown.
+        """
+        return ReleaseColumnConfig(
+            columns=[
+                ColumnSchema(
+                    key="extra.language",
+                    label="Language",
+                    render_type=ColumnRenderType.BADGE,
+                    align=ColumnAlign.CENTER,
+                    width="60px",
+                    hide_mobile=False,  # Language shown on mobile
+                    color_hint=ColumnColorHint(type="map", value="language"),
+                    uppercase=True,
+                ),
+                ColumnSchema(
+                    key="format",
+                    label="Format",
+                    render_type=ColumnRenderType.BADGE,
+                    align=ColumnAlign.CENTER,
+                    width="80px",
+                    hide_mobile=False,  # Format shown on mobile
+                    color_hint=ColumnColorHint(type="map", value="format"),
+                    uppercase=True,
+                ),
+                ColumnSchema(
+                    key="size",
+                    label="Size",
+                    render_type=ColumnRenderType.SIZE,
+                    align=ColumnAlign.CENTER,
+                    width="80px",
+                    hide_mobile=False,  # Size shown on mobile
+                ),
+            ],
+            grid_template="minmax(0,2fr) 60px 80px 80px",
+            supported_filters=["format", "language"],  # AA has reliable language metadata
+        )
+
+    def get_record(
+        self,
+        record_id: str,
+        *,
+        fetch_download_count: bool = True,
+    ) -> BrowseRecord | None:
+        """Resolve a direct-download record for direct-mode info/download flows."""
+        provider = registry.provider_for_record_id(record_id, self._providers)
+        if provider is None or not isinstance(provider, RecordLookupProvider):
+            return None
+        native_id = record_id.partition(":")[2] or record_id
+        return provider.get_record(native_id, fetch_download_count=fetch_download_count)
+
+    def search_results_are_releases(self) -> bool:
+        """Direct search results already represent concrete downloadable releases."""
+        return True
+
+    def get_destination_override(self, task: DownloadTask) -> Path | None:
+        """Apply Anna's Archive content-type routing when configured."""
+        if check_audiobook(task.content_type):
+            return None
+        return get_aa_content_type_dir(task.content_type)
+
+    def search(
+        self,
+        book: BookMetadata,
+        plan: ReleaseSearchPlan,
+        *,
+        expand_search: bool = False,
+        content_type: str = "ebook",
+    ) -> list[Release]:
+        """Search every enabled provider through the shared provider lifecycle."""
+        unavailable_reason = registry.get_unavailable_reason(self._providers)
+        if unavailable_reason:
+            raise DirectDownloadUnavailableError(unavailable_reason)
+
+        releases: list[Release] = []
+        unavailable_errors: list[SourceUnavailableError] = []
+        for provider in registry.enabled_providers(self._providers):
+            try:
+                records = provider.search(
+                    book,
+                    plan,
+                    expand_search=expand_search,
+                    content_type=content_type,
+                )
+            except SourceUnavailableError as exc:
+                unavailable_errors.append(exc)
+                continue
+            except (
+                RuntimeError,
+                TypeError,
+                ValueError,
+                requests.exceptions.RequestException,
+            ) as exc:
+                logger.warning("%s search failed: %s", provider.display_name, exc)
+                continue
+            releases.extend(_browse_record_to_release(record) for record in records)
+
+        if unavailable_errors and not releases:
+            raise unavailable_errors[0]
+        return releases
+
+    def is_available(self) -> bool:
+        """Check if Direct Download has been explicitly enabled and configured."""
+        return registry.get_unavailable_reason(self._providers) is None
diff --git a/tests/direct_download/test_handler.py b/tests/direct_download/test_handler.py
index cbf7b405..8b832e87 100644
--- a/tests/direct_download/test_handler.py
+++ b/tests/direct_download/test_handler.py
@@ -1,5 +1,6 @@
 from threading import Event
 
+from shelfmark.core.config import config
 from shelfmark.core.models import DownloadTask
 from shelfmark.release_sources.direct_download import DirectDownloadHandler
 
@@ -13,11 +14,11 @@ def test_direct_download_handler_builds_staging_filename_from_browse_record(monk
         captured["path"] = book_path
         return "https://example.com/file.epub"
 
-    import shelfmark.release_sources.direct_download as dd
+    from shelfmark.release_sources.direct_download import handler as dd
 
     monkeypatch.setattr(dd, "_download_book", fake_download_book)
     monkeypatch.setattr(
-        dd.config,
+        config,
         "get",
         lambda key, default=None: "rename" if key == "FILE_ORGANIZATION" else default,
     )
@@ -47,11 +48,11 @@ def test_direct_download_handler_uses_source_id_filename_when_organization_disab
         captured["path"] = book_path
         return "https://example.com/file.epub"
 
-    import shelfmark.release_sources.direct_download as dd
+    from shelfmark.release_sources.direct_download import handler as dd
 
     monkeypatch.setattr(dd, "_download_book", fake_download_book)
     monkeypatch.setattr(
-        dd.config,
+        config,
         "get",
         lambda key, default=None: "none" if key == "FILE_ORGANIZATION" else default,
     )
@@ -78,7 +79,7 @@ def test_direct_download_handler_skips_download_when_cancelled_before_start(monk
     def unexpected_download(*_args, **_kwargs):
         raise AssertionError("_download_book should not run when the task is already cancelled")
 
-    import shelfmark.release_sources.direct_download as dd
+    from shelfmark.release_sources.direct_download import handler as dd
 
     monkeypatch.setattr(dd, "_download_book", unexpected_download)
 
@@ -113,12 +114,12 @@ def test_direct_download_handler_removes_partial_file_when_cancelled_after_downl
         cancel_flag.set()
         return "https://example.com/file.epub"
 
-    import shelfmark.release_sources.direct_download as dd
+    from shelfmark.release_sources.direct_download import handler as dd
 
     monkeypatch.setattr(dd, "_download_book", fake_download_book)
     monkeypatch.setattr(dd, "TMP_DIR", tmp_path)
     monkeypatch.setattr(
-        dd.config,
+        config,
         "get",
         lambda key, default=None: "rename" if key == "FILE_ORGANIZATION" else default,
     )
diff --git a/tests/direct_download/test_search_budget.py b/tests/direct_download/test_search_budget.py
index 8c936da0..05efc454 100644
--- a/tests/direct_download/test_search_budget.py
+++ b/tests/direct_download/test_search_budget.py
@@ -10,13 +10,16 @@ a challenge failure became a gateway timeout (issue #1276).
 
 import pytest
 
-import shelfmark.release_sources.direct_download as dd
 from shelfmark.core import search_deadline
+from shelfmark.release_sources.direct_download import annas_archive as aa
+from shelfmark.release_sources.direct_download import source as dd
 
 
 @pytest.fixture(autouse=True)
-def _no_ambient_deadline():
+def _no_ambient_deadline(monkeypatch):
     token = search_deadline._current.set(None)
+    monkeypatch.setattr(dd.registry, "get_unavailable_reason", lambda _providers: None)
+    monkeypatch.setattr(dd.registry, "enabled_providers", lambda providers: (providers[0],))
     yield
     search_deadline._current.reset(token)
 
@@ -35,13 +38,13 @@ class _Selector:
 def test_fetch_search_table_gives_up_when_the_budget_is_spent(monkeypatch):
     """Every mirror shares the protection, so another mirror is another wasted solve."""
     fetches: list[str] = []
-    monkeypatch.setattr(dd.downloader, "html_get_page", lambda url, **_k: fetches.append(url) or "")
-    monkeypatch.setattr(dd.network, "get_available_aa_urls", lambda: ["https://annas-archive.gl"])
+    monkeypatch.setattr(aa.downloader, "html_get_page", lambda url, **_k: fetches.append(url) or "")
+    monkeypatch.setattr(aa.network, "get_available_aa_urls", lambda: ["https://annas-archive.gl"])
 
     with search_deadline.search_deadline(60) as deadline:
         deadline.event.set()
-        with pytest.raises(dd.SearchUnavailableError) as excinfo:
-            dd._fetch_search_table("https://annas-archive.gl/search?q=dune", _Selector())
+        with pytest.raises(aa.SearchUnavailableError) as excinfo:
+            aa._fetch_search_table("https://annas-archive.gl/search?q=dune", _Selector())
 
     assert fetches == [], "no fetch should have been attempted"
     assert "ran out of time" in str(excinfo.value)
@@ -49,11 +52,11 @@ def test_fetch_search_table_gives_up_when_the_budget_is_spent(monkeypatch):
 
 def test_fetch_search_table_runs_normally_within_budget(monkeypatch):
     page = "<html><body><main><table><tbody></tbody></table></main></body></html>"
-    monkeypatch.setattr(dd.downloader, "html_get_page", lambda _url, **_k: page)
-    monkeypatch.setattr(dd.network, "get_available_aa_urls", lambda: ["https://annas-archive.gl"])
+    monkeypatch.setattr(aa.downloader, "html_get_page", lambda _url, **_k: page)
+    monkeypatch.setattr(aa.network, "get_available_aa_urls", lambda: ["https://annas-archive.gl"])
 
     with search_deadline.search_deadline(60):
-        html, table = dd._fetch_search_table("https://annas-archive.gl/search?q=dune", _Selector())
+        html, table = aa._fetch_search_table("https://annas-archive.gl/search?q=dune", _Selector())
 
     assert table is not None
     assert html == page
@@ -98,8 +101,8 @@ def test_title_variants_stop_once_the_budget_is_spent(monkeypatch):
             deadline.event.set()
         return []
 
-    monkeypatch.setattr(dd, "search_books", fake_search_books)
-    monkeypatch.setattr(dd, "_ensure_direct_download_available", lambda: None)
+    monkeypatch.setattr(aa, "search_books", fake_search_books)
+    monkeypatch.setattr(aa, "ensure_available", lambda: None)
 
     source = dd.DirectDownloadSource()
     with search_deadline.search_deadline(60):
@@ -111,8 +114,8 @@ def test_title_variants_stop_once_the_budget_is_spent(monkeypatch):
 
 def test_all_title_variants_run_within_budget(monkeypatch):
     queries: list[str] = []
-    monkeypatch.setattr(dd, "search_books", lambda q, _f: queries.append(q) or [])
-    monkeypatch.setattr(dd, "_ensure_direct_download_available", lambda: None)
+    monkeypatch.setattr(aa, "search_books", lambda q, _f: queries.append(q) or [])
+    monkeypatch.setattr(aa, "ensure_available", lambda: None)
 
     source = dd.DirectDownloadSource()
     with search_deadline.search_deadline(60):
@@ -134,8 +137,8 @@ def test_language_filter_retry_is_skipped_on_a_spent_budget(monkeypatch):
                 deadline.event.set()
         return []
 
-    monkeypatch.setattr(dd, "search_books", fake_search_books)
-    monkeypatch.setattr(dd, "_ensure_direct_download_available", lambda: None)
+    monkeypatch.setattr(aa, "search_books", fake_search_books)
+    monkeypatch.setattr(aa, "ensure_available", lambda: None)
 
     source = dd.DirectDownloadSource()
     with search_deadline.search_deadline(60):
diff --git a/tests/direct_download/test_search_challenge_false_positive.py b/tests/direct_download/test_search_challenge_false_positive.py
index 9a0f8ce1..fd580864 100644
--- a/tests/direct_download/test_search_challenge_false_positive.py
+++ b/tests/direct_download/test_search_challenge_false_positive.py
@@ -10,6 +10,8 @@ unsolved challenge, telling users to go fix a bypasser that had just succeeded.
 import pytest
 from bs4 import Tag
 
+from shelfmark.release_sources.direct_download import annas_archive as aa
+
 # Verbatim from a live annas-archive.pk 403, trimmed of nothing that matters: this is
 # what an interstitial actually looks like, and it is under a kilobyte.
 DDOS_GUARD_INTERSTITIAL = (
@@ -58,7 +60,7 @@ class _Selector:
 
 
 def _patch_pages(monkeypatch, pages: list[str]):
-    import shelfmark.release_sources.direct_download as dd
+    from shelfmark.release_sources.direct_download import source as dd
 
     calls: list[str] = []
 
@@ -66,8 +68,8 @@ def _patch_pages(monkeypatch, pages: list[str]):
         calls.append(url)
         return pages[len(calls) - 1] if len(calls) <= len(pages) else ""
 
-    monkeypatch.setattr(dd.downloader, "html_get_page", fake_get)
-    monkeypatch.setattr(dd.network, "get_available_aa_urls", lambda: ["a", "b"])
+    monkeypatch.setattr(aa.downloader, "html_get_page", fake_get)
+    monkeypatch.setattr(aa.network, "get_available_aa_urls", lambda: ["a", "b"])
     return dd, calls
 
 
@@ -80,8 +82,6 @@ def search_logs():
     """
     import logging
 
-    import shelfmark.release_sources.direct_download as dd
-
     messages: list[str] = []
 
     class _Capture(logging.Handler):
@@ -89,38 +89,37 @@ def search_logs():
             messages.append(record.getMessage())
 
     handler = _Capture()
-    dd.logger.addHandler(handler)
-    previous = dd.logger.level
-    dd.logger.setLevel(logging.DEBUG)
+    aa.logger.addHandler(handler)
+    previous = aa.logger.level
+    aa.logger.setLevel(logging.DEBUG)
     # Logger.setLevel only invalidates the is-enabled cache through the manager, which
     # these loggers are not registered with; without this the DEBUG line stays filtered.
-    dd.logger._cache.clear()
+    aa.logger._cache.clear()
     try:
         yield messages
     finally:
-        dd.logger.removeHandler(handler)
-        dd.logger.setLevel(previous)
+        aa.logger.removeHandler(handler)
+        aa.logger.setLevel(previous)
 
 
 def test_the_size_guard_is_what_separates_a_real_page_from_an_interstitial():
     """The two inputs this bug turned on, checked directly."""
-    import shelfmark.release_sources.direct_download as dd
     from shelfmark.bypass.challenge import MAX_CHALLENGE_HTML_CHARS
 
     assert len(DDOS_GUARD_INTERSTITIAL) < MAX_CHALLENGE_HTML_CHARS
     assert len(AA_PAGE_WITHOUT_TABLE) > MAX_CHALLENGE_HTML_CHARS
     # Both contain "ddos-guard"; only one is a challenge.
     assert "ddos-guard" in AA_PAGE_WITHOUT_TABLE.lower()
-    assert dd._looks_like_challenge_page(DDOS_GUARD_INTERSTITIAL)
-    assert not dd._looks_like_challenge_page(AA_PAGE_WITHOUT_TABLE)
+    assert aa._looks_like_challenge_page(DDOS_GUARD_INTERSTITIAL)
+    assert not aa._looks_like_challenge_page(AA_PAGE_WITHOUT_TABLE)
 
 
 def test_real_aa_page_without_a_table_is_not_reported_as_a_challenge(monkeypatch):
     """The #1289 failure: a served AA page raised "unsolved protection challenge"."""
-    dd, calls = _patch_pages(monkeypatch, [AA_PAGE_WITHOUT_TABLE])
+    _dd, calls = _patch_pages(monkeypatch, [AA_PAGE_WITHOUT_TABLE])
     selector = _Selector(["https://real.test", "https://other.test"])
 
-    html, table = dd._fetch_search_table("https://real.test/search?q=malice", selector)
+    html, table = aa._fetch_search_table("https://real.test/search?q=malice", selector)
 
     assert table is None
     assert html == AA_PAGE_WITHOUT_TABLE
@@ -131,22 +130,22 @@ def test_real_aa_page_without_a_table_is_not_reported_as_a_challenge(monkeypatch
 
 def test_aa_markers_win_over_challenge_markers_on_the_same_page(monkeypatch):
     """Ordering, not just the size guard, keeps a marker-carrying AA page readable."""
-    dd, _calls = _patch_pages(monkeypatch, [AA_PAGE_WITHOUT_TABLE])
-    monkeypatch.setattr(dd, "_looks_like_challenge_page", lambda _html: True)
+    _dd, _calls = _patch_pages(monkeypatch, [AA_PAGE_WITHOUT_TABLE])
+    monkeypatch.setattr(aa, "_looks_like_challenge_page", lambda _html: True)
     selector = _Selector(["https://real.test", "https://other.test"])
 
-    _html, table = dd._fetch_search_table("https://real.test/search?q=malice", selector)
+    _html, table = aa._fetch_search_table("https://real.test/search?q=malice", selector)
 
     assert table is None
 
 
 def test_genuine_interstitial_still_raises(monkeypatch):
     """The behaviour the check exists for is untouched."""
-    dd, _calls = _patch_pages(monkeypatch, [DDOS_GUARD_INTERSTITIAL])
+    _dd, _calls = _patch_pages(monkeypatch, [DDOS_GUARD_INTERSTITIAL])
     selector = _Selector(["https://real.test", "https://other.test"])
 
-    with pytest.raises(dd.SearchUnavailableError, match="protection challenge"):
-        dd._fetch_search_table("https://real.test/search?q=malice", selector)
+    with pytest.raises(aa.SearchUnavailableError, match="protection challenge"):
+        aa._fetch_search_table("https://real.test/search?q=malice", selector)
 
     assert selector.quarantined == []
 
@@ -156,10 +155,10 @@ def test_results_table_is_still_returned(monkeypatch):
     page = AA_PAGE_WITHOUT_TABLE.replace(
         "<main>", "<main><table><tbody><tr><td>Malice</td></tr></tbody></table>"
     )
-    dd, _calls = _patch_pages(monkeypatch, [page])
+    _dd, _calls = _patch_pages(monkeypatch, [page])
     selector = _Selector(["https://real.test"])
 
-    _html, table = dd._fetch_search_table("https://real.test/search?q=malice", selector)
+    _html, table = aa._fetch_search_table("https://real.test/search?q=malice", selector)
 
     assert isinstance(table, Tag)
 
@@ -171,10 +170,10 @@ def test_untabled_page_is_fingerprinted_in_the_log(monkeypatch, search_logs):
     rather than reverse-engineered: size, whether the size guard applied, the AA markers
     found, and the challenge marker (or its absence).
     """
-    dd, _calls = _patch_pages(monkeypatch, [AA_PAGE_WITHOUT_TABLE])
+    _dd, _calls = _patch_pages(monkeypatch, [AA_PAGE_WITHOUT_TABLE])
     selector = _Selector(["https://real.test", "https://other.test"])
 
-    dd._fetch_search_table("https://real.test/search?q=malice", selector)
+    aa._fetch_search_table("https://real.test/search?q=malice", selector)
 
     verdict = next(m for m in search_logs if "no results table" in m)
     assert f"bytes={len(AA_PAGE_WITHOUT_TABLE)}" in verdict
@@ -190,11 +189,11 @@ def test_untabled_page_is_fingerprinted_in_the_log(monkeypatch, search_logs):
 
 def test_interstitial_fingerprint_names_the_marker_that_proved_it(monkeypatch, search_logs):
     """The same line must also settle the opposite case, without needing the body."""
-    dd, _calls = _patch_pages(monkeypatch, [DDOS_GUARD_INTERSTITIAL])
+    _dd, _calls = _patch_pages(monkeypatch, [DDOS_GUARD_INTERSTITIAL])
     selector = _Selector(["https://real.test", "https://other.test"])
 
-    with pytest.raises(dd.SearchUnavailableError):
-        dd._fetch_search_table("https://real.test/search?q=malice", selector)
+    with pytest.raises(aa.SearchUnavailableError):
+        aa._fetch_search_table("https://real.test/search?q=malice", selector)
 
     verdict = next(m for m in search_logs if "no results table" in m)
     assert "over_challenge_size_cap=False" in verdict
@@ -204,14 +203,14 @@ def test_interstitial_fingerprint_names_the_marker_that_proved_it(monkeypatch, s
 
 def test_fingerprint_failure_never_breaks_a_search(monkeypatch):
     """Diagnostics are best-effort; a bug in them must not cost the user their search."""
-    dd, _calls = _patch_pages(monkeypatch, [AA_PAGE_WITHOUT_TABLE])
+    _dd, _calls = _patch_pages(monkeypatch, [AA_PAGE_WITHOUT_TABLE])
 
     def boom(_html):
         raise RuntimeError("marker scan blew up")
 
-    monkeypatch.setattr(dd, "challenge_marker", boom)
+    monkeypatch.setattr(aa, "challenge_marker", boom)
     selector = _Selector(["https://real.test", "https://other.test"])
 
-    _html, table = dd._fetch_search_table("https://real.test/search?q=malice", selector)
+    _html, table = aa._fetch_search_table("https://real.test/search?q=malice", selector)
 
     assert table is None
diff --git a/tests/direct_download/test_search_page_reuse.py b/tests/direct_download/test_search_page_reuse.py
index 8ffe3ae2..625b7cd1 100644
--- a/tests/direct_download/test_search_page_reuse.py
+++ b/tests/direct_download/test_search_page_reuse.py
@@ -11,8 +11,10 @@ solve, tens of seconds for nothing. See issue #1285.
 import pytest
 from bs4 import BeautifulSoup
 
-import shelfmark.release_sources.direct_download as dd
 from shelfmark.core import search_deadline
+from shelfmark.core.config import config
+from shelfmark.core.models import SearchFilters
+from shelfmark.release_sources.direct_download import annas_archive as aa
 
 
 @pytest.fixture(autouse=True)
@@ -39,9 +41,9 @@ _PAGE = "<html><body><main><table><tbody></tbody></table></main></body></html>"
 def _count_fetches(monkeypatch) -> list[str]:
     fetched: list[str] = []
     monkeypatch.setattr(
-        dd.downloader, "html_get_page", lambda url, **_k: fetched.append(url) or _PAGE
+        aa.downloader, "html_get_page", lambda url, **_k: fetched.append(url) or _PAGE
     )
-    monkeypatch.setattr(dd.network, "get_available_aa_urls", lambda: ["https://annas-archive.gl"])
+    monkeypatch.setattr(aa.network, "get_available_aa_urls", lambda: ["https://annas-archive.gl"])
     return fetched
 
 
@@ -49,9 +51,9 @@ def test_repeated_url_is_fetched_once_within_one_search(monkeypatch):
     fetched = _count_fetches(monkeypatch)
     url = "https://annas-archive.gl/search?q=dune"
 
-    with dd._search_page_reuse():
-        first_html, first_table = dd._fetch_search_table(url, _Selector())
-        second_html, second_table = dd._fetch_search_table(url, _Selector())
+    with aa.search_page_reuse():
+        first_html, first_table = aa._fetch_search_table(url, _Selector())
+        second_html, second_table = aa._fetch_search_table(url, _Selector())
 
     assert fetched == [url], "the second ask should have been served from the search's cache"
     assert first_html == second_html
@@ -61,9 +63,9 @@ def test_repeated_url_is_fetched_once_within_one_search(monkeypatch):
 def test_distinct_urls_are_still_fetched_separately(monkeypatch):
     fetched = _count_fetches(monkeypatch)
 
-    with dd._search_page_reuse():
-        dd._fetch_search_table("https://annas-archive.gl/search?q=dune", _Selector())
-        dd._fetch_search_table("https://annas-archive.gl/search?q=dune&lang=en", _Selector())
+    with aa.search_page_reuse():
+        aa._fetch_search_table("https://annas-archive.gl/search?q=dune", _Selector())
+        aa._fetch_search_table("https://annas-archive.gl/search?q=dune&lang=en", _Selector())
 
     assert len(fetched) == 2
 
@@ -73,10 +75,10 @@ def test_cache_does_not_leak_between_searches(monkeypatch):
     fetched = _count_fetches(monkeypatch)
     url = "https://annas-archive.gl/search?q=dune"
 
-    with dd._search_page_reuse():
-        dd._fetch_search_table(url, _Selector())
-    with dd._search_page_reuse():
-        dd._fetch_search_table(url, _Selector())
+    with aa.search_page_reuse():
+        aa._fetch_search_table(url, _Selector())
+    with aa.search_page_reuse():
+        aa._fetch_search_table(url, _Selector())
 
     assert fetched == [url, url]
 
@@ -86,8 +88,8 @@ def test_without_the_context_every_fetch_still_goes_out(monkeypatch):
     fetched = _count_fetches(monkeypatch)
     url = "https://annas-archive.gl/search?q=dune"
 
-    dd._fetch_search_table(url, _Selector())
-    dd._fetch_search_table(url, _Selector())
+    aa._fetch_search_table(url, _Selector())
+    aa._fetch_search_table(url, _Selector())
 
     assert fetched == [url, url]
 
@@ -97,12 +99,12 @@ def test_a_failure_is_not_cached(monkeypatch):
     fetched = _count_fetches(monkeypatch)
     url = "https://annas-archive.gl/search?q=dune"
 
-    with dd._search_page_reuse():
+    with aa.search_page_reuse():
         with search_deadline.search_deadline(60) as deadline:
             deadline.event.set()
-            with pytest.raises(dd.SearchUnavailableError):
-                dd._fetch_search_table(url, _Selector())
-        dd._fetch_search_table(url, _Selector())
+            with pytest.raises(aa.SearchUnavailableError):
+                aa._fetch_search_table(url, _Selector())
+        aa._fetch_search_table(url, _Selector())
 
     assert fetched == [url], "the successful retry should be the only fetch"
 
@@ -119,14 +121,14 @@ def test_a_give_up_page_is_not_cached(monkeypatch):
     parked = "<html><body>This domain is for sale.</body></html>"
     fetched: list[str] = []
     monkeypatch.setattr(
-        dd.downloader, "html_get_page", lambda url, **_k: fetched.append(url) or parked
+        aa.downloader, "html_get_page", lambda url, **_k: fetched.append(url) or parked
     )
-    monkeypatch.setattr(dd.network, "get_available_aa_urls", lambda: ["https://annas-archive.gl"])
+    monkeypatch.setattr(aa.network, "get_available_aa_urls", lambda: ["https://annas-archive.gl"])
     url = "https://annas-archive.gl/search?q=dune"
 
-    with dd._search_page_reuse():
-        assert dd._fetch_search_table(url, _Selector()) == (parked, None)
-        assert dd._fetch_search_table(url, _Selector()) == (parked, None)
+    with aa.search_page_reuse():
+        assert aa._fetch_search_table(url, _Selector()) == (parked, None)
+        assert aa._fetch_search_table(url, _Selector()) == (parked, None)
 
     assert fetched == [url, url], "the second pass must not inherit the first's give-up"
 
@@ -136,14 +138,14 @@ def test_a_genuinely_empty_result_is_still_cached(monkeypatch):
     empty = "<html><body><main>No files found. <a href='/md5/x'>x</a></main></body></html>"
     fetched: list[str] = []
     monkeypatch.setattr(
-        dd.downloader, "html_get_page", lambda url, **_k: fetched.append(url) or empty
+        aa.downloader, "html_get_page", lambda url, **_k: fetched.append(url) or empty
     )
-    monkeypatch.setattr(dd.network, "get_available_aa_urls", lambda: ["https://annas-archive.gl"])
+    monkeypatch.setattr(aa.network, "get_available_aa_urls", lambda: ["https://annas-archive.gl"])
     url = "https://annas-archive.gl/search?q=nothing"
 
-    with dd._search_page_reuse():
-        assert dd._fetch_search_table(url, _Selector()) == (empty, None)
-        assert dd._fetch_search_table(url, _Selector()) == (empty, None)
+    with aa.search_page_reuse():
+        assert aa._fetch_search_table(url, _Selector()) == (empty, None)
+        assert aa._fetch_search_table(url, _Selector()) == (empty, None)
 
     assert fetched == [url], "an empty answer is an answer; re-solving for it buys nothing"
 
@@ -152,7 +154,7 @@ def test_language_retry_reuses_the_page_it_already_fetched(monkeypatch):
     """The end-to-end shape: language-from-path makes both passes build the same URL."""
     fetched = _count_fetches(monkeypatch)
 
-    original_get = dd.config.get
+    original_get = config.get
 
     def _fake_get(key: str, default=None, user_id=None):
         del user_id
@@ -160,15 +162,15 @@ def test_language_retry_reuses_the_page_it_already_fetched(monkeypatch):
             return True
         return original_get(key, default)
 
-    monkeypatch.setattr(dd.config, "get", _fake_get)
-    monkeypatch.setattr(dd.network, "get_aa_base_url", lambda: "https://annas-archive.gl")
+    monkeypatch.setattr(config, "get", _fake_get)
+    monkeypatch.setattr(aa.network, "get_aa_base_url", lambda: "https://annas-archive.gl")
 
-    filters_with_lang = dd.SearchFilters(lang=["en"])
-    filters_without = dd.SearchFilters()
+    filters_with_lang = SearchFilters(lang=["en"])
+    filters_without = SearchFilters()
 
-    with dd._search_page_reuse():
-        dd.search_books("dune", filters_with_lang)
-        dd.search_books("dune", filters_without)
+    with aa.search_page_reuse():
+        aa.search_books("dune", filters_with_lang)
+        aa.search_books("dune", filters_without)
 
     assert len(fetched) == 1, f"both passes build the same URL, got {fetched}"
     assert "lang=" not in fetched[0]
@@ -179,12 +181,12 @@ def test_soup_reuse_is_safe_for_repeated_parsing(monkeypatch):
     page = (
         "<html><body><main><table><tbody><tr><td>row</td></tr></tbody></table></main></body></html>"
     )
-    monkeypatch.setattr(dd.downloader, "html_get_page", lambda _url, **_k: page)
-    monkeypatch.setattr(dd.network, "get_available_aa_urls", lambda: ["https://annas-archive.gl"])
+    monkeypatch.setattr(aa.downloader, "html_get_page", lambda _url, **_k: page)
+    monkeypatch.setattr(aa.network, "get_available_aa_urls", lambda: ["https://annas-archive.gl"])
 
-    with dd._search_page_reuse():
-        _, first = dd._fetch_search_table("https://annas-archive.gl/search?q=dune", _Selector())
-        _, second = dd._fetch_search_table("https://annas-archive.gl/search?q=dune", _Selector())
+    with aa.search_page_reuse():
+        _, first = aa._fetch_search_table("https://annas-archive.gl/search?q=dune", _Selector())
+        _, second = aa._fetch_search_table("https://annas-archive.gl/search?q=dune", _Selector())
 
     assert first is not None
     assert second is not None
diff --git a/tests/direct_download/test_search_parked_mirror.py b/tests/direct_download/test_search_parked_mirror.py
index 1e2415d1..684be75e 100644
--- a/tests/direct_download/test_search_parked_mirror.py
+++ b/tests/direct_download/test_search_parked_mirror.py
@@ -8,6 +8,8 @@ rotation and every later search pays for it again.
 import pytest
 from bs4 import Tag
 
+from shelfmark.release_sources.direct_download import annas_archive as aa
+
 PARKED_PAGE = """<!doctype html><html><head><title>annas-archive.li</title></head>
 <body><h1>This domain is for sale</h1><p>Inquire now. Buy this domain.</p></body></html>"""
 
@@ -48,7 +50,7 @@ class _Selector:
 
 def _patch_pages(monkeypatch, pages: list[str]):
     """Serve `pages` in order, recording the URL each call was made against."""
-    import shelfmark.release_sources.direct_download as dd
+    from shelfmark.release_sources.direct_download import source as dd
 
     calls: list[str] = []
 
@@ -56,16 +58,16 @@ def _patch_pages(monkeypatch, pages: list[str]):
         calls.append(url)
         return pages[len(calls) - 1] if len(calls) <= len(pages) else ""
 
-    monkeypatch.setattr(dd.downloader, "html_get_page", fake_get)
-    monkeypatch.setattr(dd.network, "get_available_aa_urls", lambda: ["a", "b", "c"])
+    monkeypatch.setattr(aa.downloader, "html_get_page", fake_get)
+    monkeypatch.setattr(aa.network, "get_available_aa_urls", lambda: ["a", "b", "c"])
     return dd, calls
 
 
 def test_parked_mirror_is_quarantined_and_search_retries_next_mirror(monkeypatch):
-    dd, calls = _patch_pages(monkeypatch, [PARKED_PAGE, AA_RESULTS_PAGE])
+    _dd, calls = _patch_pages(monkeypatch, [PARKED_PAGE, AA_RESULTS_PAGE])
     selector = _Selector(["https://parked.test", "https://real.test"])
 
-    html, table = dd._fetch_search_table("https://parked.test/search?q=dune", selector)
+    html, table = aa._fetch_search_table("https://parked.test/search?q=dune", selector)
 
     assert selector.quarantined == ["https://parked.test"]
     assert isinstance(table, Tag)
@@ -76,10 +78,10 @@ def test_parked_mirror_is_quarantined_and_search_retries_next_mirror(monkeypatch
 
 def test_genuinely_empty_aa_result_does_not_quarantine(monkeypatch):
     """'No files found.' is a real answer from a healthy mirror."""
-    dd, _calls = _patch_pages(monkeypatch, [AA_EMPTY_PAGE])
+    _dd, _calls = _patch_pages(monkeypatch, [AA_EMPTY_PAGE])
     selector = _Selector(["https://real.test", "https://other.test"])
 
-    html, table = dd._fetch_search_table("https://real.test/search?q=zzz", selector)
+    html, table = aa._fetch_search_table("https://real.test/search?q=zzz", selector)
 
     assert selector.quarantined == []
     assert table is None
@@ -92,29 +94,28 @@ def test_challenge_page_is_reported_not_passed_off_as_an_empty_result(monkeypatc
     The mirror is alive and holds our clearance, so it must not be quarantined - but
     returning it as "no table" made the caller tell the user their query found nothing.
     """
-    dd, _calls = _patch_pages(monkeypatch, [DDOS_GUARD_PAGE])
+    _dd, _calls = _patch_pages(monkeypatch, [DDOS_GUARD_PAGE])
     selector = _Selector(["https://real.test", "https://other.test"])
 
-    with pytest.raises(dd.SearchUnavailableError, match="protection challenge"):
-        dd._fetch_search_table("https://real.test/search?q=dune", selector)
+    with pytest.raises(aa.SearchUnavailableError, match="protection challenge"):
+        aa._fetch_search_table("https://real.test/search?q=dune", selector)
 
     assert selector.quarantined == []
 
 
 def test_unreachable_mirror_raises_search_unavailable(monkeypatch):
-    dd, _calls = _patch_pages(monkeypatch, [""])
+    _dd, _calls = _patch_pages(monkeypatch, [""])
     selector = _Selector(["https://real.test"])
 
     try:
-        dd._fetch_search_table("https://real.test/search?q=dune", selector)
-    except dd.SearchUnavailableError:
+        aa._fetch_search_table("https://real.test/search?q=dune", selector)
+    except aa.SearchUnavailableError:
         return
     raise AssertionError("expected SearchUnavailableError")
 
 
 def test_recorded_failure_reason_is_surfaced_to_the_caller(monkeypatch):
     """The concrete give-up reason html_get_page stashed replaces the generic line."""
-    import shelfmark.release_sources.direct_download as dd
 
     reason = "Anna's Archive returned 403 (blocked) and no bypasser is enabled."
 
@@ -122,10 +123,10 @@ def test_recorded_failure_reason_is_surfaced_to_the_caller(monkeypatch):
         selector.last_failure = reason
         return ""
 
-    monkeypatch.setattr(dd.downloader, "html_get_page", fake_get)
-    monkeypatch.setattr(dd.network, "get_available_aa_urls", lambda: ["a"])
+    monkeypatch.setattr(aa.downloader, "html_get_page", fake_get)
+    monkeypatch.setattr(aa.network, "get_available_aa_urls", lambda: ["a"])
     selector = _Selector(["https://real.test"])
     selector.last_failure = None
 
-    with pytest.raises(dd.SearchUnavailableError, match="403 .blocked."):
-        dd._fetch_search_table("https://real.test/search?q=dune", selector)
+    with pytest.raises(aa.SearchUnavailableError, match="403 .blocked."):
+        aa._fetch_search_table("https://real.test/search?q=dune", selector)
diff --git a/tests/direct_download/test_search_queries.py b/tests/direct_download/test_search_queries.py
index 5e1e3944..fef3fcde 100644
--- a/tests/direct_download/test_search_queries.py
+++ b/tests/direct_download/test_search_queries.py
@@ -1,8 +1,10 @@
+from shelfmark.core.config import config
 from shelfmark.core.models import SearchFilters
 from shelfmark.core.search_plan import build_release_search_plan
 from shelfmark.metadata_providers import BookMetadata
 from shelfmark.release_sources import BrowseRecord
 from shelfmark.release_sources.direct_download import DirectDownloadSource
+from shelfmark.release_sources.direct_download import annas_archive as aa
 
 
 def _browse_record(record_id: str, title: str) -> BrowseRecord:
@@ -10,9 +12,9 @@ def _browse_record(record_id: str, title: str) -> BrowseRecord:
 
 
 def _enable_direct_download(monkeypatch):
-    import shelfmark.release_sources.direct_download as dd
+    from shelfmark.release_sources.direct_download import source as dd
 
-    original_get = dd.config.get
+    original_get = config.get
 
     def _fake_get(key: str, default=None, user_id=None):
         del user_id
@@ -20,7 +22,7 @@ def _enable_direct_download(monkeypatch):
             return True
         return original_get(key, default)
 
-    monkeypatch.setattr(dd.config, "get", _fake_get)
+    monkeypatch.setattr(config, "get", _fake_get)
     monkeypatch.setattr("shelfmark.core.mirrors.has_aa_mirror_configuration", lambda: True)
     return dd
 
@@ -33,9 +35,9 @@ class TestDirectDownloadSearchQueries:
             captured.append(query)
             return []
 
-        dd = _enable_direct_download(monkeypatch)
+        _enable_direct_download(monkeypatch)
 
-        monkeypatch.setattr(dd, "search_books", fake_search_books)
+        monkeypatch.setattr(aa, "search_books", fake_search_books)
 
         source = DirectDownloadSource()
         book = BookMetadata(
@@ -75,9 +77,9 @@ class TestDirectDownloadSearchQueries:
             captured.append((query, filters.lang))
             return records_by_query[query]
 
-        dd = _enable_direct_download(monkeypatch)
+        _enable_direct_download(monkeypatch)
 
-        monkeypatch.setattr(dd, "search_books", fake_search_books)
+        monkeypatch.setattr(aa, "search_books", fake_search_books)
 
         source = DirectDownloadSource()
         book = BookMetadata(
@@ -119,9 +121,9 @@ class TestDirectDownloadSearchQueries:
                 return []
             return fallback_results[query]
 
-        dd = _enable_direct_download(monkeypatch)
+        _enable_direct_download(monkeypatch)
 
-        monkeypatch.setattr(dd, "search_books", fake_search_books)
+        monkeypatch.setattr(aa, "search_books", fake_search_books)
 
         source = DirectDownloadSource()
         book = BookMetadata(
@@ -157,9 +159,9 @@ class TestDirectDownloadSearchQueries:
                 return []
             return [_browse_record("manual-1", "Manual result")]
 
-        dd = _enable_direct_download(monkeypatch)
+        _enable_direct_download(monkeypatch)
 
-        monkeypatch.setattr(dd, "search_books", fake_search_books)
+        monkeypatch.setattr(aa, "search_books", fake_search_books)
 
         source = DirectDownloadSource()
         book = BookMetadata(
@@ -188,9 +190,9 @@ class TestDirectDownloadSearchQueries:
 
 
 def _patch_path_language(monkeypatch, enabled: bool = True):
-    import shelfmark.release_sources.direct_download as dd
+    from shelfmark.release_sources.direct_download import source as dd
 
-    original_get = dd.config.get
+    original_get = config.get
 
     def _fake_get(key: str, default=None, user_id=None):
         del user_id
@@ -198,7 +200,7 @@ def _patch_path_language(monkeypatch, enabled: bool = True):
             return enabled
         return original_get(key, default)
 
-    monkeypatch.setattr(dd.config, "get", _fake_get)
+    monkeypatch.setattr(config, "get", _fake_get)
     return dd
 
 
@@ -228,59 +230,59 @@ def _make_row(distant_path: str, language: str = "", record_id: str = "rec-1") -
 
 
 def test_detects_bracketed_language_from_distant_path(monkeypatch):
-    dd = _patch_path_language(monkeypatch)
+    _patch_path_language(monkeypatch)
     row = _row_from_html(_make_row(r"lgli/N:\comics1\emule\2021.08.01\[BD FR] Scrameustache.cbz"))
-    record = dd._parse_search_result_row(row)
+    record = aa._parse_search_result_row(row)
     assert record is not None
     assert record.language == "fr"
     assert record.download_path is not None
 
 
 def test_detects_mixed_case_bracketed_language(monkeypatch):
-    dd = _patch_path_language(monkeypatch)
+    _patch_path_language(monkeypatch)
     row = _row_from_html(_make_row(r"lgli/V:\comics\_0DAY3\[Fr]\BDs [Fr]\!Pdf\S\Book.pdf"))
-    record = dd._parse_search_result_row(row)
+    record = aa._parse_search_result_row(row)
     assert record is not None
     assert record.language == "fr"
 
 
 def test_overrides_unknown_language_with_path_detection(monkeypatch):
-    dd = _patch_path_language(monkeypatch)
+    _patch_path_language(monkeypatch)
     row = _row_from_html(_make_row(r"lgli/V:\comics\_0DAY3\[Fr]\Book.pdf", language="unknown"))
-    record = dd._parse_search_result_row(row)
+    record = aa._parse_search_result_row(row)
     assert record is not None
     assert record.language == "fr"
 
 
 def test_sets_unknown_when_path_has_no_language(monkeypatch):
-    dd = _patch_path_language(monkeypatch)
+    _patch_path_language(monkeypatch)
     row = _row_from_html(_make_row(r"lgli/N:\comics1\emule\NoLanguageHere.epub"))
-    record = dd._parse_search_result_row(row)
+    record = aa._parse_search_result_row(row)
     assert record is not None
     assert record.language == "unknown"
 
 
 def test_avoids_en_false_positive_when_french_present(monkeypatch):
-    dd = _patch_path_language(monkeypatch)
+    _patch_path_language(monkeypatch)
     row = _row_from_html(
         _make_row(r"lgli/V:\comics\_0DAY2\Stripboeken Frans - BD en Français\[BD Fr] Book.cbr")
     )
-    record = dd._parse_search_result_row(row)
+    record = aa._parse_search_result_row(row)
     assert record is not None
     assert record.language == "fr"
 
 
 def test_keeps_row_with_missing_language_when_toggle_disabled(monkeypatch):
-    dd = _patch_path_language(monkeypatch, enabled=False)
+    _patch_path_language(monkeypatch, enabled=False)
     row = _row_from_html(_make_row(r"lgli/N:\comics1\[BD FR] Scrameustache.cbz"))
-    record = dd._parse_search_result_row(row)
+    record = aa._parse_search_result_row(row)
     assert record is not None
     assert record.language is None
 
 
 def test_keeps_sparse_lgli_row(monkeypatch):
     """lgli rows missing author/publisher/year must not be dropped."""
-    dd = _patch_path_language(monkeypatch)
+    _patch_path_language(monkeypatch)
     html = r"""
     <tr>
       <td><a href="/md5/sparse-1"><img src="cover.jpg"></a></td>
@@ -292,7 +294,7 @@ def test_keeps_sparse_lgli_row(monkeypatch):
       <td><span>lgli/N:\comics1\ftp\[BD.FR] French Comics\Book.cbz</span></td>
     </tr>
     """
-    record = dd._parse_search_result_row(_row_from_html(html))
+    record = aa._parse_search_result_row(_row_from_html(html))
     assert record is not None
     assert record.id == "sparse-1"
     assert record.language == "fr"
@@ -300,9 +302,9 @@ def test_keeps_sparse_lgli_row(monkeypatch):
 
 
 def test_search_books_filters_locally_when_path_language_enabled(monkeypatch):
-    dd = _patch_path_language(monkeypatch)
-    monkeypatch.setattr(dd.network, "get_aa_base_url", lambda: "https://mirror.example")
-    monkeypatch.setattr(dd.network, "AAMirrorSelector", lambda: object())
+    _patch_path_language(monkeypatch)
+    monkeypatch.setattr(aa.network, "get_aa_base_url", lambda: "https://mirror.example")
+    monkeypatch.setattr(aa.network, "AAMirrorSelector", lambda: object())
 
     captured_url: dict[str, str] = {}
 
@@ -332,9 +334,9 @@ def test_search_books_filters_locally_when_path_language_enabled(monkeypatch):
         </table>
         """
 
-    monkeypatch.setattr(dd.downloader, "html_get_page", _fake_html_get_page)
+    monkeypatch.setattr(aa.downloader, "html_get_page", _fake_html_get_page)
 
-    records = dd.search_books("demo", SearchFilters(lang=["fr"], format=["pdf"]))
+    records = aa.search_books("demo", SearchFilters(lang=["fr"], format=["pdf"]))
 
     assert "&lang=" not in captured_url["url"]
     assert len(records) == 1
@@ -343,9 +345,8 @@ def test_search_books_filters_locally_when_path_language_enabled(monkeypatch):
 
 
 def test_book_matches_requested_languages_logic():
-    import shelfmark.release_sources.direct_download as dd
 
-    assert dd._book_matches_requested_languages(None, {"fr"}) is True
-    assert dd._book_matches_requested_languages(None, set()) is True
-    assert dd._book_matches_requested_languages("en", {"fr"}) is False
-    assert dd._book_matches_requested_languages("fr", {"fr"}) is True
+    assert aa._book_matches_requested_languages(None, {"fr"}) is True
+    assert aa._book_matches_requested_languages(None, set()) is True
+    assert aa._book_matches_requested_languages("en", {"fr"}) is False
+    assert aa._book_matches_requested_languages("fr", {"fr"}) is True
diff --git a/tests/direct_download/test_source_availability.py b/tests/direct_download/test_source_availability.py
index 16c2dc79..079f43ea 100644
--- a/tests/direct_download/test_source_availability.py
+++ b/tests/direct_download/test_source_availability.py
@@ -2,7 +2,9 @@ from types import SimpleNamespace
 
 import pytest
 
+from shelfmark.core.config import config
 from shelfmark.release_sources.direct_download import DirectDownloadSource, SearchUnavailableError
+from shelfmark.release_sources.direct_download import annas_archive as aa
 
 
 def _fake_config_get(values: dict[str, object]):
@@ -14,9 +16,8 @@ def _fake_config_get(values: dict[str, object]):
 
 
 def test_direct_download_source_is_unavailable_when_disabled(monkeypatch):
-    import shelfmark.release_sources.direct_download as dd
 
-    monkeypatch.setattr(dd.config, "get", _fake_config_get({"DIRECT_DOWNLOAD_ENABLED": False}))
+    monkeypatch.setattr(config, "get", _fake_config_get({"DIRECT_DOWNLOAD_ENABLED": False}))
     monkeypatch.setattr("shelfmark.core.mirrors.has_aa_mirror_configuration", lambda: True)
 
     source = DirectDownloadSource()
@@ -28,9 +29,8 @@ def test_direct_download_source_is_unavailable_when_disabled(monkeypatch):
 
 
 def test_direct_download_source_is_unavailable_without_aa_mirrors(monkeypatch):
-    import shelfmark.release_sources.direct_download as dd
 
-    monkeypatch.setattr(dd.config, "get", _fake_config_get({"DIRECT_DOWNLOAD_ENABLED": True}))
+    monkeypatch.setattr(config, "get", _fake_config_get({"DIRECT_DOWNLOAD_ENABLED": True}))
     monkeypatch.setattr("shelfmark.core.mirrors.has_aa_mirror_configuration", lambda: False)
 
     source = DirectDownloadSource()
@@ -42,9 +42,8 @@ def test_direct_download_source_is_unavailable_without_aa_mirrors(monkeypatch):
 
 
 def test_direct_download_source_is_available_when_enabled_and_configured(monkeypatch):
-    import shelfmark.release_sources.direct_download as dd
 
-    monkeypatch.setattr(dd.config, "get", _fake_config_get({"DIRECT_DOWNLOAD_ENABLED": True}))
+    monkeypatch.setattr(config, "get", _fake_config_get({"DIRECT_DOWNLOAD_ENABLED": True}))
     monkeypatch.setattr("shelfmark.core.mirrors.has_aa_mirror_configuration", lambda: True)
 
     source = DirectDownloadSource()
@@ -53,10 +52,9 @@ def test_direct_download_source_is_available_when_enabled_and_configured(monkeyp
 
 
 def test_get_source_priority_disables_entries_without_required_mirrors(monkeypatch):
-    import shelfmark.release_sources.direct_download as dd
 
     monkeypatch.setattr(
-        dd.config,
+        config,
         "get",
         _fake_config_get(
             {
@@ -77,7 +75,7 @@ def test_get_source_priority_disables_entries_without_required_mirrors(monkeypat
     monkeypatch.setattr("shelfmark.core.mirrors.has_welib_mirror_configuration", lambda: False)
     monkeypatch.setattr("shelfmark.core.mirrors.has_zlib_mirror_configuration", lambda: True)
 
-    priority = {item["id"]: item["enabled"] for item in dd._get_source_priority()}
+    priority = {item["id"]: item["enabled"] for item in aa._get_source_priority()}
 
     assert priority["aa-fast"] is False
     assert priority["libgen"] is True
@@ -86,12 +84,11 @@ def test_get_source_priority_disables_entries_without_required_mirrors(monkeypat
 
 
 def test_is_configured_zlib_link_uses_configured_mirror_domains(monkeypatch):
-    import shelfmark.release_sources.direct_download as dd
 
     monkeypatch.setattr(
         "shelfmark.core.mirrors.get_zlib_cookie_domains",
         lambda: {"custom-zlib.example"},
     )
 
-    assert dd._is_configured_zlib_link("https://custom-zlib.example/books/example") is True
-    assert dd._is_configured_zlib_link("https://other-zlib.example/books/example") is False
+    assert aa._is_configured_zlib_link("https://custom-zlib.example/books/example") is True
+    assert aa._is_configured_zlib_link("https://other-zlib.example/books/example") is False
diff --git a/tests/direct_download/test_untabled_page_names_the_answering_mirror.py b/tests/direct_download/test_untabled_page_names_the_answering_mirror.py
index 34f2a76b..9837385f 100644
--- a/tests/direct_download/test_untabled_page_names_the_answering_mirror.py
+++ b/tests/direct_download/test_untabled_page_names_the_answering_mirror.py
@@ -11,6 +11,8 @@ import logging
 
 import pytest
 
+from shelfmark.release_sources.direct_download import annas_archive as aa
+
 # A protection challenge, so the fingerprint line fires without looking like AA.
 CHALLENGE_PAGE = (
     "<html><head><title>DDOS-GUARD</title>"
@@ -26,7 +28,6 @@ def search_logs():
     setup_logger builds loggers outside the standard hierarchy, so their records never
     reach the root handler caplog installs - see tests/bypass/test_ddg_cookie_reuse.py.
     """
-    import shelfmark.release_sources.direct_download as dd
 
     messages: list[str] = []
 
@@ -35,15 +36,15 @@ def search_logs():
             messages.append(record.getMessage())
 
     handler = _Capture()
-    dd.logger.addHandler(handler)
-    previous = dd.logger.level
-    dd.logger.setLevel(logging.DEBUG)
-    dd.logger._cache.clear()
+    aa.logger.addHandler(handler)
+    previous = aa.logger.level
+    aa.logger.setLevel(logging.DEBUG)
+    aa.logger._cache.clear()
     try:
         yield messages
     finally:
-        dd.logger.removeHandler(handler)
-        dd.logger.setLevel(previous)
+        aa.logger.removeHandler(handler)
+        aa.logger.setLevel(previous)
 
 
 class _Selector:
@@ -62,7 +63,6 @@ ANSWERED = "https://annas-archive.pk/search?q=Ken+follett"
 
 
 def test_the_fingerprint_names_the_mirror_that_answered(monkeypatch, search_logs):
-    import shelfmark.release_sources.direct_download as dd
 
     def fake_get(url, **kwargs):
         # The caller must ask for it, or there is nothing to report.
@@ -71,11 +71,11 @@ def test_the_fingerprint_names_the_mirror_that_answered(monkeypatch, search_logs
         # What an internal rotation looks like from the outside: a different host.
         return CHALLENGE_PAGE, ANSWERED
 
-    monkeypatch.setattr(dd.downloader, "html_get_page", fake_get)
-    monkeypatch.setattr(dd.network, "get_available_aa_urls", lambda: ["a"])
+    monkeypatch.setattr(aa.downloader, "html_get_page", fake_get)
+    monkeypatch.setattr(aa.network, "get_available_aa_urls", lambda: ["a"])
 
-    with pytest.raises(dd.SearchUnavailableError):
-        dd._fetch_search_table_uncached(REQUESTED, _Selector())
+    with pytest.raises(aa.SearchUnavailableError):
+        aa._fetch_search_table_uncached(REQUESTED, _Selector())
 
     fingerprint = [m for m in search_logs if m.startswith("Search page has no results table")]
     assert len(fingerprint) == 1
@@ -85,13 +85,12 @@ def test_the_fingerprint_names_the_mirror_that_answered(monkeypatch, search_logs
 
 def test_a_downloader_that_reports_no_url_falls_back_to_the_request(monkeypatch, search_logs):
     """The plain-string shape stays supported; the line is still worth having."""
-    import shelfmark.release_sources.direct_download as dd
 
-    monkeypatch.setattr(dd.downloader, "html_get_page", lambda _url, **_k: CHALLENGE_PAGE)
-    monkeypatch.setattr(dd.network, "get_available_aa_urls", lambda: ["a"])
+    monkeypatch.setattr(aa.downloader, "html_get_page", lambda _url, **_k: CHALLENGE_PAGE)
+    monkeypatch.setattr(aa.network, "get_available_aa_urls", lambda: ["a"])
 
-    with pytest.raises(dd.SearchUnavailableError):
-        dd._fetch_search_table_uncached(REQUESTED, _Selector())
+    with pytest.raises(aa.SearchUnavailableError):
+        aa._fetch_search_table_uncached(REQUESTED, _Selector())
 
     fingerprint = [m for m in search_logs if m.startswith("Search page has no results table")]
     assert len(fingerprint) == 1
@@ -100,15 +99,14 @@ def test_a_downloader_that_reports_no_url_falls_back_to_the_request(monkeypatch,
 
 def test_the_empty_body_give_up_survives_the_tuple_shape(monkeypatch):
     """`("", url)` is truthy, so the exhaustion check has to read the body."""
-    import shelfmark.release_sources.direct_download as dd
 
-    monkeypatch.setattr(dd.downloader, "html_get_page", lambda _url, **_k: ("", REQUESTED))
-    monkeypatch.setattr(dd.network, "get_available_aa_urls", lambda: ["a"])
+    monkeypatch.setattr(aa.downloader, "html_get_page", lambda _url, **_k: ("", REQUESTED))
+    monkeypatch.setattr(aa.network, "get_available_aa_urls", lambda: ["a"])
 
     selector = _Selector()
     selector.last_failure = "Every mirror refused the connection."
 
-    with pytest.raises(dd.SearchUnavailableError) as excinfo:
-        dd._fetch_search_table_uncached(REQUESTED, selector)
+    with pytest.raises(aa.SearchUnavailableError) as excinfo:
+        aa._fetch_search_table_uncached(REQUESTED, selector)
 
     assert "Every mirror refused the connection." in str(excinfo.value)
diff --git a/tests/direct_download/test_web_providers.py b/tests/direct_download/test_web_providers.py
new file mode 100644
index 00000000..93a77540
--- /dev/null
+++ b/tests/direct_download/test_web_providers.py
@@ -0,0 +1,185 @@
+from types import SimpleNamespace
+
+import pytest
+
+from shelfmark.core.models import SearchFilters
+from shelfmark.release_sources.direct_download import handler, registry
+from shelfmark.release_sources.direct_download.common import ParsedSearchResult, parse_search_page
+
+
+def test_shared_parser_accepts_provider_specific_extraction():
+    from bs4 import BeautifulSoup, Tag
+
+    soup = BeautifulSoup('<li data-url="https://books.example/book">Example</li>', "html.parser")
+
+    def extract(item: Tag) -> ParsedSearchResult:
+        return ParsedSearchResult(
+            key=item["data-url"],
+            title=item.get_text(strip=True),
+            formats=("epub", "pdf"),
+            language="English",
+            source_url=item["data-url"],
+        )
+
+    records = parse_search_page(
+        soup,
+        SearchFilters(format=["epub"], lang=["en"]),
+        provider_id="example",
+        item_selector="li",
+        extract_item=extract,
+    )
+
+    assert len(records) == 1
+    assert records[0].id.startswith("example:")
+    assert records[0].format == "epub"
+    assert records[0].language == "en"
+
+
+def test_web_provider_registry_dispatches_without_source_changes(monkeypatch, tmp_path):
+    from shelfmark.release_sources.direct_download import source as dd
+
+    source_url = "https://books.example/example.epub"
+
+    class ExampleProvider:
+        id = "example"
+        display_name = "Example"
+
+        def is_enabled(self):
+            return True
+
+        def handles(self, url):
+            return url.startswith("https://books.example/")
+
+        def search(self, book, plan, **kwargs):
+            del book, plan, kwargs
+            return [
+                dd.BrowseRecord(
+                    id="example:1",
+                    title="Example",
+                    source="direct_download",
+                    format="epub",
+                    source_url=source_url,
+                )
+            ]
+
+        def download(self, book_info, book_path, *callbacks):
+            del book_info, callbacks
+            book_path.write_bytes(b"x" * 12_000)
+            return source_url
+
+    monkeypatch.setattr(registry, "PROVIDER_TYPES", (ExampleProvider,))
+    monkeypatch.setattr(
+        registry.config,
+        "get",
+        lambda key, default=None: True if key == "DIRECT_DOWNLOAD_ENABLED" else default,
+    )
+    source = dd.DirectDownloadSource()
+    plan = SimpleNamespace(
+        manual_query="Example",
+        primary_query=None,
+        source_filters=None,
+        languages=None,
+    )
+    releases = source.search(SimpleNamespace(title="Example"), plan)
+    assert len(releases) == 1
+    release = releases[0]
+    record = ExampleProvider().search(SimpleNamespace(), plan)[0]
+    destination = tmp_path / "example.epub"
+
+    assert release.extra["web_provider"] == "example"
+    assert handler._download_book(record, destination) == source_url
+    assert destination.stat().st_size == 12_000
+
+
+def test_md5_record_routes_to_annas_archive_cascade(monkeypatch, tmp_path):
+    from threading import Event
+
+    from shelfmark.release_sources.direct_download import annas_archive
+    from shelfmark.release_sources.direct_download import source as dd
+
+    record = dd.BrowseRecord(
+        id="0123456789abcdef0123456789abcdef",
+        title="Example",
+        source="direct_download",
+        format="epub",
+    )
+    destination = tmp_path / "example.epub"
+    cancel_flag = Event()
+    calls = []
+
+    def download(*args):
+        calls.append(args)
+        return "https://mirror.example/file.epub"
+
+    monkeypatch.setattr(annas_archive, "download_book", download)
+
+    assert handler._download_book(record, destination, cancel_flag=cancel_flag) == (
+        "https://mirror.example/file.epub"
+    )
+    assert calls == [(record, destination, None, cancel_flag, None)]
+
+
+def test_unknown_url_is_not_silently_routed_to_annas_archive(monkeypatch, tmp_path):
+    from shelfmark.release_sources.direct_download import annas_archive
+
+    monkeypatch.setattr(
+        annas_archive,
+        "download_book",
+        lambda *_args, **_kwargs: pytest.fail("unknown provider reached Anna's Archive"),
+    )
+    record = handler.BrowseRecord(
+        id="unknown-record",
+        title="Example",
+        source="direct_download",
+        source_url="https://unknown.example/book",
+    )
+
+    with pytest.raises(RuntimeError, match="No Direct Download provider owns"):
+        handler._download_book(record, tmp_path / "example.epub")
+
+
+def test_provider_failure_is_suppressed_when_another_provider_succeeds(monkeypatch):
+    from shelfmark.release_sources import BrowseRecord
+    from shelfmark.release_sources.direct_download.common import (
+        DirectDownloadUnavailableError,
+    )
+    from shelfmark.release_sources.direct_download.source import DirectDownloadSource
+
+    class FailingProvider:
+        id = "failing"
+        display_name = "Failing"
+
+        def is_enabled(self):
+            return True
+
+        def search(self, *_args, **_kwargs):
+            raise DirectDownloadUnavailableError("provider unavailable")
+
+    class WorkingProvider:
+        id = "working"
+        display_name = "Working"
+
+        def is_enabled(self):
+            return True
+
+        def search(self, *_args, **_kwargs):
+            return [
+                BrowseRecord(
+                    id="working:1",
+                    title="Example",
+                    source="direct_download",
+                    source_url="https://working.example/book",
+                )
+            ]
+
+    monkeypatch.setattr(
+        registry.config,
+        "get",
+        lambda key, default=None: True if key == "DIRECT_DOWNLOAD_ENABLED" else default,
+    )
+    source = DirectDownloadSource()
+    source._providers = (FailingProvider(), WorkingProvider())
+
+    releases = source.search(SimpleNamespace(title="Example"), SimpleNamespace())
+
+    assert [release.source_id for release in releases] == ["working:1"]