unshackle/unshackle/core/api/download_manager.py

import asyncio
import hashlib
import json
import logging
import os
import re
import sys
import tempfile
import threading
import uuid
from collections import deque
from contextlib import suppress
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional

from unshackle.core.api.events import bus, publish_service_event
from unshackle.core.api.sanitize import sanitize_log
from unshackle.core.utils.click_types import AUDIO_CODEC_LIST, VIDEO_CODEC_LIST
from unshackle.core.utils.redact import REDACTED, URL_USERINFO_RE, redact_path, redact_text

log = logging.getLogger("download_manager")


# Job parameters may carry secrets (a raw "user:pass" credential, a proxy URL with embedded
# userinfo). These must never leave the process via the API or logs, so they are masked
# wherever parameters are serialized for a response.
SENSITIVE_PARAM_KEYS = ("credential", "credentials", "password", "token", "api_key")


def redact_parameters(parameters: Dict[str, Any]) -> Dict[str, Any]:
    """Return a copy of job parameters with secrets masked, safe to serialise."""
    if not isinstance(parameters, dict):
        return parameters
    redacted = dict(parameters)
    for key in SENSITIVE_PARAM_KEYS:
        if redacted.get(key):
            redacted[key] = REDACTED
    for key in ("proxy", "proxy_download"):
        proxy = redacted.get(key)
        if isinstance(proxy, str) and "@" in proxy:
            redacted[key] = URL_USERINFO_RE.sub(f"{REDACTED}@", proxy)
    return redacted


def secret_values(parameters: Dict[str, Any]) -> List[str]:
    """Raw secret strings carried in job parameters, longest first, for scrubbing free text."""
    if not isinstance(parameters, dict):
        return []
    secrets: List[str] = []
    for key in ("credential", "password", "token", "api_key"):
        value = parameters.get(key)
        if isinstance(value, str) and value:
            secrets.append(value)
            if key == "credential" and ":" in value:
                password = value.split(":", 1)[1]
                if len(password) >= 4:  # short passwords would blanket-replace and garble the text
                    secrets.append(password)
    creds = parameters.get("credentials")
    if isinstance(creds, dict):
        secrets.extend(v for v in creds.values() if isinstance(v, str) and v)
    elif isinstance(creds, str) and creds:
        secrets.append(creds)
    return sorted(set(secrets), key=len, reverse=True)  # longest first so substrings don't survive


def owner_id(api_key: Optional[str]) -> Optional[str]:
    """A short digest of an API key for the on-disk history, so the key itself never lands in a file."""
    if api_key is None:
        return None
    return hashlib.sha256(api_key.encode()).hexdigest()[:16]


def _redact_text(text: Optional[str], parameters: Dict[str, Any]) -> Optional[str]:
    """Mask proxy userinfo and any known parameter secrets that leaked into a free-text field
    (error message / details / traceback / job worker stderr) before the API returns the field."""
    return redact_text(text, secret_values(parameters))


class JobStatus(Enum):
    QUEUED = "queued"
    DOWNLOADING = "downloading"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"


# Statuses a job can never leave; such jobs are safe to remove or retry.
TERMINAL_STATUSES = (JobStatus.COMPLETED, JobStatus.FAILED, JobStatus.CANCELLED)


@dataclass
class DownloadJob:
    """Represents a download job with all its parameters and status."""

    job_id: str
    status: JobStatus
    created_time: datetime
    service: str
    title_id: str
    parameters: Dict[str, Any]

    started_time: Optional[datetime] = None
    completed_time: Optional[datetime] = None
    progress: float = 0.0

    output_files: List[str] = field(default_factory=list)
    error_message: Optional[str] = None
    error_details: Optional[str] = None
    error_code: Optional[str] = None
    error_traceback: Optional[str] = None
    worker_stderr: Optional[str] = None

    phase: Optional[str] = None
    title: Optional[str] = None
    current_title: Optional[str] = None
    completed_tracks: int = 0
    total_tracks: int = 0
    active_tracks: List[str] = field(default_factory=list)
    track_progress: List[Dict[str, Any]] = field(default_factory=list)
    segments_done: float = 0.0
    segments_total: float = 0.0
    speed: Optional[str] = None

    # Prompt a service raised through prompt_user() and waits on; answered by submit_input().
    input_prompt: Optional[str] = None
    # Message a service shows through show_notice() while it waits for an approval; needs no answer.
    notice: Optional[str] = None

    # Subtitles skipped under skip_subtitle_errors (non-fatal). Each entry is a dl.SkippedSubtitle
    # dict (id / language / title) so a client can report which weren't available.
    skipped_subtitles: List[Dict[str, Any]] = field(default_factory=list)

    cancel_event: threading.Event = field(default_factory=threading.Event)

    # Guards against writing the same job to the persistent history file twice.
    history_recorded: bool = False

    # X-Secret-Key that created the job; scopes per-caller access, never serialized into to_dict().
    owner_key: Optional[str] = None

    def to_dict(self, include_full_details: bool = False) -> Dict[str, Any]:
        """Convert job to dictionary for JSON response."""
        result = {
            "job_id": self.job_id,
            "status": self.status.value,
            "created_time": self.created_time.isoformat(),
            "service": self.service,
            "title_id": self.title_id,
            "title": self.title,
            "progress": self.progress,
            "phase": self.phase,
            "current_title": self.current_title,
            "completed_tracks": self.completed_tracks,
            "total_tracks": self.total_tracks,
            "active_tracks": self.active_tracks,
            "track_progress": self.track_progress,
            "segments_done": self.segments_done,
            "segments_total": self.segments_total,
            "speed": self.speed,
            "skipped_subtitles": self.skipped_subtitles,
            "input_prompt": self.input_prompt,
            "notice": self.notice,
        }

        if include_full_details:
            # Error/stderr/traceback are free text a service may have echoed a credential or proxy
            # URL into, so scrub them with the same secrets that redact_parameters masks.
            result.update(
                {
                    "parameters": redact_parameters(self.parameters),
                    "started_time": self.started_time.isoformat() if self.started_time else None,
                    "completed_time": self.completed_time.isoformat() if self.completed_time else None,
                    "output_files": self.output_files,
                    "error_message": redact_path(_redact_text(self.error_message, self.parameters)),
                    "error_details": redact_path(_redact_text(self.error_details, self.parameters)),
                    "error_code": self.error_code,
                    "error_traceback": redact_path(_redact_text(self.error_traceback, self.parameters)),
                    "worker_stderr": redact_path(_redact_text(self.worker_stderr, self.parameters)),
                }
            )

        return result


