Magent/backend/app/routers/requests.py

from ..services import manual_releases
from ..services.request_language import language_info, original_profile, is_original_profile, apply_original_to_movie, movie_search_outcome, series_search_outcome
from ..feature_guards import require_request_access
from typing import Any, Dict, List, Optional, Tuple
import asyncio
import httpx
import json
import logging
import os
import time
from urllib.parse import quote
from datetime import datetime, timezone, timedelta
from fastapi import APIRouter, HTTPException, Depends

from ..clients.jellyseerr import JellyseerrClient
from ..clients.jellyfin import JellyfinClient
from ..clients.qbittorrent import QBittorrentClient
from ..clients.radarr import RadarrClient
from ..clients.sonarr import SonarrClient
from ..clients.bazarr import BazarrClient
from ..ai.triage import triage_snapshot
from ..auth import get_current_user
from ..runtime import get_runtime_settings
from .images import cache_tmdb_image, is_tmdb_cached
from ..db import (
    get_request_stage_cache,
    save_request_stage_cache,
    add_portal_item_activity,
    get_portal_item,
    save_action,
    get_recent_actions,
    get_recent_snapshots,
    get_cached_requests,
    get_cached_requests_since,
    get_cached_request_by_media_id,
    get_request_cache_lookup,
    get_request_cache_payload,
    get_request_cache_last_updated,
    get_request_cache_count,
    get_request_cache_payloads,
    get_request_cache_payloads_missing,
    repair_request_cache_titles,
    prune_duplicate_requests_cache,
    upsert_request_cache,
    upsert_request_cache_many,
    upsert_artwork_cache_status,
    upsert_artwork_cache_status_many,
    get_artwork_cache_missing_count,
    get_artwork_cache_status_count,
    get_setting,
    set_setting,
    update_portal_item,
    update_artwork_cache_stats,
    cleanup_history,
    is_seerr_media_failure_suppressed,
    record_seerr_media_failure,
    clear_seerr_media_failure,
    get_request_download_evidence,
    start_request_repair,
    get_request_repairs,
    active_repair_request_ids,
)
from ..services.media_repair import current_cycle_torrents
from ..services.download_labels import label_episode_downloads
from ..models import Snapshot, TriageResult, RequestType
from ..services.snapshot import (
    _summarize_qbit,
    _torrent_progress,
    build_snapshot,
    jellyfin_item_matches_request,
)

router = APIRouter(prefix="/requests", tags=["requests"], dependencies=[Depends(get_current_user), Depends(require_request_access)])

CACHE_TTL_SECONDS = 600
_detail_cache: Dict[str, Tuple[float, Dict[str, Any]]] = {}
FAILED_DETAIL_CACHE_TTL_SECONDS = 3600
_failed_detail_cache: Dict[str, float] = {}
REQUEST_CACHE_TTL_SECONDS = 600
logger = logging.getLogger(__name__)
_sync_state: Dict[str, Any] = {
    "status": "idle",
    "stored": 0,
    "total": None,
    "skip": 0,
    "message": None,
    "started_at": None,
    "finished_at": None,
}
_sync_task: Optional[asyncio.Task] = None
_sync_last_key = "requests_sync_last_at"
RECENT_CACHE_MAX_DAYS = 180
RECENT_CACHE_TTL_SECONDS = 300
_recent_cache: Dict[str, Any] = {"items": [], "updated_at": None}
_artwork_prefetch_state: Dict[str, Any] = {
    "status": "idle",
    "processed": 0,
    "total": 0,
    "message": "",
    "only_missing": False,
    "started_at": None,
    "finished_at": None,
}
_artwork_prefetch_task: Optional[asyncio.Task] = None

STATUS_LABELS = {
    1: "Waiting for approval",
    2: "Approved",
    3: "Declined",
    4: "Ready to watch",
    5: "Working on it",
    6: "Partially ready",
}

REQUEST_STAGE_CODES = {
    "all": None,
    "pending": [1],
    "approved": [2],
    "declined": [3],
    "ready": [4],
    "working": [5],
    "partial": [6],
    "in_progress": [2, 5, 6],
}


def _cache_get(key: str) -> Optional[Dict[str, Any]]:
    cached = _detail_cache.get(key)
    if not cached:
        return None
    expires_at, payload = cached
    if expires_at < time.time():
        _detail_cache.pop(key, None)
        return None
    return payload


def _cache_set(key: str, payload: Dict[str, Any]) -> None:
    _detail_cache[key] = (time.time() + CACHE_TTL_SECONDS, payload)
    _failed_detail_cache.pop(key, None)


def _status_label_with_jellyfin(current_status: Any, jellyfin_available: bool) -> str:
    if not jellyfin_available:
        return _status_label(current_status)
    try:
        status_code = int(current_status)
    except (TypeError, ValueError):
        status_code = None
    if status_code == 6:
        return STATUS_LABELS[6]
    return STATUS_LABELS[4]


async def _request_is_available_in_jellyfin(
    jellyfin: JellyfinClient,
    title: Optional[str],
    year: Optional[int],
    media_type: Optional[str],
    request_payload: Optional[Dict[str, Any]],
    availability_cache: Dict[str, bool],
    raise_errors: bool = False,
) -> bool:
    if not jellyfin.configured() or not title:
        return False
    cache_key = f"{media_type or ''}:{title.lower()}:{year or ''}:{request_payload.get('id') if isinstance(request_payload, dict) else ''}"
    cached_value = availability_cache.get(cache_key)
    if cached_value is not None:
        return cached_value
    types = ["Movie"] if media_type == "movie" else ["Series"]
    try:
        search = await jellyfin.search_items(title, types, limit=50)
    except Exception:
        if raise_errors:
            raise
        availability_cache[cache_key] = False
        return False
    if isinstance(search, dict):
        items = search.get("Items") or search.get("items") or []
        request_type = RequestType.movie if media_type == "movie" else RequestType.tv
        for item in items:
            if not isinstance(item, dict):
                continue
            if jellyfin_item_matches_request(
                item,
                title=title,
                year=year,
                request_type=request_type,
                request_payload=request_payload,
            ):
                availability_cache[cache_key] = True
                return True
    availability_cache[cache_key] = False
    return False


def _failure_cache_has(key: str) -> bool:
    expires_at = _failed_detail_cache.get(key)
    if not expires_at:
        return False
    if expires_at < time.time():
        _failed_detail_cache.pop(key, None)
        return False
    return True


def _failure_cache_set(key: str, ttl_seconds: int = FAILED_DETAIL_CACHE_TTL_SECONDS) -> None:
    _failed_detail_cache[key] = time.time() + ttl_seconds


def _extract_http_error_message(exc: httpx.HTTPStatusError) -> Optional[str]:
    response = exc.response
    if response is None:
        return None
    try:
        payload = response.json()
    except ValueError:
        payload = response.text
    if isinstance(payload, dict):
        message = payload.get("message") or payload.get("error")
        return str(message).strip() if message else json.dumps(payload, ensure_ascii=True)
    if isinstance(payload, str):
        trimmed = payload.strip()
        return trimmed or None
    return str(payload)


def _should_persist_seerr_media_failure(exc: httpx.HTTPStatusError) -> bool:
    response = exc.response
    if response is None:
        return False
    return response.status_code == 404 or response.status_code >= 500


def _status_label(value: Any) -> str:
    if isinstance(value, int):
        return STATUS_LABELS.get(value, f"Status {value}")
    return "Unknown"


def normalize_request_stage_filter(value: Optional[str]) -> str:
    if not isinstance(value, str):
        return "all"
    normalized = value.strip().lower().replace("-", "_").replace(" ", "_")
    if not normalized:
        return "all"
    if normalized in {"processing", "inprogress"}:
        normalized = "in_progress"
    return normalized if normalized in REQUEST_STAGE_CODES else "all"


def request_stage_filter_codes(value: Optional[str]) -> Optional[list[int]]:
    normalized = normalize_request_stage_filter(value)
    codes = REQUEST_STAGE_CODES.get(normalized)
    return list(codes) if codes else None


def _normalize_username(value: Any) -> Optional[str]:
    if not isinstance(value, str):
        return None
    normalized = value.strip().lower()
    if not normalized:
        return None
    if "@" in normalized:
        normalized = normalized.split("@", 1)[0]
    return normalized if normalized else None


def _user_can_use_search_auto(user: Dict[str, Any]) -> bool:
    if user.get("role") == "admin":
        return True
    return bool(user.get("auto_search_enabled", True))


def _filter_snapshot_for_user(snapshot: Snapshot, user: Dict[str, Any]) -> Snapshot:
    if not _user_can_use_search_auto(user):
        snapshot.actions = [action for action in snapshot.actions if action.id != "search_auto"]
    if user.get("role") != "admin":
        # The standard request view is intentionally collaborative, but service payloads can
        # contain requester identities, internal URLs, download hashes and diagnostic errors.
        snapshot.timeline = []
        snapshot.raw = {}
    return snapshot


def _require_advanced_request_access(user: Dict[str, Any]) -> None:
    if user.get("role") != "admin":
        raise HTTPException(
            status_code=403,
            detail="Advanced request details are available to administrators only",
        )


def _quality_profile_id(value: Any) -> Optional[int]:
    if isinstance(value, int):
        return value
    if isinstance(value, str) and value.strip().isdigit():
        return int(value.strip())
    return None


def _request_matches_user(request_data: Any, username: str) -> bool:
    requested_by = None
    if isinstance(request_data, dict):
        requested_by = request_data.get("requestedBy") or request_data.get("requestedByUser")
        if requested_by is None:
            requested_by = request_data.get("requestedByName") or request_data.get("requestedByUsername")
    if isinstance(requested_by, dict):
        candidates = [
            requested_by.get("username"),
            requested_by.get("displayName"),
            requested_by.get("name"),
            requested_by.get("email"),
        ]
    else:
        candidates = [requested_by]

    username_norm = _normalize_username(username)
    if not username_norm:
        return False

    for candidate in candidates:
        candidate_norm = _normalize_username(candidate)
        if not candidate_norm:
            continue
        if "@" in candidate_norm:
            candidate_norm = candidate_norm.split("@", 1)[0]
        if candidate_norm == username_norm:
            return True
    return False


def _normalize_requested_by(request_data: Any) -> Optional[str]:
    if not isinstance(request_data, dict):
        return None
    requested_by = request_data.get("requestedBy")
    if isinstance(requested_by, dict):
        for key in ("username", "displayName", "name", "email"):
            value = requested_by.get(key)
            normalized = _normalize_username(value)
            if normalized and "@" in normalized:
                normalized = normalized.split("@", 1)[0]
            if normalized:
                return normalized
    normalized = _normalize_username(requested_by)
    if normalized and "@" in normalized:
        normalized = normalized.split("@", 1)[0]
    return normalized

def _extract_requested_by_id(request_data: Any) -> Optional[int]:
    if not isinstance(request_data, dict):
        return None
    requested_by = request_data.get("requestedBy") or request_data.get("requestedByUser")
    if isinstance(requested_by, dict):
        for key in ("id", "userId", "Id"):
            value = requested_by.get(key)
            if value is None:
                continue
            try:
                return int(value)
            except (TypeError, ValueError):
                continue
    return None


def _format_upstream_error(service: str, exc: httpx.HTTPStatusError) -> str:
    response = exc.response
    status = response.status_code if response is not None else "unknown"
    message = ""
    if response is not None:
        try:
            payload = response.json()
            if isinstance(payload, dict):
                message = str(payload.get("message") or payload.get("error") or "").strip()
            elif isinstance(payload, list):
                validation_messages = [
                    str(item.get("errorMessage") or item.get("message") or "").strip()
                    for item in payload
                    if isinstance(item, dict)
                ]
                message = "; ".join(item for item in validation_messages if item)
        except ValueError:
            message = response.text.strip()
    if message:
        compact_message = " ".join(message.split())[:500]
        return f"{service} could not complete the request ({status}): {compact_message}"
    return f"{service} could not complete the request ({status})."


def _request_display_name(request_data: Any) -> Optional[str]:
    if not isinstance(request_data, dict):
        return None
    requested_by = request_data.get("requestedBy")
    if isinstance(requested_by, dict):
        for key in ("displayName", "username", "name", "email"):
            value = requested_by.get(key)
            if isinstance(value, str) and value.strip():
                return value.strip()
    if isinstance(requested_by, str) and requested_by.strip():
        return requested_by.strip()
    return None


def _parse_request_payload(item: Dict[str, Any]) -> Dict[str, Any]:
    media = item.get("media") or {}
    media_id = media.get("id") or item.get("mediaId")
    media_type = media.get("mediaType") or item.get("type")
    tmdb_id = media.get("tmdbId") or item.get("tmdbId")
    title = media.get("title") or media.get("name") or item.get("title") or item.get("name")
    year = media.get("year") or item.get("year")
    created_at = item.get("createdAt") or item.get("addedAt") or item.get("updatedAt")
    updated_at = item.get("updatedAt") or created_at
    requested_by = _request_display_name(item)
    requested_by_norm = _normalize_requested_by(item)
    requested_by_id = _extract_requested_by_id(item)
    return {
        "request_id": item.get("id"),
        "media_id": media_id,
        "media_type": media_type,
        "tmdb_id": tmdb_id,
        "status": item.get("status"),
        "title": title,
        "year": year,
        "requested_by": requested_by,
        "requested_by_norm": requested_by_norm,
        "requested_by_id": requested_by_id,
        "created_at": created_at,
        "updated_at": updated_at,
    }


def _merge_request_media_details(
    request_payload: Dict[str, Any], details: Dict[str, Any]
) -> Dict[str, Any]:
    """Fill display metadata omitted by Seerr's request mutation/detail payloads."""
    merged = dict(request_payload)
    media = request_payload.get("media")
    media = dict(media) if isinstance(media, dict) else {}
    media_type = _normalize_media_type(
        media.get("mediaType") or request_payload.get("mediaType") or request_payload.get("type")
    )

    title = details.get("title") or details.get("name")
    if title and not (media.get("title") or media.get("name")):
        if media_type == "tv":
            media["name"] = title
        else:
            media["title"] = title

    date_value = details.get("releaseDate") or details.get("firstAirDate")
    if not media.get("year") and isinstance(date_value, str) and date_value[:4].isdigit():
        media["year"] = int(date_value[:4])

    for camel_key, snake_key in (
        ("posterPath", "poster_path"),
        ("backdropPath", "backdrop_path"),
    ):
        if not (media.get(camel_key) or media.get(snake_key)):
            value = details.get(camel_key) or details.get(snake_key)
            if value:
                media[camel_key] = value

    if media_type and not media.get("mediaType"):
        media["mediaType"] = media_type
    merged["media"] = media
    return merged


def _extract_artwork_paths(item: Dict[str, Any]) -> tuple[Optional[str], Optional[str]]:
    media = item.get("media") or {}
    poster_path = None
    backdrop_path = None
    if isinstance(media, dict):
        poster_path = media.get("posterPath") or media.get("poster_path")
        backdrop_path = media.get("backdropPath") or media.get("backdrop_path")
    if not poster_path:
        poster_path = item.get("posterPath") or item.get("poster_path")
    if not backdrop_path:
        backdrop_path = item.get("backdropPath") or item.get("backdrop_path")
    return poster_path, backdrop_path

def _extract_tmdb_lookup(payload: Dict[str, Any]) -> tuple[Optional[int], Optional[str]]:
    media = payload.get("media") or {}
    if not isinstance(media, dict):
        media = {}
    tmdb_id = media.get("tmdbId") or payload.get("tmdbId")
    media_type = (
        media.get("mediaType")
        or payload.get("mediaType")
        or payload.get("type")
    )
    try:
        tmdb_id = int(tmdb_id) if tmdb_id is not None else None
    except (TypeError, ValueError):
        tmdb_id = None
    if isinstance(media_type, str):
        media_type = media_type.strip().lower() or None
    else:
        media_type = None
    return tmdb_id, media_type


def _normalize_media_type(value: Any) -> Optional[str]:
    if not isinstance(value, str):
        return None
    normalized = value.strip().lower()
    if normalized in {"movie", "tv"}:
        return normalized
    return None


def _normalize_seasons(value: Any) -> list[int]:
    if value is None:
        return []
    if not isinstance(value, list):
        raise HTTPException(status_code=400, detail="seasons must be an array of positive integers")
    normalized: list[int] = []
    for raw in value:
        try:
            season = int(raw)
        except (TypeError, ValueError) as exc:
            raise HTTPException(
                status_code=400, detail="seasons must contain only positive integers"
            ) from exc
        if season <= 0:
            raise HTTPException(status_code=400, detail="seasons must contain only positive integers")
        normalized.append(season)
    return sorted(set(normalized))


def _normalize_request_profiles(value: Any) -> list[Dict[str, Any]]:
    if not isinstance(value, list):
        return []
    profiles: list[Dict[str, Any]] = []
    for item in value:
        if not isinstance(item, dict):
            continue
        profile_id = _quality_profile_id(item.get("id"))
        name = str(item.get("name") or "").strip()
        if profile_id is None or not name:
            continue
        profiles.append({"id": profile_id, "name": name})
    return profiles


def _normalize_request_roots(value: Any) -> list[str]:
    if not isinstance(value, list):
        return []
    roots: list[str] = []
    for item in value:
        if not isinstance(item, dict):
            continue
        path = str(item.get("path") or "").strip()
        if path:
            roots.append(path)
    return roots


