refactor: move the release routes into core/release_routes.py (#1640)

The release routes are the largest piece left in `main.py`.
`/api/releases`, `/api/releases/download`, `/api/release-sources` and
the source record route move to `core/release_routes.py`, registered
with `register_release_routes(app, login_required, ...)`. They take the
content type inference, the request policy checks that gate listing and
download, and the release serializers with them. The inspect route is
registered from the same function since it shares the policy gate.
`main.py` is about 700 lines shorter.

The first commit adds 34 tests for what the policy, budget, direct
provider and title tests did not cover: content type inference on
download, the policy refusal on download, the error statuses of the
listing route, the source list and the record route. They pass on
current `main`. The second commit is the move.

The policy checks need the auth mode, the user database and the request
policy settings, so they come in as callables that `main.py` resolves
lazily, which keeps the existing tests that patch those on
`shelfmark.main` working. One test line changed, the patch of
`get_source_content_type_capabilities`, which now points at the new
module. Log lines from these routes now name
`shelfmark.core.release_routes`.

The URL map is identical, all 77 rules. The full suite, ruff and
basedpyright are clean, and I ran the app from the branch with the other
two route PRs merged in and hit the source list, the listing route's
validation, the record route and the download route, which answer as
before. The exception text a few of them return in a 500 is unchanged,
that is for the follow-up on exception text.


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
<!-- coderabbit-ai-generated:v1 -->

## Summary by CodeRabbit

* **Release Discovery**
* Release searches can use metadata providers, manual searches, or
direct source browsing, with results and source details returned
together.
* Searches respect request policies and report unavailable sources or
partial-search errors.

* **Downloads**
* Download requests validate submitted details, infer a content type
when possible, and report whether the request was queued successfully.

* **Source Browsing**
* Available sources and individual source records can be retrieved,
subject to applicable access policies.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

shelfmark/core/release_routes.py

@@ -0,0 +1,803 @@
+"""Release routes: listing, download, release sources and source records."""
+
+from __future__ import annotations
+
+import re
+import sqlite3
+from typing import TYPE_CHECKING, Any
+
+from flask import jsonify, request, session
+
+from shelfmark.core import search_deadline
+from shelfmark.core.config import config as app_config
+from shelfmark.core.logger import setup_logger
+from shelfmark.core.models import SearchFilters
+from shelfmark.core.release_inspect_routes import register_release_inspect_routes
+from shelfmark.core.request_helpers import (
+    coerce_bool,
+    get_session_db_user_id,
+)
+from shelfmark.core.request_policy import (
+    DEFAULT_SUPPORTED_CONTENT_TYPES,
+    PolicyMode,
+    get_source_content_type_capabilities,
+    merge_request_policy_settings,
+    normalize_content_type,
+    normalize_source,
+    resolve_policy_mode,
+)
+from shelfmark.core.utils import AUDIOBOOK_FORMATS
+from shelfmark.download import orchestrator as backend
+from shelfmark.download import warmup
+from shelfmark.release_sources import BrowseRecord, Release, SourceUnavailableError
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    from flask import Flask
+    from werkzeug.wrappers import Response
+
+    from shelfmark.core.user_db import UserDB
+    from shelfmark.metadata_providers import BookMetadata
+
+logger = setup_logger(__name__)
+
+_OPERATIONAL_ERRORS = (OSError, RuntimeError, TypeError, ValueError, sqlite3.Error)
+_IMPORT_OPERATIONAL_ERRORS = (ImportError, *_OPERATIONAL_ERRORS)
+
+_AUDIOBOOK_CATEGORY_RANGE = (3030, 3049)
+_AUDIOBOOK_FORMAT_HINTS = frozenset(AUDIOBOOK_FORMATS)
+
+
+def _contains_audiobook_format_hint(value: Any) -> bool:
+    if not isinstance(value, str):
+        return False
+
+    normalized = value.strip().lower()
+    if not normalized:
+        return False
+
+    tokens = [token for token in re.split(r"[^a-z0-9]+", normalized) if token]
+    return any(token in _AUDIOBOOK_FORMAT_HINTS for token in tokens)
+
+
+def _resolve_release_content_type(data: dict[str, Any], source: Any) -> tuple[str, bool]:
+    """Resolve release content type for policy checks and queue payload normalization."""
+    extra = data.get("extra")
+    if not isinstance(extra, dict):
+        extra = {}
+
+    explicit_content_type = data.get("content_type")
+    if explicit_content_type is None:
+        explicit_content_type = extra.get("content_type")
+    if explicit_content_type is not None:
+        return normalize_content_type(explicit_content_type), False
+
+    categories = extra.get("categories")
+    if isinstance(categories, list):
+        min_cat, max_cat = _AUDIOBOOK_CATEGORY_RANGE
+        for raw_category in categories:
+            try:
+                category_id = int(raw_category)
+            except TypeError, ValueError:
+                continue
+            if min_cat <= category_id <= max_cat:
+                return "audiobook", True
+
+    candidates: list[Any] = [
+        data.get("format"),
+        extra.get("format"),
+        extra.get("formats_display"),
+        data.get("title"),
+    ]
+    formats = extra.get("formats")
+    if isinstance(formats, list):
+        candidates.extend(formats)
+    else:
+        candidates.append(formats)
+
+    if any(_contains_audiobook_format_hint(candidate) for candidate in candidates):
+        return "audiobook", True
+
+    capabilities = get_source_content_type_capabilities()
+    supported = capabilities.get(normalize_source(source))
+    if supported and len(supported) == 1:
+        return normalize_content_type(next(iter(supported))), True
+
+    return "ebook", False
+
+
+def _policy_block_response(
+    mode: PolicyMode, *, error: str = "Download not allowed by policy"
+) -> tuple[Response, int]:
+    logger.debug(
+        "download policy guard user=%s db_user_id=%s mode=%s",
+        session.get("user_id"),
+        session.get("db_user_id"),
+        mode.value,
+    )
+    if mode == PolicyMode.BLOCKED:
+        return (
+            jsonify(
+                {
+                    "error": error,
+                    "code": "policy_blocked",
+                    "required_mode": PolicyMode.BLOCKED.value,
+                }
+            ),
+            403,
+        )
+    return (
+        jsonify(
+            {
+                "error": error,
+                "code": "policy_requires_request",
+                "required_mode": mode.value,
+            }
+        ),
+        403,
+    )
+
+
+# Modes under which the user never sees a release list: request_book asks for the
+# book as a whole and blocked gets nothing. request_release still lists releases,
+# since that user picks the one they ask for. Listing goes through the shared
+# source credentials (Prowlarr, torrent clients, scrapers), so the routes that
+# list have to refuse these two modes just as the download route does.
+_RELEASE_LISTING_REFUSED_MODES = frozenset({PolicyMode.REQUEST_BOOK, PolicyMode.BLOCKED})
+
+
+def _release_listing_block_response(mode: PolicyMode) -> tuple[Response, int]:
+    return _policy_block_response(mode, error="Release listing not allowed by policy")
+
+
+class _ReleasePolicy:
+    """Request-policy checks for the current session, shared by the release routes."""
+
+    def __init__(
+        self,
+        *,
+        resolve_auth_mode: Callable[[], str],
+        get_user_db: Callable[[], UserDB | None],
+        load_policy_settings: Callable[[], dict[str, Any]],
+    ) -> None:
+        self._resolve_auth_mode = resolve_auth_mode
+        self._get_user_db = get_user_db
+        self._load_policy_settings = load_policy_settings
+
+    def mode_for_current_user(self, *, source: Any, content_type: Any) -> PolicyMode | None:
+        """Resolve policy mode for current session, or None when policy guard is bypassed."""
+        user_db = self._get_user_db()
+        auth_mode = self._resolve_auth_mode()
+        if auth_mode == "none":
+            return None
+        if session.get("is_admin", False):
+            return None
+        if user_db is None:
+            return None
+
+        global_settings = self._load_policy_settings()
+        db_user_id = session.get("db_user_id")
+        user_settings: dict[str, Any] | None = None
+        if db_user_id is not None:
+            try:
+                user_settings = user_db.get_user_settings(int(db_user_id))
+            except TypeError, ValueError:
+                user_settings = None
+
+        effective = merge_request_policy_settings(global_settings, user_settings)
+        if not coerce_bool(effective.get("REQUESTS_ENABLED"), default=False):
+            return None
+
+        resolved_mode = resolve_policy_mode(
+            source=source,
+            content_type=content_type,
+            global_settings=global_settings,
+            user_settings=user_settings,
+        )
+        logger.debug(
+            "download policy resolve user=%s db_user_id=%s is_admin=%s source=%s content_type=%s mode=%s",
+            session.get("user_id"),
+            db_user_id,
+            bool(session.get("is_admin", False)),
+            source,
+            content_type,
+            resolved_mode.value,
+        )
+        return resolved_mode
+
+    def listing_refusal(self, *, source: Any, content_type: Any) -> PolicyMode | None:
+        """Return the policy mode when the current user may not list releases from `source`."""
+        mode = self.mode_for_current_user(source=source, content_type=content_type)
+        if mode is not None and mode in _RELEASE_LISTING_REFUSED_MODES:
+            return mode
+        return None
+
+    def sources_allowed_for_listing(
+        self, sources: list[str], content_type: Any
+    ) -> tuple[list[str], PolicyMode | None]:
+        """Drop the sources the current user may not list releases from.
+
+        Returns the sources left and, when any was dropped, the mode of the first one
+        dropped, which is what the response reports when nothing is left.
+        """
+        allowed: list[str] = []
+        refused_mode: PolicyMode | None = None
+        for source_name in sources:
+            refusal = self.listing_refusal(source=source_name, content_type=content_type)
+            if refusal is None:
+                allowed.append(source_name)
+            elif refused_mode is None:
+                refused_mode = refusal
+        return allowed, refused_mode
+
+    def source_record_refusal(self, source_name: str) -> PolicyMode | None:
+        """Policy refusal for resolving one browse record of a release source.
+
+        The record's content type is not known before it is fetched, so the user is
+        refused only when every content type the source supports is refused.
+        """
+        capabilities = get_source_content_type_capabilities()
+        content_types = capabilities.get(normalize_source(source_name)) or set(
+            DEFAULT_SUPPORTED_CONTENT_TYPES
+        )
+        refused_mode: PolicyMode | None = None
+        for content_type in sorted(content_types):
+            refusal = self.listing_refusal(source=source_name, content_type=content_type)
+            if refusal is None:
+                return None
+            if refused_mode is None:
+                refused_mode = refusal
+        return refused_mode
+
+    def payload_listing_gate(self, data: dict[str, Any]) -> tuple[Response, int] | None:
+        """Policy gate for routes that take a release payload, shared with the inspect route."""
+        source = data["source"]
+        resolved_content_type, _inferred = _resolve_release_content_type(data, source)
+        refusal = self.listing_refusal(source=source, content_type=resolved_content_type)
+        if refusal is None:
+            return None
+        return _release_listing_block_response(refusal)
+
+    def resolve_download_user_context(
+        self,
+        db_user_id: Any,
+        username: Any,
+        on_behalf_of_user_id: Any,
+    ) -> tuple[Any, Any, tuple[Response, int] | None]:
+        """Resolve download queue user context, including optional admin on-behalf overrides."""
+        user_db = self._get_user_db()
+        if on_behalf_of_user_id in (None, ""):
+            return db_user_id, username, None
+
+        if not session.get("is_admin", False):
+            return db_user_id, username, (jsonify({"error": "Admin required"}), 403)
+
+        if user_db is None:
+            return db_user_id, username, (jsonify({"error": "User database unavailable"}), 503)
+
+        try:
+            target_user_id = int(on_behalf_of_user_id)
+        except TypeError, ValueError:
+            return db_user_id, username, (jsonify({"error": "Invalid on_behalf_of_user_id"}), 400)
+
+        if target_user_id <= 0:
+            return db_user_id, username, (jsonify({"error": "Invalid on_behalf_of_user_id"}), 400)
+
+        target_user = user_db.get_user(user_id=target_user_id)
+        if not target_user:
+            return db_user_id, username, (jsonify({"error": "User not found"}), 404)
+
+        return target_user["id"], target_user["username"], None
+
+
+def _parse_search_filters_from_request() -> SearchFilters:
+    """Parse direct/source browse filters from query parameters."""
+    return SearchFilters(
+        isbn=request.args.getlist("isbn"),
+        author=request.args.getlist("author"),
+        title=request.args.getlist("title"),
+        lang=request.args.getlist("lang"),
+        sort=request.args.get("sort"),
+        content=request.args.getlist("content"),
+        format=request.args.getlist("format"),
+    )
+
+
+def _build_source_query_book(query_text: str, filters: SearchFilters) -> BookMetadata:
+    """Build a synthetic book context for source-native browse searches."""
+    from shelfmark.metadata_providers import BookMetadata
+
+    author_values = [value.strip() for value in (filters.author or []) if str(value).strip()]
+    title_values = [value.strip() for value in (filters.title or []) if str(value).strip()]
+    isbn_values = [value.strip() for value in (filters.isbn or []) if str(value).strip()]
+    title = (
+        title_values[0]
+        if title_values
+        else query_text
+        or (isbn_values[0] if isbn_values else "")
+        or (author_values[0] if author_values else "Direct Search")
+    )
+    author = author_values[0] if author_values else ""
+
+    return BookMetadata(
+        provider="manual",
+        provider_id=query_text or title,
+        provider_display_name="Manual Search",
+        title=title,
+        search_title=title,
+        search_author=author or None,
+        authors=author_values,
+    )
+
+
+def _serialize_browse_record(record: BrowseRecord) -> dict:
+    """Serialize a source-native browse record for the frontend."""
+    result = {key: value for key, value in record.__dict__.items() if value is not None}
+
+    preview = result.get("preview")
+    if isinstance(preview, str) and preview:
+        from shelfmark.core.utils import transform_cover_url
+
+        result["preview"] = transform_cover_url(preview, record.id)
+
+    return result
+
+
+def _serialize_release(release: Release) -> dict:
+    """Serialize a release for the frontend, normalizing preview URLs."""
+    from dataclasses import asdict
+
+    from shelfmark.core.utils import transform_cover_url
+
+    result = asdict(release)
+    extra = result.get("extra")
+    if isinstance(extra, dict):
+        preview = extra.get("preview")
+        if isinstance(preview, str) and preview:
+            extra = dict(extra)
+            extra["preview"] = transform_cover_url(preview, release.source_id)
+            result["extra"] = extra
+
+    return result
+
+
+def register_release_routes(
+    app: Flask,
+    login_required: Callable[..., Any],
+    *,
+    resolve_auth_mode: Callable[[], str],
+    get_user_db: Callable[[], UserDB | None],
+    load_policy_settings: Callable[[], dict[str, Any]],
+    resolve_status_scope: Callable[..., tuple[bool, int | None, bool]],
+    attach_library_ownership: Callable[[Any, dict[str, Any]], None],
+) -> None:
+    """Register the release listing, download, source list and source record routes."""
+    policy = _ReleasePolicy(
+        resolve_auth_mode=resolve_auth_mode,
+        get_user_db=get_user_db,
+        load_policy_settings=load_policy_settings,
+    )
+
+    register_release_inspect_routes(app, login_required, policy_gate=policy.payload_listing_gate)
+
+    @app.route("/api/releases/download", methods=["POST"])
+    @login_required
+    def api_download_release() -> Response | tuple[Response, int]:
+        """Queue a release for download.
+
+        This endpoint is used when downloading from the ReleaseModal where the
+        frontend already has all the release data from the search results.
+
+        Request Body (JSON):
+            source (str): Release source (e.g., "direct_download")
+            source_id (str): ID within the source (e.g., AA MD5 hash)
+            title (str): Book title
+            format (str, optional): File format
+            size (str, optional): Human-readable size
+            extra (dict, optional): Additional metadata
+
+        Returns:
+            flask.Response: JSON status object indicating success or failure.
+
+        """
+        try:
+            data = request.get_json(silent=True)
+            if not data:
+                return jsonify({"error": "No data provided"}), 400
+
+            if "source_id" not in data:
+                return jsonify({"error": "source_id is required"}), 400
+            if "source" not in data:
+                return jsonify({"error": "source is required"}), 400
+
+            source = data["source"]
+            resolved_content_type, inferred_content_type = _resolve_release_content_type(
+                data, source
+            )
+            policy_mode = policy.mode_for_current_user(
+                source=source,
+                content_type=resolved_content_type,
+            )
+            if policy_mode is not None and policy_mode != PolicyMode.DOWNLOAD:
+                return _policy_block_response(policy_mode)
+
+            release_payload = data
+            if inferred_content_type and data.get("content_type") is None:
+                release_payload = dict(data)
+                release_payload["content_type"] = resolved_content_type
+
+            logger.info(
+                "Download request received. keys=%s downloads=%s extra.downloads=%s",
+                list(data.keys()),
+                data.get("downloads"),
+                data.get("extra", {}).get("downloads")
+                if isinstance(data.get("extra"), dict)
+                else None,
+            )
+
+            priority = data.get("priority", 0)
+            # Per-user download overrides
+            db_user_id = session.get("db_user_id")
+            _username = session.get("user_id")
+            db_user_id, _username, on_behalf_error = policy.resolve_download_user_context(
+                db_user_id,
+                _username,
+                data.get("on_behalf_of_user_id"),
+            )
+            if on_behalf_error:
+                return on_behalf_error
+            success, error_msg = backend.queue_release(
+                release_payload,
+                priority,
+                user_id=db_user_id,
+                username=_username,
+            )
+
+            if success:
+                return jsonify({"status": "queued", "priority": priority})
+            return jsonify({"error": error_msg or "Failed to queue release"}), 500
+        except _OPERATIONAL_ERRORS as e:
+            logger.error_trace(f"Release download error: {e}")
+            return jsonify({"error": str(e)}), 500
+
+    @app.route("/api/releases", methods=["GET"])
+    @login_required
+    def api_releases() -> Response | tuple[Response, int]:
+        """Search for downloadable releases of a book.
+
+        This endpoint takes book metadata and searches available release sources
+        (e.g., Anna's Archive, Libgen) for downloadable files.
+
+        Query Parameters:
+            provider (str): Metadata provider name (required)
+            book_id (str): Book ID from metadata provider (required)
+            source (str): Release source to search (optional, default: all)
+
+        Returns:
+            flask.Response: JSON with list of available releases.
+
+        """
+        try:
+            from dataclasses import asdict
+
+            from shelfmark.core.release_search import search_source_releases
+            from shelfmark.metadata_providers import (
+                BookMetadata,
+                get_provider,
+                get_provider_kwargs,
+                is_provider_registered,
+            )
+            from shelfmark.release_sources import (
+                apply_column_visibility,
+                browse_record_to_book_metadata,
+                get_source,
+                list_available_sources,
+                serialize_column_config,
+                source_results_are_releases,
+            )
+
+            def _search_source_releases(
+                source_name: str, search_book: BookMetadata
+            ) -> tuple[Any | None, list[Any], str | None]:
+                """Search one source and return any error message instead of raising."""
+                return search_source_releases(
+                    source_name,
+                    search_book,
+                    languages=(
+                        browse_filters.lang if source_query_filters is not None else languages
+                    ),
+                    manual_query=(query_text if source_query_filters is not None else manual_query),
+                    indexers=indexers,
+                    expand_search=expand_search,
+                    content_type=content_type,
+                    source_filters=source_query_filters,
+                    user_id=db_user_id,
+                    include_error_detail=resolve_status_scope(require_authenticated=False)[0],
+                )
+
+            provider = request.args.get("provider", "").strip()
+            book_id = request.args.get("book_id", "").strip()
+            source_filter = request.args.get("source", "").strip()
+            query_text = request.args.get("query", "").strip()
+            # Accept title/author from frontend to avoid re-fetching metadata
+            title_param = request.args.get("title", "").strip()
+            author_param = request.args.get("author", "").strip()
+            expand_search = request.args.get("expand_search", "").lower() == "true"
+            # Accept language codes for filtering (comma-separated)
+            languages_param = request.args.get("languages", "").strip()
+            languages = (
+                [lang.strip() for lang in languages_param.split(",") if lang.strip()]
+                if languages_param
+                else None
+            )
+            # Without an explicit filter the plan falls back to this user's default languages.
+            db_user_id = get_session_db_user_id(session)
+            # Content type for audiobook vs ebook search
+            content_type = request.args.get("content_type", "ebook").strip()
+
+            manual_query = request.args.get("manual_query", "").strip()
+
+            # Accept indexer names for Prowlarr filtering (comma-separated)
+            indexers_param = request.args.get("indexers", "").strip()
+            indexers = (
+                [idx.strip() for idx in indexers_param.split(",") if idx.strip()]
+                if indexers_param
+                else None
+            )
+            browse_filters = _parse_search_filters_from_request()
+            has_browse_filters = bool(query_text or any(vars(browse_filters).values()))
+
+            source_query_filters = None
+            is_source_provider = bool(provider) and source_results_are_releases(provider)
+
+            if not provider or not book_id:
+                if not source_filter or not has_browse_filters:
+                    return jsonify(
+                        {"error": "Parameters 'provider' and 'book_id' are required"}
+                    ), 400
+                if not source_results_are_releases(source_filter):
+                    return jsonify(
+                        {"error": f"Source does not support browse release search: {source_filter}"}
+                    ), 400
+                source_query_filters = browse_filters
+
+            # Determine which release sources to search
+            if source_filter:
+                sources_to_search = [source_filter]
+            elif is_source_provider:
+                # Source-backed browse flows stay within the source that produced the record.
+                sources_to_search = [provider]
+            else:
+                # Search only enabled sources
+                sources_to_search = [
+                    src["name"] for src in list_available_sources() if src["enabled"]
+                ]
+
+            # The policy gate comes before anything is asked of a source: listing goes
+            # through the shared source credentials, so a user whose policy never shows
+            # a release list (request_book, blocked) must not reach them.
+            sources_to_search, refused_mode = policy.sources_allowed_for_listing(
+                sources_to_search, content_type
+            )
+            if not sources_to_search and refused_mode is not None:
+                return _release_listing_block_response(refused_mode)
+
+            book: BookMetadata
+
+            if source_query_filters is not None:
+                book = _build_source_query_book(query_text, browse_filters)
+            elif is_source_provider:
+                # Source-backed browse flows can reopen the release modal with provider=<source name>.
+                # In that flow, treat the source-native record as release-search context instead of
+                # requiring a metadata provider registration.
+                # A source named in `source` was gated above. When provider names another one,
+                # its record lookup needs the refusal the record route applies.
+                if normalize_source(provider) not in {
+                    normalize_source(s) for s in sources_to_search
+                }:
+                    refused_record_mode = policy.source_record_refusal(provider)
+                    if refused_record_mode is not None:
+                        return _release_listing_block_response(refused_record_mode)
+                source = get_source(provider)
+                direct_record = source.get_record(book_id)
+                if direct_record is None:
+                    return jsonify({"error": "Book not found in release source"}), 404
+
+                book = browse_record_to_book_metadata(
+                    direct_record,
+                    title_override=title_param or None,
+                    author_override=author_param or None,
+                )
+            elif provider == "manual":
+                resolved_title = title_param or manual_query or "Manual Search"
+                resolved_author = author_param or ""
+                # The release modal sends `authors.join(', ')` as `author`, so the commas here
+                # are joins between contributors, not part of one name. This split is the only
+                # place that knows that, so `search_author` comes from it rather than from the
+                # joined text - see issue #1252.
+                authors = [a.strip() for a in resolved_author.split(",") if a.strip()]
+
+                book = BookMetadata(
+                    provider="manual",
+                    provider_id=book_id,
+                    provider_display_name="Manual Search",
+                    title=resolved_title,
+                    search_title=resolved_title,
+                    search_author=authors[0] if authors else None,
+                    authors=authors,
+                )
+            else:
+                if not is_provider_registered(provider):
+                    return jsonify({"error": f"Unknown metadata provider: {provider}"}), 400
+
+                # Get book metadata from provider
+                kwargs = get_provider_kwargs(provider)
+                prov = get_provider(provider, **kwargs)
+                resolved_book = prov.get_book(book_id)
+
+                if not resolved_book:
+                    return jsonify({"error": "Book not found in metadata provider"}), 404
+                book = resolved_book
+
+                # Override title from frontend if available (search results may have better data)
+                # Note: We intentionally DON'T override authors here - get_book() now returns
+                # filtered authors (primary authors only, excluding translators/narrators),
+                # which gives better release search results than the unfiltered search data
+                if title_param:
+                    if title_param != book.title:
+                        # A different title is one the search result carried and the book
+                        # record does not, typically a translation's ("Kinder des Nebels" for
+                        # "Mistborn: The Final Empire"). The provider's search_title derives
+                        # from the original and would otherwise win the release search.
+                        book.search_title = title_param
+                    book.title = title_param
+
+            # Search each source for releases.
+            #
+            # Under a wall-clock budget: this endpoint is synchronous, and the bypass path it
+            # can reach used to be allowed minutes per URL with nothing bounding the request
+            # as a whole. A search that ran into an unsolvable protection challenge therefore
+            # outlived every reverse proxy in front of it and surfaced to the user as
+            # "Server unavailable (504)" - a gateway timeout that blames their proxy for a
+            # challenge failure. The budget is shared across sources, so a stuck first source
+            # cannot spend the whole request on its own. See issue #1276.
+            all_releases = []
+            errors = []
+            source_instances = {}  # Keep source instances for column config
+
+            # A real search is under way, so a warm-up still sitting on its start-up delay
+            # should stand down rather than queue its throwaway solve in front of this one.
+            warmup.note_user_search()
+
+            with search_deadline.search_deadline():
+                for index, source_name in enumerate(sources_to_search):
+                    if search_deadline.expired():
+                        logger.warning("Release search budget spent; %s not searched", source_name)
+                        errors.append(f"{source_name}: {search_deadline.deadline_message()}")
+                        continue
+
+                    with search_deadline.unit_deadline(waiting=len(sources_to_search) - index - 1):
+                        source, releases, error = _search_source_releases(source_name, book)
+                    if source is not None:
+                        source_instances[source_name] = source
+                        all_releases.extend(releases)
+                    if error is not None:
+                        errors.append(error)
+
+            # Convert Release objects to dicts
+            releases_data = [_serialize_release(release) for release in all_releases]
+
+            # Get column config from the first source searched
+            # Reuse the same instance to get any dynamic data (e.g., online_servers for IRC)
+            column_config = None
+            if sources_to_search and sources_to_search[0] in source_instances:
+                try:
+                    first_source = source_instances[sources_to_search[0]]
+                    column_config = serialize_column_config(
+                        apply_column_visibility(
+                            first_source.get_column_config(),
+                            content_type=normalize_content_type(content_type),
+                            is_setting_enabled=lambda key: bool(
+                                app_config.get(key, True, user_id=db_user_id)
+                            ),
+                        )
+                    )
+                except _OPERATIONAL_ERRORS as e:
+                    logger.warning("Failed to get column config: %s", e)
+
+            # Convert book to dict and transform cover_url
+            book_dict = asdict(book)
+            attach_library_ownership(book, book_dict)
+            from shelfmark.core.utils import transform_cover_url
+
+            if book_dict.get("cover_url"):
+                cache_id = f"{provider}_{book_id}"
+                book_dict["cover_url"] = transform_cover_url(book_dict["cover_url"], cache_id)
+
+            search_info = {}
+            for source_name, source_instance in source_instances.items():
+                info: dict[str, str | int | None] = {}
+                if (
+                    hasattr(source_instance, "last_search_type")
+                    and source_instance.last_search_type
+                ):
+                    info["search_type"] = source_instance.last_search_type
+                if hasattr(source_instance, "total_results"):
+                    info["total_results"] = source_instance.total_results
+                if info:
+                    search_info[source_name] = info
+
+            response = {
+                "releases": releases_data,
+                "book": book_dict,
+                "sources_searched": sources_to_search,
+                "column_config": column_config,
+                "search_info": search_info,
+            }
+
+            if errors:
+                response["errors"] = errors
+
+            # If no releases found and there were errors, return 503 with the first
+            # source failure message so direct-mode source query searches surface the
+            # same unavailable-state messaging as release modal searches.
+            if not releases_data and errors:
+                # Use the first error message (typically the most relevant)
+                error_message = errors[0]
+                # Strip the source prefix if present (e.g., "direct_download: message" -> "message")
+                if ": " in error_message:
+                    error_message = error_message.split(": ", 1)[1]
+                return jsonify({"error": error_message}), 503
+
+            return jsonify(response)
+        except SourceUnavailableError as e:
+            logger.warning("Release search unavailable: %s", e)
+            return jsonify({"error": str(e)}), 503
+        except _IMPORT_OPERATIONAL_ERRORS as e:
+            logger.error_trace(f"Releases search error: {e}")
+            return jsonify({"error": str(e)}), 500
+
+    @app.route("/api/release-sources", methods=["GET"])
+    @login_required
+    def api_release_sources() -> Response | tuple[Response, int]:
+        """Get available release sources from the plugin registry.
+
+        Returns:
+            flask.Response: JSON list of available release sources.
+
+        """
+        try:
+            from shelfmark.release_sources import list_available_sources
+
+            sources = list_available_sources()
+            return jsonify(sources)
+        except _IMPORT_OPERATIONAL_ERRORS as e:
+            logger.error_trace(f"Release sources error: {e}")
+            return jsonify({"error": str(e)}), 500
+
+    @app.route("/api/release-sources/<source_name>/records/<path:record_id>", methods=["GET"])
+    @login_required
+    def api_release_source_record(
+        source_name: str, record_id: str
+    ) -> Response | tuple[Response, int]:
+        """Resolve a source-native browse record for a release source."""
+        try:
+            from shelfmark.release_sources import get_source
+
+            source = get_source(source_name)
+            refused_mode = policy.source_record_refusal(source_name)
+            if refused_mode is not None:
+                return _release_listing_block_response(refused_mode)
+            record = source.get_record(record_id)
+            if record is None:
+                return jsonify({"error": "Record not found"}), 404
+            return jsonify(_serialize_browse_record(record))
+        except ValueError:
+            return jsonify({"error": f"Unknown release source: {source_name}"}), 400
+        except SourceUnavailableError as e:
+            logger.warning("Release source record unavailable: %s", e)
+            return jsonify({"error": str(e)}), 503
+        except _OPERATIONAL_ERRORS as e:
+            logger.error_trace(f"Release source record error: {e}")
+            return jsonify({"error": str(e)}), 500

shelfmark/main.py

@@ -1,7 +1,6 @@
 """Flask app - routes, WebSocket handlers, and middleware."""
 
 import os
-import re
 import sqlite3
 import time
 from functools import wraps
@@ -34,7 +33,6 @@ from shelfmark.config.security import _migrate_security_settings
 from shelfmark.config.settings import (
     migrate_audiobook_format_settings,
 )
-from shelfmark.core import search_deadline
 from shelfmark.core.activity_view_state_service import ActivityViewStateService
 from shelfmark.core.app_setup import (
     configure_app_logging,
@@ -67,7 +65,6 @@ from shelfmark.core.metadata_routes import register_metadata_routes
 from shelfmark.core.models import (
     TERMINAL_QUEUE_STATUSES,
     QueueStatus,
-    SearchFilters,
 )
 from shelfmark.core.notifications import (
     NotificationContext,
@@ -78,9 +75,8 @@ from shelfmark.core.notifications import (
 from shelfmark.core.onboarding_routes import register_onboarding_routes
 from shelfmark.core.prefix_middleware import PrefixMiddleware
 from shelfmark.core.queue_routes import register_queue_routes
-from shelfmark.core.release_inspect_routes import register_release_inspect_routes
+from shelfmark.core.release_routes import register_release_routes
 from shelfmark.core.request_helpers import (
-    coerce_bool,
     emit_ws_event,
     get_session_db_user_id,
     load_users_request_policy_settings,
@@ -88,13 +84,8 @@ from shelfmark.core.request_helpers import (
     normalize_positive_int,
 )
 from shelfmark.core.request_policy import (
-    DEFAULT_SUPPORTED_CONTENT_TYPES,
-    PolicyMode,
-    get_source_content_type_capabilities,
-    merge_request_policy_settings,
     normalize_content_type,
     normalize_source,
-    resolve_policy_mode,
 )
 from shelfmark.core.requests_service import (
     reopen_failed_request,
@@ -106,7 +97,7 @@ from shelfmark.core.socketio_origins import build_cors_allowed_origins
 from shelfmark.core.stats_routes import register_stats_routes
 from shelfmark.core.system_routes import DebugHooks, register_system_routes
 from shelfmark.core.user_db import UserDB
-from shelfmark.core.utils import AUDIOBOOK_FORMATS, normalize_base_path
+from shelfmark.core.utils import normalize_base_path
 from shelfmark.core.viewer_redaction import (
     configure_public_id_key,
     redact_status,
@@ -115,16 +106,12 @@ from shelfmark.core.viewer_redaction import (
 from shelfmark.download import orchestrator as backend
 from shelfmark.download import warmup
 from shelfmark.release_sources import (
-    BrowseRecord,
-    Release,
-    SourceUnavailableError,
     get_source_display_name,
 )
 
 if TYPE_CHECKING:
     from werkzeug.wrappers import Response
 
-    from shelfmark.metadata_providers import BookMetadata
 
 logger = setup_logger(__name__)
 _OPERATIONAL_ERRORS = (OSError, RuntimeError, TypeError, ValueError, sqlite3.Error)
@@ -331,240 +318,6 @@ def get_auth_mode() -> str:
     return load_active_auth_mode()
 
 
-_AUDIOBOOK_CATEGORY_RANGE = (3030, 3049)
-_AUDIOBOOK_FORMAT_HINTS = frozenset(AUDIOBOOK_FORMATS)
-
-
-def _contains_audiobook_format_hint(value: Any) -> bool:
-    if not isinstance(value, str):
-        return False
-
-    normalized = value.strip().lower()
-    if not normalized:
-        return False
-
-    tokens = [token for token in re.split(r"[^a-z0-9]+", normalized) if token]
-    return any(token in _AUDIOBOOK_FORMAT_HINTS for token in tokens)
-
-
-def _resolve_release_content_type(data: dict[str, Any], source: Any) -> tuple[str, bool]:
-    """Resolve release content type for policy checks and queue payload normalization."""
-    extra = data.get("extra")
-    if not isinstance(extra, dict):
-        extra = {}
-
-    explicit_content_type = data.get("content_type")
-    if explicit_content_type is None:
-        explicit_content_type = extra.get("content_type")
-    if explicit_content_type is not None:
-        return normalize_content_type(explicit_content_type), False
-
-    categories = extra.get("categories")
-    if isinstance(categories, list):
-        min_cat, max_cat = _AUDIOBOOK_CATEGORY_RANGE
-        for raw_category in categories:
-            try:
-                category_id = int(raw_category)
-            except TypeError, ValueError:
-                continue
-            if min_cat <= category_id <= max_cat:
-                return "audiobook", True
-
-    candidates: list[Any] = [
-        data.get("format"),
-        extra.get("format"),
-        extra.get("formats_display"),
-        data.get("title"),
-    ]
-    formats = extra.get("formats")
-    if isinstance(formats, list):
-        candidates.extend(formats)
-    else:
-        candidates.append(formats)
-
-    if any(_contains_audiobook_format_hint(candidate) for candidate in candidates):
-        return "audiobook", True
-
-    capabilities = get_source_content_type_capabilities()
-    supported = capabilities.get(normalize_source(source))
-    if supported and len(supported) == 1:
-        return normalize_content_type(next(iter(supported))), True
-
-    return "ebook", False
-
-
-def _resolve_policy_mode_for_current_user(*, source: Any, content_type: Any) -> PolicyMode | None:
-    """Resolve policy mode for current session, or None when policy guard is bypassed."""
-    auth_mode = get_auth_mode()
-    if auth_mode == "none":
-        return None
-    if session.get("is_admin", False):
-        return None
-    if user_db is None:
-        return None
-
-    global_settings = load_users_request_policy_settings()
-    db_user_id = session.get("db_user_id")
-    user_settings: dict[str, Any] | None = None
-    if db_user_id is not None:
-        try:
-            user_settings = user_db.get_user_settings(int(db_user_id))
-        except TypeError, ValueError:
-            user_settings = None
-
-    effective = merge_request_policy_settings(global_settings, user_settings)
-    if not coerce_bool(effective.get("REQUESTS_ENABLED"), default=False):
-        return None
-
-    resolved_mode = resolve_policy_mode(
-        source=source,
-        content_type=content_type,
-        global_settings=global_settings,
-        user_settings=user_settings,
-    )
-    logger.debug(
-        "download policy resolve user=%s db_user_id=%s is_admin=%s source=%s content_type=%s mode=%s",
-        session.get("user_id"),
-        db_user_id,
-        bool(session.get("is_admin", False)),
-        source,
-        content_type,
-        resolved_mode.value,
-    )
-    return resolved_mode
-
-
-def _policy_block_response(
-    mode: PolicyMode, *, error: str = "Download not allowed by policy"
-) -> tuple[Response, int]:
-    logger.debug(
-        "download policy guard user=%s db_user_id=%s mode=%s",
-        session.get("user_id"),
-        session.get("db_user_id"),
-        mode.value,
-    )
-    if mode == PolicyMode.BLOCKED:
-        return (
-            jsonify(
-                {
-                    "error": error,
-                    "code": "policy_blocked",
-                    "required_mode": PolicyMode.BLOCKED.value,
-                }
-            ),
-            403,
-        )
-    return (
-        jsonify(
-            {
-                "error": error,
-                "code": "policy_requires_request",
-                "required_mode": mode.value,
-            }
-        ),
-        403,
-    )
-
-
-# Modes under which the user never sees a release list: request_book asks for the
-# book as a whole and blocked gets nothing. request_release still lists releases,
-# since that user picks the one they ask for. Listing goes through the shared
-# source credentials (Prowlarr, torrent clients, scrapers), so the routes that
-# list have to refuse these two modes just as the download route does.
-_RELEASE_LISTING_REFUSED_MODES = frozenset({PolicyMode.REQUEST_BOOK, PolicyMode.BLOCKED})
-
-
-def _release_listing_refusal(*, source: Any, content_type: Any) -> PolicyMode | None:
-    """Return the policy mode when the current user may not list releases from `source`."""
-    mode = _resolve_policy_mode_for_current_user(source=source, content_type=content_type)
-    if mode is not None and mode in _RELEASE_LISTING_REFUSED_MODES:
-        return mode
-    return None
-
-
-def _release_listing_block_response(mode: PolicyMode) -> tuple[Response, int]:
-    return _policy_block_response(mode, error="Release listing not allowed by policy")
-
-
-def _sources_allowed_for_listing(
-    sources: list[str], content_type: Any
-) -> tuple[list[str], PolicyMode | None]:
-    """Drop the sources the current user may not list releases from.
-
-    Returns the sources left and, when any was dropped, the mode of the first one
-    dropped, which is what the response reports when nothing is left.
-    """
-    allowed: list[str] = []
-    refused_mode: PolicyMode | None = None
-    for source_name in sources:
-        refusal = _release_listing_refusal(source=source_name, content_type=content_type)
-        if refusal is None:
-            allowed.append(source_name)
-        elif refused_mode is None:
-            refused_mode = refusal
-    return allowed, refused_mode
-
-
-def _release_source_record_refusal(source_name: str) -> PolicyMode | None:
-    """Policy refusal for resolving one browse record of a release source.
-
-    The record's content type is not known before it is fetched, so the user is
-    refused only when every content type the source supports is refused.
-    """
-    capabilities = get_source_content_type_capabilities()
-    content_types = capabilities.get(normalize_source(source_name)) or set(
-        DEFAULT_SUPPORTED_CONTENT_TYPES
-    )
-    refused_mode: PolicyMode | None = None
-    for content_type in sorted(content_types):
-        refusal = _release_listing_refusal(source=source_name, content_type=content_type)
-        if refusal is None:
-            return None
-        if refused_mode is None:
-            refused_mode = refusal
-    return refused_mode
-
-
-def _release_payload_listing_gate(data: dict[str, Any]) -> tuple[Response, int] | None:
-    """Policy gate for routes that take a release payload, shared with the inspect route."""
-    source = data["source"]
-    resolved_content_type, _inferred = _resolve_release_content_type(data, source)
-    refusal = _release_listing_refusal(source=source, content_type=resolved_content_type)
-    if refusal is None:
-        return None
-    return _release_listing_block_response(refusal)
-
-
-def _resolve_download_user_context(
-    db_user_id: Any,
-    username: Any,
-    on_behalf_of_user_id: Any,
-) -> tuple[Any, Any, tuple[Response, int] | None]:
-    """Resolve download queue user context, including optional admin on-behalf overrides."""
-    if on_behalf_of_user_id in (None, ""):
-        return db_user_id, username, None
-
-    if not session.get("is_admin", False):
-        return db_user_id, username, (jsonify({"error": "Admin required"}), 403)
-
-    if user_db is None:
-        return db_user_id, username, (jsonify({"error": "User database unavailable"}), 503)
-
-    try:
-        target_user_id = int(on_behalf_of_user_id)
-    except TypeError, ValueError:
-        return db_user_id, username, (jsonify({"error": "Invalid on_behalf_of_user_id"}), 400)
-
-    if target_user_id <= 0:
-        return db_user_id, username, (jsonify({"error": "Invalid on_behalf_of_user_id"}), 400)
-
-    target_user = user_db.get_user(user_id=target_user_id)
-    if not target_user:
-        return db_user_id, username, (jsonify({"error": "User not found"}), 404)
-
-    return target_user["id"], target_user["username"], None
-
-
 def _emit_request_updates(rows: list[dict[str, Any]]) -> None:
     """Defer request update emission until the runtime hook is available."""
     _emit_request_update_events(rows)
@@ -575,6 +328,11 @@ def _resolve_auth_mode_for_routes() -> str:
     return get_auth_mode()
 
 
+def _load_policy_settings_for_routes() -> dict[str, Any]:
+    """Load the request policy lazily so tests and runtime patches still take effect."""
+    return load_users_request_policy_settings()
+
+
 def _queue_release_for_routes(*args: Any, **kwargs: Any) -> Any:
     """Queue a release via the current backend instance."""
     return backend.queue_release(*args, **kwargs)
@@ -730,157 +488,6 @@ register_system_routes(
 )
 
 
-def _parse_search_filters_from_request() -> SearchFilters:
-    """Parse direct/source browse filters from query parameters."""
-    return SearchFilters(
-        isbn=request.args.getlist("isbn"),
-        author=request.args.getlist("author"),
-        title=request.args.getlist("title"),
-        lang=request.args.getlist("lang"),
-        sort=request.args.get("sort"),
-        content=request.args.getlist("content"),
-        format=request.args.getlist("format"),
-    )
-
-
-def _build_source_query_book(query_text: str, filters: SearchFilters) -> BookMetadata:
-    """Build a synthetic book context for source-native browse searches."""
-    from shelfmark.metadata_providers import BookMetadata
-
-    author_values = [value.strip() for value in (filters.author or []) if str(value).strip()]
-    title_values = [value.strip() for value in (filters.title or []) if str(value).strip()]
-    isbn_values = [value.strip() for value in (filters.isbn or []) if str(value).strip()]
-    title = (
-        title_values[0]
-        if title_values
-        else query_text
-        or (isbn_values[0] if isbn_values else "")
-        or (author_values[0] if author_values else "Direct Search")
-    )
-    author = author_values[0] if author_values else ""
-
-    return BookMetadata(
-        provider="manual",
-        provider_id=query_text or title,
-        provider_display_name="Manual Search",
-        title=title,
-        search_title=title,
-        search_author=author or None,
-        authors=author_values,
-    )
-
-
-def _serialize_browse_record(record: BrowseRecord) -> dict:
-    """Serialize a source-native browse record for the frontend."""
-    result = {key: value for key, value in record.__dict__.items() if value is not None}
-
-    preview = result.get("preview")
-    if isinstance(preview, str) and preview:
-        from shelfmark.core.utils import transform_cover_url
-
-        result["preview"] = transform_cover_url(preview, record.id)
-
-    return result
-
-
-def _serialize_release(release: Release) -> dict:
-    """Serialize a release for the frontend, normalizing preview URLs."""
-    from dataclasses import asdict
-
-    from shelfmark.core.utils import transform_cover_url
-
-    result = asdict(release)
-    extra = result.get("extra")
-    if isinstance(extra, dict):
-        preview = extra.get("preview")
-        if isinstance(preview, str) and preview:
-            extra = dict(extra)
-            extra["preview"] = transform_cover_url(preview, release.source_id)
-            result["extra"] = extra
-
-    return result
-
-
-register_release_inspect_routes(app, login_required, policy_gate=_release_payload_listing_gate)
-
-
-@app.route("/api/releases/download", methods=["POST"])
-@login_required
-def api_download_release() -> Response | tuple[Response, int]:
-    """Queue a release for download.
-
-    This endpoint is used when downloading from the ReleaseModal where the
-    frontend already has all the release data from the search results.
-
-    Request Body (JSON):
-        source (str): Release source (e.g., "direct_download")
-        source_id (str): ID within the source (e.g., AA MD5 hash)
-        title (str): Book title
-        format (str, optional): File format
-        size (str, optional): Human-readable size
-        extra (dict, optional): Additional metadata
-
-    Returns:
-        flask.Response: JSON status object indicating success or failure.
-
-    """
-    try:
-        data = request.get_json(silent=True)
-        if not data:
-            return jsonify({"error": "No data provided"}), 400
-
-        if "source_id" not in data:
-            return jsonify({"error": "source_id is required"}), 400
-        if "source" not in data:
-            return jsonify({"error": "source is required"}), 400
-
-        source = data["source"]
-        resolved_content_type, inferred_content_type = _resolve_release_content_type(data, source)
-        policy_mode = _resolve_policy_mode_for_current_user(
-            source=source,
-            content_type=resolved_content_type,
-        )
-        if policy_mode is not None and policy_mode != PolicyMode.DOWNLOAD:
-            return _policy_block_response(policy_mode)
-
-        release_payload = data
-        if inferred_content_type and data.get("content_type") is None:
-            release_payload = dict(data)
-            release_payload["content_type"] = resolved_content_type
-
-        logger.info(
-            "Download request received. keys=%s downloads=%s extra.downloads=%s",
-            list(data.keys()),
-            data.get("downloads"),
-            data.get("extra", {}).get("downloads") if isinstance(data.get("extra"), dict) else None,
-        )
-
-        priority = data.get("priority", 0)
-        # Per-user download overrides
-        db_user_id = session.get("db_user_id")
-        _username = session.get("user_id")
-        db_user_id, _username, on_behalf_error = _resolve_download_user_context(
-            db_user_id,
-            _username,
-            data.get("on_behalf_of_user_id"),
-        )
-        if on_behalf_error:
-            return on_behalf_error
-        success, error_msg = backend.queue_release(
-            release_payload,
-            priority,
-            user_id=db_user_id,
-            username=_username,
-        )
-
-        if success:
-            return jsonify({"status": "queued", "priority": priority})
-        return jsonify({"error": error_msg or "Failed to queue release"}), 500
-    except _OPERATIONAL_ERRORS as e:
-        logger.error_trace(f"Release download error: {e}")
-        return jsonify({"error": str(e)}), 500
-
-
 register_stats_routes(
     app,
     get_user_db=lambda: user_db,
@@ -1240,335 +847,15 @@ register_metadata_routes(
 )
 
 
-@app.route("/api/releases", methods=["GET"])
-@login_required
-def api_releases() -> Response | tuple[Response, int]:
-    """Search for downloadable releases of a book.
-
-    This endpoint takes book metadata and searches available release sources
-    (e.g., Anna's Archive, Libgen) for downloadable files.
-
-    Query Parameters:
-        provider (str): Metadata provider name (required)
-        book_id (str): Book ID from metadata provider (required)
-        source (str): Release source to search (optional, default: all)
-
-    Returns:
-        flask.Response: JSON with list of available releases.
-
-    """
-    try:
-        from dataclasses import asdict
-
-        from shelfmark.core.release_search import search_source_releases
-        from shelfmark.metadata_providers import (
-            BookMetadata,
-            get_provider,
-            get_provider_kwargs,
-            is_provider_registered,
-        )
-        from shelfmark.release_sources import (
-            apply_column_visibility,
-            browse_record_to_book_metadata,
-            get_source,
-            list_available_sources,
-            serialize_column_config,
-            source_results_are_releases,
-        )
-
-        def _search_source_releases(
-            source_name: str, search_book: BookMetadata
-        ) -> tuple[Any | None, list[Any], str | None]:
-            """Search one source and return any error message instead of raising."""
-            return search_source_releases(
-                source_name,
-                search_book,
-                languages=(browse_filters.lang if source_query_filters is not None else languages),
-                manual_query=(query_text if source_query_filters is not None else manual_query),
-                indexers=indexers,
-                expand_search=expand_search,
-                content_type=content_type,
-                source_filters=source_query_filters,
-                user_id=db_user_id,
-                include_error_detail=_resolve_status_scope(require_authenticated=False)[0],
-            )
-
-        provider = request.args.get("provider", "").strip()
-        book_id = request.args.get("book_id", "").strip()
-        source_filter = request.args.get("source", "").strip()
-        query_text = request.args.get("query", "").strip()
-        # Accept title/author from frontend to avoid re-fetching metadata
-        title_param = request.args.get("title", "").strip()
-        author_param = request.args.get("author", "").strip()
-        expand_search = request.args.get("expand_search", "").lower() == "true"
-        # Accept language codes for filtering (comma-separated)
-        languages_param = request.args.get("languages", "").strip()
-        languages = (
-            [lang.strip() for lang in languages_param.split(",") if lang.strip()]
-            if languages_param
-            else None
-        )
-        # Without an explicit filter the plan falls back to this user's default languages.
-        db_user_id = get_session_db_user_id(session)
-        # Content type for audiobook vs ebook search
-        content_type = request.args.get("content_type", "ebook").strip()
-
-        manual_query = request.args.get("manual_query", "").strip()
-
-        # Accept indexer names for Prowlarr filtering (comma-separated)
-        indexers_param = request.args.get("indexers", "").strip()
-        indexers = (
-            [idx.strip() for idx in indexers_param.split(",") if idx.strip()]
-            if indexers_param
-            else None
-        )
-        browse_filters = _parse_search_filters_from_request()
-        has_browse_filters = bool(query_text or any(vars(browse_filters).values()))
-
-        source_query_filters = None
-        is_source_provider = bool(provider) and source_results_are_releases(provider)
-
-        if not provider or not book_id:
-            if not source_filter or not has_browse_filters:
-                return jsonify({"error": "Parameters 'provider' and 'book_id' are required"}), 400
-            if not source_results_are_releases(source_filter):
-                return jsonify(
-                    {"error": f"Source does not support browse release search: {source_filter}"}
-                ), 400
-            source_query_filters = browse_filters
-
-        # Determine which release sources to search
-        if source_filter:
-            sources_to_search = [source_filter]
-        elif is_source_provider:
-            # Source-backed browse flows stay within the source that produced the record.
-            sources_to_search = [provider]
-        else:
-            # Search only enabled sources
-            sources_to_search = [src["name"] for src in list_available_sources() if src["enabled"]]
-
-        # The policy gate comes before anything is asked of a source: listing goes
-        # through the shared source credentials, so a user whose policy never shows
-        # a release list (request_book, blocked) must not reach them.
-        sources_to_search, refused_mode = _sources_allowed_for_listing(
-            sources_to_search, content_type
-        )
-        if not sources_to_search and refused_mode is not None:
-            return _release_listing_block_response(refused_mode)
-
-        book: BookMetadata
-
-        if source_query_filters is not None:
-            book = _build_source_query_book(query_text, browse_filters)
-        elif is_source_provider:
-            # Source-backed browse flows can reopen the release modal with provider=<source name>.
-            # In that flow, treat the source-native record as release-search context instead of
-            # requiring a metadata provider registration.
-            # A source named in `source` was gated above. When provider names another one,
-            # its record lookup needs the refusal the record route applies.
-            if normalize_source(provider) not in {normalize_source(s) for s in sources_to_search}:
-                refused_record_mode = _release_source_record_refusal(provider)
-                if refused_record_mode is not None:
-                    return _release_listing_block_response(refused_record_mode)
-            source = get_source(provider)
-            direct_record = source.get_record(book_id)
-            if direct_record is None:
-                return jsonify({"error": "Book not found in release source"}), 404
-
-            book = browse_record_to_book_metadata(
-                direct_record,
-                title_override=title_param or None,
-                author_override=author_param or None,
-            )
-        elif provider == "manual":
-            resolved_title = title_param or manual_query or "Manual Search"
-            resolved_author = author_param or ""
-            # The release modal sends `authors.join(', ')` as `author`, so the commas here
-            # are joins between contributors, not part of one name. This split is the only
-            # place that knows that, so `search_author` comes from it rather than from the
-            # joined text - see issue #1252.
-            authors = [a.strip() for a in resolved_author.split(",") if a.strip()]
-
-            book = BookMetadata(
-                provider="manual",
-                provider_id=book_id,
-                provider_display_name="Manual Search",
-                title=resolved_title,
-                search_title=resolved_title,
-                search_author=authors[0] if authors else None,
-                authors=authors,
-            )
-        else:
-            if not is_provider_registered(provider):
-                return jsonify({"error": f"Unknown metadata provider: {provider}"}), 400
-
-            # Get book metadata from provider
-            kwargs = get_provider_kwargs(provider)
-            prov = get_provider(provider, **kwargs)
-            resolved_book = prov.get_book(book_id)
-
-            if not resolved_book:
-                return jsonify({"error": "Book not found in metadata provider"}), 404
-            book = resolved_book
-
-            # Override title from frontend if available (search results may have better data)
-            # Note: We intentionally DON'T override authors here - get_book() now returns
-            # filtered authors (primary authors only, excluding translators/narrators),
-            # which gives better release search results than the unfiltered search data
-            if title_param:
-                if title_param != book.title:
-                    # A different title is one the search result carried and the book
-                    # record does not, typically a translation's ("Kinder des Nebels" for
-                    # "Mistborn: The Final Empire"). The provider's search_title derives
-                    # from the original and would otherwise win the release search.
-                    book.search_title = title_param
-                book.title = title_param
-
-        # Search each source for releases.
-        #
-        # Under a wall-clock budget: this endpoint is synchronous, and the bypass path it
-        # can reach used to be allowed minutes per URL with nothing bounding the request
-        # as a whole. A search that ran into an unsolvable protection challenge therefore
-        # outlived every reverse proxy in front of it and surfaced to the user as
-        # "Server unavailable (504)" - a gateway timeout that blames their proxy for a
-        # challenge failure. The budget is shared across sources, so a stuck first source
-        # cannot spend the whole request on its own. See issue #1276.
-        all_releases = []
-        errors = []
-        source_instances = {}  # Keep source instances for column config
-
-        # A real search is under way, so a warm-up still sitting on its start-up delay
-        # should stand down rather than queue its throwaway solve in front of this one.
-        warmup.note_user_search()
-
-        with search_deadline.search_deadline():
-            for index, source_name in enumerate(sources_to_search):
-                if search_deadline.expired():
-                    logger.warning("Release search budget spent; %s not searched", source_name)
-                    errors.append(f"{source_name}: {search_deadline.deadline_message()}")
-                    continue
-
-                with search_deadline.unit_deadline(waiting=len(sources_to_search) - index - 1):
-                    source, releases, error = _search_source_releases(source_name, book)
-                if source is not None:
-                    source_instances[source_name] = source
-                    all_releases.extend(releases)
-                if error is not None:
-                    errors.append(error)
-
-        # Convert Release objects to dicts
-        releases_data = [_serialize_release(release) for release in all_releases]
-
-        # Get column config from the first source searched
-        # Reuse the same instance to get any dynamic data (e.g., online_servers for IRC)
-        column_config = None
-        if sources_to_search and sources_to_search[0] in source_instances:
-            try:
-                first_source = source_instances[sources_to_search[0]]
-                column_config = serialize_column_config(
-                    apply_column_visibility(
-                        first_source.get_column_config(),
-                        content_type=normalize_content_type(content_type),
-                        is_setting_enabled=lambda key: bool(
-                            app_config.get(key, True, user_id=db_user_id)
-                        ),
-                    )
-                )
-            except _OPERATIONAL_ERRORS as e:
-                logger.warning("Failed to get column config: %s", e)
-
-        # Convert book to dict and transform cover_url
-        book_dict = asdict(book)
-        _attach_library_ownership(book, book_dict)
-        from shelfmark.core.utils import transform_cover_url
-
-        if book_dict.get("cover_url"):
-            cache_id = f"{provider}_{book_id}"
-            book_dict["cover_url"] = transform_cover_url(book_dict["cover_url"], cache_id)
-
-        search_info = {}
-        for source_name, source_instance in source_instances.items():
-            info: dict[str, str | int | None] = {}
-            if hasattr(source_instance, "last_search_type") and source_instance.last_search_type:
-                info["search_type"] = source_instance.last_search_type
-            if hasattr(source_instance, "total_results"):
-                info["total_results"] = source_instance.total_results
-            if info:
-                search_info[source_name] = info
-
-        response = {
-            "releases": releases_data,
-            "book": book_dict,
-            "sources_searched": sources_to_search,
-            "column_config": column_config,
-            "search_info": search_info,
-        }
-
-        if errors:
-            response["errors"] = errors
-
-        # If no releases found and there were errors, return 503 with the first
-        # source failure message so direct-mode source query searches surface the
-        # same unavailable-state messaging as release modal searches.
-        if not releases_data and errors:
-            # Use the first error message (typically the most relevant)
-            error_message = errors[0]
-            # Strip the source prefix if present (e.g., "direct_download: message" -> "message")
-            if ": " in error_message:
-                error_message = error_message.split(": ", 1)[1]
-            return jsonify({"error": error_message}), 503
-
-        return jsonify(response)
-    except SourceUnavailableError as e:
-        logger.warning("Release search unavailable: %s", e)
-        return jsonify({"error": str(e)}), 503
-    except _IMPORT_OPERATIONAL_ERRORS as e:
-        logger.error_trace(f"Releases search error: {e}")
-        return jsonify({"error": str(e)}), 500
-
-
-@app.route("/api/release-sources", methods=["GET"])
-@login_required
-def api_release_sources() -> Response | tuple[Response, int]:
-    """Get available release sources from the plugin registry.
-
-    Returns:
-        flask.Response: JSON list of available release sources.
-
-    """
-    try:
-        from shelfmark.release_sources import list_available_sources
-
-        sources = list_available_sources()
-        return jsonify(sources)
-    except _IMPORT_OPERATIONAL_ERRORS as e:
-        logger.error_trace(f"Release sources error: {e}")
-        return jsonify({"error": str(e)}), 500
-
-
-@app.route("/api/release-sources/<source_name>/records/<path:record_id>", methods=["GET"])
-@login_required
-def api_release_source_record(source_name: str, record_id: str) -> Response | tuple[Response, int]:
-    """Resolve a source-native browse record for a release source."""
-    try:
-        from shelfmark.release_sources import get_source
-
-        source = get_source(source_name)
-        refused_mode = _release_source_record_refusal(source_name)
-        if refused_mode is not None:
-            return _release_listing_block_response(refused_mode)
-        record = source.get_record(record_id)
-        if record is None:
-            return jsonify({"error": "Record not found"}), 404
-        return jsonify(_serialize_browse_record(record))
-    except ValueError:
-        return jsonify({"error": f"Unknown release source: {source_name}"}), 400
-    except SourceUnavailableError as e:
-        logger.warning("Release source record unavailable: %s", e)
-        return jsonify({"error": str(e)}), 503
-    except _OPERATIONAL_ERRORS as e:
-        logger.error_trace(f"Release source record error: {e}")
-        return jsonify({"error": str(e)}), 500
+register_release_routes(
+    app,
+    login_required,
+    resolve_auth_mode=_resolve_auth_mode_for_routes,
+    get_user_db=lambda: user_db,
+    load_policy_settings=_load_policy_settings_for_routes,
+    resolve_status_scope=_resolve_status_scope,
+    attach_library_ownership=_attach_library_ownership,
+)
 
 
 register_settings_routes(app, login_required)

tests/core/test_release_routes.py

@@ -0,0 +1,548 @@
+"""Behaviour of the release routes: download, listing, source list and source records.
+
+Pins the parts the policy, budget, direct-provider and title-override tests do not
+reach: content-type inference on download, policy refusal on download, the error
+statuses of /api/releases, the source list and the source record route.
+"""
+
+from __future__ import annotations
+
+import importlib
+import uuid
+from unittest.mock import patch
+
+import pytest
+
+from shelfmark.metadata_providers import BookMetadata
+from shelfmark.release_sources import BrowseRecord, SourceUnavailableError
+
+
+@pytest.fixture(scope="module")
+def main_module():
+    with patch("shelfmark.download.orchestrator.start"):
+        import shelfmark.main as main
+
+        importlib.reload(main)
+        return main
+
+
+@pytest.fixture
+def client(main_module):
+    return main_module.app.test_client()
+
+
+def _no_auth(main_module):
+    return patch.object(main_module, "get_auth_mode", return_value="none")
+
+
+def _policy(default_ebook: str = "download", default_audiobook: str = "download") -> dict:
+    return {
+        "REQUESTS_ENABLED": True,
+        "REQUEST_POLICY_DEFAULT_EBOOK": default_ebook,
+        "REQUEST_POLICY_DEFAULT_AUDIOBOOK": default_audiobook,
+        "MAX_PENDING_REQUESTS_PER_USER": 20,
+        "REQUESTS_ALLOW_NOTES": True,
+        "REQUEST_POLICY_RULES": [],
+    }
+
+
+def _login_user(client, main_module, *, is_admin: bool = False) -> dict:
+    user = main_module.user_db.create_user(
+        username=f"reader-{uuid.uuid4().hex[:8]}", role="admin" if is_admin else "user"
+    )
+    with client.session_transaction() as sess:
+        sess["user_id"] = user["username"]
+        sess["is_admin"] = is_admin
+        sess["db_user_id"] = user["id"]
+    return user
+
+
+def _download(client, main_module, payload: dict) -> tuple[object, dict]:
+    """POST a release to the download route; returns the response and the queued payload."""
+    captured: dict = {}
+
+    def fake_queue_release(release_data, priority, user_id=None, username=None):
+        captured["release_data"] = release_data
+        return True, None
+
+    with (
+        _no_auth(main_module),
+        patch.object(main_module.backend, "queue_release", side_effect=fake_queue_release),
+    ):
+        resp = client.post("/api/releases/download", json=payload)
+    return resp, captured
+
+
+class TestDownloadContentType:
+    def test_audiobook_torznab_category_marks_the_release_as_audiobook(self, client, main_module):
+        payload = {
+            "source": "prowlarr",
+            "source_id": "a",
+            "title": "Some Book",
+            "extra": {"categories": [3035]},
+        }
+
+        resp, captured = _download(client, main_module, payload)
+
+        assert resp.status_code == 200
+        assert captured["release_data"] == {**payload, "content_type": "audiobook"}
+
+    def test_a_category_outside_the_audiobook_range_stays_ebook(self, client, main_module):
+        payload = {
+            "source": "prowlarr",
+            "source_id": "a",
+            "title": "Some Book",
+            "extra": {"categories": [7020, "not-a-number"]},
+        }
+
+        resp, captured = _download(client, main_module, payload)
+
+        assert resp.status_code == 200
+        assert captured["release_data"] == payload
+
+    def test_audiobook_format_in_the_formats_list_is_enough(self, client, main_module):
+        payload = {
+            "source": "prowlarr",
+            "source_id": "a",
+            "title": "Some Book",
+            "extra": {"formats": ["epub", "m4b"]},
+        }
+
+        _resp, captured = _download(client, main_module, payload)
+
+        assert captured["release_data"]["content_type"] == "audiobook"
+
+    def test_an_explicit_content_type_is_forwarded_untouched(self, client, main_module):
+        payload = {
+            "source": "prowlarr",
+            "source_id": "a",
+            "title": "Some Book m4b",
+            "content_type": "ebook",
+        }
+
+        resp, captured = _download(client, main_module, payload)
+
+        assert resp.status_code == 200
+        assert captured["release_data"] == payload
+
+    def test_a_content_type_inside_extra_counts_as_explicit(self, client, main_module):
+        payload = {
+            "source": "prowlarr",
+            "source_id": "a",
+            "title": "Some Book m4b",
+            "extra": {"content_type": "ebook"},
+        }
+
+        resp, captured = _download(client, main_module, payload)
+
+        assert resp.status_code == 200
+        assert captured["release_data"] == payload
+
+    def test_a_source_that_only_serves_audiobooks_decides_the_type(self, client, main_module):
+        payload = {"source": "audio_only", "source_id": "a", "title": "Some Book"}
+
+        with patch(
+            "shelfmark.core.release_routes.get_source_content_type_capabilities",
+            return_value={"audio_only": {"audiobook"}},
+        ):
+            resp, captured = _download(client, main_module, payload)
+
+        assert resp.status_code == 200
+        assert captured["release_data"] == {**payload, "content_type": "audiobook"}
+
+
+class TestDownloadPolicyAndFailures:
+    def _post_with_policy(self, client, main_module, policy: dict):
+        with (
+            patch.object(main_module, "get_auth_mode", return_value="builtin"),
+            patch.object(main_module, "load_users_request_policy_settings", return_value=policy),
+            patch.object(main_module.backend, "queue_release") as queue_release,
+        ):
+            resp = client.post(
+                "/api/releases/download",
+                json={"source": "direct_download", "source_id": "a", "title": "Dune"},
+            )
+        return resp, queue_release
+
+    def test_request_book_user_is_told_to_request_the_release_instead(self, client, main_module):
+        _login_user(client, main_module)
+
+        resp, queue_release = self._post_with_policy(
+            client, main_module, _policy(default_ebook="request_book")
+        )
+
+        assert resp.status_code == 403
+        assert resp.get_json() == {
+            "error": "Download not allowed by policy",
+            "code": "policy_requires_request",
+            "required_mode": "request_release",
+        }
+        queue_release.assert_not_called()
+
+    def test_blocked_user_gets_policy_blocked(self, client, main_module):
+        _login_user(client, main_module)
+
+        resp, queue_release = self._post_with_policy(
+            client, main_module, _policy(default_ebook="blocked")
+        )
+
+        assert resp.status_code == 403
+        assert resp.get_json()["code"] == "policy_blocked"
+        assert resp.get_json()["required_mode"] == "blocked"
+        queue_release.assert_not_called()
+
+    def test_admin_downloads_regardless_of_policy(self, client, main_module):
+        _login_user(client, main_module, is_admin=True)
+
+        with (
+            patch.object(main_module, "get_auth_mode", return_value="builtin"),
+            patch.object(
+                main_module,
+                "load_users_request_policy_settings",
+                return_value=_policy(default_ebook="blocked"),
+            ),
+            patch.object(main_module.backend, "queue_release", return_value=(True, None)),
+        ):
+            resp = client.post(
+                "/api/releases/download",
+                json={"source": "direct_download", "source_id": "a", "title": "Dune"},
+            )
+
+        assert resp.status_code == 200
+
+    def test_a_failed_queue_returns_the_reason_with_500(self, client, main_module):
+        with (
+            _no_auth(main_module),
+            patch.object(
+                main_module.backend, "queue_release", return_value=(False, "No free slot")
+            ),
+        ):
+            resp = client.post(
+                "/api/releases/download",
+                json={"source": "direct_download", "source_id": "a", "title": "Dune"},
+            )
+
+        assert resp.status_code == 500
+        assert resp.get_json() == {"error": "No free slot"}
+
+    def test_a_failed_queue_without_a_reason_uses_a_default_message(self, client, main_module):
+        with (
+            _no_auth(main_module),
+            patch.object(main_module.backend, "queue_release", return_value=(False, None)),
+        ):
+            resp = client.post(
+                "/api/releases/download",
+                json={"source": "direct_download", "source_id": "a", "title": "Dune"},
+            )
+
+        assert resp.status_code == 500
+        assert resp.get_json() == {"error": "Failed to queue release"}
+
+    def test_an_operational_error_while_queueing_returns_500(self, client, main_module):
+        with (
+            _no_auth(main_module),
+            patch.object(main_module.backend, "queue_release", side_effect=RuntimeError("boom")),
+        ):
+            resp = client.post(
+                "/api/releases/download",
+                json={"source": "direct_download", "source_id": "a", "title": "Dune"},
+            )
+
+        assert resp.status_code == 500
+
+
+class _Provider:
+    def __init__(self, book: BookMetadata | None) -> None:
+        self._book = book
+
+    def get_book(self, book_id: str) -> BookMetadata | None:
+        return self._book
+
+
+class _Source:
+    last_search_type = "title"
+    total_results = 1
+
+    def search(self, book, plan, *, expand_search=False, content_type="ebook"):
+        return []
+
+    def get_column_config(self):
+        from shelfmark.release_sources import _default_column_config
+
+        return _default_column_config()
+
+
+def _book(**overrides) -> BookMetadata:
+    fields = {
+        "provider": "hardcover",
+        "provider_id": "42",
+        "provider_display_name": "Hardcover",
+        "title": "Dune",
+        "authors": ["Frank Herbert"],
+    }
+    fields.update(overrides)
+    return BookMetadata(**fields)
+
+
+def _list_releases(client, main_module, query: dict, *, book: BookMetadata | None = None, **extra):
+    """GET /api/releases with the provider and one release source stubbed."""
+    with (
+        _no_auth(main_module),
+        patch("shelfmark.metadata_providers.is_provider_registered", return_value=True),
+        patch("shelfmark.metadata_providers.get_provider_kwargs", return_value={}),
+        patch("shelfmark.metadata_providers.get_provider", return_value=_Provider(book)),
+        patch(
+            "shelfmark.release_sources.list_available_sources",
+            return_value=[{"name": "direct_download", "enabled": True}],
+        ),
+        patch("shelfmark.release_sources.get_source", return_value=_Source()),
+        patch("shelfmark.release_sources.source_results_are_releases", return_value=False),
+    ):
+        return client.get("/api/releases", query_string=query)
+
+
+class TestReleaseListing:
+    def test_provider_and_book_id_are_required(self, client, main_module):
+        resp = _list_releases(client, main_module, {})
+
+        assert resp.status_code == 400
+        assert resp.get_json() == {"error": "Parameters 'provider' and 'book_id' are required"}
+
+    def test_a_source_without_browse_filters_is_not_enough(self, client, main_module):
+        resp = _list_releases(client, main_module, {"source": "direct_download"})
+
+        assert resp.status_code == 400
+        assert resp.get_json() == {"error": "Parameters 'provider' and 'book_id' are required"}
+
+    def test_a_source_that_does_not_browse_is_refused(self, client, main_module):
+        resp = _list_releases(client, main_module, {"source": "direct_download", "query": "dune"})
+
+        assert resp.status_code == 400
+        assert resp.get_json() == {
+            "error": "Source does not support browse release search: direct_download"
+        }
+
+    def test_an_unregistered_provider_is_refused(self, client, main_module):
+        with (
+            _no_auth(main_module),
+            patch("shelfmark.metadata_providers.is_provider_registered", return_value=False),
+            patch(
+                "shelfmark.release_sources.list_available_sources",
+                return_value=[{"name": "direct_download", "enabled": True}],
+            ),
+            patch("shelfmark.release_sources.source_results_are_releases", return_value=False),
+        ):
+            resp = client.get("/api/releases", query_string={"provider": "nope", "book_id": "1"})
+
+        assert resp.status_code == 400
+        assert resp.get_json() == {"error": "Unknown metadata provider: nope"}
+
+    def test_a_book_the_provider_does_not_know_is_a_404(self, client, main_module):
+        resp = _list_releases(client, main_module, {"provider": "hardcover", "book_id": "42"})
+
+        assert resp.status_code == 404
+        assert resp.get_json() == {"error": "Book not found in metadata provider"}
+
+    def test_a_source_record_that_does_not_exist_is_a_404(self, client, main_module):
+        class _EmptySource(_Source):
+            def get_record(self, record_id, *, fetch_download_count=True):
+                return None
+
+        with (
+            _no_auth(main_module),
+            patch("shelfmark.release_sources.get_source", return_value=_EmptySource()),
+            patch("shelfmark.release_sources.source_results_are_releases", return_value=True),
+            patch(
+                "shelfmark.release_sources.list_available_sources",
+                return_value=[{"name": "direct_download", "enabled": True}],
+            ),
+        ):
+            resp = client.get(
+                "/api/releases",
+                query_string={"provider": "direct_download", "book_id": "gone"},
+            )
+
+        assert resp.status_code == 404
+        assert resp.get_json() == {"error": "Book not found in release source"}
+
+    def test_the_response_carries_book_sources_and_search_info(self, client, main_module):
+        resp = _list_releases(
+            client, main_module, {"provider": "hardcover", "book_id": "42"}, book=_book()
+        )
+
+        assert resp.status_code == 200
+        body = resp.get_json()
+        assert body["releases"] == []
+        assert body["book"]["title"] == "Dune"
+        assert body["sources_searched"] == ["direct_download"]
+        assert body["column_config"] is not None
+        assert body["search_info"] == {
+            "direct_download": {"search_type": "title", "total_results": 1}
+        }
+        assert "errors" not in body
+
+    def test_library_ownership_is_attached_to_the_book(self, client, main_module):
+        with patch("shelfmark.core.library_index.ownership", return_value={"epub": True}):
+            resp = _list_releases(
+                client, main_module, {"provider": "hardcover", "book_id": "42"}, book=_book()
+            )
+
+        assert resp.status_code == 200
+        assert resp.get_json()["book"]["library"] == {"epub": True}
+
+    def test_no_library_key_when_the_check_is_off(self, client, main_module):
+        with patch("shelfmark.core.library_index.ownership", return_value=None):
+            resp = _list_releases(
+                client, main_module, {"provider": "hardcover", "book_id": "42"}, book=_book()
+            )
+
+        assert "library" not in resp.get_json()["book"]
+
+    def test_every_source_failing_returns_503_without_the_source_prefix(self, client, main_module):
+        with patch(
+            "shelfmark.core.release_search.search_source_releases",
+            return_value=(None, [], "direct_download: Bypass failed"),
+        ):
+            resp = _list_releases(
+                client, main_module, {"provider": "hardcover", "book_id": "42"}, book=_book()
+            )
+
+        assert resp.status_code == 503
+        assert resp.get_json() == {"error": "Bypass failed"}
+
+    def test_a_failing_source_with_releases_elsewhere_still_returns_them(self, client, main_module):
+        from shelfmark.release_sources import Release
+
+        release = Release(source="direct_download", source_id="r1", title="Dune", format="epub")
+        with patch(
+            "shelfmark.core.release_search.search_source_releases",
+            return_value=(_Source(), [release], "prowlarr: down"),
+        ):
+            resp = _list_releases(
+                client, main_module, {"provider": "hardcover", "book_id": "42"}, book=_book()
+            )
+
+        assert resp.status_code == 200
+        body = resp.get_json()
+        assert [r["source_id"] for r in body["releases"]] == ["r1"]
+        assert body["errors"] == ["prowlarr: down"]
+
+    def test_an_unavailable_source_returns_503(self, client, main_module):
+        with patch(
+            "shelfmark.core.release_search.search_source_releases",
+            side_effect=SourceUnavailableError("Source is down"),
+        ):
+            resp = _list_releases(
+                client, main_module, {"provider": "hardcover", "book_id": "42"}, book=_book()
+            )
+
+        assert resp.status_code == 503
+        assert resp.get_json() == {"error": "Source is down"}
+
+    def test_the_manual_provider_splits_joined_authors(self, client, main_module):
+        resp = _list_releases(
+            client,
+            main_module,
+            {
+                "provider": "manual",
+                "book_id": "m1",
+                "title": "Good Omens",
+                "author": "Terry Pratchett, Neil Gaiman",
+            },
+        )
+
+        assert resp.status_code == 200
+        book = resp.get_json()["book"]
+        assert book["title"] == "Good Omens"
+        assert book["authors"] == ["Terry Pratchett", "Neil Gaiman"]
+        assert book["search_author"] == "Terry Pratchett"
+
+
+class TestReleaseSourcesAndRecords:
+    def test_release_sources_lists_the_registry(self, client, main_module):
+        sources = [{"name": "direct_download", "enabled": True}]
+        with (
+            _no_auth(main_module),
+            patch("shelfmark.release_sources.list_available_sources", return_value=sources),
+        ):
+            resp = client.get("/api/release-sources")
+
+        assert resp.status_code == 200
+        assert resp.get_json() == sources
+
+    def test_release_sources_failure_returns_500(self, client, main_module):
+        with (
+            _no_auth(main_module),
+            patch(
+                "shelfmark.release_sources.list_available_sources",
+                side_effect=RuntimeError("registry broke"),
+            ),
+        ):
+            resp = client.get("/api/release-sources")
+
+        assert resp.status_code == 500
+
+    def test_an_unknown_source_is_a_400(self, client, main_module):
+        with (
+            _no_auth(main_module),
+            patch("shelfmark.release_sources.get_source", side_effect=ValueError("unknown")),
+        ):
+            resp = client.get("/api/release-sources/nope/records/abc")
+
+        assert resp.status_code == 400
+        assert resp.get_json() == {"error": "Unknown release source: nope"}
+
+    def test_a_missing_record_is_a_404(self, client, main_module):
+        class _NoRecord:
+            def get_record(self, record_id):
+                return None
+
+        with (
+            _no_auth(main_module),
+            patch("shelfmark.release_sources.get_source", return_value=_NoRecord()),
+        ):
+            resp = client.get("/api/release-sources/direct_download/records/abc")
+
+        assert resp.status_code == 404
+        assert resp.get_json() == {"error": "Record not found"}
+
+    def test_record_ids_may_contain_slashes(self, client, main_module):
+        seen: list[str] = []
+
+        class _Records:
+            def get_record(self, record_id):
+                seen.append(record_id)
+                return BrowseRecord(id=record_id, title="Browsed", source="stub")
+
+        with (
+            _no_auth(main_module),
+            patch("shelfmark.release_sources.get_source", return_value=_Records()),
+        ):
+            resp = client.get("/api/release-sources/stub/records/a/b/c")
+
+        assert resp.status_code == 200
+        assert seen == ["a/b/c"]
+        body = resp.get_json()
+        assert body["id"] == "a/b/c"
+        assert body["title"] == "Browsed"
+        assert None not in body.values()
+
+
+class TestReleaseRoutesNeedLogin:
+    @pytest.mark.parametrize(
+        ("method", "path"),
+        [
+            ("get", "/api/releases"),
+            ("post", "/api/releases/download"),
+            ("get", "/api/release-sources"),
+            ("get", "/api/release-sources/stub/records/abc"),
+        ],
+    )
+    def test_anonymous_requests_are_refused(self, client, main_module, method, path):
+        with client.session_transaction() as sess:
+            sess.clear()
+        with patch.object(main_module, "get_auth_mode", return_value="builtin"):
+            resp = getattr(client, method)(path)
+
+        assert resp.status_code == 401
+        assert resp.get_json() == {"error": "Unauthorized"}

tests/core/test_release_routes_policy.py

@@ -282,8 +282,9 @@ def _get_record(
     with (
         auth_patch,
         policy_patch,
-        patch.object(
-            main_module, "get_source_content_type_capabilities", return_value=capabilities
+        patch(
+            "shelfmark.core.release_routes.get_source_content_type_capabilities",
+            return_value=capabilities,
         ),
         patch("shelfmark.release_sources.get_source", return_value=_Source(searched)),
     ):