def history_path() -> Path:
    """Path of the persistent job history file (under the cache dir, kept out of the config/data tree)."""
    from unshackle.core.config import config

    return config.directories.cache / "api_history.jsonl"


def history_limit() -> int:
    """Max history entries to retain (serve.history_limit, default 100). A value of 0 or less means no limit."""
    from unshackle.core.config import config

    try:
        return int((config.serve or {}).get("history_limit", 100))
    except (TypeError, ValueError):
        return 100


_history_appends_since_trim = 0  # module-level; record_job_history runs serialized on the event loop


def record_job_history(job: DownloadJob) -> None:
    """Append a terminal job's summary as one JSON line to the history file (best effort)."""
    global _history_appends_since_trim
    if job.history_recorded or job.status not in TERMINAL_STATUSES:
        return
    job.history_recorded = True
    entry = {
        "job_id": job.job_id,
        "owner": owner_id(job.owner_key),
        "service": job.service,
        "title_id": job.title_id,
        "title": job.title,
        "status": job.status.value,
        "created_time": job.created_time.isoformat(),
        "completed_time": job.completed_time.isoformat() if job.completed_time else None,
        "output_files": job.output_files,
        "parameters": redact_parameters(job.parameters),
        "error_message": _redact_text(job.error_message, job.parameters),
    }
    try:
        path = history_path()
        path.parent.mkdir(parents=True, exist_ok=True)
        line = json.dumps(entry)
        limit = history_limit()
        # Serialized on the manager's event loop, so append + periodic trim doesn't race.
        with open(path, "a", encoding="utf-8") as handle:
            handle.write(line + "\n")
        # Appending is O(1); only pay the full read-rewrite once appends drift past a slack
        # margin, keeping the newest N.
        if limit > 0:
            _history_appends_since_trim += 1
            if _history_appends_since_trim >= max(16, limit // 4):
                _history_appends_since_trim = 0
                existing = [ln for ln in path.read_text(encoding="utf-8").splitlines() if ln.strip()]
                if len(existing) > limit:
                    path.write_text("\n".join(existing[-limit:]) + "\n", encoding="utf-8")  # keep newest N
    except OSError as e:
        log.warning(f"Could not write job history: {e}")


def read_job_history(
    limit: int = 100, service: Optional[str] = None, owner: Optional[str] = None
) -> List[Dict[str, Any]]:
    """Read persisted job history, newest first.

    With ``owner`` (an :func:`owner_id`), only that owner's entries and ownerless legacy
    entries come back. This function ignores corrupt lines. A missing file gives no entries.
    """
    path = history_path()
    if not path.exists():
        return []
    try:
        lines = path.read_text(encoding="utf-8").splitlines()
    except OSError as e:
        log.warning(f"Could not read job history: {e}")
        return []
    entries: List[Dict[str, Any]] = []
    for line in lines:
        line = line.strip()
        if not line:
            continue
        try:
            entry = json.loads(line)
        except json.JSONDecodeError:
            continue
        if not isinstance(entry, dict):
            continue
        if service and str(entry.get("service") or "").upper() != service.upper():
            continue
        if owner is not None and entry.get("owner") not in (None, owner):
            continue
        entries.append(entry)
    entries.reverse()
    return entries[:limit] if limit and limit > 0 else entries


def delete_job_history(job_id: str, allowed: Optional[set] = None, owner: Optional[str] = None) -> bool:
    """Remove a history entry by job_id (rewriting the file). Returns True if it removed one.

    When you give `allowed`, this function treats an entry outside the caller's service
    allowlist as absent (returns False), so a restricted API key cannot delete the entry.
    """
    path = history_path()
    if not path.exists():
        return False
    try:
        lines = path.read_text(encoding="utf-8").splitlines()
    except OSError as e:
        log.warning(f"Could not read job history: {e}")
        return False
    kept: List[str] = []
    deleted = False
    for line in lines:
        stripped = line.strip()
        if not stripped:
            continue
        try:
            entry = json.loads(stripped)
        except json.JSONDecodeError:
            kept.append(stripped)  # preserve corrupt lines untouched
            continue
        match = isinstance(entry, dict) and entry.get("job_id") == job_id
        owned = owner is None or entry.get("owner") in (None, owner)
        if match and owned and (allowed is None or str(entry.get("service") or "").upper() in allowed):
            deleted = True
            continue
        kept.append(stripped)
    if not deleted:
        return False
    try:
        path.write_text("\n".join(kept) + ("\n" if kept else ""), encoding="utf-8")
    except OSError as e:
        log.warning(f"Could not write job history: {e}")
        return False
    return True


def to_enum(values: List[str], enum_cls: type[Enum]) -> List[Enum]:
    """Map case-insensitive strings (by member name or value) to enum members, dropping unknowns."""
    lookup = {m.name.upper(): m for m in enum_cls}
    lookup.update({str(m.value).upper(): m for m in enum_cls})
    return [lookup[v.upper()] for v in values if v.upper() in lookup]


def normalize_sub_format(sub_format_raw: str) -> Any:
    """Normalise an API sub_format string to what dl.py expects.

    "original" is a sentinel meaning "keep source format" (dl.py skips conversion), so it must
    pass through unchanged to match the CLI (SubtitleCodecChoice returns the same string). Any
    other value maps to a Subtitle.Codec, or None when unrecognized.
    """
    from unshackle.core.tracks import Subtitle

    if sub_format_raw.lower() == "original":
        return "original"
    mapped = to_enum([sub_format_raw], Subtitle.Codec)
    return mapped[0] if mapped else None


def perform_download(
    job_id: str,
    service: str,
    title_id: str,
    params: Dict[str, Any],
    progress_callback: Optional[Callable[[Dict[str, Any]], None]] = None,
) -> List[str]:
    """Do the synchronous download work for a job.

    `params` holds the API's string form of the `dl` options. This function normalizes `params` in
    place before dl.result() sees it: vcodec and range names become enums, `slow` accepts "MIN-MAX", a two-item
    list, or True (which means 60-120), and `wanted` accepts forms such as "S01E01", "S01-S03",
    "s1e1", or "1x1" as well as the internal "SxE" form. Music takes track numbers in the same
    field, such as "1-5" or "2x3" for disc 2 track 3.
    """

    from contextlib import redirect_stderr, redirect_stdout
    from io import StringIO

    # Import dl.py components lazily to avoid circular deps during module import
    import click
    import yaml

    from unshackle.commands.dl import dl
    from unshackle.core.api.errors import APIError, APIErrorCode
    from unshackle.core.config import config
    from unshackle.core.services import Services
    from unshackle.core.tracks import Video
    from unshackle.core.utils.click_types import ContextData
    from unshackle.core.utils.collections import merge_dict

    log.info(f"Starting sync download for job {job_id}")

    # A service caches tokens under cache/<Service>/, keyed by service name only, so two jobs on
    # one service with different credentials would share a cache. When a per-job credential is set,
    # namespace the cache dir by a hash of it so the sessions can't cross.
    job_credential = params.get("credential")
    if job_credential:
        import hashlib

        cred_hash = hashlib.pbkdf2_hmac("sha256", job_credential.encode("utf-8"), b"unshackle-job-ns", 100_000).hex()[
            :12
        ]
        config.directories.cache = config.directories.cache / "_jobs" / cred_hash

    for banned in ("postscript", "post_script", "post_scripts"):
        if params.pop(banned, None) is not None:
            log.warning(f"Ignoring '{banned}' in API download parameters; post-scripts are config-only")

    vcodec_raw = params.get("vcodec")
    if vcodec_raw:
        params["vcodec"] = VIDEO_CODEC_LIST.convert(vcodec_raw)
    else:
        params["vcodec"] = []

    if params.get("acodec"):
        params["acodec"] = AUDIO_CODEC_LIST.convert(params["acodec"])

    range_raw = params.get("range")
    if range_raw:
        if isinstance(range_raw, str):
            range_raw = [range_raw]
        if isinstance(range_raw, list) and range_raw and not isinstance(range_raw[0], Video.Range):
            params["range"] = to_enum(range_raw, Video.Range)
    else:
        params["range"] = [Video.Range.SDR]

    sub_format_raw = params.get("sub_format")
    if sub_format_raw and isinstance(sub_format_raw, str):
        params["sub_format"] = normalize_sub_format(sub_format_raw)

    if params.get("export"):
        params["export"] = bool(params["export"])

    slow_raw = params.get("slow")
    if slow_raw is not None and not isinstance(slow_raw, tuple):
        if isinstance(slow_raw, bool):
            params["slow"] = (60, 120) if slow_raw else None
        elif isinstance(slow_raw, list) and len(slow_raw) == 2:
            params["slow"] = (int(slow_raw[0]), int(slow_raw[1]))
        elif isinstance(slow_raw, str):
            from unshackle.core.utils.click_types import SLOW_DELAY_RANGE

            try:
                params["slow"] = SLOW_DELAY_RANGE.convert(slow_raw, None, None)
            except click.BadParameter as exc:
                raise Exception(f"Invalid slow parameter: {exc}")

    wanted_raw = params.get("wanted")
    if wanted_raw:
        from unshackle.core.utils.click_types import SeasonRange

        if isinstance(wanted_raw, str):
            # the CLI splits a comma-separated value before parsing, so a raw string must too
            wanted_raw = re.split(r"\s*[,;]\s*", wanted_raw)
        # the !? keeps a pre-parsed part exclusion from being re-fed through parse_tokens
        needs_conversion = any(not re.match(r"^!?\d+x\d+(\.\d+)?$", w) for w in wanted_raw)
        if needs_conversion:
            season_range = SeasonRange()
            params["wanted"] = season_range.parse_tokens(*wanted_raw)

    service_config_path = Services.get_path(service) / config.filenames.config
    if service_config_path.exists():
        service_config = yaml.safe_load(service_config_path.read_text(encoding="utf8"))
    else:
        service_config = {}
    merge_dict(config.services.get(service), service_config)

    from unshackle.commands.dl import dl as dl_command

    ctx = click.Context(dl_command.cli)
    ctx.invoked_subcommand = service
    from unshackle.core.api.handlers import load_full_cdm

    cdm = load_full_cdm(service, params.get("profile"), params.get("cdm_type"))
    ctx.obj = ContextData(config=service_config, cdm=cdm, proxy_providers=[], profile=params.get("profile"))
    ctx.params = {
        "proxy": params.get("proxy"),
        "no_proxy": params.get("no_proxy", False),
        "no_proxy_download": params.get("no_proxy_download", False),
        "proxy_download": params.get("proxy_download"),
        "profile": params.get("profile"),
        "cdm_name": params.get("cdm"),
        "all_drm": params.get("all_drm", False),
        "repack": params.get("repack", False),
        "tag": params.get("tag"),
        "tmdb_id": params.get("tmdb_id"),
        "imdb_id": params.get("imdb_id"),
        "tvdb_id": params.get("tvdb_id"),
        "tvdb_order": params.get("tvdb_order"),
        "anilist_id": params.get("anilist_id"),
        "enrich": params.get("enrich", False),
        "daily": params.get("daily", False),
        "output_dir": Path(params["output_dir"]) if params.get("output_dir") else None,
        "no_cache": params.get("no_cache", False),
        "reset_cache": params.get("reset_cache", False),
        # Track-selection params read by Service.__init__ via ctx.parent.params
        "quality": params.get("quality", []),
        "vcodec": params.get("vcodec", []),
        "range_": params.get("range", [Video.Range.SDR]),
        "best_available": params.get("best_available", False),
        # services read these here, not from dl.result(), to pick the manifests they fetch
        "acodec": params.get("acodec") or [],
        "lang": params.get("lang", ["orig"]),
        "v_lang": params.get("v_lang", []),
        "a_lang": params.get("a_lang", []),
        "forced_subs": params.get("forced_subs", False),
    }
    # Hand-built context: record parameter sources so service dl overrides
    # apply to defaults but never clobber client-sent values.
    from click.core import ParameterSource

    for param_name in ctx.params:
        source_key = "range" if param_name == "range_" else param_name
        ctx.set_parameter_source(
            param_name, ParameterSource.COMMANDLINE if source_key in params else ParameterSource.DEFAULT
        )

    dl_instance = dl(
        ctx=ctx,
        no_proxy=params.get("no_proxy", False),
        proxy_providers=None if params.get("server_proxy") is True else [],
        profile=params.get("profile"),
        proxy=params.get("proxy"),
        repack=params.get("repack", False),
        tag=params.get("tag"),
        tmdb_id=params.get("tmdb_id"),
        imdb_id=params.get("imdb_id"),
        tvdb_id=params.get("tvdb_id"),
        tvdb_order=params.get("tvdb_order"),
        anilist_id=params.get("anilist_id"),
        enrich=params.get("enrich", False),
        daily=params.get("daily", False),
        output_dir=Path(params["output_dir"]) if params.get("output_dir") else None,
    )
    # Per-request credential ("user:pass"); feed it into the map get_credentials() reads so a
    # client can authenticate without anything being persisted to disk. Without a profile,
    # get_credentials() falls back to "default", so store it there too rather than dropping it
    # (which would silently authenticate as the server's own default account).
    if params.get("credential"):
        svc_creds = config.credentials.get(service)
        if not isinstance(svc_creds, dict):
            config.credentials[service] = svc_creds = {}
        svc_creds[params.get("profile") or "default"] = params["credential"]

    service_module = Services.load(service)

    try:
        import inspect

        service_init_params = inspect.signature(service_module.__init__).parameters

        service_ctx = click.Context(click.Command(service))
        service_ctx.parent = ctx
        service_ctx.obj = ctx.obj

        service_kwargs = {}

        if "title" in service_init_params:
            service_kwargs["title"] = title_id

        for key, value in params.items():
            if key in service_init_params and key not in ["service", "title_id", "profile", "service_params"]:
                service_kwargs[key] = value
        service_params = params.get("service_params")
        if isinstance(service_params, dict):
            for key, value in service_params.items():
                if key in service_init_params and key != "title":
                    service_kwargs[key] = value

        for param_name, param_info in service_init_params.items():
            if param_name not in service_kwargs and param_name not in ["self", "ctx"]:
                if param_info.default is inspect.Parameter.empty:
                    if param_name == "movie":
                        service_kwargs[param_name] = "/movies/" in title_id
                    elif param_name == "meta_lang":
                        service_kwargs[param_name] = None
                    else:
                        log.warning(f"Unknown required parameter '{param_name}' for service {service}, using None")
                        service_kwargs[param_name] = None

        service_instance = service_module(service_ctx, **service_kwargs)

    except Exception as exc:  # noqa: BLE001 - propagate meaningful failure
        log.error(f"Failed to create service instance: {exc}")
        raise

    original_download_dir = config.directories.downloads

    stdout_capture = StringIO()
    stderr_capture = StringIO()

    # progress_sink (dl.build_job_progress_callables) owns percentage; terminal status
    # (success/failure) is reported by the worker's result file.
    try:
        with redirect_stdout(stdout_capture), redirect_stderr(stderr_capture):
            dl_instance.result(
                service=service_instance,
                quality=params.get("quality", []),
                vcodec=params.get("vcodec", []),
                acodec=ctx.params["acodec"],
                vbitrate=params.get("vbitrate"),
                abitrate=params.get("abitrate"),
                vbitrate_range=params.get("vbitrate_range"),
                abitrate_range=params.get("abitrate_range"),
                range_=params.get("range", ["SDR"]),
                channels=params.get("channels"),
                no_atmos=params.get("no_atmos", False),
                select_titles=False,
                wanted=params.get("wanted", []),
                latest_episode=params.get("latest_episode", False),
                latest_episodes=params.get("latest_episodes"),
                lang=ctx.params["lang"],
                v_lang=ctx.params["v_lang"],
                a_lang=ctx.params["a_lang"],
                s_lang=params.get("s_lang", ["all"]),
                require_audio=params.get("require_audio", []),
                require_video=params.get("require_video", []),
                require_subs=params.get("require_subs", []),
                forced_subs=ctx.params["forced_subs"],
                forced_subs_only=params.get("forced_subs_only", False),
                forced_s_lang=params.get("forced_s_lang", []),
                exact_lang=params.get("exact_lang", False),
                sub_format=params.get("sub_format"),
                video_only=params.get("video_only", False),
                audio_only=params.get("audio_only", False),
                subs_only=params.get("subs_only", False),
                chapters_only=params.get("chapters_only", False),
                no_subs=params.get("no_subs", False),
                skip_subtitle_errors=params.get("skip_subtitle_errors", False),
                no_audio=params.get("no_audio", False),
                no_chapters=params.get("no_chapters", False),
                no_video=params.get("no_video", False),
                no_attachments=params.get("no_attachments", False),
                audio_description=params.get("audio_description", False),
                audio_description_only=params.get("audio_description_only", False),
                slow=params.get("slow", None),
                list_=False,
                list_titles=False,
                skip_dl=params.get("skip_dl", False),
                all_drm=params.get("all_drm", False),
                export=params.get("export"),
                cdm_only=params.get("cdm_only"),
                no_proxy=params.get("no_proxy", False),
                no_proxy_download=params.get("no_proxy_download", False),
                proxy_download=params.get("proxy_download"),
                no_folder=params.get("no_folder", False),
                no_source=params.get("no_source", False),
                no_mux=params.get("no_mux", False),
                workers=params.get("workers"),
                adaptive_workers=params.get("adaptive_workers", False),
                download_processes=params.get("download_processes", 1),
                continue_downloads=params.get("continue_downloads", False),
                downloads=params.get("downloads", 1),
                worst=params.get("worst", False),
                best_available=params.get("best_available", False),
                split_audio=params.get("split_audio"),
                progress_sink=progress_callback,
            )

    except SystemExit as exc:
        if exc.code != 0:
            stdout_str = stdout_capture.getvalue()
            stderr_str = stderr_capture.getvalue()
            log.error(f"Download exited with code {exc.code}")
            log.error(f"Stdout: {stdout_str}")
            log.error(f"Stderr: {stderr_str}")
            raise APIError(APIErrorCode.DOWNLOAD_ERROR, f"Download failed with exit code {exc.code}")

    except Exception as exc:  # noqa: BLE001 - propagate to caller
        stdout_str = stdout_capture.getvalue()
        stderr_str = stderr_capture.getvalue()
        log.error(f"Download execution failed: {exc}")
        log.error(f"Stdout: {stdout_str}")
        log.error(f"Stderr: {stderr_str}")
        raise

    # dl.result() catches a download-worker exception, reports it, but returns normally (exit 0).
    # It sets download_failed in that case, so the job isn't reported as completed with no output.
    if getattr(dl_instance, "download_failed", False):
        detail = (stdout_capture.getvalue() + stderr_capture.getvalue())[-200:].strip()
        raise APIError(APIErrorCode.WORKER_ERROR, "download worker failed: " + (detail or "see logs"))

    if progress_callback:
        skipped_subs = getattr(dl_instance, "skipped_subtitles", None)
        if skipped_subs:
            progress_callback({"skipped_subtitles": list(skipped_subs)})

    output_files = [str(p) for p in dl_instance.completed_files]
    log.info(f"Download completed for job {job_id}, {len(output_files)} file(s) in {original_download_dir}")

    return output_files


class DownloadQueueManager:
    """Manages download job queue with configurable concurrency limits."""

    def __init__(self, max_concurrent_downloads: int = 2, job_retention_hours: int = 24):
        self.max_concurrent_downloads = max_concurrent_downloads
        self.job_retention_hours = job_retention_hours

        self._jobs: Dict[str, DownloadJob] = {}
        self._job_queue: asyncio.Queue = asyncio.Queue()
        # asyncio.Queue has no reorder, so prioritized jobs go in a deque workers drain first.
        self._priority_jobs: deque = deque()
        self._promoted_ids: set = set()  # their stale FIFO-queue copies get skipped via this set
        self._active_downloads: Dict[str, asyncio.Task] = {}
        self._download_processes: Dict[str, asyncio.subprocess.Process] = {}
        self._job_temp_files: Dict[str, Dict[str, str]] = {}
        self._workers_started = False
        self._shutdown_event = asyncio.Event()
        self._subscribers: Dict[str, List[asyncio.Queue]] = {}

        log.info(
            f"Initialized download queue manager: max_concurrent={max_concurrent_downloads}, retention_hours={job_retention_hours}"
        )

    def create_job(self, service: str, title_id: str, owner_key: Optional[str] = None, **parameters) -> DownloadJob:
        """Make a new download job and add it to the queue."""
        job_id = str(uuid.uuid4())
        job = DownloadJob(
            job_id=job_id,
            status=JobStatus.QUEUED,
            created_time=datetime.now(),
            service=service,
            title_id=title_id,
            parameters=parameters,
            owner_key=owner_key,
        )

        self._jobs[job_id] = job
        self._job_queue.put_nowait(job)

        log.info(f"Created download job {job_id} for {sanitize_log(service)}:{sanitize_log(title_id)}")
        bus.publish("job", {"event": "queued", **job.to_dict(include_full_details=True)})
        return job

    def get_job(self, job_id: str) -> Optional[DownloadJob]:
        """Get job by ID."""
        return self._jobs.get(job_id)

    def list_jobs(self) -> List[DownloadJob]:
        """Get all jobs."""
        return list(self._jobs.values())

    def subscribe(self, job_id: str) -> asyncio.Queue:
        """Add a listener for a job's events, and return the queue it receives them on."""
        queue: asyncio.Queue = asyncio.Queue(maxsize=100)
        self._subscribers.setdefault(job_id, []).append(queue)
        return queue

    def unsubscribe(self, job_id: str, queue: asyncio.Queue) -> None:
        """Drop a listener registered by subscribe()."""
        queues = self._subscribers.get(job_id)
        if not queues:
            return
        with suppress(ValueError):
            queues.remove(queue)
        if not queues:
            del self._subscribers[job_id]

    def publish(self, job: DownloadJob, event: str) -> None:
        """Fan an event carrying the job's full detail dict out to every listener."""
        bus.publish("job", {"event": event, **job.to_dict(include_full_details=True)})
        queues = self._subscribers.get(job.job_id)
        if not queues:
            return
        payload = {"event": event, "data": job.to_dict(include_full_details=True)}
        for queue in queues:
            self.put(queue, payload)

    def publish_terminal(self, job: DownloadJob) -> None:
        """Publish a job's final state, close every listener's event stream, then apply staged service reloads."""
        self.publish(job, job.status.value)
        for queue in self._subscribers.pop(job.job_id, []):
            self.put(queue, None)
        self.schedule_pending_reload()

    def busy_services(self) -> set[str]:
        """Tags with a job that has not reached a terminal status; queued jobs count so none starts on half-swapped code."""
        return {job.service for job in self._jobs.values() if job.status not in TERMINAL_STATUSES}

    def schedule_pending_reload(self) -> None:
        """Swap in staged service updates whose last job has finished, off the event loop."""
        schedule_pending_reload()

    @staticmethod
    def put(queue: asyncio.Queue, item: Optional[Dict[str, Any]]) -> None:
        """Enqueue an item, discarding the oldest when a slow listener has filled the queue."""
        try:
            queue.put_nowait(item)
        except asyncio.QueueFull:
            with suppress(asyncio.QueueEmpty):
                queue.get_nowait()
            with suppress(asyncio.QueueFull):
                queue.put_nowait(item)

    def cancel_job(self, job_id: str) -> bool:
        """Cancel a job that has the queued or the downloading status."""
        job = self._jobs.get(job_id)
        if not job:
            return False

        if job.status == JobStatus.QUEUED:
            job.status = JobStatus.CANCELLED
            job.cancel_event.set()  # Signal cancellation
            job.completed_time = datetime.now()
            record_job_history(job)  # queued jobs never reach a worker, so persist here
            self.publish_terminal(job)
            log.info(f"Cancelled queued job {sanitize_log(job_id)}")
            return True
        elif job.status == JobStatus.DOWNLOADING:
            job.cancel_event.set()
            job.status = JobStatus.CANCELLED
            log.info(f"Signaled cancellation for downloading job {sanitize_log(job_id)}")

            task = self._active_downloads.get(job_id)
            if task:
                task.cancel()
                log.info(f"Cancelled download task for job {sanitize_log(job_id)}")

            process = self._download_processes.get(job_id)
            if process:
                try:
                    process.terminate()
                    log.info(f"Terminated worker process for job {sanitize_log(job_id)}")
                except ProcessLookupError:
                    log.debug(f"Worker process for job {sanitize_log(job_id)} already exited")

            return True

        return False

    def submit_input(self, job: DownloadJob, response: str) -> bool:
        """Send *response* to the job's worker when it waits on a prompt; ``False`` when no prompt is pending."""
        process = self._download_processes.get(job.job_id)
        if not job.input_prompt or process is None or process.stdin is None:
            return False
        process.stdin.write((json.dumps(response) + "\n").encode("utf-8"))
        # Cleared here, not on the worker's next progress write, so a second submit cannot queue a stale answer.
        job.input_prompt = None
        self.publish(job, "progress")
        return True

    def remove_job(self, job_id: str) -> bool:
        """Remove a terminal (completed/failed/cancelled) job entirely."""
        job = self._jobs.get(job_id)
        if not job or job.status not in TERMINAL_STATUSES:
            return False
        del self._jobs[job_id]
        log.info(f"Removed job {sanitize_log(job_id)}")
        return True

    def clear_finished_jobs(self, owner_key: Optional[str] = None) -> int:
        """Remove terminal (completed/failed/cancelled) jobs, returning the count removed.

        With ``owner_key`` only that key's jobs and ownerless legacy jobs go.
        """
        finished = [
            job_id
            for job_id, job in self._jobs.items()
            if job.status in TERMINAL_STATUSES and (owner_key is None or job.owner_key in (None, owner_key))
        ]
        for job_id in finished:
            del self._jobs[job_id]
        if finished:
            log.info(f"Cleared {len(finished)} finished jobs")
        return len(finished)

    def prioritize_job(self, job_id: str) -> bool:
        """Move a queued job to the front of the line."""
        job = self._jobs.get(job_id)
        if not job or job.status != JobStatus.QUEUED:
            return False
        if job in self._priority_jobs:
            return True
        self._priority_jobs.append(job)
        self._promoted_ids.add(job_id)
        log.info(f"Moved job {sanitize_log(job_id)} to front of queue")
        return True

    def cleanup_old_jobs(self) -> int:
        """Remove jobs older than retention period."""
        cutoff_time = datetime.now() - timedelta(hours=self.job_retention_hours)
        jobs_to_remove = []

        for job_id, job in self._jobs.items():
            if job.status in [JobStatus.COMPLETED, JobStatus.FAILED, JobStatus.CANCELLED]:
                if job.completed_time and job.completed_time < cutoff_time:
                    jobs_to_remove.append(job_id)
                elif not job.completed_time and job.created_time < cutoff_time:
                    jobs_to_remove.append(job_id)

        for job_id in jobs_to_remove:
            del self._jobs[job_id]

        if jobs_to_remove:
            log.info(f"Cleaned up {len(jobs_to_remove)} old jobs")

        return len(jobs_to_remove)

    async def start_workers(self):
        """Start job worker tasks to process the download queue."""
        if self._workers_started:
            return

        self._workers_started = True

        # Start worker tasks
        for i in range(self.max_concurrent_downloads):
            asyncio.create_task(self.download_worker(f"worker-{i}"))

        # Start cleanup task
        asyncio.create_task(self.cleanup_worker())

        log.info(f"Started {self.max_concurrent_downloads} download workers")

    async def shutdown(self):
        """Shutdown the queue manager and cancel all active downloads."""
        log.info("Shutting down download queue manager")
        self._shutdown_event.set()

        for queues in list(self._subscribers.values()):
            for queue in queues:
                self.put(queue, None)
        self._subscribers.clear()

        # Cancel all active downloads
        for task in self._active_downloads.values():
            task.cancel()

        # Terminate worker processes
        for job_id, process in list(self._download_processes.items()):
            try:
                process.terminate()
            except ProcessLookupError:
                log.debug(f"Worker process for job {job_id} already exited during shutdown")

        for job_id, process in list(self._download_processes.items()):
            try:
                await asyncio.wait_for(process.wait(), timeout=5)
            except asyncio.TimeoutError:
                log.warning(f"Worker process for job {job_id} did not exit, killing")
                process.kill()
                await process.wait()
            finally:
                self._download_processes.pop(job_id, None)

        # Clean up any remaining temp files
        for paths in self._job_temp_files.values():
            for path in paths.values():
                try:
                    os.remove(path)
                except OSError:
                    pass
        self._job_temp_files.clear()

        # Wait for workers to finish
        if self._active_downloads:
            await asyncio.gather(*self._active_downloads.values(), return_exceptions=True)

    async def download_worker(self, worker_name: str):
        """Job worker task that processes jobs from the queue."""
        log.debug(f"Download worker {worker_name} started")

        while not self._shutdown_event.is_set():
            try:
                # Prioritized jobs run first; their stale copies still in the FIFO queue are skipped.
                if self._priority_jobs:
                    job = self._priority_jobs.popleft()
                else:
                    job = await asyncio.wait_for(self._job_queue.get(), timeout=1.0)
                    if job.job_id in self._promoted_ids:
                        self._promoted_ids.discard(job.job_id)
                        continue

                if job.status == JobStatus.CANCELLED:
                    continue

                # Start processing the job
                job.status = JobStatus.DOWNLOADING
                job.started_time = datetime.now()
                self.publish(job, "status")

                log.info(f"Worker {worker_name} starting job {job.job_id}")

                # Create download task
                download_task = asyncio.create_task(self.execute_download(job))
                self._active_downloads[job.job_id] = download_task

                try:
                    await download_task
                except asyncio.CancelledError:
                    job.status = JobStatus.CANCELLED
                    log.info(f"Job {job.job_id} was cancelled")
                except Exception as e:
                    job.status = JobStatus.FAILED
                    job.error_message = str(e)
                    log.error(f"Job {job.job_id} failed: {e}")
                finally:
                    job.completed_time = datetime.now()
                    record_job_history(job)
                    self.publish_terminal(job)
                    if job.job_id in self._active_downloads:
                        del self._active_downloads[job.job_id]

            except asyncio.TimeoutError:
                continue
            except Exception as e:
                log.error(f"Worker {worker_name} error: {e}")

    async def execute_download(self, job: DownloadJob):
        """Do the actual download for a job."""
        log.info(f"Executing download for job {job.job_id}")

        try:
            output_files = await self.run_download_async(job)
            job.status = JobStatus.COMPLETED
            job.output_files = output_files
            job.progress = 100.0
            log.info(f"Download completed for job {job.job_id}: {len(output_files)} files")
        except Exception as e:
            import traceback

            from unshackle.core.api.errors import categorize_exception

            job.status = JobStatus.FAILED
            job.error_message = str(e)

            api_error = categorize_exception(
                e, context={"service": job.service, "title_id": job.title_id, "job_id": job.job_id}
            )
            job.error_details = job.error_details or str(e)
            job.error_code = job.error_code or api_error.error_code.value
            job.error_traceback = job.error_traceback or traceback.format_exc()

            log.error(f"Download failed for job {job.job_id}: {e}")
            raise

    async def run_download_async(self, job: DownloadJob) -> List[str]:
        """Invoke a job worker subprocess to do the download."""

        payload = {
            "job_id": job.job_id,
            "service": job.service,
            "title_id": job.title_id,
            "parameters": job.parameters,
        }

        payload_fd, payload_path = tempfile.mkstemp(prefix=f"unshackle_job_{job.job_id}_", suffix="_payload.json")
        os.close(payload_fd)
        result_fd, result_path = tempfile.mkstemp(prefix=f"unshackle_job_{job.job_id}_", suffix="_result.json")
        os.close(result_fd)
        progress_fd, progress_path = tempfile.mkstemp(prefix=f"unshackle_job_{job.job_id}_", suffix="_progress.json")
        os.close(progress_fd)

        with open(payload_path, "w", encoding="utf-8") as handle:
            json.dump(payload, handle)

        process = await asyncio.create_subprocess_exec(
            sys.executable,
            "-m",
            "unshackle.core.api.download_worker",
            payload_path,
            result_path,
            progress_path,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )

        self._download_processes[job.job_id] = process
        self._job_temp_files[job.job_id] = {"payload": payload_path, "result": result_path, "progress": progress_path}

        async def drain() -> tuple[bytes, bytes]:
            """Collect the worker's output; unlike communicate(), leave stdin open for prompt answers."""
            assert process.stdout is not None and process.stderr is not None
            out, err, _ = await asyncio.gather(process.stdout.read(), process.stderr.read(), process.wait())
            return out, err

        communicate_task = asyncio.create_task(drain())

        stdout_bytes = b""
        stderr_bytes = b""
        last_progress_stat: Optional[tuple[int, int]] = None  # (st_mtime_ns, st_size) of last parse
        last_published: Optional[Dict[str, Any]] = None

        try:
            while True:
                done, _ = await asyncio.wait({communicate_task}, timeout=0.5)
                if communicate_task in done:
                    stdout_bytes, stderr_bytes = communicate_task.result()
                    break

                # Check for progress updates (skip re-parse when the file hasn't changed since last tick)
                try:
                    stat = os.stat(progress_path)
                    stat_key = (stat.st_mtime_ns, stat.st_size)
                    if stat_key != last_progress_stat:
                        with open(progress_path, "r", encoding="utf-8") as handle:
                            progress_data = json.load(handle)
                            prompt_before = job.input_prompt
                            if progress_data.get("phase") and progress_data["phase"] != job.phase:
                                job.phase = progress_data["phase"]
                            if progress_data.get("current_title"):
                                job.current_title = str(progress_data["current_title"])
                            if progress_data.get("title"):
                                job.title = str(progress_data["title"])
                            if progress_data.get("output_files"):
                                job.output_files = [str(f) for f in progress_data["output_files"]]
                            if progress_data.get("total_tracks"):
                                job.total_tracks = int(progress_data["total_tracks"])
                            if progress_data.get("completed_tracks") is not None:
                                job.completed_tracks = int(progress_data["completed_tracks"])
                            if "active_tracks" in progress_data:
                                job.active_tracks = list(progress_data["active_tracks"])
                            if "track_progress" in progress_data:
                                job.track_progress = list(progress_data["track_progress"])
                            for _sk in ("segments_done", "segments_total"):
                                if _sk in progress_data:
                                    try:
                                        setattr(job, _sk, float(progress_data[_sk]))
                                    except (TypeError, ValueError):
                                        pass
                            if progress_data.get("speed"):
                                job.speed = str(progress_data["speed"])
                            if "input_prompt" in progress_data:
                                job.input_prompt = progress_data["input_prompt"]
                            if "notice" in progress_data:
                                job.notice = progress_data["notice"]
                            if progress_data.get("skipped_subtitles"):
                                job.skipped_subtitles = progress_data["skipped_subtitles"]
                            if "progress" in progress_data:
                                new_progress = float(progress_data["progress"])
                                if new_progress != job.progress:
                                    job.progress = new_progress
                                    log.info(f"Job {job.job_id} progress updated: {job.progress}%")
                            # submit_input() clears the prompt out of band, so a re-prompt with the same text
                            # leaves the file equal to last_published; publish the prompt change anyway.
                            if progress_data != last_published or job.input_prompt != prompt_before:
                                last_published = progress_data
                                self.publish(job, "progress")
                        last_progress_stat = stat_key
                except (OSError, json.JSONDecodeError, ValueError) as e:
                    log.debug(f"Could not read progress for job {job.job_id}: {e}")

                if job.cancel_event.is_set() or job.status == JobStatus.CANCELLED:
                    log.info(f"Cancellation detected for job {job.job_id}, terminating worker process")
                    process.terminate()
                    try:
                        await asyncio.wait_for(communicate_task, timeout=5)
                    except asyncio.TimeoutError:
                        log.warning(f"Worker process for job {job.job_id} did not terminate, killing")
                        process.kill()
                        await asyncio.wait_for(communicate_task, timeout=5)
                    raise asyncio.CancelledError("Job was cancelled")

            returncode = process.returncode
            stdout = stdout_bytes.decode("utf-8", errors="ignore")
            stderr = stderr_bytes.decode("utf-8", errors="ignore")

            # A service can echo a credential or a proxy URL into its output, so scrub it before
            # it reaches the log as well, not only the API response.
            safe_stdout = _redact_text(stdout.strip(), job.parameters)
            safe_stderr = _redact_text(stderr.strip(), job.parameters)
            if stdout.strip():
                log.debug(f"Worker stdout for job {job.job_id}: {safe_stdout}")
            if stderr.strip():
                job.worker_stderr = stderr.strip()
                if returncode != 0:
                    log.warning(f"Worker stderr for job {job.job_id}: {safe_stderr}")
                else:
                    log.debug(f"Worker stderr for job {job.job_id}: {safe_stderr}")

            result_data: Optional[Dict[str, Any]] = None
            for attempt in range(5):
                try:
                    with open(result_path, "r", encoding="utf-8") as handle:
                        result_data = json.load(handle)
                except FileNotFoundError:
                    log.error(f"Result file missing for job {job.job_id}")
                except PermissionError as exc:
                    log.debug(f"Result file for job {job.job_id} denied, retrying: {exc}")
                    await asyncio.sleep(0.02 * (attempt + 1))
                    continue
                except json.JSONDecodeError as exc:
                    log.error(f"Failed to parse worker result for job {job.job_id}: {exc}")
                break

            if result_data:
                job.error_traceback = result_data.get("traceback")

            if returncode != 0:
                message = result_data.get("message") if result_data else "unknown error"
                if result_data:
                    job.error_details = result_data.get("error_details", message)
                    job.error_code = result_data.get("error_code")
                raise Exception(f"Worker exited with code {returncode}: {message}")

            if not result_data or result_data.get("status") != "success":
                message = result_data.get("message") if result_data else "worker did not report success"
                if result_data:
                    job.error_details = result_data.get("error_details", message)
                    job.error_code = result_data.get("error_code")
                raise Exception(f"Worker failure: {message}")

            return result_data.get("output_files", [])

        finally:
            if not communicate_task.done():
                communicate_task.cancel()
                with suppress(asyncio.CancelledError):
                    await communicate_task

            self._download_processes.pop(job.job_id, None)
            job.input_prompt = None
            job.notice = None

            temp_paths = self._job_temp_files.pop(job.job_id, {})
            for path in temp_paths.values():
                try:
                    os.remove(path)
                except OSError:
                    pass

    async def cleanup_worker(self):
        """Job worker that periodically cleans up old jobs."""
        while not self._shutdown_event.is_set():
            try:
                await asyncio.sleep(3600)
                self.cleanup_old_jobs()
            except Exception as e:
                log.error(f"Cleanup worker error: {e}")


# Global instance
download_manager: Optional[DownloadQueueManager] = None


def get_download_manager() -> DownloadQueueManager:
    """Get the global download manager instance."""
    global download_manager
    if download_manager is None:
        from unshackle.core.config import config

        max_concurrent = getattr(config, "max_concurrent_downloads", 2)
        retention_hours = getattr(config, "download_job_retention_hours", 24)

        download_manager = DownloadQueueManager(max_concurrent, retention_hours)

    return download_manager


def busy_services() -> set[str]:
    """Tags a hot reload must not swap: those with an unfinished job and those a live remote session uses."""
    from unshackle.core.api.session_store import get_session_store
    from unshackle.core.api.stats import stats

    busy = {entry.service_tag for entry in get_session_store().list()}
    if stats.mode != "remote_only":
        busy |= get_download_manager().busy_services()
    return busy


_reload_task: Optional[asyncio.Task[None]] = None


def schedule_pending_reload(tag: Optional[str] = None) -> None:
    """Swap in staged service updates whose last job or remote session has ended, off the event loop.

    Runs on every job completion and remote session removal, because a busy service on a popular
    tag can otherwise stay staged across every periodic refresh tick. With ``tag``, the reload only
    runs when ``PENDING`` holds that tag.
    """
    global _reload_task
    from unshackle.core import services

    if not services.PENDING or (tag is not None and tag not in services.PENDING):
        return
    # apply_pending computes its ready set before it takes RELOAD_LOCK, so concurrent
    # session ends would each re-import the same tag; one in-flight task covers them all.
    if _reload_task is not None and not _reload_task.done():
        return

    async def run() -> None:
        try:
            applied = await asyncio.to_thread(services.apply_pending, busy_services())
            if applied:
                log.info(f"Services reloaded after their jobs and sessions ended: {', '.join(applied)}")
                publish_service_event("applied", applied)
        except Exception:
            log.exception("Applying pending service reloads failed")

    _reload_task = asyncio.get_running_loop().create_task(run())

    def clear(_: asyncio.Task[None]) -> None:
        global _reload_task
        _reload_task = None

    _reload_task.add_done_callback(clear)