def _normalize_seerr_servers(value: Any) -> list[Dict[str, Any]]:
    if isinstance(value, list):
        return [item for item in value if isinstance(item, dict)]
    if isinstance(value, dict):
        results = value.get("results")
        if isinstance(results, list):
            return [item for item in results if isinstance(item, dict)]
    return []


async def _resolve_request_destination(
    runtime: Any,
    seerr: JellyseerrClient,
    media_type: str,
) -> Dict[str, Any]:
    if media_type == "tv":
        collector_name = "Sonarr"
        collector = SonarrClient(runtime.sonarr_base_url, runtime.sonarr_api_key)
        configured_profile_id = _quality_profile_id(runtime.sonarr_quality_profile_id)
        configured_root = str(runtime.sonarr_root_folder or "").strip()
    else:
        collector_name = "Radarr"
        collector = RadarrClient(runtime.radarr_base_url, runtime.radarr_api_key)
        configured_profile_id = _quality_profile_id(runtime.radarr_quality_profile_id)
        configured_root = str(runtime.radarr_root_folder or "").strip()

    if not collector.configured():
        raise HTTPException(status_code=400, detail=f"{collector_name} is not configured")

    try:
        server_settings, profile_payload, root_payload = await asyncio.gather(
            seerr.get_service_settings(media_type),
            collector.get_quality_profiles(),
            collector.get_root_folders(),
        )
    except httpx.HTTPStatusError as exc:
        service = "Seerr" if "/settings/" in str(exc.request.url) else collector_name
        raise HTTPException(status_code=502, detail=_format_upstream_error(service, exc)) from exc

    servers = [item for item in _normalize_seerr_servers(server_settings) if not item.get("is4k")]
    if not servers:
        raise HTTPException(
            status_code=409,
            detail=f"Seerr has no standard {collector_name} destination configured.",
        )
    server = next((item for item in servers if item.get("isDefault")), servers[0])

    profiles = _normalize_request_profiles(profile_payload)
    if not profiles:
        raise HTTPException(status_code=409, detail=f"{collector_name} has no quality profiles available.")
    profile_ids = {int(item["id"]) for item in profiles}

    # Magent's administrator default is authoritative for every new request.
    # An unset default inherits Seerr's profile; a stale default must be repaired.
    default_profile_id = configured_profile_id
    if default_profile_id is None:
        default_profile_id = _quality_profile_id(server.get("activeProfileId"))
    if default_profile_id not in profile_ids:
        raise HTTPException(
            status_code=409,
            detail=f"The default quality profile is not available in {collector_name}. Ask an administrator to select a valid default in Admin settings.",
        )
    selected_profile_id = default_profile_id

    roots = _normalize_request_roots(root_payload)
    root_folder = str(server.get("activeDirectory") or "").strip()
    if root_folder not in roots:
        root_folder = configured_root if configured_root in roots else ""
    if not root_folder:
        raise HTTPException(
            status_code=409,
            detail=f"Seerr's {collector_name} library location does not match an active {collector_name} root folder.",
        )

    server_id = _quality_profile_id(server.get("id"))
    if server_id is None:
        raise HTTPException(status_code=409, detail=f"Seerr's {collector_name} destination is invalid.")

    return {
        "collector": collector_name,
        "server_id": server_id,
        "server_name": str(server.get("name") or collector_name),
        "profile_id": int(selected_profile_id),
        "default_profile_id": int(default_profile_id),
        "profiles": profiles,
        "root_folder": root_folder,
    }


def _artwork_missing_for_payload(payload: Dict[str, Any]) -> bool:
    poster_path, backdrop_path = _extract_artwork_paths(payload)
    tmdb_id, media_type = _extract_tmdb_lookup(payload)
    can_hydrate = bool(tmdb_id and media_type)
    if poster_path:
        if not is_tmdb_cached(poster_path, "w185") or not is_tmdb_cached(poster_path, "w342"):
            return True
    elif can_hydrate:
        return True
    if backdrop_path:
        if not is_tmdb_cached(backdrop_path, "w780"):
            return True
    elif can_hydrate:
        return True
    return False


def _compute_cached_flags(
    poster_path: Optional[str],
    backdrop_path: Optional[str],
    cache_mode: str,
    poster_cached: Optional[bool] = None,
    backdrop_cached: Optional[bool] = None,
) -> tuple[bool, bool]:
    if cache_mode != "cache":
        return True, True
    poster = poster_cached
    backdrop = backdrop_cached
    if poster is None:
        poster = bool(poster_path) and is_tmdb_cached(poster_path, "w185") and is_tmdb_cached(
            poster_path, "w342"
        )
    if backdrop is None:
        backdrop = bool(backdrop_path) and is_tmdb_cached(backdrop_path, "w780")
    return bool(poster), bool(backdrop)


def _upsert_artwork_status(
    payload: Dict[str, Any],
    cache_mode: str,
    poster_cached: Optional[bool] = None,
    backdrop_cached: Optional[bool] = None,
) -> None:
    record = _build_artwork_status_record(payload, cache_mode, poster_cached, backdrop_cached)
    if not record:
        return
    upsert_artwork_cache_status(**record)


def _build_request_cache_record(payload: Dict[str, Any], request_payload: Dict[str, Any]) -> Dict[str, Any]:
    return {
        "request_id": payload.get("request_id"),
        "media_id": payload.get("media_id"),
        "media_type": payload.get("media_type"),
        "status": payload.get("status"),
        "title": payload.get("title"),
        "year": payload.get("year"),
        "requested_by": payload.get("requested_by"),
        "requested_by_norm": payload.get("requested_by_norm"),
        "requested_by_id": payload.get("requested_by_id"),
        "created_at": payload.get("created_at"),
        "updated_at": payload.get("updated_at"),
        "payload_json": json.dumps(request_payload, ensure_ascii=True),
    }


def _build_artwork_status_record(
    payload: Dict[str, Any],
    cache_mode: str,
    poster_cached: Optional[bool] = None,
    backdrop_cached: Optional[bool] = None,
) -> Optional[Dict[str, Any]]:
    parsed = _parse_request_payload(payload)
    request_id = parsed.get("request_id")
    if not isinstance(request_id, int):
        return None
    tmdb_id, media_type = _extract_tmdb_lookup(payload)
    poster_path, backdrop_path = _extract_artwork_paths(payload)
    has_tmdb = bool(tmdb_id and media_type)
    poster_cached_flag, backdrop_cached_flag = _compute_cached_flags(
        poster_path, backdrop_path, cache_mode, poster_cached, backdrop_cached
    )
    return {
        "request_id": request_id,
        "tmdb_id": tmdb_id,
        "media_type": media_type,
        "poster_path": poster_path,
        "backdrop_path": backdrop_path,
        "has_tmdb": has_tmdb,
        "poster_cached": poster_cached_flag,
        "backdrop_cached": backdrop_cached_flag,
    }


def _collect_artwork_cache_disk_stats() -> tuple[int, int]:
    cache_root = os.path.join(os.getcwd(), "data", "artwork")
    total_bytes = 0
    total_files = 0
    if not os.path.isdir(cache_root):
        return 0, 0
    for root, _, files in os.walk(cache_root):
        for name in files:
            path = os.path.join(root, name)
            try:
                total_bytes += os.path.getsize(path)
                total_files += 1
            except OSError:
                continue
    return total_bytes, total_files


async def _get_request_details(client: JellyseerrClient, request_id: int) -> Optional[Dict[str, Any]]:
    cache_key = f"request:{request_id}"
    cached = _cache_get(cache_key)
    if isinstance(cached, dict):
        parsed_cached = _parse_request_payload(cached)
        if parsed_cached.get("title"):
            return cached
        details = await _get_media_details(
            client, parsed_cached.get("media_type"), parsed_cached.get("tmdb_id")
        )
        if isinstance(details, dict):
            cached = _merge_request_media_details(cached, details)
            _cache_set(cache_key, cached)
        return cached
    if _failure_cache_has(cache_key):
        return None
    try:
        fetched = await client.get_request(str(request_id))
    except httpx.HTTPStatusError:
        _failure_cache_set(cache_key)
        return None
    if isinstance(fetched, dict):
        parsed_fetched = _parse_request_payload(fetched)
        if not parsed_fetched.get("title"):
            details = await _get_media_details(
                client, parsed_fetched.get("media_type"), parsed_fetched.get("tmdb_id")
            )
            if isinstance(details, dict):
                fetched = _merge_request_media_details(fetched, details)
        _cache_set(cache_key, fetched)
        return fetched
    return None


async def _get_media_details(
    client: JellyseerrClient, media_type: Optional[str], tmdb_id: Optional[int]
) -> Optional[Dict[str, Any]]:
    if not tmdb_id or not media_type:
        return None
    normalized_media_type = str(media_type).strip().lower()
    if normalized_media_type not in {"movie", "tv"}:
        return None
    cache_key = f"media:{normalized_media_type}:{int(tmdb_id)}"
    cached = _cache_get(cache_key)
    if isinstance(cached, dict):
        return cached
    if is_seerr_media_failure_suppressed(normalized_media_type, int(tmdb_id)):
        logger.debug(
            "Seerr media hydration suppressed from db: media_type=%s tmdb_id=%s",
            normalized_media_type,
            tmdb_id,
        )
        _failure_cache_set(cache_key, ttl_seconds=FAILED_DETAIL_CACHE_TTL_SECONDS)
        return None
    if _failure_cache_has(cache_key):
        return None
    try:
        if normalized_media_type == "movie":
            fetched = await client.get_movie(int(tmdb_id))
        else:
            fetched = await client.get_tv(int(tmdb_id))
    except httpx.HTTPStatusError as exc:
        _failure_cache_set(cache_key)
        if _should_persist_seerr_media_failure(exc):
            record_seerr_media_failure(
                normalized_media_type,
                int(tmdb_id),
                status_code=exc.response.status_code if exc.response is not None else None,
                error_message=_extract_http_error_message(exc),
            )
        return None
    if isinstance(fetched, dict):
        clear_seerr_media_failure(normalized_media_type, int(tmdb_id))
        _cache_set(cache_key, fetched)
        return fetched
    return None


async def _hydrate_title_from_tmdb(
    client: JellyseerrClient, media_type: Optional[str], tmdb_id: Optional[int]
) -> tuple[Optional[str], Optional[int]]:
    details = await _get_media_details(client, media_type, tmdb_id)
    if not isinstance(details, dict):
        return None, None
    normalized_media_type = str(media_type).strip().lower() if media_type else None
    if normalized_media_type == "movie":
        title = details.get("title")
        release_date = details.get("releaseDate")
        year = int(release_date[:4]) if release_date else None
        return title, year
    if normalized_media_type == "tv":
        title = details.get("name") or details.get("title")
        first_air = details.get("firstAirDate")
        year = int(first_air[:4]) if first_air else None
        return title, year
    return None, None


async def _hydrate_artwork_from_tmdb(
    client: JellyseerrClient, media_type: Optional[str], tmdb_id: Optional[int]
) -> tuple[Optional[str], Optional[str]]:
    details = await _get_media_details(client, media_type, tmdb_id)
    if not isinstance(details, dict):
        return None, None
    return (
        details.get("posterPath") or details.get("poster_path"),
        details.get("backdropPath") or details.get("backdrop_path"),
    )


def _artwork_url(path: Optional[str], size: str, cache_mode: str) -> Optional[str]:
    if not path:
        return None
    if not path.startswith("/"):
        path = f"/{path}"
    if cache_mode == "cache":
        return f"/images/tmdb?path={quote(path)}&size={size}"
    return f"https://image.tmdb.org/t/p/{size}{path}"


def _cache_is_stale(last_updated: Optional[str]) -> bool:
    if not last_updated:
        return True
    runtime = get_runtime_settings()
    ttl_seconds = max(60, int(runtime.requests_sync_ttl_minutes or 1440) * 60)
    try:
        parsed = datetime.fromisoformat(last_updated.replace("Z", "+00:00"))
        now = datetime.now(timezone.utc)
        return (now - parsed).total_seconds() > ttl_seconds
    except ValueError:
        return True


def _parse_time(value: Optional[str], fallback_hour: int, fallback_minute: int) -> tuple[int, int]:
    if isinstance(value, str) and ":" in value:
        parts = value.strip().split(":")
        if len(parts) == 2:
            try:
                hour = int(parts[0])
                minute = int(parts[1])
                if 0 <= hour <= 23 and 0 <= minute <= 59:
                    return hour, minute
            except ValueError:
                pass
    return fallback_hour, fallback_minute


def _seconds_until(hour: int, minute: int) -> int:
    now = datetime.now(timezone.utc).astimezone()
    target = now.replace(hour=hour, minute=minute, second=0, microsecond=0)
    if target <= now:
        target = target + timedelta(days=1)
    return int((target - now).total_seconds())


async def _sync_all_requests(client: JellyseerrClient) -> int:
    take = 50
    skip = 0
    stored = 0
    cache_mode = (get_runtime_settings().artwork_cache_mode or "remote").lower()
    logger.info("Seerr sync starting: take=%s", take)
    _sync_state.update(
        {
            "status": "running",
            "stored": 0,
            "total": None,
            "skip": 0,
            "message": "Starting sync",
            "started_at": datetime.now(timezone.utc).isoformat(),
            "finished_at": None,
        }
    )
    while True:
        try:
            response = await client.get_recent_requests(take=take, skip=skip)
        except httpx.HTTPError as exc:
            logger.warning("Seerr sync failed at skip=%s: %s", skip, exc)
            _sync_state.update({"status": "failed", "message": f"Sync failed: {exc}"})
            break
        if not isinstance(response, dict):
            logger.warning("Seerr sync stopped: non-dict response at skip=%s", skip)
            _sync_state.update({"status": "failed", "message": "Invalid response"})
            break
        if _sync_state["total"] is None:
            page_info = response.get("pageInfo") or {}
            total = (
                page_info.get("totalResults")
                or page_info.get("total")
                or response.get("totalResults")
                or response.get("total")
            )
            if isinstance(total, int):
                _sync_state["total"] = total
        items = response.get("results") or []
        if not isinstance(items, list) or not items:
            logger.info("Seerr sync completed: no more results at skip=%s", skip)
            break
        page_request_ids = [
            payload.get("request_id")
            for item in items
            if isinstance(item, dict)
            for payload in [_parse_request_payload(item)]
            if isinstance(payload.get("request_id"), int)
        ]
        cached_by_request_id = get_request_cache_lookup(page_request_ids)
        page_cache_records: list[Dict[str, Any]] = []
        page_artwork_records: list[Dict[str, Any]] = []
        for item in items:
            if not isinstance(item, dict):
                continue
            payload = _parse_request_payload(item)
            request_id = payload.get("request_id")
            cached_title = None
            if isinstance(request_id, int):
                cached = cached_by_request_id.get(request_id)
                if not payload.get("title") and cached and cached.get("title"):
                    cached_title = cached.get("title")
                needs_details = (
                    not payload.get("title")
                    or not payload.get("media_id")
                    or not payload.get("tmdb_id")
                    or not payload.get("media_type")
                )
                if needs_details:
                    logger.debug("Seerr sync hydrate request_id=%s", request_id)
                    details = await _get_request_details(client, request_id)
                    if isinstance(details, dict):
                        payload = _parse_request_payload(details)
                        item = details
                poster_path, backdrop_path = _extract_artwork_paths(item)
                if cache_mode == "cache" and not (poster_path or backdrop_path):
                    details = await _get_request_details(client, request_id)
                    if isinstance(details, dict):
                        item = details
                        payload = _parse_request_payload(details)
                if not payload.get("title") and payload.get("tmdb_id") and payload.get("media_type"):
                    hydrated_title, hydrated_year = await _hydrate_title_from_tmdb(
                        client, payload.get("media_type"), payload.get("tmdb_id")
                    )
                    if hydrated_title:
                        payload["title"] = hydrated_title
                    if hydrated_year:
                        payload["year"] = hydrated_year
                if not payload.get("title") and cached_title:
                    payload["title"] = cached_title
            if not isinstance(payload.get("request_id"), int):
                continue
            page_cache_records.append(_build_request_cache_record(payload, item))
            if isinstance(item, dict):
                artwork_record = _build_artwork_status_record(item, cache_mode)
                if artwork_record:
                    page_artwork_records.append(artwork_record)
            stored += 1
            _sync_state["stored"] = stored
        if page_cache_records:
            upsert_request_cache_many(page_cache_records)
        if page_artwork_records:
            upsert_artwork_cache_status_many(page_artwork_records)
        if len(items) < take:
            logger.info("Seerr sync completed: stored=%s", stored)
            break
        skip += take
        _sync_state["skip"] = skip
        _sync_state["message"] = f"Synced {stored} requests"
        logger.debug("Seerr sync progress: stored=%s skip=%s", stored, skip)
    _sync_state.update(
        {
            "status": "completed",
            "stored": stored,
            "message": f"Sync complete: {stored} requests",
            "finished_at": datetime.now(timezone.utc).isoformat(),
        }
    )
    set_setting(_sync_last_key, datetime.now(timezone.utc).isoformat())
    _refresh_recent_cache_from_db()
    if cache_mode == "cache":
        update_artwork_cache_stats(
            missing_count=get_artwork_cache_missing_count(),
            total_requests=get_request_cache_count(),
        )
    return stored


async def _sync_delta_requests(client: JellyseerrClient) -> int:
    take = 50
    skip = 0
    stored = 0
    unchanged_pages = 0
    cache_mode = (get_runtime_settings().artwork_cache_mode or "remote").lower()
    logger.info("Seerr delta sync starting: take=%s", take)
    _sync_state.update(
        {
            "status": "running",
            "stored": 0,
            "total": None,
            "skip": 0,
            "message": "Starting delta sync",
            "started_at": datetime.now(timezone.utc).isoformat(),
            "finished_at": None,
        }
    )
    while True:
        try:
            response = await client.get_recent_requests(take=take, skip=skip)
        except httpx.HTTPError as exc:
            logger.warning("Seerr delta sync failed at skip=%s: %s", skip, exc)
            _sync_state.update({"status": "failed", "message": f"Delta sync failed: {exc}"})
            break
        if not isinstance(response, dict):
            logger.warning("Seerr delta sync stopped: non-dict response at skip=%s", skip)
            _sync_state.update({"status": "failed", "message": "Invalid response"})
            break
        items = response.get("results") or []
        if not isinstance(items, list) or not items:
            logger.info("Seerr delta sync completed: no more results at skip=%s", skip)
            break
        page_request_ids = [
            payload.get("request_id")
            for item in items
            if isinstance(item, dict)
            for payload in [_parse_request_payload(item)]
            if isinstance(payload.get("request_id"), int)
        ]
        cached_by_request_id = get_request_cache_lookup(page_request_ids)
        page_cache_records: list[Dict[str, Any]] = []
        page_artwork_records: list[Dict[str, Any]] = []
        page_changed = False
        for item in items:
            if not isinstance(item, dict):
                continue
            payload = _parse_request_payload(item)
            request_id = payload.get("request_id")
            if isinstance(request_id, int):
                cached = cached_by_request_id.get(request_id)
                incoming_updated = payload.get("updated_at")
                cached_title = cached.get("title") if cached else None
                if cached and incoming_updated and cached.get("updated_at") == incoming_updated and cached.get("title"):
                    continue
                needs_details = (
                    not payload.get("title")
                    or not payload.get("media_id")
                    or not payload.get("tmdb_id")
                    or not payload.get("media_type")
                )
                if needs_details:
                    details = await _get_request_details(client, request_id)
                    if isinstance(details, dict):
                        payload = _parse_request_payload(details)
                        item = details
                poster_path, backdrop_path = _extract_artwork_paths(item)
                if cache_mode == "cache" and not (poster_path or backdrop_path):
                    details = await _get_request_details(client, request_id)
                    if isinstance(details, dict):
                        payload = _parse_request_payload(details)
                        item = details
                if not payload.get("title") and payload.get("tmdb_id") and payload.get("media_type"):
                    hydrated_title, hydrated_year = await _hydrate_title_from_tmdb(
                        client, payload.get("media_type"), payload.get("tmdb_id")
                    )
                    if hydrated_title:
                        payload["title"] = hydrated_title
                    if hydrated_year:
                        payload["year"] = hydrated_year
                if not payload.get("title") and cached_title:
                    payload["title"] = cached_title
            if not isinstance(payload.get("request_id"), int):
                continue
            page_cache_records.append(_build_request_cache_record(payload, item))
            if isinstance(item, dict):
                artwork_record = _build_artwork_status_record(item, cache_mode)
                if artwork_record:
                    page_artwork_records.append(artwork_record)
            stored += 1
            page_changed = True
            _sync_state["stored"] = stored
        if page_cache_records:
            upsert_request_cache_many(page_cache_records)
        if page_artwork_records:
            upsert_artwork_cache_status_many(page_artwork_records)
        if not page_changed:
            unchanged_pages += 1
        else:
            unchanged_pages = 0
        if len(items) < take or unchanged_pages >= 2:
            logger.info("Seerr delta sync completed: stored=%s", stored)
            break
        skip += take
        _sync_state["skip"] = skip
        _sync_state["message"] = f"Delta synced {stored} requests"
        logger.debug("Seerr delta sync progress: stored=%s skip=%s", stored, skip)
    deduped = prune_duplicate_requests_cache()
    if deduped:
        logger.info("Seerr delta sync removed duplicate rows: %s", deduped)
    _sync_state.update(
        {
            "status": "completed",
            "stored": stored,
            "message": f"Delta sync complete: {stored} updated",
            "finished_at": datetime.now(timezone.utc).isoformat(),
        }
    )
    set_setting(_sync_last_key, datetime.now(timezone.utc).isoformat())
    _refresh_recent_cache_from_db()
    if cache_mode == "cache":
        update_artwork_cache_stats(
            missing_count=get_artwork_cache_missing_count(),
            total_requests=get_request_cache_count(),
        )
    return stored


async def _prefetch_artwork_cache(
    client: JellyseerrClient,
    only_missing: bool = False,
    total: Optional[int] = None,
    use_missing_query: bool = False,
) -> None:
    runtime = get_runtime_settings()
    cache_mode = (runtime.artwork_cache_mode or "remote").lower()
    if cache_mode != "cache":
        _artwork_prefetch_state.update(
            {
                "status": "failed",
                "message": "Artwork cache mode is not set to cache.",
                "finished_at": datetime.now(timezone.utc).isoformat(),
            }
        )
        return

    total = total if total is not None else get_request_cache_count()
    _artwork_prefetch_state.update(
        {
            "status": "running",
            "processed": 0,
            "total": total,
            "message": "Starting missing artwork prefetch"
            if only_missing
            else "Starting artwork prefetch",
            "only_missing": only_missing,
            "started_at": datetime.now(timezone.utc).isoformat(),
            "finished_at": None,
        }
    )
    if only_missing and total == 0:
        _artwork_prefetch_state.update(
            {
                "status": "completed",
                "processed": 0,
                "message": "No missing artwork to cache.",
                "finished_at": datetime.now(timezone.utc).isoformat(),
            }
        )
        return
    offset = 0
    limit = 200
    processed = 0
    while True:
        if use_missing_query:
            batch = get_request_cache_payloads_missing(limit=limit, offset=offset)
        else:
            batch = get_request_cache_payloads(limit=limit, offset=offset)
        if not batch:
            break
        page_cache_records: list[Dict[str, Any]] = []
        page_artwork_records: list[Dict[str, Any]] = []
        for row in batch:
            payload = row.get("payload")
            if not isinstance(payload, dict):
                if not only_missing:
                    processed += 1
                continue
            if only_missing and not use_missing_query and not _artwork_missing_for_payload(payload):
                continue
            poster_path, backdrop_path = _extract_artwork_paths(payload)
            tmdb_id, media_type = _extract_tmdb_lookup(payload)
            if (not poster_path or not backdrop_path) and client.configured() and tmdb_id and media_type:
                media = payload.get("media") or {}
                hydrated_poster, hydrated_backdrop = await _hydrate_artwork_from_tmdb(
                    client, media_type, tmdb_id
                )
                poster_path = poster_path or hydrated_poster
                backdrop_path = backdrop_path or hydrated_backdrop
                if hydrated_poster or hydrated_backdrop:
                    media = dict(media) if isinstance(media, dict) else {}
                    if hydrated_poster:
                        media["posterPath"] = hydrated_poster
                    if hydrated_backdrop:
                        media["backdropPath"] = hydrated_backdrop
                    payload["media"] = media
                    parsed = _parse_request_payload(payload)
                    request_id = parsed.get("request_id")
                    if isinstance(request_id, int):
                        page_cache_records.append(_build_request_cache_record(parsed, payload))
            poster_cached_flag = False
            backdrop_cached_flag = False
            if poster_path:
                try:
                    poster_cached_flag = bool(
                        await cache_tmdb_image(poster_path, "w185")
                    ) and bool(await cache_tmdb_image(poster_path, "w342"))
                except httpx.HTTPError:
                    poster_cached_flag = False
            if backdrop_path:
                try:
                    backdrop_cached_flag = bool(await cache_tmdb_image(backdrop_path, "w780"))
                except httpx.HTTPError:
                    backdrop_cached_flag = False
            artwork_record = _build_artwork_status_record(
                payload,
                cache_mode,
                poster_cached=poster_cached_flag if poster_path else None,
                backdrop_cached=backdrop_cached_flag if backdrop_path else None,
            )
            if artwork_record:
                page_artwork_records.append(artwork_record)
            processed += 1
            if processed % 25 == 0:
                _artwork_prefetch_state.update(
                    {"processed": processed, "message": f"Cached artwork for {processed} requests"}
                )
        if page_cache_records:
            upsert_request_cache_many(page_cache_records)
        if page_artwork_records:
            upsert_artwork_cache_status_many(page_artwork_records)
        offset += limit

    total_requests = get_request_cache_count()
    missing_count = get_artwork_cache_missing_count()
    cache_bytes, cache_files = _collect_artwork_cache_disk_stats()
    update_artwork_cache_stats(
        cache_bytes=cache_bytes,
        cache_files=cache_files,
        missing_count=missing_count,
        total_requests=total_requests,
    )
    _artwork_prefetch_state.update(
        {
            "status": "completed",
            "processed": processed,
            "message": f"Artwork cached for {processed} requests",
            "finished_at": datetime.now(timezone.utc).isoformat(),
        }
    )


async def start_artwork_prefetch(
    base_url: Optional[str], api_key: Optional[str], only_missing: bool = False
) -> Dict[str, Any]:
    global _artwork_prefetch_task
    if _artwork_prefetch_task and not _artwork_prefetch_task.done():
        return dict(_artwork_prefetch_state)
    client = JellyseerrClient(base_url, api_key)
    status_count = get_artwork_cache_status_count()
    total_requests = get_request_cache_count()
    use_missing_query = only_missing and status_count >= total_requests and total_requests > 0
    if only_missing and use_missing_query:
        total = get_artwork_cache_missing_count()
    else:
        total = total_requests
    _artwork_prefetch_state.update(
        {
            "status": "running",
            "processed": 0,
            "total": total,
            "message": "Seeding artwork cache status"
            if only_missing and not use_missing_query
            else ("Starting missing artwork prefetch" if only_missing else "Starting artwork prefetch"),
            "only_missing": only_missing,
            "started_at": datetime.now(timezone.utc).isoformat(),
            "finished_at": None,
        }
    )
    if only_missing and total == 0:
        _artwork_prefetch_state.update(
            {
                "status": "completed",
                "processed": 0,
                "message": "No missing artwork to cache.",
                "finished_at": datetime.now(timezone.utc).isoformat(),
            }
        )
        return dict(_artwork_prefetch_state)

    async def _runner() -> None:
        try:
            await _prefetch_artwork_cache(
                client,
                only_missing=only_missing,
                total=total,
                use_missing_query=use_missing_query,
            )
        except Exception:
            logger.exception("Artwork prefetch failed")
            _artwork_prefetch_state.update(
                {
                    "status": "failed",
                    "message": "Artwork prefetch failed.",
                    "finished_at": datetime.now(timezone.utc).isoformat(),
                }
            )

    _artwork_prefetch_task = asyncio.create_task(_runner())
    return dict(_artwork_prefetch_state)


def get_artwork_prefetch_state() -> Dict[str, Any]:
    return dict(_artwork_prefetch_state)


async def _ensure_requests_cache(client: JellyseerrClient) -> None:
    last_sync = get_setting(_sync_last_key)
    last_updated = last_sync or get_request_cache_last_updated()
    if _cache_is_stale(last_updated):
        logger.info("Requests cache stale or empty, starting sync.")
        await _sync_all_requests(client)
    else:
        logger.debug("Requests cache fresh: last_sync=%s", last_updated)


def _refresh_recent_cache_from_db() -> None:
    since_iso = (datetime.now(timezone.utc) - timedelta(days=RECENT_CACHE_MAX_DAYS)).isoformat()
    items = get_cached_requests_since(since_iso)
    _recent_cache["items"] = items
    _recent_cache["updated_at"] = datetime.now(timezone.utc).isoformat()


def _recent_cache_stale() -> bool:
    updated_at = _recent_cache.get("updated_at")
    if not updated_at:
        return True
    try:
        parsed = datetime.fromisoformat(updated_at)
    except ValueError:
        return True
    return (datetime.now(timezone.utc) - parsed).total_seconds() > RECENT_CACHE_TTL_SECONDS


def _parse_iso_datetime(value: Optional[str]) -> Optional[datetime]:
    if not value:
        return None
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return None
    if parsed.tzinfo is None:
        return parsed.replace(tzinfo=timezone.utc)
    return parsed


def _get_recent_from_cache(
    requested_by_norm: Optional[str],
    requested_by_id: Optional[int],
    limit: int,
    offset: int,
    since_iso: Optional[str],
    status_codes: Optional[list[int]] = None,
) -> List[Dict[str, Any]]:
    items = _recent_cache.get("items") or []
    repairing = active_repair_request_ids()
    results = []
    since_dt = _parse_iso_datetime(since_iso)
    for item in items:
        if requested_by_id is not None:
            if item.get("requested_by_id") != requested_by_id:
                continue
        elif requested_by_norm and item.get("requested_by_norm") != requested_by_norm:
            continue
        if since_dt:
            candidate = item.get("created_at") or item.get("updated_at")
            item_dt = _parse_iso_datetime(candidate)
            if not item_dt or item_dt < since_dt:
                continue
        if str(item.get("request_id")) in repairing:
            item = {**item, "status": 5, "repairing": True}
        if status_codes and item.get("status") not in status_codes:
            continue
        results.append(item)
    return results[offset : offset + limit]


async def startup_warmup_requests_cache() -> None:
    runtime = get_runtime_settings()
    client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
    if client.configured():
        try:
            await _ensure_requests_cache(client)
        except httpx.HTTPError as exc:
            logger.warning("Requests warmup skipped: %s", exc)
    repaired = repair_request_cache_titles()
    if repaired:
        logger.info("Requests cache titles repaired: %s", repaired)
    _refresh_recent_cache_from_db()


async def run_requests_poll_loop() -> None:
    while True:
        runtime = get_runtime_settings()
        interval = max(60, int(runtime.requests_poll_interval_seconds or 300))
        client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
        if client.configured():
            try:
                await _ensure_requests_cache(client)
            except httpx.HTTPError as exc:
                logger.debug("Requests poll skipped: %s", exc)
        await asyncio.sleep(interval)


async def run_requests_delta_loop() -> None:
    while True:
        runtime = get_runtime_settings()
        interval = max(60, int(runtime.requests_delta_sync_interval_minutes or 5) * 60)
        client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
        if client.configured():
            if _sync_task and not _sync_task.done():
                logger.debug("Delta sync skipped: another sync is running.")
            else:
                try:
                    await _sync_delta_requests(client)
                except httpx.HTTPError as exc:
                    logger.debug("Delta sync skipped: %s", exc)
        await asyncio.sleep(interval)


async def run_daily_requests_full_sync() -> None:
    while True:
        runtime = get_runtime_settings()
        hour, minute = _parse_time(runtime.requests_full_sync_time, 0, 0)
        await asyncio.sleep(_seconds_until(hour, minute))
        runtime = get_runtime_settings()
        client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
        if not client.configured():
            logger.info("Daily full sync skipped: Seerr not configured.")
            continue
        if _sync_task and not _sync_task.done():
            logger.info("Daily full sync skipped: another sync is running.")
            continue
        try:
            await _sync_all_requests(client)
        except httpx.HTTPError as exc:
            logger.warning("Daily full sync failed: %s", exc)


async def run_daily_db_cleanup() -> None:
    while True:
        runtime = get_runtime_settings()
        hour, minute = _parse_time(runtime.requests_cleanup_time, 2, 0)
        await asyncio.sleep(_seconds_until(hour, minute))
        runtime = get_runtime_settings()
        result = cleanup_history(int(runtime.requests_cleanup_days or 90))
        logger.info("Daily cleanup complete: %s", result)


async def start_requests_sync(base_url: Optional[str], api_key: Optional[str]) -> Dict[str, Any]:
    global _sync_task
    if _sync_task and not _sync_task.done():
        return dict(_sync_state)
    if not base_url:
        _sync_state.update({"status": "failed", "message": "Seerr not configured"})
        return dict(_sync_state)
    client = JellyseerrClient(base_url, api_key)
    _sync_state.update(
        {
            "status": "running",
            "stored": 0,
            "total": None,
            "skip": 0,
            "message": "Starting sync",
            "started_at": datetime.now(timezone.utc).isoformat(),
            "finished_at": None,
        }
    )

    async def _runner() -> None:
        try:
            await _sync_all_requests(client)
        except Exception as exc:
            logger.exception("Seerr sync failed")
            _sync_state.update(
                {
                    "status": "failed",
                    "message": f"Sync failed: {exc}",
                    "finished_at": datetime.now(timezone.utc).isoformat(),
                }
            )

    _sync_task = asyncio.create_task(_runner())
    return dict(_sync_state)


async def start_requests_delta_sync(base_url: Optional[str], api_key: Optional[str]) -> Dict[str, Any]:
    global _sync_task
    if _sync_task and not _sync_task.done():
        return dict(_sync_state)
    if not base_url:
        _sync_state.update({"status": "failed", "message": "Seerr not configured"})
        return dict(_sync_state)
    client = JellyseerrClient(base_url, api_key)
    _sync_state.update(
        {
            "status": "running",
            "stored": 0,
            "total": None,
            "skip": 0,
            "message": "Starting delta sync",
            "started_at": datetime.now(timezone.utc).isoformat(),
            "finished_at": None,
        }
    )

    async def _runner() -> None:
        try:
            await _sync_delta_requests(client)
        except Exception as exc:
            logger.exception("Seerr delta sync failed")
            _sync_state.update(
                {
                    "status": "failed",
                    "message": f"Delta sync failed: {exc}",
                    "finished_at": datetime.now(timezone.utc).isoformat(),
                }
            )

    _sync_task = asyncio.create_task(_runner())
    return dict(_sync_state)


def get_requests_sync_state() -> Dict[str, Any]:
    return dict(_sync_state)


async def _ensure_request_access(
    client: JellyseerrClient, request_id: int, user: Dict[str, str]
) -> None:
    if user.get("role") == "admin" or user.get("username"):
        return
    raise HTTPException(status_code=403, detail="Request not accessible for this user")


def _build_recent_map(response: Dict[str, Any]) -> Dict[int, Dict[str, Any]]:
    mapping: Dict[int, Dict[str, Any]] = {}
    for item in response.get("results", []):
        media = item.get("media") or {}
        media_id = media.get("id") or item.get("mediaId")
        request_id = item.get("id")
        status = item.get("status")
        if isinstance(media_id, int) and isinstance(request_id, int):
            mapping[media_id] = {
                "requestId": request_id,
                "status": status,
                "statusLabel": _status_label(status),
            }
    return mapping


def _queue_records(queue: Any) -> List[Dict[str, Any]]:
    if isinstance(queue, dict):
        records = queue.get("records")
        if isinstance(records, list):
            return records
    if isinstance(queue, list):
        return queue
    return []


def _download_ids(records: List[Dict[str, Any]]) -> List[str]:
    ids = []
    for record in records:
        download_id = record.get("downloadId") or record.get("download_id")
        if isinstance(download_id, str) and download_id:
            ids.append(download_id)
    return ids


def _log_arr_http_error(service_label: str, action: str, exc: httpx.HTTPStatusError) -> None:
    if exc.response is None:
        logger.warning("%s %s failed: %s", service_label, action, exc)
        return
    status = exc.response.status_code
    body = exc.response.text
    if isinstance(body, str):
        body = body.strip()
        if len(body) > 800:
            body = f"{body[:800]}...(truncated)"
    logger.warning("%s %s failed: status=%s body=%s", service_label, action, status, body)


def _format_rejections(rejections: Any) -> Optional[str]:
    if isinstance(rejections, str):
        return rejections.strip() or None
    if isinstance(rejections, list):
        reasons = []
        for item in rejections:
            reason = None
            if isinstance(item, dict):
                reason = (
                    item.get("reason")
                    or item.get("message")
                    or item.get("errorMessage")
                )
            if not reason and item is not None:
                reason = str(item)
            if isinstance(reason, str) and reason.strip():
                reasons.append(reason.strip())
        if reasons:
            return "; ".join(reasons)
    return None


def _filter_arr_release_results(results: Any, include_rejected: bool = False) -> List[Dict[str, Any]]:
    if not isinstance(results, list):
        return []
    keep: List[Dict[str, Any]] = []
    seen: set[tuple[Any, Any]] = set()
    for item in results:
        if not isinstance(item, dict):
            continue
        key = (item.get("indexerId"), item.get("guid"))
        if not key[0] or not key[1] or key in seen:
            continue
        accepted, override, reasons = manual_releases.decision(item)
        if not accepted and not include_rejected:
            continue
        seen.add(key)
        quality_payload = item.get("quality")
        quality_name = None
        if isinstance(quality_payload, dict):
            quality_value = quality_payload.get("quality")
            if isinstance(quality_value, dict):
                quality_name = str(quality_value.get("name") or "").strip() or None
            elif isinstance(quality_payload.get("name"), str):
                quality_name = quality_payload["name"].strip() or None
        keep.append(
            {
                "title": item.get("title"),
                "indexer": item.get("indexer"),
                "indexerId": item.get("indexerId"),
                "guid": item.get("guid"),
                "size": item.get("size"),
                "seeders": item.get("seeders"),
                "leechers": item.get("leechers"),
                "publishDate": item.get("publishDate"),
                "infoUrl": item.get("infoUrl"),
                "downloadUrl": item.get("downloadUrl"),
                "magnetUrl": item.get("magnetUrl"),
                "protocol": item.get("protocol"),
                "approved": accepted,
                "rejected": item.get("rejected"),
                "temporarilyRejected": item.get("temporarilyRejected"),
                "rejections": item.get("rejections"),
                "downloadAllowed": item.get("downloadAllowed"),
                "fullSeason": item.get("fullSeason"),
                "seasonNumber": item.get("seasonNumber"),
                "quality": quality_name,
                "requiresOverride": override,
                "selectable": accepted or override,
                "rejections": reasons,
                "episodeNumbers": item.get("mappedEpisodeNumbers") or item.get("episodeNumbers"),
                "customFormatScore": item.get("customFormatScore"),
            }
        )
    keep.sort(key=lambda item: (not bool(item.get("approved")), not item["requiresOverride"]))
    releases = keep[:200]
    for index, release in enumerate(releases):
        release["bestPick"] = index == 0 and release.get("approved") is True
    return releases


def _missing_episode_ids_by_season(episodes: Any) -> Dict[int, List[int]]:
    if not isinstance(episodes, list):
        return {}
    grouped: Dict[int, List[int]] = {}
    for episode in episodes:
        if not isinstance(episode, dict):
            continue
        if not episode.get("monitored", True):
            continue
        if episode.get("hasFile"):
            continue
        season_number = episode.get("seasonNumber")
        episode_id = episode.get("id")
        if isinstance(season_number, int) and isinstance(episode_id, int):
            grouped.setdefault(season_number, []).append(episode_id)
    return grouped


def _replacement_file_name(file_data: Dict[str, Any]) -> str:
    raw = file_data.get("relativePath") or file_data.get("path") or file_data.get("sceneName")
    if isinstance(raw, str) and raw.strip():
        return raw.strip().replace("\\", "/").rsplit("/", 1)[-1]
    return "Managed media file"


def _replacement_quality_name(file_data: Dict[str, Any]) -> Optional[str]:
    quality = file_data.get("quality")
    if not isinstance(quality, dict):
        return None
    nested = quality.get("quality")
    if isinstance(nested, dict):
        value = nested.get("name")
        return str(value).strip() if value is not None and str(value).strip() else None
    value = quality.get("name")
    return str(value).strip() if value is not None and str(value).strip() else None


def _jellyfin_media_signature(item: Any) -> Dict[str, Any]:
    if not isinstance(item, dict):
        return {}
    return {
        key: item.get(key)
        for key in ("Id", "Etag", "Path", "DateCreated", "MediaSources")
        if item.get(key) is not None
    }


def _replacement_file_payload(
    file_data: Dict[str, Any],
    *,
    episode_numbers: Optional[List[str]] = None,
) -> Optional[Dict[str, Any]]:
    file_id = file_data.get("id")
    if not isinstance(file_id, int) or file_id <= 0:
        return None
    size = file_data.get("size")
    return {
        "id": file_id,
        "name": _replacement_file_name(file_data),
        "quality": _replacement_quality_name(file_data),
        "size": int(size) if isinstance(size, (int, float)) and size >= 0 else None,
        "season_number": file_data.get("seasonNumber") if isinstance(file_data.get("seasonNumber"), int) else None,
        "episodes": episode_numbers or [],
    }


def _linked_issue_for_replacement(
    issue_id: Any,
    *,
    request_id: str,
    user: Dict[str, str],
) -> Optional[Dict[str, Any]]:
    if issue_id is None:
        return None
    if not isinstance(issue_id, int) or issue_id <= 0:
        raise HTTPException(status_code=400, detail="A valid linked issue is required")
    issue = get_portal_item(issue_id)
    if not issue or str(issue.get("kind") or "").lower() != "issue":
        raise HTTPException(status_code=404, detail="Linked issue not found")
    if str(issue.get("external_ref") or "") != f"/requests/{request_id}":
        raise HTTPException(status_code=409, detail="The issue is not linked to this request")
    is_admin = str(user.get("role") or "").lower() == "admin"
    is_owner = str(issue.get("created_by_username") or "").lower() == str(user.get("username") or "").lower()
    if not (is_admin or is_owner):
        raise HTTPException(status_code=403, detail="You cannot update this linked issue")
    return issue


def _record_replacement_activity(
    issue: Optional[Dict[str, Any]],
    *,
    user: Dict[str, str],
    event_type: str,
    message: str,
    metadata: Optional[Dict[str, Any]] = None,
) -> None:
    tracking = (metadata or {}).get("repairTracking")
    if event_type in {"replacement_started", "missing_search_started"} and isinstance(tracking, dict):
        start_request_repair(tracking)
    if not issue:
        return
    current_status = str(issue.get("status") or "new").strip().lower()
    next_status: Optional[str] = None
    if event_type.endswith("_started"):
        next_status = "in_progress"
    elif event_type.endswith("_failed"):
        next_status = "blocked"
    if next_status and current_status not in {"done", "closed"}:
        update_portal_item(
            int(issue["id"]),
            status=next_status,
            issue_resolved_at=None,
        )
    add_portal_item_activity(
        int(issue["id"]),
        event_type=event_type,
        actor_username=str(user.get("username") or "unknown"),
        actor_role=str(user.get("role") or "user"),
        message=message,
        metadata_json=(
            json.dumps(metadata, separators=(",", ":"), sort_keys=True)
            if metadata
            else None
        ),
    )


def _released_episode(episode: Dict[str, Any]) -> bool:
    if episode.get("hasFile") is True:
        return True
    raw_date = episode.get("airDateUtc") or episode.get("airDate")
    if not isinstance(raw_date, str) or not raw_date.strip():
        return False
    try:
        parsed = datetime.fromisoformat(raw_date.strip().replace("Z", "+00:00"))
    except ValueError:
        return False
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return parsed.astimezone(timezone.utc) <= datetime.now(timezone.utc)


def _issue_episode_payloads(
    episodes: Any,
) -> List[Dict[str, Any]]:
    results: List[Dict[str, Any]] = []
    if not isinstance(episodes, list):
        return results
    for episode in episodes:
        if not isinstance(episode, dict):
            continue
        episode_id = episode.get("id")
        season_number = episode.get("seasonNumber")
        episode_number = episode.get("episodeNumber")
        if not all(isinstance(value, int) for value in (episode_id, season_number, episode_number)):
            continue
        released = _released_episode(episode)
        has_file = episode.get("hasFile") is True or (
            isinstance(episode.get("episodeFileId"), int) and episode.get("episodeFileId") > 0
        )
        monitored = episode.get("monitored") is not False
        results.append(
            {
                "id": episode_id,
                "season_number": season_number,
                "episode_number": episode_number,
                "code": f"S{season_number:02d}E{episode_number:02d}",
                "title": str(episode.get("title") or f"Episode {episode_number}").strip(),
                "released": released,
                "monitored": monitored,
                "has_file": has_file,
                "missing": released and not has_file,
                "best_fit": released and not has_file,
                "file_id": episode.get("episodeFileId") if has_file else None,
            }
        )
    results.sort(key=lambda item: (item["season_number"], item["episode_number"]))
    return results


def _issue_season_payloads(episodes: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    grouped: Dict[int, List[Dict[str, Any]]] = {}
    for episode in episodes:
        grouped.setdefault(int(episode["season_number"]), []).append(episode)
    return [
        {
            "season_number": season_number,
            "label": "Specials" if season_number == 0 else f"Season {season_number}",
            "episode_count": len(items),
            "available_count": sum(1 for item in items if item["has_file"]),
            "missing_count": sum(1 for item in items if item["missing"]),
            "best_fit": any(item["best_fit"] for item in items),
        }
        for season_number, items in sorted(grouped.items())
        if any(item["released"] for item in items)
    ]


async def _resolve_root_folder_path(client: Any, root_folder: str, service_name: str) -> str:
    if root_folder.isdigit():
        folders = await client.get_root_folders()
        if isinstance(folders, list):
            for folder in folders:
                if folder.get("id") == int(root_folder):
                    path = folder.get("path")
                    if isinstance(path, str) and path:
                        return path
        raise HTTPException(status_code=400, detail=f"{service_name} root folder id {root_folder} not found")
    return root_folder


@router.get("/{request_id}/issue-options")
async def issue_target_options(
    request_id: str,
    user: Dict[str, str] = Depends(get_current_user),
) -> Dict[str, Any]:
    if not request_id.isdigit():
        raise HTTPException(status_code=400, detail="Invalid request id")
    runtime = get_runtime_settings()
    seerr = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
    if seerr.configured():
        await _ensure_request_access(seerr, int(request_id), user)
    snapshot = await build_snapshot(request_id)
    arr_item = snapshot.raw.get("arr", {}).get("item")
    if not isinstance(arr_item, dict):
        return {
            "request_id": request_id,
            "request_type": snapshot.request_type.value,
            "title": snapshot.title,
            "collector_id": None,
            "movie": None,
            "seasons": [],
            "episodes": [],
            "can_act": False,
            "message": "This title is not currently linked to Sonarr or Radarr.",
        }

    collector_id = arr_item.get("id")
    if not isinstance(collector_id, int):
        raise HTTPException(status_code=502, detail="Sonarr/Radarr returned an invalid media record")
    if snapshot.request_type == RequestType.movie:
        movie_file = arr_item.get("movieFile") if isinstance(arr_item.get("movieFile"), dict) else None
        return {
            "request_id": request_id,
            "request_type": "movie",
            "title": snapshot.title,
            "collector_id": collector_id,
            "movie": {
                "selected_label": snapshot.title,
                "has_file": bool(movie_file),
                "missing": not bool(movie_file),
                "best_fit": not bool(movie_file),
                "file_id": movie_file.get("id") if movie_file else None,
            },
            "seasons": [],
            "episodes": [],
            "can_act": _user_can_use_search_auto(user),
            "message": "Choose the movie to continue.",
        }

    sonarr = SonarrClient(runtime.sonarr_base_url, runtime.sonarr_api_key)
    if not sonarr.configured():
        raise HTTPException(status_code=400, detail="Sonarr is not configured")
    try:
        episodes = await sonarr.get_episodes(collector_id)
    except Exception as exc:
        logger.warning("Sonarr issue options failed request_id=%s error=%s", request_id, exc)
        raise HTTPException(
            status_code=502,
            detail="Magent could not read the seasons and episodes from Sonarr.",
        ) from exc
    episode_options = _issue_episode_payloads(episodes)
    return {
        "request_id": request_id,
        "request_type": "tv",
        "title": snapshot.title,
        "collector_id": collector_id,
        "movie": None,
        "seasons": _issue_season_payloads(episode_options),
        "episodes": episode_options,
        "can_act": _user_can_use_search_auto(user),
        "message": "Choose a season, then select every affected episode.",
    }


@router.get("/{request_id}/replacement-options")
async def replacement_options(
    request_id: str,
    user: Dict[str, str] = Depends(get_current_user),
) -> Dict[str, Any]:
    if not request_id.isdigit():
        raise HTTPException(status_code=400, detail="Invalid request id")
    runtime = get_runtime_settings()
    seerr = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
    if seerr.configured():
        await _ensure_request_access(seerr, int(request_id), user)
    snapshot = await build_snapshot(request_id)
    arr_item = snapshot.raw.get("arr", {}).get("item")
    if not isinstance(arr_item, dict):
        return {
            "request_id": request_id,
            "request_type": snapshot.request_type.value,
            "title": snapshot.title,
            "files": [],
            "message": "This title is not currently linked to a Sonarr/Radarr library item.",
        }

    files: List[Dict[str, Any]] = []
    if snapshot.request_type == RequestType.movie:
        movie_file = arr_item.get("movieFile")
        if isinstance(movie_file, dict):
            option = _replacement_file_payload(movie_file)
            if option:
                files.append(option)
    elif snapshot.request_type == RequestType.tv:
        series_id = arr_item.get("id")
        if not isinstance(series_id, int):
            raise HTTPException(status_code=502, detail="Sonarr returned an invalid series record")
        sonarr = SonarrClient(runtime.sonarr_base_url, runtime.sonarr_api_key)
        if not sonarr.configured():
            raise HTTPException(status_code=400, detail="Sonarr is not configured")
        try:
            episode_files, episodes = await asyncio.gather(
                sonarr.get_episode_files(series_id),
                sonarr.get_episodes(series_id),
            )
        except Exception as exc:
            logger.warning("Sonarr replacement options failed request_id=%s error=%s", request_id, exc)
            raise HTTPException(
                status_code=502,
                detail="Magent could not read the managed episode files from Sonarr.",
            ) from exc
        episode_labels: Dict[int, List[str]] = {}
        if isinstance(episodes, list):
            for episode in episodes:
                if not isinstance(episode, dict):
                    continue
                file_id = episode.get("episodeFileId")
                season_number = episode.get("seasonNumber")
                episode_number = episode.get("episodeNumber")
                if not all(isinstance(value, int) for value in (file_id, season_number, episode_number)):
                    continue
                label = f"S{season_number:02d}E{episode_number:02d}"
                episode_labels.setdefault(file_id, []).append(label)
        if isinstance(episode_files, list):
            for file_data in episode_files:
                if not isinstance(file_data, dict):
                    continue
                option = _replacement_file_payload(
                    file_data,
                    episode_numbers=episode_labels.get(file_data.get("id"), []),
                )
                if option:
                    files.append(option)
        files.sort(
            key=lambda item: (
                item.get("season_number") if isinstance(item.get("season_number"), int) else 9999,
                ",".join(item.get("episodes") or []),
                str(item.get("name") or ""),
            )
        )

    return {
        "request_id": request_id,
        "request_type": snapshot.request_type.value,
        "title": snapshot.title,
        "files": files,
        "can_replace": _user_can_use_search_auto(user),
        "message": (
            "Choose the exact managed file to remove and replace."
            if files
            else "Sonarr/Radarr does not currently report a managed file for this title."
        ),
    }


@router.post("/{request_id}/actions/replace")
async def action_replace_media(
    request_id: str,
    payload: Dict[str, Any],
    user: Dict[str, str] = Depends(get_current_user),
) -> Dict[str, Any]:
    if not request_id.isdigit():
        raise HTTPException(status_code=400, detail="Invalid request id")
    if not _user_can_use_search_auto(user):
        raise HTTPException(status_code=403, detail="Media replacement is disabled for this user")
    if payload.get("confirmed") is not True:
        raise HTTPException(status_code=400, detail="Replacement confirmation is required")
    raw_file_ids = payload.get("file_ids")
    if raw_file_ids is None:
        raw_file_ids = [payload.get("file_id")]
    if not isinstance(raw_file_ids, list):
        raise HTTPException(status_code=400, detail="Managed files must be supplied as a list")
    file_ids = list(dict.fromkeys(
        value
        for value in raw_file_ids
        if isinstance(value, int) and not isinstance(value, bool) and value > 0
    ))
    if not file_ids or len(file_ids) != len(raw_file_ids) or len(file_ids) > 100:
        raise HTTPException(status_code=400, detail="Choose between 1 and 100 valid managed files")
    linked_issue = _linked_issue_for_replacement(
        payload.get("issue_id"),
        request_id=request_id,
        user=user,
    )

    runtime = get_runtime_settings()
    seerr = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
    if seerr.configured():
        await _ensure_request_access(seerr, int(request_id), user)
    snapshot = await build_snapshot(request_id)
    arr_item = snapshot.raw.get("arr", {}).get("item")
    if not isinstance(arr_item, dict):
        raise HTTPException(status_code=404, detail="Item not found in Sonarr/Radarr")

    collector = "Sonarr" if snapshot.request_type == RequestType.tv else "Radarr"
    target_names: List[str] = []
    collector_id: Optional[int] = None
    target_episodes: List[Dict[str, int]] = []
    jellyfin_baseline: List[Dict[str, Any]] = []
    repair_tracking: Dict[str, Any] = {}

    def record_cycle() -> None:
        repair_tracking.update({
            "requestId": request_id,
            "actionId": "replace_media",
            "mediaType": snapshot.request_type.value,
            "collectorId": collector_id,
            "originalFileIds": file_ids,
            "previousDownloadIds": list(dict.fromkeys(
                list(snapshot.raw.get("qbittorrent", {}).get("downloadIds") or [])
                + [str(t.get("hash")) for t in snapshot.raw.get("qbittorrent", {}).get("torrents", []) if t.get("hash")]
            )),
            "episodes": target_episodes,
            "jellyfinBaseline": jellyfin_baseline,
            "jellyfinFoundAtStart": bool(snapshot.raw.get("jellyfin", {}).get("catalogFound",
                snapshot.raw.get("jellyfin", {}).get("found"))),
            "startedAt": datetime.now(timezone.utc).isoformat(),
        })
        start_request_repair(repair_tracking)

    try:
        if snapshot.request_type == RequestType.movie:
            movie_id = arr_item.get("id")
            movie_file = arr_item.get("movieFile")
            if not isinstance(movie_id, int) or not isinstance(movie_file, dict):
                raise HTTPException(status_code=409, detail="Radarr does not report a replaceable movie file")
            if len(file_ids) != 1 or movie_file.get("id") != file_ids[0]:
                raise HTTPException(status_code=409, detail="The selected movie file is no longer current")
            collector_id = movie_id
            jellyfin_baseline = [
                _jellyfin_media_signature(snapshot.raw.get("jellyfin", {}).get("item"))
            ]
            jellyfin_baseline = [item for item in jellyfin_baseline if item]
            target_names = [_replacement_file_name(movie_file)]
            radarr = RadarrClient(runtime.radarr_base_url, runtime.radarr_api_key)
            if not radarr.configured():
                raise HTTPException(status_code=400, detail="Radarr is not configured")
            await radarr.monitor_movie(movie_id, True)
            await asyncio.to_thread(record_cycle)
            await radarr.delete_movie_file(file_ids[0])
            await radarr.search(movie_id)
        elif snapshot.request_type == RequestType.tv:
            series_id = arr_item.get("id")
            if not isinstance(series_id, int):
                raise HTTPException(status_code=502, detail="Sonarr returned an invalid series record")
            sonarr = SonarrClient(runtime.sonarr_base_url, runtime.sonarr_api_key)
            if not sonarr.configured():
                raise HTTPException(status_code=400, detail="Sonarr is not configured")
            episode_files, episodes = await asyncio.gather(
                sonarr.get_episode_files(series_id),
                sonarr.get_episodes(series_id),
            )
            selected_files = [
                item
                for item in episode_files
                if isinstance(item, dict) and item.get("id") in file_ids
            ] if isinstance(episode_files, list) else []
            if len(selected_files) != len(file_ids):
                raise HTTPException(status_code=409, detail="One or more selected episode files are no longer current")
            target_episodes = [
                {
                    "id": int(episode["id"]),
                    "seasonNumber": int(episode["seasonNumber"]),
                    "episodeNumber": int(episode["episodeNumber"]),
                }
                for episode in episodes
                if isinstance(episode, dict)
                and episode.get("episodeFileId") in file_ids
                and isinstance(episode.get("id"), int)
                and isinstance(episode.get("seasonNumber"), int)
                and isinstance(episode.get("episodeNumber"), int)
            ] if isinstance(episodes, list) else []
            episode_ids = [episode["id"] for episode in target_episodes]
            if not episode_ids:
                raise HTTPException(status_code=409, detail="Sonarr could not match episodes to this file")
            collector_id = series_id
            jellyfin_series = snapshot.raw.get("jellyfin", {}).get("item")
            jellyfin_series_id = jellyfin_series.get("Id") if isinstance(jellyfin_series, dict) else None
            if jellyfin_series_id:
                try:
                    jellyfin = JellyfinClient(runtime.jellyfin_base_url, runtime.jellyfin_api_key)
                    jellyfin_episodes = await jellyfin.get_series_episodes(str(jellyfin_series_id))
                    target_pairs = {
                        (episode["seasonNumber"], episode["episodeNumber"])
                        for episode in target_episodes
                    }
                    jellyfin_baseline = [
                        {
                            "seasonNumber": int(item["ParentIndexNumber"]),
                            "episodeNumber": int(item["IndexNumber"]),
                            **_jellyfin_media_signature(item),
                        }
                        for item in jellyfin_episodes
                        if isinstance(item.get("ParentIndexNumber"), int)
                        and isinstance(item.get("IndexNumber"), int)
                        and (item["ParentIndexNumber"], item["IndexNumber"]) in target_pairs
                    ]
                except Exception:
                    logger.warning("Could not capture Jellyfin episode baseline request_id=%s", request_id)
            target_names = [_replacement_file_name(file_data) for file_data in selected_files]
            await sonarr.monitor_episodes(episode_ids, True)
            await asyncio.to_thread(record_cycle)
            for selected_file_id in file_ids:
                await sonarr.delete_episode_file(selected_file_id)
            await sonarr.search_episodes(episode_ids)
        else:
            raise HTTPException(status_code=400, detail="Unknown request type")
    except HTTPException as exc:
        detail = f"The media replacement could not be started: {exc.detail}"
        await asyncio.to_thread(
            save_action, request_id, "replace_media", "Replace media file", "failed", detail
        )
        _record_replacement_activity(
            linked_issue,
            user=user,
            event_type="replacement_failed",
            message=detail,
        )
        raise
    except Exception as exc:
        logger.exception("%s media replacement failed request_id=%s file_ids=%s", collector, request_id, file_ids)
        detail = (
            f"{collector} could not complete the replacement. Check the request action history "
            "before trying again."
        )
        await asyncio.to_thread(
            save_action,
            request_id,
            "replace_media",
            "Replace media file",
            "failed",
            detail,
        )
        _record_replacement_activity(
            linked_issue,
            user=user,
            event_type="replacement_failed",
            message=detail,
        )
        raise HTTPException(status_code=502, detail=detail) from exc

    file_label = "the selected managed file" if len(target_names) == 1 else f"{len(target_names)} selected managed files"
    message = f"{collector} removed {file_label} and started a replacement search."
    await asyncio.to_thread(
        save_action,
        request_id,
        "replace_media",
        "Replace media file",
        "ok",
        message,
    )
    _record_replacement_activity(
        linked_issue,
        user=user,
        event_type="replacement_started",
        message=message,
        metadata={
            "repairTracking": repair_tracking
        },
    )
    return {
        "status": "ok",
        "message": message,
        "collector": collector,
        "request_id": request_id,
        "file_ids": file_ids,
    }


def _positive_id_list(
    value: Any,
    *,
    field: str,
    maximum: int = 200,
    minimum: int = 1,
) -> List[int]:
    if value is None:
        return []
    if not isinstance(value, list):
        raise HTTPException(status_code=400, detail=f"{field} must be a list")
    normalized = list(
        dict.fromkeys(
            item
            for item in value
            if isinstance(item, int) and not isinstance(item, bool) and item >= minimum
        )
    )
    if len(normalized) != len(value) or len(normalized) > maximum:
        raise HTTPException(status_code=400, detail=f"Choose up to {maximum} valid {field.replace('_', ' ')}")
    return normalized


@router.post("/{request_id}/actions/search-missing")
async def action_search_missing_media(
    request_id: str,
    payload: Dict[str, Any],
    user: Dict[str, str] = Depends(get_current_user),
) -> Dict[str, Any]:
    if not request_id.isdigit():
        raise HTTPException(status_code=400, detail="Invalid request id")
    if not _user_can_use_search_auto(user):
        raise HTTPException(status_code=403, detail="Collection searches are disabled for this user")
    linked_issue = _linked_issue_for_replacement(
        payload.get("issue_id"), request_id=request_id, user=user
    )
    episode_ids = _positive_id_list(payload.get("episode_ids"), field="episode_ids")
    season_numbers = _positive_id_list(
        payload.get("season_numbers"), field="season_numbers", maximum=100, minimum=0
    )
    runtime = get_runtime_settings()
    snapshot = await build_snapshot(request_id)
    arr_item = snapshot.raw.get("arr", {}).get("item")
    if not isinstance(arr_item, dict) or not isinstance(arr_item.get("id"), int):
        raise HTTPException(status_code=404, detail="Item not found in Sonarr/Radarr")
    collector_id = int(arr_item["id"])
    target_episodes: List[Dict[str, int]] = []
    try:
        if snapshot.request_type == RequestType.movie:
            radarr = RadarrClient(runtime.radarr_base_url, runtime.radarr_api_key)
            if not radarr.configured():
                raise HTTPException(status_code=400, detail="Radarr is not configured")
            await radarr.monitor_movie(collector_id, True)
            await radarr.search(collector_id)
            message = "Radarr started searching for the missing movie."
            searched_ids: List[int] = []
        else:
            sonarr = SonarrClient(runtime.sonarr_base_url, runtime.sonarr_api_key)
            if not sonarr.configured():
                raise HTTPException(status_code=400, detail="Sonarr is not configured")
            episodes = await sonarr.get_episodes(collector_id)
            if not isinstance(episodes, list):
                raise HTTPException(status_code=502, detail="Sonarr did not return an episode list")
            episode_map = {
                int(item["id"]): item
                for item in episodes
                if isinstance(item, dict) and isinstance(item.get("id"), int)
            }
            if episode_ids:
                if any(item_id not in episode_map for item_id in episode_ids):
                    raise HTTPException(status_code=409, detail="One or more selected episodes are no longer in Sonarr")
                searched_ids = episode_ids
            else:
                searched_ids = [
                    item_id
                    for item_id, episode in episode_map.items()
                    if _released_episode(episode)
                    and not (
                        episode.get("hasFile") is True
                        or (isinstance(episode.get("episodeFileId"), int) and episode.get("episodeFileId") > 0)
                    )
                    and (
                        (season_numbers and episode.get("seasonNumber") in season_numbers)
                        or (not season_numbers and episode.get("monitored") is not False)
                    )
                ]
            if searched_ids:
                target_episodes = [
                    {
                        "id": int(episode_map[episode_id]["id"]),
                        "seasonNumber": int(episode_map[episode_id]["seasonNumber"]),
                        "episodeNumber": int(episode_map[episode_id]["episodeNumber"]),
                    }
                    for episode_id in searched_ids
                    if isinstance(episode_map.get(episode_id), dict)
                    and isinstance(episode_map[episode_id].get("seasonNumber"), int)
                    and isinstance(episode_map[episode_id].get("episodeNumber"), int)
                ]
                await sonarr.monitor_episodes(searched_ids, True)
                await sonarr.search_episodes(searched_ids)
                message = f"Sonarr started searching for {len(searched_ids)} selected missing episode(s)."
            else:
                await sonarr.search(collector_id)
                message = "Sonarr refreshed the series and started a full missing-episode search."
    except HTTPException as exc:
        detail = f"The missing-content search could not start: {exc.detail}"
        await asyncio.to_thread(
            save_action,
            request_id,
            "search_missing",
            "Search for missing content",
            "failed",
            detail,
        )
        _record_replacement_activity(
            linked_issue,
            user=user,
            event_type="missing_search_failed",
            message=detail,
        )
        raise
    except Exception as exc:
        logger.exception("missing content search failed request_id=%s", request_id)
        detail = "Sonarr/Radarr could not start the missing-content search."
        await asyncio.to_thread(
            save_action,
            request_id,
            "search_missing",
            "Search for missing content",
            "failed",
            detail,
        )
        _record_replacement_activity(
            linked_issue, user=user, event_type="missing_search_failed", message=detail
        )
        raise HTTPException(status_code=502, detail=detail) from exc
    await asyncio.to_thread(
        save_action, request_id, "search_missing", "Search for missing content", "ok", message
    )
    _record_replacement_activity(
        linked_issue,
        user=user,
        event_type="missing_search_started",
        message=message,
        metadata={
            "repairTracking": {
                "requestId": request_id,
                "actionId": "search_missing",
                "mediaType": snapshot.request_type.value,
                "collectorId": collector_id,
                "originalFileIds": [],
                "episodes": target_episodes,
                "jellyfinBaseline": [],
                "jellyfinFoundAtStart": bool(snapshot.raw.get("jellyfin", {}).get("found")),
                "startedAt": datetime.now(timezone.utc).isoformat(),
            }
        },
    )
    return {"status": "ok", "message": message, "episode_ids": searched_ids}


@router.post("/{request_id}/actions/repair-subtitles")
async def action_repair_subtitles(
    request_id: str,
    payload: Dict[str, Any],
    user: Dict[str, str] = Depends(get_current_user),
) -> Dict[str, Any]:
    if not request_id.isdigit():
        raise HTTPException(status_code=400, detail="Invalid request id")
    if not _user_can_use_search_auto(user):
        raise HTTPException(status_code=403, detail="Subtitle repairs are disabled for this user")
    linked_issue = _linked_issue_for_replacement(
        payload.get("issue_id"), request_id=request_id, user=user
    )
    episode_ids = _positive_id_list(payload.get("episode_ids"), field="episode_ids", maximum=100)
    forced = payload.get("forced") is True
    runtime = get_runtime_settings()
    bazarr = BazarrClient(runtime.bazarr_base_url, runtime.bazarr_api_key)
    if not bazarr.configured() or not runtime.bazarr_api_key:
        raise HTTPException(status_code=400, detail="Bazarr is not configured")
    language = str(runtime.bazarr_default_language or "en").strip().lower() or "en"
    snapshot = await build_snapshot(request_id)
    arr_item = snapshot.raw.get("arr", {}).get("item")
    if not isinstance(arr_item, dict) or not isinstance(arr_item.get("id"), int):
        raise HTTPException(status_code=404, detail="Item not found in Sonarr/Radarr")
    collector_id = int(arr_item["id"])
    try:
        if snapshot.request_type == RequestType.movie:
            await bazarr.search_movie_subtitles(
                collector_id, language=language, forced=forced
            )
            repaired_count = 1
            message = f"Bazarr started a fresh {language.upper()} subtitle search for the movie."
        else:
            if not episode_ids:
                raise HTTPException(status_code=400, detail="Choose at least one episode")
            sonarr = SonarrClient(runtime.sonarr_base_url, runtime.sonarr_api_key)
            episodes = await sonarr.get_episodes(collector_id)
            valid_ids = {
                int(item["id"])
                for item in episodes
                if isinstance(item, dict) and isinstance(item.get("id"), int)
            } if isinstance(episodes, list) else set()
            if any(episode_id not in valid_ids for episode_id in episode_ids):
                raise HTTPException(status_code=409, detail="One or more selected episodes are no longer in Sonarr")
            for episode_id in episode_ids:
                await bazarr.search_episode_subtitles(
                    collector_id,
                    episode_id,
                    language=language,
                    forced=forced,
                )
            repaired_count = len(episode_ids)
            message = f"Bazarr started fresh {language.upper()} subtitle searches for {repaired_count} episode(s)."
    except HTTPException as exc:
        detail = f"The subtitle repair could not start: {exc.detail}"
        await asyncio.to_thread(
            save_action,
            request_id,
            "repair_subtitles",
            "Repair subtitles",
            "failed",
            detail,
        )
        _record_replacement_activity(
            linked_issue,
            user=user,
            event_type="subtitle_repair_failed",
            message=detail,
        )
        raise
    except Exception as exc:
        logger.exception("Bazarr subtitle repair failed request_id=%s", request_id)
        detail = "Bazarr could not start the subtitle repair."
        await asyncio.to_thread(
            save_action,
            request_id,
            "repair_subtitles",
            "Repair subtitles",
            "failed",
            detail,
        )
        _record_replacement_activity(
            linked_issue, user=user, event_type="subtitle_repair_failed", message=detail
        )
        raise HTTPException(status_code=502, detail=detail) from exc
    await asyncio.to_thread(
        save_action, request_id, "repair_subtitles", "Repair subtitles", "ok", message
    )
    _record_replacement_activity(
        linked_issue, user=user, event_type="subtitle_repair_started", message=message
    )
    return {"status": "ok", "message": message, "count": repaired_count}


@router.get("/{request_id}/snapshot", response_model=Snapshot)
async def get_snapshot(request_id: str, user: Dict[str, str] = Depends(get_current_user)) -> Snapshot:
    runtime = get_runtime_settings()
    client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
    if client.configured():
        await _ensure_request_access(client, int(request_id), user)
    snapshot = await build_snapshot(request_id)
    return _filter_snapshot_for_user(snapshot, user)


async def _restore_request_monitoring(snapshot: Snapshot, request: dict) -> bool:
    # Pending or declined requests must not gain collection access via Recheck.
    if request.get('status') != 2:
        return False
    item = (snapshot.raw.get('arr') or {}).get('item')
    if not isinstance(item, dict) or not isinstance(item.get('id'), int):
        return False
    runtime = get_runtime_settings()
    if snapshot.request_type == RequestType.movie:
        client = RadarrClient(runtime.radarr_base_url, runtime.radarr_api_key)
        fresh = await client.get_movie(item['id'])
        if not isinstance(fresh, dict):
            raise ValueError('Radarr did not return the requested movie')
        if fresh.get('monitored') is True:
            return False
        await client.update_movie({**fresh, 'monitored': True})
        verified = await client.get_movie(item['id'])
        if not isinstance(verified, dict) or verified.get('monitored') is not True:
            raise ValueError('Radarr did not enable monitoring')
        return True
    client = SonarrClient(runtime.sonarr_base_url, runtime.sonarr_api_key)
    fresh = await client.get_series(item['id'])
    if not isinstance(fresh, dict):
        raise ValueError('Sonarr did not return the requested series')
    requested = {season['seasonNumber'] for season in (request.get('seasons') or [])
                 if isinstance(season, dict) and isinstance(season.get('seasonNumber'), int)}
    seasons = [{**season, 'monitored': True} if season.get('seasonNumber') in requested else season
               for season in (fresh.get('seasons') or [])]
    changed = fresh.get('monitored') is not True or seasons != fresh.get('seasons', [])
    if changed:
        await client.update_series({**fresh, 'monitored': True, 'seasons': seasons})
    episodes = await client.get_episodes(item['id']) if requested else []
    ids = [episode['id'] for episode in episodes if episode.get('seasonNumber') in requested
           and episode.get('monitored') is not True and isinstance(episode.get('id'), int)]
    if ids:
        await client.monitor_episodes(ids, True)
        verified_episodes = await client.get_episodes(item['id'])
        if any(e.get('id') in ids and e.get('monitored') is not True for e in verified_episodes):
            raise ValueError('Sonarr did not enable episode monitoring')
    if changed:
        verified = await client.get_series(item['id'])
        if not isinstance(verified, dict) or verified.get('monitored') is not True or any(
            season.get('seasonNumber') in requested and season.get('monitored') is not True
            for season in verified.get('seasons', [])):
            raise ValueError('Sonarr did not enable monitoring')
    return changed or bool(ids)


@router.post("/{request_id}/actions/recheck")
async def action_recheck(request_id: str, user: Dict[str, str] = Depends(get_current_user)) -> dict:
    if not request_id.isdigit():
        raise HTTPException(status_code=400, detail="Invalid request id")

    runtime = get_runtime_settings()
    seerr = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
    if not seerr.configured():
        raise HTTPException(status_code=400, detail="Seerr is not configured")
    await _ensure_request_access(seerr, int(request_id), user)

    try:
        fresh_request = await seerr.get_request(request_id)
    except httpx.HTTPStatusError as exc:
        detail = _format_upstream_error("Seerr", exc)
        await asyncio.to_thread(
            save_action,
            request_id,
            "recheck_pipeline",
            "Recheck request status",
            "failed",
            detail,
        )
        raise HTTPException(status_code=502, detail=detail) from exc
    except Exception as exc:
        logger.warning("Request recheck failed while reading Seerr: request_id=%s error=%s", request_id, exc)
        detail = "Magent could not reach Seerr to recheck this request."
        await asyncio.to_thread(
            save_action,
            request_id,
            "recheck_pipeline",
            "Recheck request status",
            "failed",
            detail,
        )
        raise HTTPException(status_code=502, detail=detail) from exc

    if not isinstance(fresh_request, dict):
        raise HTTPException(status_code=404, detail="Request not found in Seerr")

    parsed = _parse_request_payload(fresh_request)
    if not parsed.get("title"):
        details = await _get_media_details(
            seerr, parsed.get("media_type"), parsed.get("tmdb_id")
        )
        if isinstance(details, dict):
            fresh_request = _merge_request_media_details(fresh_request, details)
    parsed = _parse_request_payload(fresh_request)
    if parsed.get("request_id") != int(request_id):
        raise HTTPException(status_code=502, detail="Seerr returned an unexpected request record")

    cache_record = _build_request_cache_record(parsed, fresh_request)
    await asyncio.to_thread(upsert_request_cache, **cache_record)
    _cache_set(f"request:{request_id}", fresh_request)
    _refresh_recent_cache_from_db()

    snapshot = await build_snapshot(request_id)
    try:
        restored = await _restore_request_monitoring(snapshot, fresh_request)
    except Exception as exc:
        logger.warning('Recheck monitoring failed request_id=%s: %s', request_id, exc)
        raise HTTPException(502, 'Could not restore monitoring in Sonarr/Radarr. Please try Recheck again.') from exc
    if restored:
        snapshot = await build_snapshot(request_id)
    snapshot = _filter_snapshot_for_user(snapshot, user)
    status_label = str((snapshot.presentation.get("status") or {}).get("label") or "Status updated")
    message = ("Monitoring restored. " if restored else "") + f"Recheck complete. {status_label}."
    await asyncio.to_thread(
        save_action,
        request_id,
        "recheck_pipeline",
        "Recheck request status",
        "ok",
        message,
    )
    return {"status": "ok", "message": message, "snapshot": snapshot}


@router.get("/{request_id}/download-progress")
async def get_download_progress(
    request_id: str, user: Dict[str, str] = Depends(get_current_user)
) -> Dict[str, Any]:
    """Return a lightweight qBittorrent update for an open request page."""
    if not request_id.isdigit():
        raise HTTPException(status_code=400, detail="Invalid request id")

    runtime = get_runtime_settings()
    seerr = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
    if seerr.configured():
        await _ensure_request_access(seerr, int(request_id), user)

    repairs = await asyncio.to_thread(get_request_repairs, request_id, active_only=False)
    cycle = repairs[-1]["startedAt"] if repairs else None
    evidence = await asyncio.to_thread(get_request_download_evidence, request_id, 20)
    historical_torrents = evidence.get("torrents") if isinstance(evidence, dict) else []
    hashes: List[str] = []
    if isinstance(historical_torrents, list):
        hashes = list(
            dict.fromkeys(
                str(torrent.get("hash") or "").strip()
                for torrent in historical_torrents
                if isinstance(torrent, dict) and torrent.get("hash")
            )
        )

    qbittorrent = QBittorrentClient(
        runtime.qbittorrent_base_url,
        runtime.qbittorrent_username,
        runtime.qbittorrent_password,
    )
    if not qbittorrent.configured():
        raise HTTPException(status_code=503, detail="qBittorrent is not configured")

    try:
        # Discover new jobs from the collector, not only yesterday's hashes or
        # legacy Magent tags. Sonarr-owned downloads do not have those tags.
        queue = None
        request = await asyncio.to_thread(get_request_cache_payload, int(request_id))
        if not isinstance(request, dict) and seerr.configured():
            request = await seerr.get_request(request_id)
        media = (request or {}).get("media") or {}
        if (request or {}).get("type") == "tv" and media.get("tvdbId"):
            collector = SonarrClient(runtime.sonarr_base_url, runtime.sonarr_api_key)
            items = await collector.get_series_by_tvdb_id(int(media["tvdbId"]))
            item = items[0] if isinstance(items, list) and items else None
            if item and item.get("id"):
                queue = await collector.get_queue(int(item["id"]))
                queue = {**queue, "records": [r for r in _queue_records(queue) if r.get("seriesId") == item["id"]]}
                hashes.extend(_download_ids(_queue_records(queue)))
        hashes = list(dict.fromkeys(h.strip().lower() for h in hashes if h.strip()))
        if hashes:
            result = await qbittorrent.get_torrents_by_hashes("|".join(hashes))
        else:
            result = await qbittorrent.get_torrents_by_tag(f"magent-{request_id}")
    except Exception as exc:
        logger.warning("Live qBittorrent progress failed request_id=%s: %s", request_id, exc)
        raise HTTPException(status_code=502, detail="Live download progress is temporarily unavailable") from exc

    torrents = label_episode_downloads(current_cycle_torrents(result, cycle), queue)
    for torrent in torrents:
        if isinstance(torrent, dict):
            torrent["progressPercent"] = _torrent_progress(torrent)

    if torrents:
        summary = _summarize_qbit(torrents)
        state = str(summary.get("state") or "idle")
        message = str(summary.get("message") or "Download found in qBittorrent.")
    elif evidence.get("observed"):
        state = "missing"
        message = "The previous download is no longer visible in qBittorrent."
    else:
        state = "not_started"
        message = "No download attempt has been observed."

    return {
        "request_id": request_id,
        "state": state,
        "summary": message,
        "torrents": torrents,
        "updated_at": datetime.now(timezone.utc).isoformat(),
        "repairCycle": cycle,
        "visible": bool(torrents) or bool(evidence.get("observed")),
    }


@router.get("/recent")
async def recent_requests(
    take: int = 6,
    skip: int = 0,
    days: int = 90,
    stage: str = "all",
    user: Dict[str, str] = Depends(get_current_user),
) -> dict:
    runtime = get_runtime_settings()
    client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
    mode = (runtime.requests_data_source or "prefer_cache").lower()
    # Browsing is always local. Synchronization is owned by background workers.
    allow_remote = False
    username_norm = _normalize_username(user.get("username", ""))
    requested_by_id = user.get("jellyseerr_user_id")
    requested_by = None if user.get("role") == "admin" else username_norm
    requested_by_id = None if user.get("role") == "admin" else requested_by_id
    since_iso = None
    if days > 0:
        since_iso = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
    status_codes = request_stage_filter_codes(stage)
    if _recent_cache_stale():
        _refresh_recent_cache_from_db()
    # Jellyfin can promote working requests to ready. Filter the displayed
    # stage before pagination, including working candidates in the ready view.
    candidate_codes = [4, 5] if status_codes == [4] else status_codes
    take = max(1, min(int(take), 200))
    skip = max(0, int(skip))
    rows = _get_recent_from_cache(
        requested_by, requested_by_id,
        len(_recent_cache.get("items") or []), 0, since_iso,
        status_codes=candidate_codes,
    )
    matched = 0
    cache_mode = (runtime.artwork_cache_mode or "remote").lower()
    allow_title_hydrate = False
    allow_artwork_hydrate = False
    stage_cache = await asyncio.to_thread(get_request_stage_cache)
    jellyfin = JellyfinClient(runtime.jellyfin_base_url, runtime.jellyfin_api_key)
    jellyfin_cache: Dict[str, bool] = {}
    results = []
    for row in rows:
        status = row.get("status")
        title = row.get("title")
        title_is_placeholder = (
            isinstance(title, str)
            and row.get("request_id") is not None
            and title.strip().lower() == f"request {row.get('request_id')}"
        )
        year = row.get("year")
        details = None
        if row.get("request_id"):
            cached_payload = get_request_cache_payload(int(row["request_id"]))
            if isinstance(cached_payload, dict):
                details = cached_payload
        if (not title or title_is_placeholder) and row.get("request_id"):
            if details is None and (allow_remote or allow_title_hydrate):
                details = await _get_request_details(client, int(row["request_id"]))
            if isinstance(details, dict):
                payload = _parse_request_payload(details)
                title = payload.get("title") or title
                year = payload.get("year") or year
                if not title and payload.get("tmdb_id") and (allow_remote or allow_title_hydrate):
                    hydrated_title, hydrated_year = await _hydrate_title_from_tmdb(
                        client, payload.get("media_type"), payload.get("tmdb_id")
                    )
                    if hydrated_title:
                        title = hydrated_title
                    if hydrated_year:
                        year = hydrated_year
                if allow_remote and isinstance(payload.get("request_id"), int):
                    upsert_request_cache(
                        request_id=payload.get("request_id"),
                        media_id=payload.get("media_id"),
                        media_type=payload.get("media_type"),
                        status=payload.get("status"),
                        title=title or payload.get("title"),
                        year=year or payload.get("year"),
                        requested_by=payload.get("requested_by"),
                        requested_by_norm=payload.get("requested_by_norm"),
                        requested_by_id=payload.get("requested_by_id"),
                        created_at=payload.get("created_at"),
                        updated_at=payload.get("updated_at"),
                        payload_json=json.dumps(details, ensure_ascii=True),
                    )
                    row["title"] = title
                    row["year"] = year
                    row["media_type"] = payload.get("media_type") or row.get("media_type")
                    row["status"] = payload.get("status") or row.get("status")
        if details is None and row.get("request_id") and allow_remote:
            details = await _get_request_details(client, int(row["request_id"]))

        poster_path = None
        backdrop_path = None
        if isinstance(details, dict):
            media = details.get("media") or {}
            if isinstance(media, dict):
                poster_path = media.get("posterPath") or media.get("poster_path")
                backdrop_path = media.get("backdropPath") or media.get("backdrop_path")
                tmdb_id = media.get("tmdbId") or details.get("tmdbId")
            else:
                tmdb_id = details.get("tmdbId")
            media_type = media.get("mediaType") if isinstance(media, dict) else None
            media_type = media_type or details.get("type") or row.get("media_type")
            if not poster_path and tmdb_id and allow_artwork_hydrate:
                hydrated_poster, hydrated_backdrop = await _hydrate_artwork_from_tmdb(
                    client, media_type, tmdb_id
                )
                poster_path = poster_path or hydrated_poster
                backdrop_path = backdrop_path or hydrated_backdrop
                if (hydrated_poster or hydrated_backdrop) and isinstance(details, dict):
                    media = dict(media) if isinstance(media, dict) else {}
                    if hydrated_poster:
                        media["posterPath"] = hydrated_poster
                    if hydrated_backdrop:
                        media["backdropPath"] = hydrated_backdrop
                    details["media"] = media
                    payload = _parse_request_payload(details)
                    if isinstance(payload.get("request_id"), int):
                        upsert_request_cache(
                            request_id=payload.get("request_id"),
                            media_id=payload.get("media_id"),
                            media_type=payload.get("media_type"),
                            status=payload.get("status"),
                            title=payload.get("title"),
                            year=payload.get("year"),
                            requested_by=payload.get("requested_by"),
                            requested_by_norm=payload.get("requested_by_norm"),
                            requested_by_id=payload.get("requested_by_id"),
                            created_at=payload.get("created_at"),
                            updated_at=payload.get("updated_at"),
                            payload_json=json.dumps(details, ensure_ascii=True),
                        )
        status_label = _status_label(status)
        if row.get("repairing"):
            status = 5
            status_label = "Repair in progress"
        elif status_label in {"Working on it", "Ready to watch", "Partially ready"}:
            saved = stage_cache.get(row.get("request_id")) or {}
            is_available = bool(saved.get("ready"))
            status_label = _status_label_with_jellyfin(status, is_available)
        if status_label == STATUS_LABELS[4]:
            status = 4
        if status_codes and status not in status_codes:
            continue
        matched += 1
        if matched <= skip:
            continue
        results.append(
            {
                "id": row.get("request_id"),
                "title": title,
                "year": year,
                "type": row.get("media_type"),
                "status": status,
                "statusLabel": status_label,
                "mediaId": row.get("media_id"),
                "createdAt": row.get("created_at") or row.get("updated_at"),
                "artwork": {
                    "poster_url": _artwork_url(poster_path, "w185", cache_mode),
                    "backdrop_url": _artwork_url(backdrop_path, "w780", cache_mode),
                },
            }
        )
        if len(results) >= take:
            break

    return {"results": results}


@router.get("/search")
async def search_requests(
    query: str,
    page: int = 1,
    media_type: Optional[str] = None,
    user: Dict[str, str] = Depends(get_current_user),
) -> dict:
    runtime = get_runtime_settings()
    client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
    if not client.configured():
        raise HTTPException(status_code=400, detail="Seerr not configured")

    try:
        response = await client.search(query=query, page=page)
    except httpx.HTTPStatusError as exc:
        raise HTTPException(status_code=502, detail=str(exc)) from exc

    if not isinstance(response, dict):
        return {"results": []}

    try:
        await _ensure_requests_cache(client)
    except httpx.HTTPStatusError:
        pass

    requested_media_type = _normalize_media_type(media_type) if media_type is not None else None
    if media_type is not None and requested_media_type is None:
        raise HTTPException(status_code=400, detail="media_type must be 'movie' or 'tv'")

    results = []
    jellyfin = JellyfinClient(runtime.jellyfin_base_url, runtime.jellyfin_api_key)
    jellyfin_cache: Dict[str, bool] = {}
    for item in response.get("results", []):
        media_type = item.get("mediaType")
        if requested_media_type is not None and media_type != requested_media_type:
            continue
        title = item.get("title") or item.get("name")
        year = None
        if item.get("releaseDate"):
            year = int(item["releaseDate"][:4])
        if item.get("firstAirDate"):
            year = int(item["firstAirDate"][:4])

        request_id = None
        status = None
        status_label = None
        requested_by = None
        accessible = False
        media_info = item.get("mediaInfo") or {}
        media_info_id = media_info.get("id")
        requests = media_info.get("requests")
        if isinstance(requests, list) and requests:
            request_id = requests[0].get("id")
            status = requests[0].get("status")
            status_label = _status_label(status)
        elif isinstance(media_info_id, int):
            cached = get_cached_request_by_media_id(
                media_info_id,
            )
            if cached:
                request_id = cached.get("request_id")
                status = cached.get("status")
                status_label = _status_label(status)

        if isinstance(request_id, int):
            details = get_request_cache_payload(request_id)
            if not isinstance(details, dict):
                details = await _get_request_details(client, request_id)
            if user.get("role") == "admin":
                requested_by = _request_display_name(details)
            accessible = True
            if status is not None:
                is_available = await _request_is_available_in_jellyfin(
                    jellyfin,
                    title,
                    year,
                    media_type,
                    details if isinstance(details, dict) else None,
                    jellyfin_cache,
                )
                status_label = _status_label_with_jellyfin(status, is_available)

        results.append(
            {
                "title": title,
                "year": year,
                "type": media_type,
                "tmdbId": item.get("id"),
                "requestId": request_id,
                "status": status,
                "statusLabel": status_label,
                "requestedBy": requested_by,
                "accessible": accessible,
                "overview": item.get("overview"),
                "posterPath": item.get("posterPath") or item.get("poster_path"),
                "backdropPath": item.get("backdropPath") or item.get("backdrop_path"),
            }
        )

    return {"results": results}


@router.get("/request-options")
async def request_options(
    media_type: str,
    tmdb_id: int,
    user: Dict[str, str] = Depends(get_current_user),
) -> Dict[str, Any]:
    del user
    normalized_media_type = _normalize_media_type(media_type)
    if normalized_media_type is None:
        raise HTTPException(status_code=400, detail="media_type must be 'movie' or 'tv'")
    if tmdb_id <= 0:
        raise HTTPException(status_code=400, detail="tmdb_id must be a positive integer")

    runtime = get_runtime_settings()
    client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
    if not client.configured():
        raise HTTPException(status_code=400, detail="Seerr not configured")

    try:
        details, destination = await asyncio.gather(
            client.get_movie(tmdb_id) if normalized_media_type == "movie" else client.get_tv(tmdb_id),
            _resolve_request_destination(runtime, client, normalized_media_type),
        )
    except HTTPException:
        raise
    except httpx.HTTPStatusError as exc:
        raise HTTPException(status_code=502, detail=_format_upstream_error("Seerr", exc)) from exc

    if not isinstance(details, dict):
        raise HTTPException(status_code=502, detail="Seerr returned invalid media details")

    title = str(details.get("title") or details.get("name") or "Untitled")
    date_value = details.get("releaseDate") or details.get("firstAirDate")
    year = int(date_value[:4]) if isinstance(date_value, str) and len(date_value) >= 4 and date_value[:4].isdigit() else None
    seasons: list[Dict[str, Any]] = []
    if normalized_media_type == "tv":
        for season in details.get("seasons", []):
            if not isinstance(season, dict):
                continue
            season_number = _quality_profile_id(season.get("seasonNumber"))
            if season_number is None or season_number <= 0:
                continue
            seasons.append(
                {
                    "seasonNumber": season_number,
                    "name": str(season.get("name") or f"Season {season_number}"),
                    "episodeCount": _quality_profile_id(season.get("episodeCount")) or 0,
                    "airDate": season.get("airDate"),
                }
            )

    media_info = details.get("mediaInfo") if isinstance(details.get("mediaInfo"), dict) else {}
    requests_list = media_info.get("requests")
    existing_request_id = None
    if isinstance(requests_list, list) and requests_list and isinstance(requests_list[0], dict):
        existing_request_id = _quality_profile_id(requests_list[0].get("id"))

    return {
        "media": {
            "title": title,
            "year": year,
            "type": normalized_media_type,
            "tmdbId": tmdb_id,
            "overview": details.get("overview"),
            "posterPath": details.get("posterPath") or details.get("poster_path"),
            "backdropPath": details.get("backdropPath") or details.get("backdrop_path"),
            "seasons": seasons,
            "existingRequestId": existing_request_id,
            "originalLanguage": language_info(details),
        },
        "destination": {
            "collector": destination["collector"],
            "serverName": destination["server_name"],
            "defaultProfileId": destination["default_profile_id"],
            "profiles": destination["profiles"],
        },
    }


@router.post("/create")
async def create_request(
    payload: Dict[str, Any], user: Dict[str, Any] = Depends(get_current_user)
) -> Dict[str, Any]:
    runtime = get_runtime_settings()
    client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
    if not client.configured():
        raise HTTPException(status_code=400, detail="Seerr not configured")

    media_type = _normalize_media_type(
        payload.get("mediaType") or payload.get("type") or payload.get("media_type")
    )
    if media_type is None:
        raise HTTPException(status_code=400, detail="mediaType must be 'movie' or 'tv'")

    raw_tmdb_id = payload.get("tmdbId")
    if raw_tmdb_id is None:
        raw_tmdb_id = payload.get("mediaId")
    if raw_tmdb_id is None:
        raw_tmdb_id = payload.get("id")
    try:
        tmdb_id = int(raw_tmdb_id)
    except (TypeError, ValueError) as exc:
        raise HTTPException(status_code=400, detail="tmdbId must be a valid integer") from exc
    if tmdb_id <= 0:
        raise HTTPException(status_code=400, detail="tmdbId must be a positive integer")

    seasons = _normalize_seasons(payload.get("seasons")) if media_type == "tv" else []
    raw_is_4k = payload.get("is4k")
    if raw_is_4k is not None and not isinstance(raw_is_4k, bool):
        raise HTTPException(status_code=400, detail="is4k must be true or false")
    is_4k = raw_is_4k if isinstance(raw_is_4k, bool) else None

    try:
        details = await (client.get_movie(tmdb_id) if media_type == "movie" else client.get_tv(tmdb_id))
    except httpx.HTTPStatusError as exc:
        raise HTTPException(status_code=502, detail=_format_upstream_error("Seerr", exc)) from exc

    if not isinstance(details, dict):
        raise HTTPException(status_code=502, detail="Invalid response from Seerr media lookup")

    language = language_info(details)
    accept_original = payload.get("acceptOriginalLanguage", False)
    if not isinstance(accept_original, bool):
        raise HTTPException(400, "The language choice must be true or false.")
    if accept_original and not language:
        raise HTTPException(409, "The original language could not be verified. Reload this title.")

    media_info = details.get("mediaInfo") if isinstance(details.get("mediaInfo"), dict) else {}
    requests_list = media_info.get("requests")
    existing_request: Optional[Dict[str, Any]] = None
    if isinstance(requests_list, list) and requests_list:
        first_request = requests_list[0]
        if isinstance(first_request, dict):
            existing_request = first_request

    title = details.get("title") or details.get("name")
    year: Optional[int] = None
    date_value = details.get("releaseDate") or details.get("firstAirDate")
    if isinstance(date_value, str) and len(date_value) >= 4 and date_value[:4].isdigit():
        year = int(date_value[:4])

    if isinstance(existing_request, dict):
        if accept_original:
            raise HTTPException(409, 'This title is already requested. Open its request and choose Use original audio & search to update the existing movie.')
        existing_request_id = _quality_profile_id(existing_request.get("id"))
        existing_status = existing_request.get("status")
        if existing_request_id is not None:
            request_payload = await _get_request_details(client, existing_request_id)
            if isinstance(request_payload, dict):
                parsed_payload = _parse_request_payload(request_payload)
                upsert_request_cache(**_build_request_cache_record(parsed_payload, request_payload))
                _cache_set(f"request:{existing_request_id}", request_payload)
                title = parsed_payload.get("title") or title
                year = parsed_payload.get("year") or year
            return {
                "status": "exists",
                "requestId": existing_request_id,
                "type": media_type,
                "tmdbId": tmdb_id,
                "title": title,
                "year": year,
                "statusCode": existing_status,
                "statusLabel": _status_label(existing_status),
            }

    if media_type == "tv" and seasons:
        valid_seasons = {
            _quality_profile_id(item.get("seasonNumber"))
            for item in details.get("seasons", [])
            if isinstance(item, dict)
        }
        invalid_seasons = [season for season in seasons if season not in valid_seasons]
        if invalid_seasons:
            raise HTTPException(
                status_code=400,
                detail=f"Season selection is not available for this series: {invalid_seasons}",
            )

    destination = await _resolve_request_destination(runtime, client, media_type)
    if accept_original and media_type == "movie":
        destination["profile_id"] = await original_profile(
            RadarrClient(runtime.radarr_base_url, runtime.radarr_api_key), destination["profile_id"])
        # Seerr does not update an already-existing Radarr movie's profile on request creation.
        await apply_original_to_movie(RadarrClient(runtime.radarr_base_url, runtime.radarr_api_key), tmdb_id)

    try:
        created = await client.create_request(
            media_type=media_type,
            media_id=tmdb_id,
            seasons=seasons if media_type == "tv" else None,
            is_4k=is_4k,
            server_id=destination["server_id"],
            profile_id=destination["profile_id"],
            root_folder=destination["root_folder"],
        )
    except httpx.HTTPStatusError as exc:
        raise HTTPException(status_code=502, detail=_format_upstream_error("Seerr", exc)) from exc

    if not isinstance(created, dict):
        raise HTTPException(status_code=502, detail="Invalid response from Seerr request create")

    created = _merge_request_media_details(created, details)
    parsed = _parse_request_payload(created)
    request_id = _quality_profile_id(parsed.get("request_id"))
    status_code = parsed.get("status")
    title = parsed.get("title") or title
    year = parsed.get("year") or year

    if request_id is not None:
        upsert_request_cache(**_build_request_cache_record(parsed, created))
        _cache_set(f"request:{request_id}", created)
        _recent_cache["updated_at"] = None
        await asyncio.to_thread(
            save_action,
            str(request_id),
            "request_created",
            "Create request",
            "ok",
            f"{media_type} request created from discovery by {user.get('username')}."
            + (f" Original-language audio accepted ({language['code']})." if accept_original else ""),
        )

    return {
        "status": "created",
        "requestId": request_id,
        "type": media_type,
        "tmdbId": tmdb_id,
        "title": title,
        "year": year,
        "statusCode": status_code,
        "statusLabel": _status_label(status_code),
    }


@router.post("/{request_id}/ai/triage", response_model=TriageResult)
async def ai_triage(request_id: str, user: Dict[str, str] = Depends(get_current_user)) -> TriageResult:
    runtime = get_runtime_settings()
    client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
    if client.configured():
        await _ensure_request_access(client, int(request_id), user)
    snapshot = _filter_snapshot_for_user(await build_snapshot(request_id), user)
    return triage_snapshot(snapshot)


async def _request_language_context(request_id, user):
    runtime = get_runtime_settings()
    seerr = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
    await _ensure_request_access(seerr, int(request_id), user)
    request = await seerr.get_request(request_id)
    if not isinstance(request, dict) or request.get('type') != 'movie':
        return runtime, None, None
    tmdb_id = (request.get('media') or {}).get('tmdbId')
    if not isinstance(tmdb_id, int):
        raise HTTPException(502, 'Seerr did not return the movie identity.')
    details = await seerr.get_movie(tmdb_id)
    return runtime, tmdb_id, language_info(details or {})


@router.get("/{request_id}/language")
async def request_language(request_id: str, user: dict = Depends(get_current_user)):
    runtime, tmdb_id, language = await _request_language_context(request_id, user)
    if not language:
        return {'language': None}
    radarr = RadarrClient(runtime.radarr_base_url, runtime.radarr_api_key)
    movies = await radarr.get_movie_by_tmdb_id(tmdb_id)
    movie = next((m for m in (movies or []) if m.get('tmdbId') == tmdb_id), None)
    profiles = await radarr.get_quality_profiles() if movie else []
    profile = next((p for p in profiles if p['id'] == movie['qualityProfileId']), {}) if movie else {}
    return {'language': language, 'originalEnabled': is_original_profile(profile),
            'canChange': bool(movie) and _user_can_use_search_auto(user),
            'profileLanguage': (profile.get('language') or {}).get('name')}


@router.post("/{request_id}/actions/language")
async def accept_request_language(request_id: str, payload: dict, user: dict = Depends(get_current_user)):
    if not _user_can_use_search_auto(user):
        raise HTTPException(403, 'Search and download changes are disabled for this account.')
    if payload.get('acceptOriginalLanguage') is not True:
        raise HTTPException(400, 'Explicitly accept original-language audio before continuing.')
    runtime, tmdb_id, language = await _request_language_context(request_id, user)
    if not language:
        raise HTTPException(409, 'This request has no verified non-English original language.')
    if payload.get('languageCode') != language['code']:
        raise HTTPException(409, 'The language metadata changed. Reload the request and review it again.')
    radarr = RadarrClient(runtime.radarr_base_url, runtime.radarr_api_key)
    profile_id = await apply_original_to_movie(radarr, tmdb_id)
    if profile_id is None:
        raise HTTPException(409, 'The movie is not in Radarr yet. Recheck the pipeline first.')
    await asyncio.to_thread(save_action, request_id, 'original_language', 'Accept original-language audio',
                            'ok', f"Original-language audio accepted ({language['code']}); Radarr profile {profile_id} verified.")
    result = await action_search_auto(request_id, user)
    result['message'] = 'Original-language audio enabled. ' + result['message']
    return result


@router.post("/{request_id}/actions/search")
async def action_search(request_id: str, user: Dict[str, str] = Depends(get_current_user), offset: int = 0) -> dict:
    if offset < 0:
        raise HTTPException(400, 'Search offset must be zero or greater.')
    total_missing = 0
    next_offset = None
    runtime = get_runtime_settings()
    client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
    if client.configured():
        await _ensure_request_access(client, int(request_id), user)
    snapshot = await build_snapshot(request_id)
    arr_item = snapshot.raw.get("arr", {}).get("item")
    if not isinstance(arr_item, dict) or not isinstance(arr_item.get("id"), int):
        raise HTTPException(status_code=404, detail="Item not found in Sonarr/Radarr")

    results: List[Dict[str, Any]] = []
    collector = "Sonarr" if snapshot.request_type == RequestType.tv else "Radarr"
    try:
        if snapshot.request_type == RequestType.tv:
            sonarr = SonarrClient(runtime.sonarr_base_url, runtime.sonarr_api_key)
            if not sonarr.configured():
                raise HTTPException(status_code=400, detail="Sonarr not configured")
            episodes = await sonarr.get_episodes(int(arr_item["id"]))
            missing_by_season = _missing_episode_ids_by_season(episodes)
            season_numbers = sorted(missing_by_season)
            if not season_numbers:
                message = "Sonarr has no missing monitored episodes to search for."
                await asyncio.to_thread(
                    save_action,
                    request_id,
                    "search_releases",
                    "Search and choose a download",
                    "ok",
                    message,
                )
                return {"status": "ok", "message": message, "collector": collector, "releases": []}
            missing_ids = [identity for season in season_numbers for identity in missing_by_season[season]]
            total_missing = len(missing_ids)
            batch = missing_ids[offset:offset + 3]
            next_offset = offset + len(batch) if offset + len(batch) < total_missing else None
            semaphore = asyncio.Semaphore(3)
            async def search_episode(identity):
                async with semaphore:
                    found = await sonarr.search_episode_releases(identity)
                    if not isinstance(found, list):
                        raise HTTPException(502, 'Sonarr did not return valid episode search results. Try again.')
                    return found
            searches = await asyncio.gather(*(search_episode(identity) for identity in batch))
            # Interleave per-episode rankings so a prolific episode cannot hide the others.
            for position in range(max((len(items) for items in searches), default=0)):
                for items in searches:
                    if position < len(items):
                        results.append(items[position])
        elif snapshot.request_type == RequestType.movie:
            radarr = RadarrClient(runtime.radarr_base_url, runtime.radarr_api_key)
            if not radarr.configured():
                raise HTTPException(status_code=400, detail="Radarr not configured")
            movie_results = await radarr.search_releases(int(arr_item["id"]))
            if isinstance(movie_results, list):
                results = movie_results
        else:
            raise HTTPException(status_code=400, detail="Unknown request type")
    except HTTPException:
        raise
    except httpx.HTTPStatusError as exc:
        _log_arr_http_error(collector, "interactive release search", exc)
        detail = _format_upstream_error(collector, exc)
        await asyncio.to_thread(
            save_action,
            request_id,
            "search_releases",
            "Search and choose a download",
            "failed",
            detail,
        )
        raise HTTPException(status_code=502, detail=detail) from exc
    except Exception as exc:
        logger.exception("%s interactive release search failed request_id=%s", collector, request_id)
        detail = f"{collector} could not complete the release search: {exc}"
        await asyncio.to_thread(
            save_action,
            request_id,
            "search_releases",
            "Search and choose a download",
            "failed",
            detail,
        )
        raise HTTPException(status_code=502, detail=detail) from exc

    releases = _filter_arr_release_results(results, include_rejected=True)
    approved = sum(not r['requiresOverride'] and r['selectable'] for r in releases)
    source = runtime.sonarr_base_url if collector == 'Sonarr' else runtime.radarr_base_url
    override_allowed = manual_releases.can_override(user)
    for release in releases:
        if release['selectable'] and (not release['requiresOverride'] or override_allowed):
            release['selectionToken'] = manual_releases.issue_selection(release, request_id, user, source, arr_item['id'])
        for key in ('downloadUrl', 'magnetUrl'):
            release.pop(key, None)
    rejection_reasons = sorted({str(reason) for result in results for reason in (result.get('rejections') or [])})[:8]
    message = (f'{len(releases)} releases shown; {approved} meet the assigned profile. Review the reasons on other releases.'
               if releases else 'No releases were returned for the missing content. Try again later or check the indexers.')
    if len(results) > len(releases):
        message += ' Duplicate results are combined; up to 200 ranked releases are shown.'
    if total_missing:
        message += f' Searched {min(3, max(0, total_missing - offset))} of {total_missing} missing monitored episodes.'
    await asyncio.to_thread(save_action, request_id, 'search_releases', 'Search and choose a download', 'ok', message)
    return {'status': 'ok', 'collector': collector, 'qualityFiltered': False, 'message': message,
            'outcome': 'matches' if approved else 'attention', 'rejectionReasons': rejection_reasons,
            'canIgnoreProfileLimits': override_allowed, 'nextOffset': next_offset,
            'totalMissingEpisodes': total_missing, 'releases': releases}


@router.post("/{request_id}/actions/search_auto")
async def action_search_auto(request_id: str, user: Dict[str, str] = Depends(get_current_user)) -> dict:
    if not _user_can_use_search_auto(user):
        raise HTTPException(status_code=403, detail="Auto search and download is disabled for this user")
    runtime = get_runtime_settings()
    client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
    if client.configured():
        await _ensure_request_access(client, int(request_id), user)
    snapshot = await build_snapshot(request_id)
    arr_item = snapshot.raw.get("arr", {}).get("item")
    if not isinstance(arr_item, dict):
        raise HTTPException(status_code=404, detail="Item not found in Sonarr/Radarr")

    if snapshot.request_type.value == "tv":
        client = SonarrClient(runtime.sonarr_base_url, runtime.sonarr_api_key)
        if not client.configured():
            raise HTTPException(status_code=400, detail="Sonarr not configured")
        episodes = await client.get_episodes(int(arr_item["id"]))
        missing_by_season = _missing_episode_ids_by_season(episodes)
        if not missing_by_season:
            message = "No missing monitored episodes found."
            await asyncio.to_thread(
                save_action, request_id, "search_auto", "Search and auto-download", "ok", message
            )
            return {"status": "ok", "message": message, "searched": []}
        responses = []
        for season_number in sorted(missing_by_season.keys()):
            episode_ids = missing_by_season[season_number]
            if episode_ids:
                response = await client.search_episodes(episode_ids)
                responses.append(
                    {"season": season_number, "episodeCount": len(episode_ids), "response": response}
                )
        outcome = await series_search_outcome(client, int(arr_item["id"]), [item['response'] for item in responses])
        message = outcome['message']
        await asyncio.to_thread(
            save_action, request_id, "search_auto", "Search and auto-download", "ok", message
        )
        return {"status": outcome["status"], "message": message, "searched": responses}
    if snapshot.request_type.value == "movie":
        client = RadarrClient(runtime.radarr_base_url, runtime.radarr_api_key)
        if not client.configured():
            raise HTTPException(status_code=400, detail="Radarr not configured")
        response = await client.search(int(arr_item["id"]))
        outcome = await movie_search_outcome(client, int(arr_item["id"]), response)
        message = outcome['message']
        await asyncio.to_thread(
            save_action, request_id, "search_auto", "Search and auto-download", "ok", message
        )
        return {"status": outcome["status"], "message": message, "response": response}

    raise HTTPException(status_code=400, detail="Unknown request type")


@router.post("/{request_id}/actions/qbit/resume")
async def action_resume(request_id: str, user: Dict[str, str] = Depends(get_current_user)) -> dict:
    runtime = get_runtime_settings()
    client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
    if client.configured():
        await _ensure_request_access(client, int(request_id), user)
    snapshot = await build_snapshot(request_id)
    queue = snapshot.raw.get("arr", {}).get("queue")
    download_ids = _download_ids(_queue_records(queue))
    if not download_ids:
        message = "Nothing to force resume."
        await asyncio.to_thread(
            save_action, request_id, "resume_torrent", "Resume torrent", "ok", message
        )
        return {"status": "ok", "message": message}

    runtime = get_runtime_settings()
    client = QBittorrentClient(
        runtime.qbittorrent_base_url,
        runtime.qbittorrent_username,
        runtime.qbittorrent_password,
    )
    if not client.configured():
        raise HTTPException(status_code=400, detail="qBittorrent not configured")

    try:
        torrents = await client.get_torrents_by_hashes("|".join(download_ids))
        torrent_list = torrents if isinstance(torrents, list) else []
        downloading_states = {"downloading", "stalleddl", "queueddl", "checkingdl", "forceddl"}
        if torrent_list and all(
            str(t.get("state", "")).lower() in downloading_states for t in torrent_list
        ):
            message = "No need to force resume. Already downloading."
            await asyncio.to_thread(
                save_action, request_id, "resume_torrent", "Resume torrent", "ok", message
            )
            return {"status": "ok", "message": message}
        await client.resume_torrents("|".join(download_ids))
    except httpx.HTTPStatusError as exc:
        raise HTTPException(status_code=502, detail=str(exc)) from exc
    message = "Resume sent to qBittorrent."
    await asyncio.to_thread(
        save_action, request_id, "resume_torrent", "Resume torrent", "ok", message
    )
    return {"status": "ok", "resumed": download_ids, "message": message}


@router.post("/{request_id}/actions/readd")
async def action_readd(request_id: str, user: Dict[str, str] = Depends(get_current_user)) -> dict:
    runtime = get_runtime_settings()
    client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
    if client.configured():
        await _ensure_request_access(client, int(request_id), user)
    snapshot = await build_snapshot(request_id)
    jelly = snapshot.raw.get("jellyseerr") or {}
    media = jelly.get("media") or {}

    if snapshot.request_type.value == "tv":
        tvdb_id = media.get("tvdbId")
        if not tvdb_id:
            raise HTTPException(status_code=400, detail="Missing tvdbId for series")
        title = snapshot.title
        if title in {None, "", "Unknown"}:
            title = (
                media.get("name")
                or media.get("title")
                or jelly.get("title")
                or jelly.get("name")
            )
        if not runtime.sonarr_quality_profile_id or not runtime.sonarr_root_folder:
            raise HTTPException(status_code=400, detail="Sonarr profile/root not configured")
        client = SonarrClient(runtime.sonarr_base_url, runtime.sonarr_api_key)
        if not client.configured():
            raise HTTPException(status_code=400, detail="Sonarr not configured")
        try:
            existing = await client.get_series_by_tvdb_id(int(tvdb_id))
        except httpx.HTTPStatusError as exc:
            detail = _format_upstream_error("Sonarr", exc)
            await asyncio.to_thread(
                save_action, request_id, "readd_to_arr", "Re-add to Sonarr/Radarr", "failed", detail
            )
            raise HTTPException(status_code=502, detail=detail) from exc
        if isinstance(existing, list) and existing:
            series_id = existing[0].get("id")
            message = f"Already in Sonarr (seriesId {series_id})."
            await asyncio.to_thread(
                save_action, request_id, "readd_to_arr", "Re-add to Sonarr/Radarr", "ok", message
            )
            return {"status": "ok", "message": message, "seriesId": series_id}
        root_folder = await _resolve_root_folder_path(client, runtime.sonarr_root_folder, "Sonarr")
        try:
            response = await client.add_series(
                int(tvdb_id), runtime.sonarr_quality_profile_id, root_folder, title=title
            )
        except httpx.HTTPStatusError as exc:
            detail = _format_upstream_error("Sonarr", exc)
            await asyncio.to_thread(
                save_action, request_id, "readd_to_arr", "Re-add to Sonarr/Radarr", "failed", detail
            )
            raise HTTPException(status_code=502, detail=detail) from exc
        except ValueError as exc:
            detail = f"Sonarr could not add this series: {exc}"
            await asyncio.to_thread(
                save_action, request_id, "readd_to_arr", "Re-add to Sonarr/Radarr", "failed", detail
            )
            raise HTTPException(status_code=502, detail=detail) from exc
        await asyncio.to_thread(
            save_action,
            request_id,
            "readd_to_arr",
            "Re-add to Sonarr/Radarr",
            "ok",
            f"Re-added in Sonarr to {root_folder}.",
        )
        return {"status": "ok", "response": response, "rootFolder": root_folder}

    if snapshot.request_type.value == "movie":
        tmdb_id = media.get("tmdbId")
        if not tmdb_id:
            raise HTTPException(status_code=400, detail="Missing tmdbId for movie")
        if not runtime.radarr_quality_profile_id or not runtime.radarr_root_folder:
            raise HTTPException(status_code=400, detail="Radarr profile/root not configured")
        client = RadarrClient(runtime.radarr_base_url, runtime.radarr_api_key)
        if not client.configured():
            raise HTTPException(status_code=400, detail="Radarr not configured")
        try:
            existing = await client.get_movie_by_tmdb_id(int(tmdb_id))
        except httpx.HTTPStatusError as exc:
            detail = _format_upstream_error("Radarr", exc)
            await asyncio.to_thread(
                save_action, request_id, "readd_to_arr", "Re-add to Sonarr/Radarr", "failed", detail
            )
            raise HTTPException(status_code=502, detail=detail) from exc
        if isinstance(existing, list) and existing:
            movie_id = existing[0].get("id")
            message = f"Already in Radarr (movieId {movie_id})."
            await asyncio.to_thread(
                save_action, request_id, "readd_to_arr", "Re-add to Sonarr/Radarr", "ok", message
            )
            return {"status": "ok", "message": message, "movieId": movie_id}
        root_folder = await _resolve_root_folder_path(client, runtime.radarr_root_folder, "Radarr")
        title = snapshot.title
        if title in {None, "", "Unknown"}:
            title = (
                media.get("title")
                or media.get("name")
                or jelly.get("title")
                or jelly.get("name")
            )
        try:
            response = await client.add_movie(
                int(tmdb_id), runtime.radarr_quality_profile_id, root_folder, title=title
            )
        except httpx.HTTPStatusError as exc:
            detail = _format_upstream_error("Radarr", exc)
            await asyncio.to_thread(
                save_action, request_id, "readd_to_arr", "Re-add to Sonarr/Radarr", "failed", detail
            )
            raise HTTPException(status_code=502, detail=detail) from exc
        except ValueError as exc:
            detail = f"Radarr could not add this movie: {exc}"
            await asyncio.to_thread(
                save_action, request_id, "readd_to_arr", "Re-add to Sonarr/Radarr", "failed", detail
            )
            raise HTTPException(status_code=502, detail=detail) from exc
        await asyncio.to_thread(
            save_action,
            request_id,
            "readd_to_arr",
            "Re-add to Sonarr/Radarr",
            "ok",
            f"Re-added in Radarr to {root_folder}.",
        )
        return {"status": "ok", "response": response, "rootFolder": root_folder}

    raise HTTPException(status_code=400, detail="Unknown request type")


@router.get("/{request_id}/history")
async def request_history(
    request_id: str, limit: int = 10, user: Dict[str, str] = Depends(get_current_user)
) -> dict:
    _require_advanced_request_access(user)
    runtime = get_runtime_settings()
    client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
    if client.configured():
        await _ensure_request_access(client, int(request_id), user)
    snapshots = await asyncio.to_thread(get_recent_snapshots, request_id, limit)
    return {"snapshots": snapshots}


@router.get("/{request_id}/actions")
async def request_actions(
    request_id: str, limit: int = 10, user: Dict[str, str] = Depends(get_current_user)
) -> dict:
    _require_advanced_request_access(user)
    runtime = get_runtime_settings()
    client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
    if client.configured():
        await _ensure_request_access(client, int(request_id), user)
    actions = await asyncio.to_thread(get_recent_actions, request_id, limit)
    return {"actions": actions}


@router.post("/{request_id}/actions/grab")
async def action_grab(
    request_id: str, payload: Dict[str, Any], user: Dict[str, str] = Depends(get_current_user)
) -> dict:
    runtime = get_runtime_settings()
    client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
    if client.configured():
        await _ensure_request_access(client, int(request_id), user)
    snapshot = await build_snapshot(request_id)
    guid = payload.get("guid")
    indexer_id = payload.get("indexerId")
    release_title = payload.get("title")
    if not guid or not indexer_id:
        raise HTTPException(status_code=400, detail="Missing guid or indexerId")
    try:
        arr_indexer_id = int(indexer_id)
    except (TypeError, ValueError) as exc:
        raise HTTPException(status_code=400, detail="indexerId must be an integer") from exc

    logger.info(
        "Collector grab requested: request_id=%s guid=%s indexer_id=%s has_title=%s",
        request_id,
        guid,
        indexer_id,
        bool(release_title),
    )

    if snapshot.request_type.value == "tv":
        arr_client: Any = SonarrClient(runtime.sonarr_base_url, runtime.sonarr_api_key)
        service_label = "Sonarr"
    elif snapshot.request_type.value == "movie":
        arr_client = RadarrClient(runtime.radarr_base_url, runtime.radarr_api_key)
        service_label = "Radarr"
    else:
        raise HTTPException(status_code=400, detail="Unknown request type")

    if not arr_client.configured():
        raise HTTPException(status_code=400, detail=f"{service_label} not configured")

    arr_item = snapshot.raw.get('arr', {}).get('item') or {}
    source = runtime.sonarr_base_url if service_label == 'Sonarr' else runtime.radarr_base_url
    receipt = manual_releases.verify_selection(payload, request_id, user, source, arr_item.get('id'))
    release_title = receipt.get('title')
    arr_error: Optional[str] = None
    try:
        response = await arr_client.grab_release(str(guid), arr_indexer_id)
        action_message = (
            f"{release_title or 'Selected release'} was sent through {service_label} for download and import."
            + (' Profile limits explicitly overridden: ' + '; '.join(receipt['rejections']) if receipt['override'] else '')
        )
        await asyncio.to_thread(
            save_action, request_id, "grab", "Download selected release", "ok", action_message
        )
        return {
            "status": "ok",
            "message": action_message,
            "response": {"collector": service_label, "queued": True},
        }
    except httpx.HTTPStatusError as exc:
        _log_arr_http_error(service_label, "release grab", exc)
        status_code = exc.response.status_code if exc.response is not None else None
        arr_error = _format_upstream_error(service_label, exc)
        if status_code == 404:
            raise HTTPException(409, 'The collector no longer has this release cached. Search again before downloading.') from exc
    except Exception as exc:
        logger.exception("%s release grab failed request_id=%s", service_label, request_id)
        arr_error = str(exc)

    failure_message = (
        f"The selected release could not be started through {service_label}. "
        + (arr_error or f"{service_label} did not accept the release.")
    )
    await asyncio.to_thread(
        save_action, request_id, "grab", "Download selected release", "failed", failure_message
    )
    raise HTTPException(status_code=502, detail=failure_message)


async def refresh_local_request_stages():
    runtime = get_runtime_settings()
    jellyfin = JellyfinClient(runtime.jellyfin_base_url, runtime.jellyfin_api_key)
    if not jellyfin.configured():
        return
    rows = await asyncio.to_thread(get_cached_requests_since, (datetime.now(timezone.utc) - timedelta(days=RECENT_CACHE_MAX_DAYS)).isoformat())
    saved = await asyncio.to_thread(get_request_stage_cache)
    interval = max(1, min(1440, int(runtime.requests_stage_refresh_minutes))) * 60
    now = time.time()
    due = [row for row in rows if row.get('status') == 5 and
           (now - (saved.get(row['request_id']) or {}).get('checked_at', 0) >= interval or
            (saved.get(row['request_id']) or {}).get('source_updated') != row.get('updated_at'))]
    semaphore = asyncio.Semaphore(4)
    async def check(row):
        async with semaphore:
            payload = await asyncio.to_thread(get_request_cache_payload, row['request_id'])
            try:
                ready = await _request_is_available_in_jellyfin(jellyfin, row.get('title'), row.get('year'), row.get('media_type'), payload, {}, raise_errors=True)
                return (row['request_id'], row.get('updated_at'), int(ready), time.time())
            except Exception:
                logger.warning('Local request stage refresh failed request_id=%s', row['request_id'])
                return None
    checked = await asyncio.gather(*(check(row) for row in due))
    await asyncio.to_thread(save_request_stage_cache, [row for row in checked if row is not None])


async def run_local_request_stage_loop():
    while True:
        try:
            await refresh_local_request_stages()
        except Exception:
            logger.exception('Local request stage refresh failed')
        await asyncio.sleep(30)