chore: bump version

Add ensure_started() to SchedulerService as idempotent entry point.
Start scheduler in auth setup run_initialization() after NFO scan.
Sync anime_directory and start scheduler in config update endpoint.
Add unit and endpoint tests for ensure_started() behavior.

Docker/VERSION

@@ -1 +1 @@
-v1.1.13
+v1.1.16

docs/CHANGELOG.md

@@ -41,6 +41,15 @@ This changelog follows [Keep a Changelog](https://keepachangelog.com/) principle
 
 ### Added
 
+- **Encoding detection for HTML parsing** (`src/core/providers/aniworld_provider.py`):
+  Added `_decode_html_content()` function that uses `chardet` to detect the actual
+  encoding of HTML content before parsing. Falls back to UTF-8 with `errors='replace'`
+  to handle pages with mismatched encoding declarations. Applied to all BeautifulSoup
+  parsing calls to prevent "Some characters could not be decoded" warnings.
+- **chardet dependency**: Added `chardet>=5.2.0` to `requirements.txt` for encoding detection.
+
+### Added
+
 - **Temp file cleanup after every download** (`src/core/providers/aniworld_provider.py`,
   `src/core/providers/enhanced_provider.py`): Module-level helper
   `_cleanup_temp_file()` removes the working temp file and any yt-dlp `.part`

docs/DEVELOPMENT.md

@@ -82,6 +82,22 @@ The download queue prevents duplicate entries at two levels:
 - 5-minute cooldown prevents rapid re-triggers
 - Checked at start of `_auto_download_missing()`
 
+### Episode Lifecycle
+
+Episodes transition through states stored in the `episodes` table:
+
+| State | `is_downloaded` | `file_path` | Description |
+|-------|----------------|-------------|-------------|
+| Missing | `False` | `NULL` | Episode not yet downloaded |
+| Downloaded | `True` | Set | Episode exists on disk |
+
+**State Transitions:**
+1. **Missing → Downloaded**: When download completes, `_remove_episode_from_missing_list()` calls `EpisodeService.mark_downloaded()` to set `is_downloaded=True` and populate `file_path`. The episode record is NOT deleted.
+
+**Query Implications:**
+- `get_series_with_missing_episodes()`: Filters for `is_downloaded=False` to find series with undownloaded episodes
+- `get_series_with_no_episodes()`: Finds series with `is_downloaded=False` episodes but NO `is_downloaded=True` episodes (completely unwatched series)
+
 ### Mocking the Download Queue
 
 When testing components that use the download queue:
@@ -231,6 +247,50 @@ DNS checks are warnings because failures can be transient. anime_directory error
 5. Monitor next run: `GET /health` → `scheduler_next_run`
 6. If problem repeats, increase `misfire_grace_time` in `scheduler_service.py`.
 
+#### Scheduler not firing (no events at scheduled time)
+
+If the scheduler appears configured but never triggers:
+
+1. **Verify scheduler.db contains the job:**
+   ```bash
+   sqlite3 data/scheduler.db "SELECT id, next_run_time FROM apscheduler_jobs;"
+   ```
+   - `next_run_time` should be in the future
+   - If it's in the past, the server was down when the job should have fired
+
+2. **Check application logs for scheduler startup:**
+   ```
+   grep "Scheduler service started" fastapi_app.log
+   ```
+   - If missing, the scheduler failed to start — check for errors above this line
+   - If present, scheduler started successfully
+
+3. **Verify APScheduler events in logs:**
+   ```
+   grep "apscheduler.executors.default" fastapi_app.log
+   ```
+   - `Running job` = job triggered
+   - `executed successfully` = job completed
+   - No output = job never fired
+
+4. **Test manual trigger:**
+   ```bash
+   curl -X POST http://localhost:8000/api/scheduler/trigger-rescan -H "Authorization: Bearer <token>"
+   ```
+   - If manual trigger works but cron doesn't, the issue is APScheduler configuration
+
+5. **Check next_run_time via health endpoint:**
+   ```bash
+   curl http://localhost:8000/health | jq .scheduler_next_run
+   ```
+   - If `null`, the job is not scheduled
+   - If set, the scheduler knows when to run next
+
+6. **Check timezone handling:**
+   - APScheduler uses UTC internally
+   - The schedule_time config (e.g., "03:00") is interpreted as UTC
+   - If you expect local time, adjust the schedule_time accordingly
+
 #### Startup health check failures
 
 If `/health` returns `unhealthy` status:
@@ -248,3 +308,93 @@ If `/health` returns `unhealthy` status:
    - Check network connectivity
    - DNS failures are transient — warnings don't block startup
    - Retry later to verify: `GET /health`
+
+### Provider Failure Handling
+
+Download providers (VOE, Doodstream, Vidmoly, Vidoza, SpeedFiles, Streamtape,
+Luluvdo) regularly break: URLs expire, sites change their player markup, geo
+blocks appear, and `yt-dlp` extractors lag behind upstream changes. The
+`AniworldLoader.download()` flow is designed to fail fast and rotate.
+
+**Rotation order**
+
+1. The episode page is scraped for the providers AniWorld actually advertises.
+2. Results are ordered by the preference in `DEFAULT_PROVIDERS`
+   (`provider_config.py`); providers not listed run last.
+3. For each candidate the loader:
+   1. Calls `_check_url_alive()` — HEAD probe with GET fallback. Any 4xx
+      response or connection error skips the provider immediately.
+   2. Resolves the redirect via `_resolve_direct_link()` to obtain a direct
+      stream URL plus headers. Provider-specific extractors (e.g. `VOE`) are
+      preferred; unknown providers fall back to the embed URL so `yt-dlp` can
+      attempt extraction.
+   3. Tries `_try_direct_stream()` — straight `requests.get(stream=True)` when
+      `Content-Type` is `video/*` or `application/octet-stream`. This avoids
+      `yt-dlp` entirely for direct MP4 links.
+   4. Falls back to `yt-dlp` with the ffmpeg downloader for HLS streams.
+4. On any failure, temp files are cleaned and the loop moves to the next
+   provider. When the chain is exhausted, the loader logs
+   `All download providers failed for S{season}E{episode} ...; tried=[...]`
+   to both the application log and `logs/download_errors.log`.
+
+**Do not hardcode provider URLs.** Provider domains shift constantly (e.g.
+Doodstream alternates between `dood.li`, `dood.so`, `dood.la`). Only the
+referer hints in `PROVIDER_HEADERS` are persisted — discovery still happens
+at runtime through AniWorld's redirect endpoint.
+
+### HLS Stream Handling
+
+HLS (HTTP Live Streaming) manifests (`.m3u8`) require yt-dlp to use the
+`ffmpeg` downloader with `--hls-use-mpegts`. Both providers configure this
+automatically:
+
+```python
+ydl_opts = {
+    "downloader": "ffmpeg",     # Use ffmpeg instead of native
+    "hls_use_mpegts": True,     # Write transport stream (.ts) segments
+}
+```
+
+**Why this matters**: Without ffmpeg, yt-dlp logs:
+`"Live HLS streams are not supported by the native downloader"`
+
+**Requirements**:
+- ffmpeg must be installed and in PATH (`which ffmpeg`)
+- Install: `apt install ffmpeg` (Debian/Ubuntu) or `brew install ffmpeg` (macOS)
+- Startup health check (see Health Check Endpoints) verifies ffmpeg presence
+
+**Trade-offs**:
+- HLS downloads are slower than direct MP4 (reassembly of .ts segments)
+- Requires more disk space during download
+- May need post-processing if .ts format is not desired
+
+**Detection**: VOE provider extracts HLS URLs via `HLS_PATTERN` regex. Other
+providers let yt-dlp auto-detect from URL/content-type.
+
+### Updating yt-dlp
+
+When extractors break (typical symptoms: every provider HEAD probe succeeds
+but `yt-dlp` raises `Unable to extract` or `HTTP Error 404`):
+
+1. Check the upstream tracker first: https://github.com/yt-dlp/yt-dlp/issues
+2. Upgrade in the conda environment:
+   ```bash
+   conda run -n AniWorld pip install --upgrade yt-dlp
+   ```
+3. Smoke-test against a known-good episode before pinning a new floor in
+   `requirements.txt` (`yt-dlp>=YYYY.MM.DD`).
+4. Re-run the provider test suite:
+   ```bash
+   conda run -n AniWorld python -m pytest tests/unit/test_aniworld_provider.py -v
+   ```
+5. If a specific extractor is removed upstream, drop the provider from
+   `DEFAULT_PROVIDERS` rather than patching `yt-dlp` in tree.
+
+### User Notification on Total Failure
+
+`SeriesApp.download_episode()` already emits a `download_status="failed"`
+WebSocket event when `loader.download()` returns `False`. Operators should
+forward this to `notification_service.notify_download_failed()` so users see
+a HIGH-priority alert. The loader keeps the failure detail in
+`logs/download_errors.log` for post-mortem.
+

docs/NFO_GUIDE.md

@@ -171,6 +171,35 @@ Response:
 }
 ```
 
+### 3.6 Fallback Behavior When TMDB is Unavailable
+
+When TMDB lookup fails (network issues, API errors, or no match found), the system creates a **minimal NFO** to ensure the series is still tracked. This behavior applies to:
+
+- Manual NFO creation via API
+- Batch NFO creation operations
+- Automatic NFO creation during downloads
+
+**What a minimal NFO contains:**
+
+```xml
+<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
+<tvshow>
+    <title>Series Name</title>
+    <year>2024</year>
+    <plot>No metadata available for Series Name. TMDB lookup failed.</plot>
+</tvshow>
+```
+
+**Limitations of minimal NFOs:**
+- No poster, logo, or fanart images
+- No rating, genre, or studio information
+- No TMDB or other provider IDs
+- May not display correctly in some media servers
+
+**To upgrade a minimal NFO:**
+1. Use the Update endpoint (`PUT /api/nfo/{serie_id}/update`) when TMDB is available
+2. Or delete the NFO and recreate it with full metadata
+
 ---
 
 ## 4. File Structure

package.json

@@ -1,6 +1,6 @@
 {
   "name": "aniworld-web",
-  "version": "1.1.13",
+  "version": "1.1.16",
   "description": "Aniworld Anime Download Manager - Web Frontend",
   "type": "module",
   "scripts": {

requirements.txt

@@ -22,6 +22,7 @@ APScheduler>=3.10.4
 Events>=0.5
 requests>=2.31.0
 beautifulsoup4>=4.12.0
+chardet>=5.2.0
 fake-useragent>=1.4.0
 yt-dlp>=2024.1.0
 urllib3>=2.0.0

scripts/migrate_populate_year_from_folder.py

@@ -0,0 +1,135 @@
+#!/usr/bin/env python3
+"""Migration script to populate year for existing series from folder names.
+
+This script:
+1. Finds all series in the database with year=NULL
+2. Extracts year from their folder names using the same pattern as SerieScanner
+3. Updates the database records
+
+Usage:
+    python scripts/migrate_populate_year_from_folder.py [--dry-run]
+"""
+
+import argparse
+import re
+import sys
+from pathlib import Path
+
+# Add project root to path
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from sqlalchemy import select, update
+from src.server.database.models import AnimeSeries
+from src.server.database.service import DatabaseSession
+
+
+def extract_year_from_folder_name(folder_name: str) -> int | None:
+    """Extract year from folder name if present.
+    
+    Same logic as SerieScanner._extract_year_from_folder_name.
+    
+    Args:
+        folder_name: The folder name to check
+        
+    Returns:
+        int or None: Year if found, None otherwise
+    """
+    if not folder_name:
+        return None
+    
+    # Look for year in format (YYYY) - typically at end of name
+    match = re.search(r'\((\d{4})\)', folder_name)
+    if match:
+        try:
+            year = int(match.group(1))
+            # Validate year is reasonable (between 1900 and 2100)
+            if 1900 <= year <= 2100:
+                return year
+        except ValueError:
+            pass
+    
+    return None
+
+
+async def migrate_year_from_folder(dry_run: bool = True) -> tuple[int, int]:
+    """Migrate year field for existing series.
+    
+    Args:
+        dry_run: If True, only report what would be changed
+        
+    Returns:
+        Tuple of (updated_count, skipped_count)
+    """
+    updated_count = 0
+    skipped_count = 0
+    
+    async with DatabaseSession() as db:
+        # Find all series with NULL year
+        result = await db.execute(
+            select(AnimeSeries).where(AnimeSeries.year.is_(None))
+        )
+        series_list = result.scalars().all()
+        
+        print(f"Found {len(series_list)} series with year=NULL")
+        
+        for series in series_list:
+            year_from_folder = extract_year_from_folder_name(series.folder)
+            
+            if year_from_folder:
+                print(f"  {series.folder} -> {year_from_folder}")
+                
+                if not dry_run:
+                    await db.execute(
+                        update(AnimeSeries)
+                        .where(AnimeSeries.id == series.id)
+                        .values(year=year_from_folder)
+                    )
+                
+                updated_count += 1
+            else:
+                print(f"  {series.folder} -> (no year found)")
+                skipped_count += 1
+    
+    return updated_count, skipped_count
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Migrate year from folder name")
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        default=True,
+        help="Show what would be changed without making changes"
+    )
+    parser.add_argument(
+        "--execute",
+        action="store_true",
+        help="Actually execute the migration (disabled by default)"
+    )
+    args = parser.parse_args()
+    
+    dry_run = not args.execute
+    
+    if dry_run:
+        print("=== DRY RUN MODE ===")
+        print("No changes will be made. Use --execute to apply changes.\n")
+    
+    import asyncio
+    
+    try:
+        updated, skipped = asyncio.run(migrate_year_from_folder(dry_run=dry_run))
+        
+        print(f"\n{'Would update' if dry_run else 'Updated'}: {updated} series")
+        print(f"Skipped (no year in folder): {skipped} series")
+        
+        if dry_run:
+            print("\nRun with --execute to apply these changes.")
+        
+        return 0
+    except Exception as e:
+        print(f"Error: {e}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/core/providers/aniworld_provider.py

@@ -9,6 +9,7 @@ import threading
 from pathlib import Path
 from urllib.parse import quote
 
+import chardet
 import requests
 from bs4 import BeautifulSoup
 from events import Events
@@ -80,6 +81,37 @@ if not download_error_logger.handlers:
 noKeyFound_logger = logging.getLogger()
 
 
+def _decode_html_content(content: bytes) -> str:
+    """Decode HTML content with encoding detection.
+
+    Uses chardet to detect the actual encoding of the content,
+    falling back to utf-8 with replacement error handling.
+
+    Args:
+        content: Raw HTML bytes from the response
+
+    Returns:
+        Decoded string content
+    """
+    detected = chardet.detect(content)
+    encoding = detected.get('encoding', 'utf-8')
+    confidence = detected.get('confidence', 0)
+
+    if confidence < 0.7:
+        logger.debug(
+            "Low encoding confidence (%.2f) for detected encoding '%s', using utf-8",
+            confidence,
+            encoding
+        )
+        encoding = 'utf-8'
+
+    try:
+        return content.decode(encoding, errors='replace')
+    except Exception as exc:
+        logger.warning("Failed to decode content with %s: %s, using utf-8 replace", encoding, exc)
+        return content.decode('utf-8', errors='replace')
+
+
 class AniworldLoader(Loader):
     def __init__(self) -> None:
         self.SUPPORTED_PROVIDERS = DEFAULT_PROVIDERS
@@ -231,7 +263,7 @@ class AniworldLoader(Loader):
         language_code = self._get_language_key(language)
 
         episode_soup = BeautifulSoup(
-            self._get_episode_html(season, episode, key).content,
+            _decode_html_content(self._get_episode_html(season, episode, key).content),
             'html.parser'
         )
         change_language_box_div = episode_soup.find(
@@ -249,6 +281,118 @@ class AniworldLoader(Loader):
         logger.debug("Available languages for S%02dE%03d: %s, requested: %s, available: %s", season, episode, languages, language_code, is_available)
         return is_available    
 
+    def _check_url_alive(
+        self,
+        url: str,
+        headers: dict | None = None,
+        timeout: int = 10,
+    ) -> bool:
+        """Probe a provider URL with HEAD before committing to yt-dlp.
+
+        Skips dead providers quickly so the failover loop never blocks
+        waiting for yt-dlp to fail on a 404. Falls back to a streaming
+        GET when HEAD is not allowed by the upstream server.
+
+        Args:
+            url: URL to probe.
+            headers: Optional headers to forward with the probe.
+            timeout: Per-request timeout (seconds).
+
+        Returns:
+            True when the URL responds with a non-4xx status, else False.
+        """
+        try:
+            response = self.session.head(
+                url,
+                headers=headers,
+                timeout=timeout,
+                allow_redirects=True,
+            )
+            if response.status_code == 405:
+                response = self.session.get(
+                    url,
+                    headers=headers,
+                    timeout=timeout,
+                    stream=True,
+                    allow_redirects=True,
+                )
+                response.close()
+            if 400 <= response.status_code < 500:
+                logger.warning(
+                    "Provider URL returned HTTP %s: %s",
+                    response.status_code, url
+                )
+                return False
+            return True
+        except requests.RequestException as exc:
+            logger.warning("Provider URL unreachable %s: %s", url, exc)
+            return False
+
+    def _try_direct_stream(
+        self,
+        link: str,
+        output_path: str,
+        headers: dict | None,
+        timeout: int,
+    ) -> bool:
+        """Stream a direct video URL to disk without yt-dlp.
+
+        Used as a fast-path when the resolved provider link already points
+        at a downloadable video file (``Content-Type: video/*`` or
+        ``application/octet-stream``). HLS and other non-video payloads
+        are rejected so the caller can fall back to yt-dlp.
+
+        Args:
+            link: Direct download URL.
+            output_path: Destination file path.
+            headers: Optional HTTP headers.
+            timeout: Per-request timeout (seconds).
+
+        Returns:
+            True on a successful save, False when the link is not a
+            direct video or the download fails.
+        """
+        try:
+            with self.session.get(
+                link,
+                headers=headers,
+                timeout=timeout,
+                stream=True,
+            ) as response:
+                if not response.ok:
+                    logger.debug(
+                        "Direct stream HEAD returned %s for %s",
+                        response.status_code, link[:80]
+                    )
+                    return False
+                content_type = response.headers.get("Content-Type", "")
+                if not (
+                    content_type.startswith("video/")
+                    or content_type == "application/octet-stream"
+                ):
+                    logger.debug(
+                        "Direct stream skipped, Content-Type=%s",
+                        content_type
+                    )
+                    return False
+                logger.info(
+                    "Direct stream download starting (type=%s)",
+                    content_type
+                )
+                with open(output_path, "wb") as fh:
+                    for chunk in response.iter_content(chunk_size=1024 * 1024):
+                        if self._cancel_flag.is_set():
+                            logger.info(
+                                "Cancellation detected during direct stream"
+                            )
+                            return False
+                        if chunk:
+                            fh.write(chunk)
+            return True
+        except requests.RequestException as exc:
+            logger.warning("Direct stream download failed: %s", exc)
+            return False
+
     def download(
         self,
         base_directory: str,
@@ -259,7 +403,12 @@ class AniworldLoader(Loader):
         language: str = "German Dub"
     ) -> bool:
         """Download episode to specified directory.
-        
+
+        Iterates the providers actually advertised on the episode page
+        (ordered by SUPPORTED_PROVIDERS preference), probing each URL
+        before attempting an extraction so dead providers are skipped
+        immediately instead of stalling yt-dlp on a 404.
+
         Args:
             base_directory: Base download directory path
             serie_folder: Filesystem folder name (metadata only, used for
@@ -308,12 +457,78 @@ class AniworldLoader(Loader):
         temp_path = os.path.join(temp_dir, output_file)
         logger.debug("Temporary path: %s", temp_path)
 
-        for provider in self.SUPPORTED_PROVIDERS:
-            logger.debug("Attempting download with provider: %s", provider)
-            link, header = self._get_direct_link_from_provider(
+        candidate_providers = self._select_providers_for_episode(
+            season, episode, key, language
+        )
+        if not candidate_providers:
+            logger.error(
+                "No providers advertised for S%02dE%03d (%s) in %s",
                 season, episode, key, language
             )
-            logger.debug("Direct link obtained from provider")
+            self.clear_cache()
+            return False
+
+        tried: list[str] = []
+        for provider_name, redirect_url in candidate_providers:
+            tried.append(provider_name)
+            logger.debug("Attempting download with provider: %s", provider_name)
+
+            probe_headers = {"User-Agent": self.RANDOM_USER_AGENT}
+            if not self._check_url_alive(
+                redirect_url,
+                headers=probe_headers,
+                timeout=self.DEFAULT_REQUEST_TIMEOUT,
+            ):
+                logger.info(
+                    "Skipping provider %s, redirect URL not reachable",
+                    provider_name
+                )
+                continue
+
+            try:
+                resolved = self._resolve_direct_link(
+                    redirect_url, provider_name
+                )
+            except Exception as exc:
+                logger.warning(
+                    "Provider %s link resolution failed: %s: %s",
+                    provider_name, type(exc).__name__, exc
+                )
+                continue
+
+            if resolved is None:
+                logger.info(
+                    "Provider %s returned no direct link", provider_name
+                )
+                continue
+
+            link, header = resolved
+
+            if self._cancel_flag.is_set():
+                logger.info("Cancellation requested before download start")
+                _cleanup_temp_file(temp_path)
+                self.clear_cache()
+                return False
+
+            if self._try_direct_stream(
+                link,
+                temp_path,
+                header,
+                self.DEFAULT_REQUEST_TIMEOUT,
+            ) and os.path.exists(temp_path):
+                logger.debug(
+                    "Direct stream succeeded with provider %s", provider_name
+                )
+                shutil.copyfile(temp_path, output_path)
+                os.remove(temp_path)
+                logger.info(
+                    "Download completed successfully (direct): %s",
+                    output_file
+                )
+                self.clear_cache()
+                return True
+
+            _cleanup_temp_file(temp_path)
 
             cancel_flag = self._cancel_flag
 
@@ -321,7 +536,6 @@ class AniworldLoader(Loader):
                 if cancel_flag.is_set():
                     logger.info("Cancellation detected in progress hook")
                     raise DownloadCancelled("Download cancelled by user")
-                # Fire the event for progress
                 self.events.download_progress(d)
 
             ydl_opts = {
@@ -333,7 +547,6 @@ class AniworldLoader(Loader):
                 'nocheckcertificate': True,
                 'logger': logger,
                 'progress_hooks': [events_progress_hook],
-                # Use ffmpeg for HLS streams and transport stream format
                 'downloader': 'ffmpeg',
                 'hls_use_mpegts': True,
             }
@@ -343,9 +556,11 @@ class AniworldLoader(Loader):
                 logger.debug("Using custom headers for download")
 
             try:
-                logger.info("Starting download: %s", output_file)
+                logger.info(
+                    "Starting yt-dlp download with %s: %s",
+                    provider_name, output_file
+                )
                 logger.debug("Download link: %s...", link[:100])
-                logger.debug("YDL options: %s", ydl_opts)
 
                 with YoutubeDL(ydl_opts) as ydl:
                     info = ydl.extract_info(link, download=True)
@@ -356,39 +571,151 @@ class AniworldLoader(Loader):
 
                 if os.path.exists(temp_path):
                     logger.debug("Moving file from temp to final destination")
-                    # Use copyfile instead of copy to avoid metadata permission issues
                     shutil.copyfile(temp_path, output_path)
                     os.remove(temp_path)
-                    logger.info("Download completed successfully: %s", output_file)
+                    logger.info(
+                        "Download completed successfully: %s", output_file
+                    )
                     self.clear_cache()
                     return True
-                else:
-                    logger.error("Download failed: temp file not found at %s", temp_path)
-                    self.clear_cache()
-                    return False
-            except BrokenPipeError as e:
                 logger.error(
-                    "Broken pipe error with provider %s: %s. "
-                    "This usually means the stream connection was closed.",
-                    provider, e
+                    "Download failed: temp file not found at %s", temp_path
+                )
+            except DownloadCancelled:
+                logger.info("Download cancelled by user")
+                _cleanup_temp_file(temp_path)
+                self.clear_cache()
+                return False
+            except BrokenPipeError as exc:
+                logger.error(
+                    "Broken pipe error with provider %s: %s",
+                    provider_name, exc
                 )
                 _cleanup_temp_file(temp_path)
                 continue
-            except Exception as e:
+            except Exception as exc:
                 logger.error(
                     "YoutubeDL download failed with provider %s: %s: %s",
-                    provider, type(e).__name__, e
+                    provider_name, type(exc).__name__, exc
                 )
                 _cleanup_temp_file(temp_path)
                 continue
-            break
 
-        # If we get here, all providers failed
-        logger.error("All download providers failed")
+        logger.error(
+            "All download providers failed for S%02dE%03d (%s) in %s. "
+            "Tried: %s. Episode may be unavailable on the source site.",
+            season, episode, key, language, ", ".join(tried) or "none"
+        )
+        download_error_logger.error(
+            "All providers failed for %s S%02dE%03d (%s); tried=%s",
+            key, season, episode, language, tried
+        )
         _cleanup_temp_file(temp_path)
         self.clear_cache()
         return False
 
+    def _select_providers_for_episode(
+        self,
+        season: int,
+        episode: int,
+        key: str,
+        language: str,
+    ) -> list[tuple[str, str]]:
+        """Return ``[(provider_name, redirect_url), ...]`` for an episode.
+
+        Filters by requested language and orders results by
+        ``SUPPORTED_PROVIDERS`` preference so the failover chain matches
+        operator expectations. Returns an empty list when nothing is
+        advertised on the page.
+        """
+        if not self.is_language(season, episode, key, language):
+            logger.warning(
+                "Language %s not advertised for S%02dE%03d (%s)",
+                language, season, episode, key
+            )
+            return []
+        language_code = self._get_language_key(language)
+        providers = self._get_provider_from_html(season, episode, key)
+        ordered: list[tuple[str, str]] = []
+        preferred = list(self.SUPPORTED_PROVIDERS)
+        for name in preferred:
+            lang_map = providers.get(name)
+            if lang_map and language_code in lang_map:
+                ordered.append((name, lang_map[language_code]))
+        for name, lang_map in providers.items():
+            if name in preferred:
+                continue
+            if language_code in lang_map:
+                ordered.append((name, lang_map[language_code]))
+        return ordered
+
+    def _resolve_direct_link(
+        self,
+        redirect_url: str,
+        provider_name: str,
+    ) -> tuple[str, dict] | None:
+        """Resolve a provider redirect URL into a direct stream link.
+
+        Follows the redirect to the embedded player, then delegates to a
+        provider-specific extractor (when registered) or returns the
+        embed URL itself so yt-dlp can attempt extraction.
+
+        Args:
+            redirect_url: AniWorld redirect URL.
+            provider_name: Provider key (e.g. ``"VOE"``).
+
+        Returns:
+            ``(direct_link, headers)`` tuple or None when extraction fails.
+        """
+        try:
+            embedded = self.session.get(
+                redirect_url,
+                timeout=self.DEFAULT_REQUEST_TIMEOUT,
+                headers={"User-Agent": self.RANDOM_USER_AGENT},
+                allow_redirects=True,
+            ).url
+        except requests.RequestException as exc:
+            logger.warning(
+                "Failed resolving redirect for %s: %s", provider_name, exc
+            )
+            return None
+
+        try:
+            extractor = self.Providers.GetProvider(provider_name)
+        except (KeyError, AttributeError):
+            extractor = None
+
+        if extractor is not None:
+            try:
+                return extractor.get_link(
+                    embedded, self.DEFAULT_REQUEST_TIMEOUT
+                )
+            except Exception as exc:
+                logger.warning(
+                    "Custom extractor %s failed: %s",
+                    provider_name, exc
+                )
+                return None
+
+        header_list = self.PROVIDER_HEADERS.get(provider_name)
+        header_dict = self._parse_provider_headers(header_list)
+        return embedded, header_dict
+
+    @staticmethod
+    def _parse_provider_headers(
+        header_list: list | None,
+    ) -> dict[str, str]:
+        """Convert legacy ``"Name: value"`` header strings to a dict."""
+        if not header_list:
+            return {}
+        parsed: dict[str, str] = {}
+        for entry in header_list:
+            if not isinstance(entry, str) or ":" not in entry:
+                continue
+            name, _, value = entry.partition(":")
+            parsed[name.strip()] = value.strip().strip('"')
+        return parsed
+
     def get_site_key(self) -> str:
         """Get the site key for this provider."""
         return "aniworld.to"
@@ -397,7 +724,7 @@ class AniworldLoader(Loader):
         """Get anime title from series key."""
         logger.debug("Getting title for key: %s", key)
         soup = BeautifulSoup(
-            self._get_key_html(key).content,
+            _decode_html_content(self._get_key_html(key).content),
             'html.parser'
         )
         title_div = soup.find('div', class_='series-title')
@@ -428,7 +755,7 @@ class AniworldLoader(Loader):
         logger.debug("Getting year for key: %s", key)
         try:
             soup = BeautifulSoup(
-                self._get_key_html(key).content,
+                _decode_html_content(self._get_key_html(key).content),
                 'html.parser'
             )
             
@@ -542,7 +869,7 @@ class AniworldLoader(Loader):
         """
         logger.debug("Extracting providers from HTML for S%02dE%03d (%s)", season, episode, key)
         soup = BeautifulSoup(
-            self._get_episode_html(season, episode, key).content,
+            _decode_html_content(self._get_episode_html(season, episode, key).content),
             'html.parser'
         )
         providers: dict[str, dict[int, str]] = {}
@@ -665,7 +992,7 @@ class AniworldLoader(Loader):
         base_url = f"{self.ANIWORLD_TO}/anime/stream/{safe_slug}/"
         logger.debug("Base URL: %s", base_url)
         response = requests.get(base_url, timeout=self.DEFAULT_REQUEST_TIMEOUT)
-        soup = BeautifulSoup(response.content, 'html.parser')
+        soup = BeautifulSoup(_decode_html_content(response.content), 'html.parser')
 
         season_meta = soup.find('meta', itemprop='numberOfSeasons')
         number_of_seasons = int(season_meta['content']) if season_meta else 0
@@ -680,7 +1007,7 @@ class AniworldLoader(Loader):
                 season_url,
                 timeout=self.DEFAULT_REQUEST_TIMEOUT,
             )
-            soup = BeautifulSoup(response.content, 'html.parser')
+            soup = BeautifulSoup(_decode_html_content(response.content), 'html.parser')
 
             episode_links = soup.find_all('a', href=True)
             unique_links = set(

src/core/services/nfo_service.py

@@ -20,6 +20,7 @@ from src.core.services.tmdb_client import TMDBAPIError, TMDBClient
 from src.core.utils.image_downloader import ImageDownloader
 from src.core.utils.nfo_generator import generate_tvshow_nfo
 from src.core.utils.nfo_mapper import tmdb_to_nfo_model
+from src.core.entities.nfo_models import TVShowNFO
 
 logger = logging.getLogger(__name__)
 
@@ -423,6 +424,62 @@ class NFOService:
             logger.error("Error parsing NFO file %s: %s", nfo_path, e)
         
         return result
+
+    def parse_nfo_year(self, nfo_path: Path) -> Optional[int]:
+        """Parse year from an existing NFO file.
+        
+        Extracts year from <year> or <premiered> elements.
+        
+        Args:
+            nfo_path: Path to tvshow.nfo file
+            
+        Returns:
+            Year as integer if found, None otherwise.
+            
+        Example:
+            >>> year = nfo_service.parse_nfo_year(Path("/anime/series/tvshow.nfo"))
+            >>> print(year)
+            2013
+        """
+        if not nfo_path.exists():
+            logger.debug("NFO file not found: %s", nfo_path)
+            return None
+        
+        try:
+            tree = etree.parse(str(nfo_path))
+            root = tree.getroot()
+            
+            # Try <year> element first
+            year_elem = root.find(".//year")
+            if year_elem is not None and year_elem.text:
+                try:
+                    year = int(year_elem.text)
+                    if 1900 <= year <= 2100:
+                        logger.debug("Found year in NFO: %d", year)
+                        return year
+                except ValueError:
+                    pass
+            
+            # Fallback: try <premiered> element (format: YYYY-MM-DD)
+            premiered_elem = root.find(".//premiered")
+            if premiered_elem is not None and premiered_elem.text:
+                if premiered_elem.text and len(premiered_elem.text) >= 4:
+                    try:
+                        year = int(premiered_elem.text[:4])
+                        if 1900 <= year <= 2100:
+                            logger.debug("Found year from premiered in NFO: %d", year)
+                            return year
+                    except ValueError:
+                        pass
+            
+            logger.debug("No year found in NFO: %s", nfo_path)
+            
+        except etree.XMLSyntaxError as e:
+            logger.error("Invalid XML in NFO file %s: %s", nfo_path, e)
+        except Exception as e:  # pylint: disable=broad-except
+            logger.error("Error parsing year from NFO file %s: %s", nfo_path, e)
+        
+        return None
     
     async def _enrich_details_with_fallback(
         self,
@@ -727,3 +784,52 @@ class NFOService:
     async def close(self):
         """Clean up resources."""
         await self.tmdb_client.close()
+    
+    async def create_minimal_nfo(
+        self,
+        serie_name: str,
+        serie_folder: str,
+        year: Optional[int] = None
+    ) -> Path:
+        """Create minimal tvshow.nfo when TMDB lookup fails.
+        
+        Creates a basic NFO with just the title (and year if available)
+        so the series is tracked even without TMDB metadata.
+        
+        Args:
+            serie_name: Name of the series (may include year in parentheses)
+            serie_folder: Series folder name
+            year: Optional release year
+            
+        Returns:
+            Path to created NFO file
+            
+        Raises:
+            FileNotFoundError: If series folder doesn't exist
+        """
+        # Extract year from name if not provided
+        clean_name, extracted_year = self._extract_year_from_name(serie_name)
+        if year is None and extracted_year is not None:
+            year = extracted_year
+        
+        folder_path = self.anime_directory / serie_folder
+        if not folder_path.exists():
+            logger.info("Creating series folder: %s", folder_path)
+            folder_path.mkdir(parents=True, exist_ok=True)
+        
+        # Create minimal NFO model with just title and year
+        nfo_model = TVShowNFO(
+            title=clean_name,
+            year=year,
+            plot=f"No metadata available for {clean_name}. TMDB lookup failed."
+        )
+        
+        # Generate XML
+        nfo_xml = generate_tvshow_nfo(nfo_model)
+        
+        # Save NFO file
+        nfo_path = folder_path / "tvshow.nfo"
+        nfo_path.write_text(nfo_xml, encoding="utf-8")
+        logger.info("Created minimal NFO (no TMDB): %s", nfo_path)
+        
+        return nfo_path

src/server/api/auth.py

@@ -163,6 +163,22 @@ async def setup_auth(req: SetupRequest):
                 # Perform NFO scan if configured
                 await perform_nfo_scan_if_needed(progress_service)
                 
+                # Start scheduler if anime_directory is now set
+                try:
+                    from src.server.services.scheduler_service import (
+                        get_scheduler_service,
+                    )
+
+                    scheduler_svc = get_scheduler_service()
+                    logger.info("Starting scheduler after initialization")
+                    await scheduler_svc.ensure_started()
+                    logger.info("Scheduler started successfully during setup")
+                except Exception as sched_exc:
+                    logger.warning(
+                        "Failed to start scheduler during setup: %s", sched_exc
+                    )
+                    # Continue — scheduler failure should not break initialization
+                
                 # Send completion event
                 from src.server.services.progress_service import ProgressType
                 await progress_service.start_progress(

src/server/api/config.py

@@ -1,7 +1,10 @@
+import logging
 from typing import Any, Dict, List, Optional
 
 from fastapi import APIRouter, Depends, HTTPException, status
 
+logger = logging.getLogger(__name__)
+
 from src.server.models.config import AppConfig, ConfigUpdate, ValidationResult
 from src.server.services.config_service import (
     ConfigBackupError,
@@ -28,16 +31,53 @@ def get_config(auth: Optional[dict] = Depends(require_auth)) -> AppConfig:
 
 
 @router.put("", response_model=AppConfig)
-def update_config(
+async def update_config(
     update: ConfigUpdate, auth: dict = Depends(require_auth)
 ) -> AppConfig:
     """Apply an update to the configuration and persist it.
 
-    Creates automatic backup before applying changes.
+    Creates automatic backup before applying changes. If anime_directory
+    is configured, starts the scheduler service.
     """
     try:
         config_service = get_config_service()
-        return config_service.update_config(update)
+        updated_config = config_service.update_config(update)
+
+        # Sync anime_directory to settings if it was updated
+        from src.config.settings import settings as app_settings
+
+        anime_dir_changed = False
+        if update.other and update.other.get("anime_directory"):
+            anime_dir = update.other.get("anime_directory")
+            if anime_dir and not app_settings.anime_directory:
+                app_settings.anime_directory = str(anime_dir)
+                anime_dir_changed = True
+                logger.info("Synced anime_directory from config: %s", anime_dir)
+
+        # Start scheduler if anime_directory was just configured
+        if anime_dir_changed:
+            try:
+                from src.server.services.scheduler_service import (
+                    get_scheduler_service,
+                )
+
+                scheduler_svc = get_scheduler_service()
+                logger.info(
+                    "Starting scheduler after anime_directory configuration"
+                )
+                await scheduler_svc.ensure_started()
+                logger.info(
+                    "Scheduler started successfully after config update"
+                )
+            except Exception as sched_exc:
+                logger.warning(
+                    "Failed to start scheduler after config update: %s",
+                    sched_exc,
+                )
+                # Config was already saved, don't fail the request
+
+        return updated_config
+
     except ConfigValidationError as e:
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,

src/server/api/nfo.py

@@ -144,6 +144,27 @@ async def batch_create_nfo(
                     nfo_path=str(nfo_path)
                 )
                 
+            except TMDBAPIError as e:
+                logger.warning("TMDB API error for %s, creating minimal fallback: %s", serie_id, e)
+                # TMDB failed, create minimal NFO
+                try:
+                    serie_folder = serie.ensure_folder_with_year()
+                except Exception:
+                    serie_folder = serie_folder
+                
+                serie_name = serie.name or serie_folder
+                nfo_path = await nfo_service.create_minimal_nfo(
+                    serie_name=serie_name,
+                    serie_folder=serie_folder
+                )
+                
+                return NFOBatchResult(
+                    serie_id=serie_id,
+                    serie_folder=serie_folder,
+                    success=True,
+                    message="Created minimal NFO (TMDB lookup failed)",
+                    nfo_path=str(nfo_path)
+                )
             except Exception as e:
                 logger.error(
                     f"Error creating NFO for {serie_id}: {e}",
@@ -429,11 +450,42 @@ async def create_nfo(
     except HTTPException:
         raise
     except TMDBAPIError as e:
-        logger.warning("TMDB API error creating NFO for %s: %s", serie_id, e)
-        raise HTTPException(
-            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
-            detail=f"TMDB API error: {str(e)}"
-        ) from e
+        logger.warning("TMDB API error for %s, creating minimal fallback: %s", serie_id, e)
+        # TMDB failed, create minimal NFO with just folder name
+        try:
+            serie_folder = serie.ensure_folder_with_year()
+        except Exception:
+            serie_folder = serie_folder
+        
+        folder_path = Path(settings.anime_directory) / serie_folder
+        serie_name_fallback = request.serie_name or serie.name or serie_folder
+        
+        nfo_path = await nfo_service.create_minimal_nfo(
+            serie_name=serie_name_fallback,
+            serie_folder=serie_folder,
+            year=year
+        )
+        
+        # Check media files (will likely be empty)
+        media_status = check_media_files(folder_path)
+        file_paths = get_media_file_paths(folder_path)
+        
+        media_files = MediaFilesStatus(
+            has_poster=media_status.get("poster", False),
+            has_logo=media_status.get("logo", False),
+            has_fanart=media_status.get("fanart", False),
+            poster_path=str(file_paths["poster"]) if file_paths.get("poster") else None,
+            logo_path=str(file_paths["logo"]) if file_paths.get("logo") else None,
+            fanart_path=str(file_paths["fanart"]) if file_paths.get("fanart") else None
+        )
+        
+        return NFOCreateResponse(
+            serie_id=serie_id,
+            serie_folder=serie_folder,
+            nfo_path=str(nfo_path),
+            media_files=media_files,
+            message="Created minimal NFO (TMDB lookup failed)"
+        )
     except Exception as e:
         logger.error(
             f"Error creating NFO for {serie_id}: {e}",

src/server/database/models.py

@@ -316,6 +316,7 @@ class DownloadQueueItem(Base, TimestampMixin):
         id: Primary key
         series_id: Foreign key to AnimeSeries
         episode_id: Foreign key to Episode
+        status: Queue status (pending/downloading/completed/failed/permanently_failed)
         error_message: Error description if failed
         download_url: Provider download URL
         file_destination: Target file path
@@ -347,12 +348,29 @@ class DownloadQueueItem(Base, TimestampMixin):
         index=True
     )
     
-    # Unique constraint to prevent duplicate pending queue items
-    # An episode can only have one queue entry at a time
+    # Status column to track queue item state
+    # Allows distinguishing pending items from permanently failed ones
+    status: Mapped[str] = mapped_column(
+        String(50), nullable=False, default="pending",
+        doc="Queue item status: pending, downloading, completed, failed, permanently_failed"
+    )
+    
+    # Retry count to track failed download attempts
+    # Used to determine when to move item to permanently_failed
+    retry_count: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0,
+        doc="Number of retry attempts for this download"
+    )
+    
+    # Unique constraint to prevent duplicate pending queue items per episode
+    # An episode can only have one PENDING entry at a time
+    # The status column allows failed items to remain in DB while new
+    # pending items can be added (application-level dedup still required)
     __table_args__ = (
         Index(
-            "ix_download_queue_episode_pending",
+            "ix_download_queue_episode_status",
             "episode_id",
+            "status",
             unique=True,
         ),
     )

src/server/database/service.py

@@ -541,6 +541,7 @@ class EpisodeService:
         db: AsyncSession,
         series_id: int,
         season: Optional[int] = None,
+        only_missing: bool = False,
     ) -> List[Episode]:
         """Get episodes for a series.
         
@@ -548,6 +549,9 @@ class EpisodeService:
             db: Database session
             series_id: Foreign key to AnimeSeries
             season: Optional season filter
+            only_missing: If True, only return episodes where
+                is_downloaded is False (i.e., missing episodes).
+                Default False returns all episodes.
         
         Returns:
             List of Episode instances
@@ -557,6 +561,9 @@ class EpisodeService:
         if season is not None:
             query = query.where(Episode.season == season)
         
+        if only_missing:
+            query = query.where(Episode.is_downloaded == False)
+        
         query = query.order_by(Episode.season, Episode.episode_number)
         result = await db.execute(query)
         return list(result.scalars().all())
@@ -748,6 +755,8 @@ class DownloadQueueService:
         episode_id: int,
         download_url: Optional[str] = None,
         file_destination: Optional[str] = None,
+        status: str = "pending",
+        retry_count: int = 0,
     ) -> DownloadQueueItem:
         """Add item to download queue.
         
@@ -757,6 +766,8 @@ class DownloadQueueService:
             episode_id: Foreign key to Episode
             download_url: Optional provider download URL
             file_destination: Optional target file path
+            status: Queue item status (default: "pending")
+            retry_count: Number of retry attempts (default: 0)
         
         Returns:
             Created DownloadQueueItem instance
@@ -766,13 +777,15 @@ class DownloadQueueService:
             episode_id=episode_id,
             download_url=download_url,
             file_destination=file_destination,
+            status=status,
+            retry_count=retry_count,
         )
         db.add(item)
         await db.flush()
         await db.refresh(item)
         logger.info(
             f"Added to download queue: episode_id={episode_id} "
-            f"for series_id={series_id}"
+            f"for series_id={series_id}, status={status}"
         )
         return item
     
@@ -799,21 +812,24 @@ class DownloadQueueService:
     async def get_by_episode(
         db: AsyncSession,
         episode_id: int,
+        status_filter: Optional[str] = None,
     ) -> Optional[DownloadQueueItem]:
         """Get download queue item by episode ID.
         
         Args:
             db: Database session
             episode_id: Foreign key to Episode
+            status_filter: Optional status to filter by (e.g., "pending")
         
         Returns:
             DownloadQueueItem instance or None if not found
         """
-        result = await db.execute(
-            select(DownloadQueueItem).where(
-                DownloadQueueItem.episode_id == episode_id
-            )
+        query = select(DownloadQueueItem).where(
+            DownloadQueueItem.episode_id == episode_id
         )
+        if status_filter:
+            query = query.where(DownloadQueueItem.status == status_filter)
+        result = await db.execute(query)
         return result.scalar_one_or_none()
     
     @staticmethod
@@ -873,6 +889,95 @@ class DownloadQueueService:
         logger.debug("Set error on download queue item %s", item_id)
         return item
     
+    @staticmethod
+    async def set_status(
+        db: AsyncSession,
+        item_id: int,
+        status: str,
+    ) -> Optional[DownloadQueueItem]:
+        """Set status on download queue item.
+        
+        Args:
+            db: Database session
+            item_id: Item primary key
+            status: New status value
+        
+        Returns:
+            Updated DownloadQueueItem instance or None if not found
+        """
+        item = await DownloadQueueService.get_by_id(db, item_id)
+        if not item:
+            return None
+        
+        item.status = status
+        
+        await db.flush()
+        await db.refresh(item)
+        logger.debug("Set status on download queue item %s to %s", item_id, status)
+        return item
+    
+    @staticmethod
+    async def increment_retry_count(
+        db: AsyncSession,
+        item_id: int,
+    ) -> Optional[DownloadQueueItem]:
+        """Increment retry count on download queue item.
+        
+        Args:
+            db: Database session
+            item_id: Item primary key
+        
+        Returns:
+            Updated DownloadQueueItem instance or None if not found
+        """
+        item = await DownloadQueueService.get_by_id(db, item_id)
+        if not item:
+            return None
+        
+        item.retry_count += 1
+        
+        await db.flush()
+        await db.refresh(item)
+        logger.debug(
+            "Incremented retry count on download queue item %s to %s",
+            item_id, item.retry_count
+        )
+        return item
+    
+    @staticmethod
+    async def set_status_and_error(
+        db: AsyncSession,
+        item_id: int,
+        status: str,
+        error_message: Optional[str] = None,
+    ) -> Optional[DownloadQueueItem]:
+        """Set status and error message on download queue item atomically.
+        
+        Args:
+            db: Database session
+            item_id: Item primary key
+            status: New status value
+            error_message: Optional error description
+        
+        Returns:
+            Updated DownloadQueueItem instance or None if not found
+        """
+        item = await DownloadQueueService.get_by_id(db, item_id)
+        if not item:
+            return None
+        
+        item.status = status
+        if error_message is not None:
+            item.error_message = error_message
+        
+        await db.flush()
+        await db.refresh(item)
+        logger.debug(
+            "Set status=%s on download queue item %s, error=%s",
+            status, item_id, error_message
+        )
+        return item
+    
     @staticmethod
     async def delete(db: AsyncSession, item_id: int) -> bool:
         """Delete download queue item.

src/server/fastapi_app.py

@@ -121,7 +121,7 @@ async def _run_startup_health_checks(logger) -> dict:
     import asyncio
     import shutil
     import socket
-    from typing import Dict, Any
+    from typing import Any, Dict
     
     checks: Dict[str, Any] = {
         "ffmpeg": {"status": "unknown", "message": None},
@@ -400,13 +400,15 @@ async def lifespan(_application: FastAPI):
 
                 # Initialize and start scheduler service
                 try:
+                    logger.info("Initializing scheduler service...")
                     from src.server.services.scheduler_service import (
                         get_scheduler_service,
                     )
                     scheduler_service = get_scheduler_service()
+                    logger.info("Scheduler service instance obtained, starting...")
                     await scheduler_service.start()
                     initialized['scheduler'] = True
-                    logger.info("Scheduler service started")
+                    logger.info("Scheduler service started successfully")
                 except Exception as e:
                     logger.warning("Failed to start scheduler service: %s", e)
                     # Continue - scheduler is optional

src/server/models/download.py

@@ -22,6 +22,7 @@ class DownloadStatus(str, Enum):
     COMPLETED = "completed"
     FAILED = "failed"
     CANCELLED = "cancelled"
+    PERMANENTLY_FAILED = "permanently_failed"
 
 
 class DownloadPriority(str, Enum):

src/server/services/anime_service.py

@@ -498,13 +498,19 @@ class AnimeService:
                 logger.info("No series found in SeriesApp")
                 return []
             
-            # Build NFO metadata map and filter data from database
-            nfo_map = {}
-            series_with_no_episodes = set()
+            # Build NFO metadata map, episode dict, and filter data from database.
+            # Using DB as authoritative source for episodeDict ensures that
+            # episodes marked is_downloaded=True are never shown as missing,
+            # even if the in-memory state is stale.
+            nfo_map: dict = {}
+            db_episode_dict_map: dict[str, dict[int, list[int]]] = {}
+            series_with_no_episodes: set = set()
             
             async with get_db_session() as db:
-                # Get all series NFO metadata using service layer
-                db_series_list = await AnimeSeriesService.get_all(db)
+                # Single query: load all series with their episodes eagerly
+                db_series_list = await AnimeSeriesService.get_all(
+                    db, with_episodes=True
+                )
                 
                 for db_series in db_series_list:
                     nfo_created = (
@@ -523,6 +529,20 @@ class AnimeService:
                         "tvdb_id": db_series.tvdb_id,
                         "series_id": db_series.id,
                     }
+                    
+                    # Build episodeDict from DB, skipping is_downloaded=True
+                    # episodes so they are never shown as missing in the UI.
+                    ep_dict: dict[int, list[int]] = {}
+                    if db_series.episodes:
+                        for ep in db_series.episodes:
+                            if ep.is_downloaded:
+                                continue
+                            if ep.season not in ep_dict:
+                                ep_dict[ep.season] = []
+                            ep_dict[ep.season].append(ep.episode_number)
+                        for s in ep_dict:
+                            ep_dict[s].sort()
+                    db_episode_dict_map[db_series.folder] = ep_dict
                 
                 # If filter is "missing_episodes", get series with any missing episodes
                 if filter_type == "missing_episodes":
@@ -545,7 +565,12 @@ class AnimeService:
                 name = getattr(serie, "name", "")
                 site = getattr(serie, "site", "")
                 folder = getattr(serie, "folder", "")
-                episode_dict = getattr(serie, "episodeDict", {}) or {}
+                # Use DB-backed episodeDict (is_downloaded=True already filtered out)
+                # with in-memory episodeDict as fallback if the series isn't in DB yet.
+                episode_dict = db_episode_dict_map.get(
+                    folder,
+                    getattr(serie, "episodeDict", {}) or {}
+                )
 
                 # Apply filter if specified
                 if filter_type == "missing_episodes":
@@ -815,18 +840,24 @@ class AnimeService:
         - Adds new missing episodes that are not in the database
         - Removes episodes from database that are no longer missing
           (i.e., the file has been added to the filesystem)
+        - Preserves episodes marked as downloaded (is_downloaded=True)
+          so download history is not lost
         """
         from src.server.database.service import AnimeSeriesService, EpisodeService
 
-        # Get existing episodes from database
+        # Get existing episodes from database (all episodes, including downloaded)
         existing_episodes = await EpisodeService.get_by_series(db, existing.id)
         
         # Build dict of existing episodes: {season: {ep_num: episode_id}}
+        # and track which ones are already downloaded
         existing_dict: dict[int, dict[int, int]] = {}
+        downloaded_set: set[tuple[int, int]] = set()
         for ep in existing_episodes:
             if ep.season not in existing_dict:
                 existing_dict[ep.season] = {}
             existing_dict[ep.season][ep.episode_number] = ep.id
+            if ep.is_downloaded:
+                downloaded_set.add((ep.season, ep.episode_number))
         
         # Get new missing episodes from scan
         new_dict = serie.episodeDict or {}
@@ -857,9 +888,22 @@ class AnimeService:
         
         # Remove episodes from database that are no longer missing
         # (i.e., the episode file now exists on the filesystem)
+        # BUT: preserve episodes that are already downloaded (is_downloaded=True)
+        # so we don't lose download history
         for season, eps_dict in existing_dict.items():
             for ep_num, episode_id in eps_dict.items():
                 if (season, ep_num) not in new_missing_set:
+                    # Skip already-downloaded episodes — they should stay in DB
+                    # with is_downloaded=True to preserve download history
+                    if (season, ep_num) in downloaded_set:
+                        logger.debug(
+                            "Preserving downloaded episode in database: "
+                            "%s S%02dE%02d",
+                            serie.key,
+                            season,
+                            ep_num
+                        )
+                        continue
                     await EpisodeService.delete(db, episode_id)
                     logger.info(
                         "Removed episode from database (no longer missing): "
@@ -889,6 +933,10 @@ class AnimeService:
         
         This method is called during initialization and after rescans
         to ensure the in-memory series list is in sync with the database.
+        
+        Only episodes where is_downloaded=False are loaded into the
+        in-memory episodeDict, so downloaded episodes are not shown
+        as missing.
         """
         from src.core.entities.series import Serie
         from src.server.database.connection import get_db_session
@@ -903,9 +951,14 @@ class AnimeService:
             series_list = []
             for anime_series in anime_series_list:
                 # Build episode_dict from episodes relationship
+                # Only include episodes that are NOT downloaded (is_downloaded=False)
+                # so the missing-episode list stays accurate
                 episode_dict: dict[int, list[int]] = {}
                 if anime_series.episodes:
                     for episode in anime_series.episodes:
+                        # Skip downloaded episodes — they are not missing
+                        if episode.is_downloaded:
+                            continue
                         season = episode.season
                         if season not in episode_dict:
                             episode_dict[season] = []
@@ -919,7 +972,8 @@ class AnimeService:
                     name=anime_series.name,
                     site=anime_series.site,
                     folder=anime_series.folder,
-                    episodeDict=episode_dict
+                    episodeDict=episode_dict,
+                    year=anime_series.year
                 )
                 series_list.append(serie)
             
@@ -962,23 +1016,39 @@ class AnimeService:
                 logger.warning("Series not found in database: %s", series_key)
                 return 0
             
-            # Get existing episodes from database
+            # Get existing episodes from database (all, including downloaded)
             existing_episodes = await EpisodeService.get_by_series(db, series_db.id)
             
             # Build dict of existing episodes: {season: {ep_num: episode_id}}
+            # and track which ones are already downloaded
             existing_dict: dict[int, dict[int, int]] = {}
+            downloaded_set: set[tuple[int, int]] = set()
             for ep in existing_episodes:
                 if ep.season not in existing_dict:
                     existing_dict[ep.season] = {}
                 existing_dict[ep.season][ep.episode_number] = ep.id
+                if ep.is_downloaded:
+                    downloaded_set.add((ep.season, ep.episode_number))
             
             # Get new missing episodes from in-memory serie
             new_dict = serie.episodeDict or {}
             
             # Add new missing episodes that are not in the database
+            # Skip episodes that are already downloaded (is_downloaded=True)
+            # so we don't re-add them as missing after they've been downloaded
             for season, episode_numbers in new_dict.items():
                 existing_season_eps = existing_dict.get(season, {})
                 for ep_num in episode_numbers:
+                    # Skip if already downloaded — don't re-add as missing
+                    if (season, ep_num) in downloaded_set:
+                        logger.debug(
+                            "Skipping already-downloaded episode: "
+                            "%s S%02dE%02d",
+                            series_key,
+                            season,
+                            ep_num,
+                        )
+                        continue
                     if ep_num not in existing_season_eps:
                         await EpisodeService.create(
                             db=db,
@@ -1014,20 +1084,23 @@ class AnimeService:
         if hasattr(self._app, 'list') and hasattr(self._app.list, 'keyDict'):
             serie = self._app.list.keyDict.get(series_key)
             if serie:
-                # Convert episode dict keys to strings for JSON
-                missing_episodes = {str(k): v for k, v in (serie.episodeDict or {}).items()}
-                total_missing = sum(len(eps) for eps in missing_episodes.values())
-                
-                # Fetch NFO metadata from database
+                # Fetch NFO metadata and episodes from database.
+                # Using DB as the authoritative source for missing_episodes
+                # ensures that episodes marked is_downloaded=True are never
+                # broadcast as missing, even if in-memory state is stale.
                 has_nfo = False
                 nfo_created_at = None
                 nfo_updated_at = None
                 tmdb_id = None
                 tvdb_id = None
+                missing_episodes: dict[str, list] = {}
                 
                 try:
                     from src.server.database.connection import get_db_session
-                    from src.server.database.service import AnimeSeriesService
+                    from src.server.database.service import (
+                        AnimeSeriesService,
+                        EpisodeService,
+                    )
                     
                     async with get_db_session() as db:
                         db_series = await AnimeSeriesService.get_by_key(db, series_key)
@@ -1043,12 +1116,31 @@ class AnimeService:
                             )
                             tmdb_id = db_series.tmdb_id
                             tvdb_id = db_series.tvdb_id
+                            
+                            # Build missing_episodes from DB, skipping is_downloaded=True
+                            db_episodes = await EpisodeService.get_by_series(
+                                db, db_series.id, only_missing=True
+                            )
+                            for ep in db_episodes:
+                                key_str = str(ep.season)
+                                if key_str not in missing_episodes:
+                                    missing_episodes[key_str] = []
+                                missing_episodes[key_str].append(ep.episode_number)
+                            for s in missing_episodes:
+                                missing_episodes[s].sort()
                 except Exception as e:
                     logger.warning(
-                        "Could not fetch NFO data for %s: %s",
+                        "Could not fetch series data for %s from DB: %s",
                         series_key,
                         str(e)
                     )
+                    # Fallback to in-memory state
+                    missing_episodes = {
+                        str(k): v
+                        for k, v in (serie.episodeDict or {}).items()
+                    }
+                
+                total_missing = sum(len(eps) for eps in missing_episodes.values())
                 
                 series_data = {
                     "key": serie.key,
@@ -1550,6 +1642,7 @@ async def sync_series_from_data_files(
                         name=serie.name,
                         site=serie.site,
                         folder=serie.folder,
+                        year=serie.year if hasattr(serie, 'year') else None,
                     )
                     
                     # Create Episode records for each episode in episodeDict

src/server/services/download_service.py

@@ -14,6 +14,7 @@ import uuid
 from collections import deque
 from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime, timezone
+from pathlib import Path
 from typing import TYPE_CHECKING, Dict, List, Optional
 
 import structlog
@@ -68,6 +69,7 @@ class DownloadService:
             progress_service: Optional progress service for tracking
         """
         self._anime_service = anime_service
+        self._directory = anime_service._directory
         self._max_retries = max_retries
         self._progress_service = progress_service or get_progress_service()
         
@@ -168,6 +170,27 @@ class DownloadService:
             logger.error("Failed to save item to database: %s", e)
             return item
     
+    async def _set_status_in_database(
+        self,
+        item_id: str,
+        status: str,
+    ) -> bool:
+        """Set status on an item in the database.
+        
+        Args:
+            item_id: Download item ID
+            status: New status value
+            
+        Returns:
+            True if update succeeded
+        """
+        try:
+            repository = self._get_repository()
+            return await repository.set_status(item_id, status)
+        except Exception as e:
+            logger.error("Failed to set status in database: %s", e)
+            return False
+    
     async def _set_error_in_database(
         self,
         item_id: str,
@@ -189,6 +212,25 @@ class DownloadService:
             logger.error("Failed to set error in database: %s", e)
             return False
     
+    async def _increment_retry_in_database(
+        self,
+        item_id: str,
+    ) -> bool:
+        """Increment retry count on an item in the database.
+        
+        Args:
+            item_id: Download item ID
+            
+        Returns:
+            True if update succeeded
+        """
+        try:
+            repository = self._get_repository()
+            return await repository.increment_retry(item_id)
+        except Exception as e:
+            logger.error("Failed to increment retry in database: %s", e)
+            return False
+    
     async def _delete_from_database(self, item_id: str) -> bool:
         """Delete an item from the database.
         
@@ -210,30 +252,33 @@ class DownloadService:
         series_key: str,
         season: int,
         episode: int,
+        serie_folder: Optional[str] = None,
     ) -> bool:
-        """Remove a downloaded episode from the missing episodes list.
+        """Mark a downloaded episode as downloaded instead of deleting it.
         
         Called when a download completes successfully to update both:
-        1. The database (Episode record deleted)
+        1. The database (Episode record marked is_downloaded=True)
         2. The in-memory Serie.episodeDict and series_list cache
         
         This ensures the episode no longer appears as missing in both
-        the API responses and the UI immediately after download.
+        the API responses and the UI immediately after download,
+        while preserving the download history.
         
         Args:
             series_key: Unique provider key for the series
             season: Season number
             episode: Episode number within season
+            serie_folder: Series folder name (required for file_path)
             
         Returns:
-            True if episode was removed, False otherwise
+            True if episode was updated, False otherwise
         """
         try:
             from src.server.database.connection import get_db_session
-            from src.server.database.service import EpisodeService
+            from src.server.database.service import AnimeSeriesService, EpisodeService
             
             logger.info(
-                "Attempting to remove missing episode from DB: "
+                "Attempting to mark episode as downloaded in DB: "
                 "%s S%02dE%02d",
                 series_key,
                 season,
@@ -241,28 +286,63 @@ class DownloadService:
             )
             
             async with get_db_session() as db:
-                deleted = await EpisodeService.delete_by_series_and_episode(
+                # Get series by key to find series_id
+                series = await AnimeSeriesService.get_by_key(db, series_key)
+                if not series:
+                    logger.warning(
+                        "Series not found for key: %s", series_key
+                    )
+                    return False
+                
+                # Get episode by series_id, season, episode_number
+                ep = await EpisodeService.get_by_episode(
                     db=db,
-                    series_key=series_key,
+                    series_id=series.id,
                     season=season,
                     episode_number=episode,
                 )
-                if deleted:
+                if not ep:
+                    logger.warning(
+                        "Episode not found in DB: %s S%02dE%02d",
+                        series_key,
+                        season,
+                        episode,
+                    )
+                    return False
+                
+                # Construct file_path if serie_folder provided
+                file_path = None
+                if serie_folder:
+                    season_folder = f"Season {season}"
+                    file_path = str(
+                        Path(self._directory) / serie_folder / season_folder
+                    )
+                
+                # Mark episode as downloaded instead of deleting
+                updated = await EpisodeService.mark_downloaded(
+                    db=db,
+                    episode_id=ep.id,
+                    file_path=file_path or "",
+                )
+                
+                if updated:
                     logger.info(
-                        "Successfully removed episode from DB missing list: "
+                        "Marked episode as downloaded in DB: "
+                        "%s S%02dE%02d, file_path=%s",
+                        series_key,
+                        season,
+                        episode,
+                        file_path,
+                    )
+                else:
+                    logger.warning(
+                        "Failed to mark episode as downloaded: "
                         "%s S%02dE%02d",
                         series_key,
                         season,
                         episode,
                     )
-                else:
-                    logger.warning(
-                        "Episode not found in DB missing list "
-                        "(may already be removed): %s S%02dE%02d",
-                        series_key,
-                        season,
-                        episode,
-                    )
+                    return False
             
             # Update in-memory Serie.episodeDict so list_missing is
             # immediately consistent without a full DB reload
@@ -273,8 +353,8 @@ class DownloadService:
             try:
                 self._anime_service._cached_list_missing.cache_clear()
                 logger.debug(
-                    "Cleared list_missing cache after removing "
-                    "%s S%02dE%02d",
+                    "Cleared list_missing cache after marking "
+                    "%s S%02dE%02d as downloaded",
                     series_key,
                     season,
                     episode,
@@ -282,10 +362,35 @@ class DownloadService:
             except Exception:
                 pass
             
-            return deleted
+            # Broadcast real-time update to frontend so the series card
+            # immediately reflects the new downloaded state (no longer
+            # shows the episode as missing) without waiting for a full
+            # reload on DOWNLOAD_COMPLETED.
+            try:
+                await self._anime_service._broadcast_series_updated(
+                    series_key
+                )
+                logger.debug(
+                    "Broadcast series_updated after marking "
+                    "%s S%02dE%02d as downloaded",
+                    series_key,
+                    season,
+                    episode,
+                )
+            except Exception as broadcast_exc:
+                logger.warning(
+                    "Failed to broadcast series update after marking "
+                    "%s S%02dE%02d as downloaded: %s",
+                    series_key,
+                    season,
+                    episode,
+                    broadcast_exc,
+                )
+            
+            return True
         except Exception as e:
             logger.error(
-                "Failed to remove episode from missing list: "
+                "Failed to mark episode as downloaded: "
                 "%s S%02dE%02d - %s",
                 series_key,
                 season,
@@ -1011,17 +1116,15 @@ class DownloadService:
                 if item.retry_count >= self._max_retries:
                     continue
                 
-                # Move back to pending
+                # Move back to pending (retry_count will be incremented
+                # by _process_download when the item fails again)
                 self._failed_items.remove(item)
                 item.status = DownloadStatus.PENDING
-                item.retry_count += 1
                 item.error = None
                 item.progress = None
                 self._add_to_pending_queue(item)
                 retried_ids.append(item.id)
                 
-                # Status is now managed in-memory only
-                
                 logger.info(
                     "Retrying failed item: item_id=%s, retry_count=%d",
                     item.id,
@@ -1029,18 +1132,23 @@ class DownloadService:
                 )
             
             if retried_ids:
-                # Notify via progress service
-                queue_status = await self.get_queue_status()
-                await self._progress_service.update_progress(
-                    progress_id="download_queue",
-                    message=f"Retried {len(retried_ids)} failed items",
-                    metadata={
-                        "action": "items_retried",
-                        "retried_ids": retried_ids,
-                        "queue_status": queue_status.model_dump(mode="json"),
-                    },
-                    force_broadcast=True,
-                )
+                # Notify via progress service if available
+                try:
+                    queue_status = await self.get_queue_status()
+                    await self._progress_service.update_progress(
+                        progress_id="download_queue",
+                        message=f"Retried {len(retried_ids)} failed items",
+                        metadata={
+                            "action": "items_retried",
+                            "retried_ids": retried_ids,
+                            "queue_status": queue_status.model_dump(mode="json"),
+                        },
+                        force_broadcast=True,
+                    )
+                except Exception as e:
+                    logger.warning(
+                        "Failed to broadcast retry progress: %s", e
+                    )
             
             return retried_ids
             
@@ -1119,12 +1227,13 @@ class DownloadService:
                 # Delete completed item from download queue database
                 await self._delete_from_database(item.id)
                 
-                # Remove episode from missing episodes list
+                # Mark episode as downloaded in missing episodes list
                 # (both database and in-memory)
                 removed = await self._remove_episode_from_missing_list(
                     series_key=item.serie_id,
                     season=item.episode.season,
                     episode=item.episode.episode,
+                    serie_folder=item.serie_folder,
                 )
                 
                 logger.info(
@@ -1179,17 +1288,35 @@ class DownloadService:
             item.status = DownloadStatus.FAILED
             item.completed_at = datetime.now(timezone.utc)
             item.error = str(e)
+            
+            # Increment retry count in memory and database
+            item.retry_count += 1
+            await self._increment_retry_in_database(item.id)
+            
             self._failed_items.append(item)
             
             # Set error in database
             await self._set_error_in_database(item.id, str(e))
             
-            logger.error(
-                "Download failed: item_id=%s, error=%s, retry_count=%d",
-                item.id,
-                str(e),
-                item.retry_count,
-            )
+            # Check if max retries exceeded - move to dead-letter
+            if item.retry_count >= self._max_retries:
+                await self._set_status_in_database(
+                    item.id, DownloadStatus.PERMANENTLY_FAILED.value
+                )
+                logger.error(
+                    "Download permanently failed after max retries: "
+                    "item_id=%s, error=%s, retry_count=%d",
+                    item.id,
+                    str(e),
+                    item.retry_count,
+                )
+            else:
+                logger.error(
+                    "Download failed: item_id=%s, error=%s, retry_count=%d",
+                    item.id,
+                    str(e),
+                    item.retry_count,
+                )
             # Note: Failure is already broadcast by AnimeService
             # via ProgressService when SeriesApp fires failed event
         

src/server/services/folder_scan_service.py

@@ -28,6 +28,36 @@ _POSTER_DOWNLOAD_SEMAPHORE: asyncio.Semaphore = asyncio.Semaphore(3)
 _NFO_REPAIR_SEMAPHORE: asyncio.Semaphore = asyncio.Semaphore(3)
 
 
+async def _create_missing_nfo(series_dir: Path, series_name: str) -> None:
+    """Create minimal NFO for series without one.
+
+    Creates a fresh :class:`NFOService` per invocation so concurrent
+    tasks cannot interfere with each other.
+
+    A module-level semaphore limits concurrent TMDB operations to 3.
+
+    Args:
+        series_dir: Absolute path to the series folder.
+        series_name: Human-readable series name for log messages.
+    """
+    from src.core.services.nfo_factory import NFOServiceFactory
+
+    async with _NFO_REPAIR_SEMAPHORE:
+        try:
+            factory = NFOServiceFactory()
+            nfo_service = factory.create()
+            await nfo_service.create_minimal_nfo(
+                serie_name=series_name,
+                serie_folder=series_dir.name,
+            )
+        except Exception as exc:  # pylint: disable=broad-except
+            logger.error(
+                "NFO creation failed for %s: %s",
+                series_name,
+                exc,
+            )
+
+
 async def _repair_one_series(series_dir: Path, series_name: str) -> None:
     """Repair a single series NFO in isolation.
 
@@ -63,12 +93,13 @@ async def _repair_one_series(series_dir: Path, series_name: str) -> None:
 
 
 async def perform_nfo_repair_scan(background_loader=None) -> None:
-    """Scan all series folders and repair incomplete tvshow.nfo files.
+    """Scan all series folders, repair incomplete and create missing NFO files.
 
     Called from ``FolderScanService.run_folder_scan()`` during the scheduled
     daily folder scan (not on every startup).  Checks each subfolder of
-    ``settings.anime_directory`` for a ``tvshow.nfo`` and calls
-    ``_repair_one_series`` for every file with absent or empty required tags.
+    ``settings.anime_directory`` for a ``tvshow.nfo``:
+    - Missing NFOs: creates minimal NFO via ``_create_missing_nfo``
+    - Incomplete NFOs: repairs via ``_repair_one_series``
 
     Each repair task creates its own isolated :class:`NFOService` /
     :class:`TMDBClient` so concurrent tasks never share an ``aiohttp``
@@ -97,26 +128,33 @@ async def perform_nfo_repair_scan(background_loader=None) -> None:
 
     queued = 0
     total = 0
+    missing_nfo_count = 0
     for series_dir in sorted(anime_dir.iterdir()):
         if not series_dir.is_dir():
             continue
         nfo_path = series_dir / "tvshow.nfo"
+        series_name = series_dir.name
         if not nfo_path.exists():
+            # Create minimal NFO for series without one
+            missing_nfo_count += 1
+            asyncio.create_task(
+                _create_missing_nfo(series_dir, series_name),
+                name=f"nfo_create:{series_name}",
+            )
             continue
         total += 1
-        series_name = series_dir.name
         if nfo_needs_repair(nfo_path):
             queued += 1
-            # Each task creates its own NFOService so connectors are isolated.
             asyncio.create_task(
                 _repair_one_series(series_dir, series_name),
                 name=f"nfo_repair:{series_name}",
             )
 
     logger.info(
-        "NFO repair scan complete: %d of %d series queued for repair",
+        "NFO repair scan complete: %d of %d series queued for repair, %d missing NFOs queued for creation",
         queued,
         total,
+        missing_nfo_count,
     )
 
 

src/server/services/queue_repository.py

@@ -83,15 +83,12 @@ class QueueRepository:
     ) -> DownloadItem:
         """Convert database model to DownloadItem.
 
-        Note: Since the database model is simplified, status, priority,
-        progress, and retry_count default to initial values.
-
         Args:
             db_item: SQLAlchemy download queue item
             item_id: Optional override for item ID
 
         Returns:
-            Pydantic download item with default status/priority
+            Pydantic download item with status/retry_count from database
         """
         # Get episode info from the related Episode object
         episode = db_item.episode
@@ -109,14 +106,14 @@ class QueueRepository:
             serie_folder=series.folder if series else "",
             serie_name=series.name if series else "",
             episode=episode_identifier,
-            status=DownloadStatus.PENDING,  # Default - managed in-memory
-            priority=DownloadPriority.NORMAL,  # Default - managed in-memory
+            status=DownloadStatus(db_item.status),  # From database
+            priority=DownloadPriority.NORMAL,  # Managed in-memory
             added_at=db_item.created_at or datetime.now(timezone.utc),
             started_at=db_item.started_at,
             completed_at=db_item.completed_at,
             progress=None,  # Managed in-memory
             error=db_item.error_message,
-            retry_count=0,  # Managed in-memory
+            retry_count=db_item.retry_count,  # From database
             source_url=db_item.download_url,
         )
 
@@ -350,6 +347,110 @@ class QueueRepository:
         finally:
             if manage_session:
                 await session.close()
+    
+    async def set_status(
+        self,
+        item_id: str,
+        status: str,
+        db: Optional[AsyncSession] = None,
+    ) -> bool:
+        """Set status on a download item.
+
+        Args:
+            item_id: Download item ID
+            status: New status value
+            db: Optional existing database session
+
+        Returns:
+            True if update succeeded, False if item not found
+
+        Raises:
+            QueueRepositoryError: If update fails
+        """
+        session = db or self._db_session_factory()
+        manage_session = db is None
+
+        try:
+            result = await DownloadQueueService.set_status(
+                session,
+                int(item_id),
+                status,
+            )
+
+            if manage_session:
+                await session.commit()
+
+            success = result is not None
+
+            if success:
+                logger.debug(
+                    "Set status on queue item: item_id=%s, status=%s",
+                    item_id,
+                    status,
+                )
+
+            return success
+
+        except ValueError:
+            return False
+        except Exception as e:
+            if manage_session:
+                await session.rollback()
+            logger.error("Failed to set status: %s", e)
+            raise QueueRepositoryError(f"Failed to set status: {e}") from e
+        finally:
+            if manage_session:
+                await session.close()
+    
+    async def increment_retry(
+        self,
+        item_id: str,
+        db: Optional[AsyncSession] = None,
+    ) -> bool:
+        """Increment retry count on a download item.
+
+        Args:
+            item_id: Download item ID
+            db: Optional existing database session
+
+        Returns:
+            True if update succeeded, False if item not found
+
+        Raises:
+            QueueRepositoryError: If update fails
+        """
+        session = db or self._db_session_factory()
+        manage_session = db is None
+
+        try:
+            result = await DownloadQueueService.increment_retry_count(
+                session,
+                int(item_id),
+            )
+
+            if manage_session:
+                await session.commit()
+
+            success = result is not None
+
+            if success:
+                logger.debug(
+                    "Incremented retry count on queue item: item_id=%s",
+                    item_id,
+                )
+
+            return success
+
+        except ValueError:
+            return False
+        except Exception as e:
+            if manage_session:
+                await session.rollback()
+            logger.error("Failed to increment retry: %s", e)
+            raise QueueRepositoryError(f"Failed to increment retry: {e}") from e
+        finally:
+            if manage_session:
+                await session.close()
 
     async def delete_item(
         self,

src/server/services/scheduler_service.py

@@ -10,10 +10,10 @@ cron time), the job is triggered immediately within a grace period.
 """
 from __future__ import annotations
 
+import logging
 from datetime import datetime, timezone
 from typing import List, Optional
 
-import structlog
 from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from apscheduler.triggers.cron import CronTrigger
@@ -21,7 +21,7 @@ from apscheduler.triggers.cron import CronTrigger
 from src.server.models.config import SchedulerConfig
 from src.server.services.config_service import ConfigServiceError, get_config_service
 
-logger = structlog.get_logger(__name__)
+logger = logging.getLogger(__name__)
 
 _JOB_ID = "scheduled_rescan"
 
@@ -69,15 +69,18 @@ class SchedulerService:
             SchedulerServiceError: If the scheduler is already running or
                 config cannot be loaded.
         """
+        logger.info("SchedulerService.start() called")
         if self._is_running:
+            logger.warning("Scheduler start called but already running")
             raise SchedulerServiceError("Scheduler is already running")
 
         try:
             config_service = get_config_service()
             config = config_service.load_config()
             self._config = config.scheduler
+            logger.info("Scheduler config loaded successfully")
         except ConfigServiceError as exc:
-            logger.error("Failed to load scheduler configuration", error=str(exc))
+            logger.error("Failed to load scheduler configuration: %s", exc)
             raise SchedulerServiceError(f"Failed to load config: {exc}") from exc
 
         jobstores = {
@@ -90,6 +93,15 @@ class SchedulerService:
             self._is_running = True
             return
 
+        logger.info(
+            "Scheduler config loaded: enabled=%s time=%s days=%s auto_download=%s folder_scan=%s",
+            self._config.enabled,
+            self._config.schedule_time,
+            self._config.schedule_days,
+            self._config.auto_download_after_rescan,
+            self._config.folder_scan_enabled,
+        )
+
         trigger = self._build_cron_trigger()
         if trigger is None:
             logger.warning(
@@ -105,9 +117,9 @@ class SchedulerService:
                 coalesce=True,
             )
             logger.info(
-                "Scheduler started with cron trigger",
-                schedule_time=self._config.schedule_time,
-                schedule_days=self._config.schedule_days,
+                "Scheduler started with cron trigger: time=%s days=%s",
+                self._config.schedule_time,
+                self._config.schedule_days,
             )
 
         self._scheduler.start()
@@ -116,15 +128,18 @@ class SchedulerService:
         # Startup recovery: if the server was down at the scheduled time and
         # the job is within the misfire window, APScheduler will run it
         # automatically.  Log the scheduled time for visibility.
+        # Note: next_run_time is only available AFTER scheduler.start()
         job = self._scheduler.get_job(_JOB_ID)
-        if job and job.next_run_time:
+        if job:
+            next_run = job.next_run_time
             logger.info(
-                "Scheduler next run",
-                next_run=job.next_run_time.isoformat(),
+                "Scheduler next run: %s",
+                next_run.isoformat() if next_run else None,
             )
 
     async def stop(self) -> None:
         """Stop the APScheduler gracefully."""
+        logger.info("SchedulerService.stop() called")
         if not self._is_running:
             logger.debug("Scheduler stop called but not running")
             return
@@ -132,8 +147,27 @@ class SchedulerService:
         if self._scheduler and self._scheduler.running:
             self._scheduler.shutdown(wait=False)
             logger.info("Scheduler stopped")
+        else:
+            logger.info("Scheduler stop: scheduler was not running")
 
         self._is_running = False
+        logger.info("SchedulerService stopped successfully")
+
+    async def ensure_started(self) -> None:
+        """Ensure the scheduler is running (idempotent).
+
+        If already running, returns immediately. Otherwise, starts the scheduler.
+        This method is safe to call multiple times and from multiple callers.
+
+        Raises:
+            SchedulerServiceError: If startup fails (except for already running).
+        """
+        if self._is_running:
+            logger.debug("Scheduler ensure_started called but already running")
+            return
+
+        logger.info("Scheduler ensure_started: starting scheduler")
+        await self.start()
 
     async def trigger_rescan(self) -> bool:
         """Manually trigger a library rescan.
@@ -166,12 +200,12 @@ class SchedulerService:
         """
         self._config = config
         logger.info(
-            "Scheduler config reloaded",
-            enabled=config.enabled,
-            schedule_time=config.schedule_time,
-            schedule_days=config.schedule_days,
-            auto_download=config.auto_download_after_rescan,
-            folder_scan=config.folder_scan_enabled,
+            "Scheduler config reloaded: enabled=%s time=%s days=%s auto_download=%s folder_scan=%s",
+            config.enabled,
+            config.schedule_time,
+            config.schedule_days,
+            config.auto_download_after_rescan,
+            config.folder_scan_enabled,
         )
 
         if not self._scheduler or not self._scheduler.running:
@@ -192,9 +226,9 @@ class SchedulerService:
             if self._scheduler.get_job(_JOB_ID):
                 self._scheduler.reschedule_job(_JOB_ID, trigger=trigger)
                 logger.info(
-                    "Scheduler rescheduled with cron trigger",
-                    schedule_time=config.schedule_time,
-                    schedule_days=config.schedule_days,
+                    "Scheduler rescheduled with cron trigger: time=%s days=%s",
+                    config.schedule_time,
+                    config.schedule_days,
                 )
             else:
                 self._scheduler.add_job(
@@ -206,9 +240,9 @@ class SchedulerService:
                     coalesce=True,
                 )
                 logger.info(
-                    "Scheduler job added with cron trigger",
-                    schedule_time=config.schedule_time,
-                    schedule_days=config.schedule_days,
+                    "Scheduler job added with cron trigger: time=%s days=%s",
+                    config.schedule_time,
+                    config.schedule_days,
                 )
 
     def get_status(self) -> dict:
@@ -262,10 +296,10 @@ class SchedulerService:
             day_of_week=day_of_week,
         )
         logger.debug(
-            "CronTrigger built",
-            hour=hour_str,
-            minute=minute_str,
-            day_of_week=day_of_week,
+            "CronTrigger built: hour=%s minute=%s day_of_week=%s",
+            hour_str,
+            minute_str,
+            day_of_week,
         )
         return trigger
 
@@ -279,7 +313,7 @@ class SchedulerService:
             ws_service = get_websocket_service()
             await ws_service.manager.broadcast({"type": event_type, "data": data})
         except Exception as exc:  # pylint: disable=broad-exception-caught
-            logger.warning("WebSocket broadcast failed", event=event_type, error=str(exc))
+            logger.warning("WebSocket broadcast failed: event=%s error=%s", event_type, exc)
 
     async def _auto_download_missing(self) -> None:
         """Queue and start downloads for all series with missing episodes."""
@@ -297,9 +331,9 @@ class SchedulerService:
             elapsed = now - self._last_auto_download_time
             if elapsed < timedelta(seconds=self._auto_download_cooldown_seconds):
                 logger.debug(
-                    "Auto-download skipped: cooldown active",
-                    elapsed_seconds=elapsed.total_seconds(),
-                    cooldown_seconds=self._auto_download_cooldown_seconds,
+                    "Auto-download skipped: cooldown active (elapsed=%.1fs cooldown=%ds)",
+                    elapsed.total_seconds(),
+                    self._auto_download_cooldown_seconds,
                 )
                 return
 
@@ -332,29 +366,31 @@ class SchedulerService:
             )
             queued_count += len(episodes)
             logger.info(
-                "Auto-download queued episodes",
-                series=series.get("key"),
-                count=len(episodes),
+                "Auto-download queued episodes for series=%s count=%d",
+                series.get("key"),
+                len(episodes),
             )
 
         if queued_count:
             await download_service.start_queue_processing()
-            logger.info("Auto-download queue processing started", queued=queued_count)
+            logger.info("Auto-download queue processing started: queued=%d", queued_count)
 
         await self._broadcast("auto_download_started", {"queued_count": queued_count})
-        logger.info("Auto-download completed", queued_count=queued_count)
+        logger.info("Auto-download completed: queued_count=%d", queued_count)
 
         # Update cooldown timestamp after successful auto-download
         self._last_auto_download_time = datetime.now(timezone.utc)
 
     async def _perform_rescan(self) -> None:
         """Execute a library rescan and optionally trigger auto-download."""
+        logger.info("Scheduler _perform_rescan entered: scan_in_progress=%s", self._scan_in_progress)
         if self._scan_in_progress:
             logger.warning("Skipping rescan: previous scan still in progress")
             return
 
         self._scan_in_progress = True
         scan_start = datetime.now(timezone.utc)
+        logger.info("Scheduled rescan started at %s", scan_start.isoformat())
 
         try:
             logger.info("Starting scheduled library rescan")
@@ -362,18 +398,20 @@ class SchedulerService:
             from src.server.utils.dependencies import get_anime_service  # noqa: PLC0415
 
             anime_service = get_anime_service()
+            logger.info("Anime service obtained for rescan")
 
             await self._broadcast(
                 "scheduled_rescan_started",
                 {"timestamp": scan_start.isoformat()},
             )
 
+            logger.info("Calling anime_service.rescan()...")
             await anime_service.rescan()
 
             self._last_scan_time = datetime.now(timezone.utc)
             duration = (self._last_scan_time - scan_start).total_seconds()
 
-            logger.info("Scheduled library rescan completed", duration_seconds=duration)
+            logger.info("Scheduled library rescan completed: duration=%.2fs", duration)
 
             await self._broadcast(
                 "scheduled_rescan_completed",
@@ -390,8 +428,8 @@ class SchedulerService:
                     await self._auto_download_missing()
                 except Exception as dl_exc:  # pylint: disable=broad-exception-caught
                     logger.error(
-                        "Auto-download after rescan failed",
-                        error=str(dl_exc),
+                        "Auto-download after rescan failed: %s",
+                        dl_exc,
                         exc_info=True,
                     )
                     await self._broadcast(
@@ -410,10 +448,11 @@ class SchedulerService:
 
                     folder_scan_service = FolderScanService()
                     await folder_scan_service.run_folder_scan()
+                    logger.info("Folder scan completed successfully")
                 except Exception as fs_exc:  # pylint: disable=broad-exception-caught
                     logger.error(
-                        "Folder scan failed",
-                        error=str(fs_exc),
+                        "Folder scan failed: %s",
+                        fs_exc,
                         exc_info=True,
                     )
                     await self._broadcast(
@@ -423,7 +462,7 @@ class SchedulerService:
                 logger.debug("Folder scan is disabled — skipping")
 
         except Exception as exc:  # pylint: disable=broad-exception-caught
-            logger.error("Scheduled rescan failed", error=str(exc), exc_info=True)
+            logger.error("Scheduled rescan failed: %s", exc, exc_info=True)
             await self._broadcast(
                 "scheduled_rescan_error",
                 {"error": str(exc), "timestamp": datetime.now(timezone.utc).isoformat()},
@@ -431,6 +470,7 @@ class SchedulerService:
 
         finally:
             self._scan_in_progress = False
+            logger.info("Scheduled rescan finished: scan_in_progress reset to False")
 
 
 # ---------------------------------------------------------------------------
@@ -443,8 +483,14 @@ class SchedulerService:
 
 async def _run_rescan_job() -> None:
     """Module-level job entry point — delegates to the current service."""
+    logger.info("=" * 60)
+    logger.info("APScheduler triggered _run_rescan_job")
+    logger.info("Getting scheduler service singleton...")
     svc = get_scheduler_service()
+    logger.info("Scheduler service obtained, calling _perform_rescan()")
     await svc._perform_rescan()
+    logger.info("_run_rescan_job completed")
+    logger.info("=" * 60)
 
 
 # ---------------------------------------------------------------------------
@@ -458,7 +504,10 @@ def get_scheduler_service() -> SchedulerService:
     """Return the singleton SchedulerService instance."""
     global _scheduler_service
     if _scheduler_service is None:
+        logger.info("Creating new SchedulerService singleton")
         _scheduler_service = SchedulerService()
+    else:
+        logger.debug("Returning existing SchedulerService singleton")
     return _scheduler_service
 
 

src/server/web/static/js/index/series-manager.js

@@ -203,6 +203,17 @@ AniWorld.SeriesManager = (function() {
     function applyFiltersAndSort() {
         let filtered = seriesData.slice();
 
+        // Apply client-side filter so that real-time WebSocket updates
+        // (e.g. an episode being marked downloaded) are immediately
+        // reflected without a full server reload.
+        if (filterMode === 'missing_episodes') {
+            filtered = filtered.filter(function(s) {
+                return s.missing_episodes > 0;
+            });
+        }
+        // 'no_episodes' filter state is maintained server-side;
+        // don't try to replicate it client-side here.
+
         // Sort based on the current sorting mode
         filtered.sort(function(a, b) {
             if (sortAlphabetical) {
@@ -233,8 +244,12 @@ AniWorld.SeriesManager = (function() {
      */
     function renderSeries() {
         const grid = document.getElementById('series-grid');
-        const dataToRender = filteredSeriesData.length > 0 ? filteredSeriesData :
-            (seriesData.length > 0 ? seriesData : []);
+        // Always use filteredSeriesData — applyFiltersAndSort() is always
+        // called before renderSeries(), so filteredSeriesData is current.
+        // The old fallback to seriesData was incorrect: when a filter is
+        // active and filteredSeriesData is empty it must show the empty-state
+        // message, not fall through to unfiltered seriesData.
+        const dataToRender = filteredSeriesData;
 
         if (dataToRender.length === 0) {
             let message;

tests/api/test_config_endpoints.py

@@ -2,7 +2,7 @@
 
 import tempfile
 from pathlib import Path
-from unittest.mock import patch
+from unittest.mock import AsyncMock, patch
 
 import pytest
 from httpx import ASGITransport, AsyncClient
@@ -207,3 +207,46 @@ async def test_tmdb_validation_endpoint_exists(authenticated_client):
     assert "message" in data
     assert data["valid"] is False  # Empty key should be invalid
     assert "required" in data["message"].lower()
+
+
+@pytest.mark.asyncio
+async def test_update_config_with_anime_directory_starts_scheduler(
+    authenticated_client, mock_config_service
+):
+    """PUT /api/config with anime_directory syncs and starts scheduler."""
+    mock_scheduler = AsyncMock()
+    
+    with patch("src.server.services.scheduler_service.get_scheduler_service") as mock_sched_fn:
+        mock_sched_fn.return_value = mock_scheduler
+
+        with patch("src.config.settings.settings") as mock_settings:
+            mock_settings.anime_directory = None
+
+            resp = await authenticated_client.put(
+                "/api/config",
+                json={"other": {"anime_directory": "/data/anime"}},
+            )
+
+            assert resp.status_code == 200
+            mock_scheduler.ensure_started.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_update_config_without_anime_directory_does_not_start_scheduler(
+    authenticated_client, mock_config_service
+):
+    """PUT /api/config without new anime_directory does not call scheduler.ensure_started()."""
+    mock_scheduler = AsyncMock()
+    
+    with patch("src.server.services.scheduler_service.get_scheduler_service") as mock_sched_fn:
+        mock_sched_fn.return_value = mock_scheduler
+
+        with patch("src.config.settings.settings") as mock_settings:
+            mock_settings.anime_directory = "/already/set"
+
+            resp = await authenticated_client.put(
+                "/api/config", json={"other": {}}
+            )
+
+            assert resp.status_code == 200
+            mock_scheduler.ensure_started.assert_not_called()

tests/unit/test_add_series_episodes.py

@@ -392,23 +392,33 @@ class TestAddSeriesWithEpisodes:
             nfo_created_at=datetime(2024, 1, 1, 12, 0, 0),
             nfo_updated_at=datetime(2024, 1, 2, 12, 0, 0)
         )
-        
+        mock_db_series.id = 1
+
         # Create service with mocked WebSocket
         anime_service = AnimeService(mock_series_app)
         mock_websocket = AsyncMock()
         anime_service._websocket_service = mock_websocket
-        
+
         # Mock database session and service
         mock_db_session = AsyncMock()
         mock_db_session.__aenter__ = AsyncMock(return_value=mock_db_session)
         mock_db_session.__aexit__ = AsyncMock()
-        
+
+        # Mock episodes that match the in-memory episodeDict
+        mock_episodes = [
+            MagicMock(season=1, episode_number=1),
+            MagicMock(season=1, episode_number=2),
+            MagicMock(season=1, episode_number=3),
+        ]
+
         with patch('src.server.database.connection.get_db_session', return_value=mock_db_session):
             with patch('src.server.database.service.AnimeSeriesService') as MockAnimeSeriesService:
                 MockAnimeSeriesService.get_by_key = AsyncMock(return_value=mock_db_series)
-                
-                # Act
-                await anime_service._broadcast_series_updated(key)
+                with patch('src.server.database.service.EpisodeService') as MockEpisodeService:
+                    MockEpisodeService.get_by_series = AsyncMock(return_value=mock_episodes)
+
+                    # Act
+                    await anime_service._broadcast_series_updated(key)
         
         # Assert
         mock_websocket.broadcast.assert_called_once()

tests/unit/test_aniworld_provider.py

@@ -1,9 +1,11 @@
 """Unit tests for aniworld_provider.py - Anime catalog scraping, episode listing, streaming link extraction."""
 
 import json
+import os
 from unittest.mock import MagicMock, Mock, patch
 
 import pytest
+import requests
 
 from src.core.providers.aniworld_provider import AniworldLoader
 
@@ -472,3 +474,284 @@ class TestAniworldEvents:
         # Fire event - handler should NOT be called
         loader.events.download_progress({"status": "downloading"})
         handler.assert_not_called()
+
+
+class TestAniworldHealthCheck:
+    """Tests for the _check_url_alive HEAD probe."""
+
+    def test_returns_true_on_200(self, loader):
+        loader.session.head.return_value = MagicMock(status_code=200)
+        assert loader._check_url_alive("https://provider/x") is True
+
+    def test_returns_false_on_404(self, loader):
+        loader.session.head.return_value = MagicMock(status_code=404)
+        assert loader._check_url_alive("https://provider/x") is False
+
+    def test_returns_false_on_403(self, loader):
+        loader.session.head.return_value = MagicMock(status_code=403)
+        assert loader._check_url_alive("https://provider/x") is False
+
+    def test_falls_back_to_get_when_head_disallowed(self, loader):
+        loader.session.head.return_value = MagicMock(status_code=405)
+        get_resp = MagicMock(status_code=200)
+        get_resp.close = MagicMock()
+        loader.session.get.return_value = get_resp
+        assert loader._check_url_alive("https://provider/x") is True
+        loader.session.get.assert_called_once()
+
+    def test_returns_false_on_connection_error(self, loader):
+        loader.session.head.side_effect = requests.ConnectionError("boom")
+        assert loader._check_url_alive("https://provider/x") is False
+
+
+class TestAniworldDirectStream:
+    """Tests for the _try_direct_stream fast-path."""
+
+    def _build_response(self, status, content_type, body=b""):
+        resp = MagicMock()
+        resp.ok = status < 400
+        resp.status_code = status
+        resp.headers = {"Content-Type": content_type}
+        resp.iter_content = MagicMock(return_value=[body])
+        resp.__enter__ = MagicMock(return_value=resp)
+        resp.__exit__ = MagicMock(return_value=False)
+        return resp
+
+    def test_skips_non_video_content(self, loader, tmp_path):
+        target = tmp_path / "out.mp4"
+        loader.session.get.return_value = self._build_response(
+            200, "text/html"
+        )
+        assert loader._try_direct_stream(
+            "https://x", str(target), None, 10
+        ) is False
+        assert not target.exists()
+
+    def test_writes_video_content(self, loader, tmp_path):
+        target = tmp_path / "out.mp4"
+        loader.session.get.return_value = self._build_response(
+            200, "video/mp4", body=b"abc123"
+        )
+        assert loader._try_direct_stream(
+            "https://x", str(target), None, 10
+        ) is True
+        assert target.read_bytes() == b"abc123"
+
+    def test_returns_false_on_http_error(self, loader, tmp_path):
+        target = tmp_path / "out.mp4"
+        loader.session.get.return_value = self._build_response(
+            404, "video/mp4"
+        )
+        assert loader._try_direct_stream(
+            "https://x", str(target), None, 10
+        ) is False
+
+    def test_returns_false_on_request_exception(self, loader, tmp_path):
+        loader.session.get.side_effect = requests.RequestException("nope")
+        assert loader._try_direct_stream(
+            "https://x", str(tmp_path / "out.mp4"), None, 10
+        ) is False
+
+
+class TestAniworldProviderSelection:
+    """Tests for _select_providers_for_episode ordering and filtering."""
+
+    def test_orders_by_supported_preference(self, loader):
+        loader.is_language = MagicMock(return_value=True)
+        loader._get_provider_from_html = MagicMock(return_value={
+            "Vidoza": {1: "https://aniworld.to/redirect/2"},
+            "VOE": {1: "https://aniworld.to/redirect/1"},
+        })
+        result = loader._select_providers_for_episode(1, 1, "k", "German Dub")
+        assert [name for name, _ in result] == ["VOE", "Vidoza"]
+
+    def test_filters_by_language(self, loader):
+        loader.is_language = MagicMock(return_value=True)
+        loader._get_provider_from_html = MagicMock(return_value={
+            "VOE": {2: "https://aniworld.to/redirect/1"},  # English only
+        })
+        result = loader._select_providers_for_episode(1, 1, "k", "German Dub")
+        assert result == []
+
+    def test_returns_empty_when_language_unavailable(self, loader):
+        loader.is_language = MagicMock(return_value=False)
+        loader._get_provider_from_html = MagicMock()
+        result = loader._select_providers_for_episode(1, 1, "k", "German Dub")
+        assert result == []
+        loader._get_provider_from_html.assert_not_called()
+
+
+class TestAniworldDownloadFailover:
+    """Tests for the failover rotation in download()."""
+
+    @pytest.fixture
+    def patched_loader(self, loader, tmp_path):
+        """Loader with side-effect heavy methods stubbed."""
+        loader.get_title = MagicMock(return_value="Anime")
+        loader._select_providers_for_episode = MagicMock(return_value=[
+            ("VOE", "https://aniworld.to/redirect/1"),
+            ("Doodstream", "https://aniworld.to/redirect/2"),
+        ])
+        loader._check_url_alive = MagicMock(return_value=True)
+        loader._try_direct_stream = MagicMock(return_value=False)
+        loader.clear_cache = MagicMock()
+        loader._resolve_direct_link = MagicMock(
+            return_value=("https://cdn/video.m3u8", {"Referer": "https://x"})
+        )
+        return loader
+
+    def test_skips_provider_when_url_dead(self, patched_loader, tmp_path):
+        # First provider URL fails health check, second succeeds and downloads
+        patched_loader._check_url_alive.side_effect = [False, True]
+
+        def fake_ytdl(opts):
+            outpath = opts["outtmpl"]
+            os.makedirs(os.path.dirname(outpath), exist_ok=True)
+            with open(outpath, "wb") as fh:
+                fh.write(b"data")
+            ydl = MagicMock()
+            ydl.__enter__ = MagicMock(return_value=ydl)
+            ydl.__exit__ = MagicMock(return_value=False)
+            ydl.extract_info = MagicMock(return_value={"title": "t"})
+            return ydl
+
+        with patch(
+            "src.core.providers.aniworld_provider.YoutubeDL",
+            side_effect=fake_ytdl,
+        ):
+            result = patched_loader.download(
+                str(tmp_path), "Anime", 1, 1, "k", "German Dub"
+            )
+        assert result is True
+        assert patched_loader._check_url_alive.call_count == 2
+        # Only second provider (Doodstream) attempted resolve
+        patched_loader._resolve_direct_link.assert_called_once_with(
+            "https://aniworld.to/redirect/2", "Doodstream"
+        )
+
+    def test_falls_back_to_next_provider_on_ytdl_error(
+        self, patched_loader, tmp_path
+    ):
+        calls = {"n": 0}
+
+        def fake_ytdl(opts):
+            calls["n"] += 1
+            if calls["n"] == 1:
+                raise Exception("HTTP 404 from VOE")
+            outpath = opts["outtmpl"]
+            os.makedirs(os.path.dirname(outpath), exist_ok=True)
+            with open(outpath, "wb") as fh:
+                fh.write(b"ok")
+            ydl = MagicMock()
+            ydl.__enter__ = MagicMock(return_value=ydl)
+            ydl.__exit__ = MagicMock(return_value=False)
+            ydl.extract_info = MagicMock(return_value={"title": "t"})
+            return ydl
+
+        with patch(
+            "src.core.providers.aniworld_provider.YoutubeDL",
+            side_effect=fake_ytdl,
+        ):
+            result = patched_loader.download(
+                str(tmp_path), "Anime", 1, 1, "k", "German Dub"
+            )
+        assert result is True
+        assert calls["n"] == 2
+
+    def test_uses_direct_stream_when_available(
+        self, patched_loader, tmp_path
+    ):
+        def write_direct(link, output, headers, timeout):
+            os.makedirs(os.path.dirname(output), exist_ok=True)
+            with open(output, "wb") as fh:
+                fh.write(b"vid")
+            return True
+
+        patched_loader._try_direct_stream.side_effect = write_direct
+
+        with patch(
+            "src.core.providers.aniworld_provider.YoutubeDL"
+        ) as mock_ydl:
+            result = patched_loader.download(
+                str(tmp_path), "Anime", 1, 1, "k", "German Dub"
+            )
+        assert result is True
+        mock_ydl.assert_not_called()
+
+    def test_returns_false_when_all_providers_fail(
+        self, patched_loader, tmp_path, caplog
+    ):
+        with patch(
+            "src.core.providers.aniworld_provider.YoutubeDL",
+            side_effect=Exception("HTTP 404"),
+        ):
+            result = patched_loader.download(
+                str(tmp_path), "Anime", 1, 1, "k", "German Dub"
+            )
+        assert result is False
+        assert "All download providers failed" in caplog.text
+        # Both providers attempted
+        assert patched_loader._resolve_direct_link.call_count == 2
+
+    def test_returns_false_when_no_providers_advertised(
+        self, patched_loader, tmp_path, caplog
+    ):
+        patched_loader._select_providers_for_episode.return_value = []
+        result = patched_loader.download(
+            str(tmp_path), "Anime", 1, 1, "k", "German Dub"
+        )
+        assert result is False
+        assert "No providers advertised" in caplog.text
+
+
+class TestAniworldHeaderParsing:
+    """_parse_provider_headers normalizes legacy strings to dict."""
+
+    def test_parses_referer(self):
+        result = AniworldLoader._parse_provider_headers(
+            ['Referer: "https://vidmoly.to"']
+        )
+        assert result == {"Referer": "https://vidmoly.to"}
+
+    def test_handles_none(self):
+        assert AniworldLoader._parse_provider_headers(None) == {}
+
+    def test_skips_malformed_entries(self):
+        result = AniworldLoader._parse_provider_headers(
+            ["not-a-header", "Key: value"]
+        )
+        assert result == {"Key": "value"}
+
+
+class TestDecodeHtmlContent:
+    """Test _decode_html_content function."""
+
+    def test_decodes_utf8_content(self):
+        """Should correctly decode UTF-8 content."""
+        from src.core.providers.aniworld_provider import _decode_html_content
+        html = '<html><body><h1>Titel mit Ümläüten</h1></body></html>'
+        content = html.encode('utf-8')
+        result = _decode_html_content(content)
+        assert 'Titel mit Ümläüten' in result
+
+    def test_decodes_latin1_content(self):
+        """Should correctly decode Latin-1 content when chardet detects it."""
+        from src.core.providers.aniworld_provider import _decode_html_content
+        # Longer content for more reliable chardet detection
+        html = '<html><body><h1>CafÉ and more text here</h1></body></html>'
+        content = html.encode('latin-1')
+        result = _decode_html_content(content)
+        assert 'Caf' in result  # Decoded content contains expected substring
+
+    def test_replaces_invalid_bytes(self):
+        """Should replace invalid bytes with replacement character."""
+        from src.core.providers.aniworld_provider import _decode_html_content
+        content = b'\xff\xfe Invalid \x80\x81'
+        result = _decode_html_content(content)
+        assert isinstance(result, str)
+
+    def test_handles_empty_content(self):
+        """Should handle empty content gracefully."""
+        from src.core.providers.aniworld_provider import _decode_html_content
+        result = _decode_html_content(b'')
+        assert result == ''

tests/unit/test_download_service.py

@@ -60,6 +60,27 @@ class MockQueueRepository:
         self._items[item_id].error = error
         return True
 
+    async def set_status(
+        self,
+        item_id: str,
+        status: str,
+    ) -> bool:
+        """Set status on an item."""
+        if item_id not in self._items:
+            return False
+        self._items[item_id].status = DownloadStatus(status)
+        return True
+
+    async def increment_retry(
+        self,
+        item_id: str,
+    ) -> bool:
+        """Increment retry count on an item."""
+        if item_id not in self._items:
+            return False
+        self._items[item_id].retry_count += 1
+        return True
+
     async def delete_item(self, item_id: str) -> bool:
         """Delete item from storage."""
         if item_id in self._items:
@@ -79,6 +100,8 @@ def mock_anime_service():
     """Create a mock AnimeService."""
     service = MagicMock(spec=AnimeService)
     service.download = AsyncMock(return_value=True)
+    service._directory = "/mock/anime/directory"
+    service._broadcast_series_updated = AsyncMock(return_value=None)
     return service
 
 
@@ -503,7 +526,9 @@ class TestRetryLogic:
         assert len(retried_ids) == 1
         assert len(download_service._failed_items) == 0
         assert len(download_service._pending_queue) == 1
-        assert download_service._pending_queue[0].retry_count == 1
+        # retry_count stays same when retrying; incremented only on failure
+        assert download_service._pending_queue[0].retry_count == 0
+        assert download_service._pending_queue[0].status == DownloadStatus.PENDING
 
     @pytest.mark.asyncio
     async def test_max_retries_not_exceeded(self, download_service):
@@ -526,6 +551,45 @@ class TestRetryLogic:
         assert len(retried_ids) == 0
         assert len(download_service._failed_items) == 1
         assert len(download_service._pending_queue) == 0
+    
+    @pytest.mark.asyncio
+    async def test_permanently_failed_after_max_retries(self, download_service):
+        """Test that item is marked permanently_failed after max retries."""
+        # Mock download to fail
+        download_service._anime_service.download = AsyncMock(
+            side_effect=Exception("Download failed")
+        )
+        
+        # Create item with max_retries - 1 already used
+        item = DownloadItem(
+            id="perm-failed-1",
+            serie_id="series-1",
+            serie_folder="Test Series (2023)",
+            serie_name="Test Series",
+            episode=EpisodeIdentifier(season=1, episode=1),
+            status=DownloadStatus.PENDING,
+            retry_count=2,  # Already 2 retries, max is 3
+            error=None,
+        )
+        download_service._pending_queue.append(item)
+        
+        # Process download - will fail and reach max retries
+        await download_service._process_download(item)
+        
+        # Item should be in failed_items with permanently_failed status
+        assert len(download_service._failed_items) == 1
+        assert download_service._failed_items[0].retry_count == 3
+
+
+class TestDeadLetterQueue:
+    """Test dead-letter queue behavior for permanently failed items."""
+    
+    @pytest.mark.asyncio
+    async def test_requeue_permanently_failed_item(self, download_service):
+        """Test that a permanently failed item can be re-queued."""
+        # The unique constraint now includes status, so a permanently_failed
+        # item doesn't block re-queuing the same episode
+        pass  # Implementation depends on UI/API behavior
 
 
 class TestBroadcastCallbacks:
@@ -731,13 +795,22 @@ class TestRemoveEpisodeFromMissingList:
         download_service._anime_service._app = mock_app
         download_service._anime_service._cached_list_missing = MagicMock()
 
-        # Mock DB call
+        # Mock DB session
         mock_db_session = AsyncMock()
-        mock_delete = AsyncMock(return_value=True)
+        
+        # Mock series returned by get_by_key
+        mock_series = MagicMock()
+        mock_series.id = 1
+        
+        # Mock episode returned by get_by_episode
+        mock_episode = MagicMock()
+        mock_episode.id = 100
 
         with patch(
             "src.server.database.connection.get_db_session"
         ) as mock_get_db, patch(
+            "src.server.database.service.AnimeSeriesService"
+        ) as mock_series_svc, patch(
             "src.server.database.service.EpisodeService"
         ) as mock_ep_svc:
             mock_get_db.return_value.__aenter__ = AsyncMock(
@@ -746,26 +819,40 @@ class TestRemoveEpisodeFromMissingList:
             mock_get_db.return_value.__aexit__ = AsyncMock(
                 return_value=False
             )
-            mock_ep_svc.delete_by_series_and_episode = mock_delete
+            
+            # Mock get_by_key to return series
+            mock_series_svc.get_by_key = AsyncMock(return_value=mock_series)
+            
+            # Mock get_by_episode to return episode
+            mock_ep_svc.get_by_episode = AsyncMock(return_value=mock_episode)
+            
+            # Mock mark_downloaded to succeed
+            mock_ep_svc.mark_downloaded = AsyncMock(return_value=mock_episode)
 
             result = await download_service._remove_episode_from_missing_list(
                 series_key="test-series",
                 season=1,
                 episode=2,
+                serie_folder="Test Series (2024)",
             )
 
-        # DB deletion was called
-        mock_delete.assert_awaited_once_with(
+        # mark_downloaded was called instead of delete
+        mock_ep_svc.mark_downloaded.assert_awaited_once_with(
             db=mock_db_session,
-            series_key="test-series",
-            season=1,
-            episode_number=2,
+            episode_id=100,
+            file_path=(
+                f"{download_service._directory}/Test Series (2024)/Season 1"
+            ),
         )
         # In-memory update happened
         assert 2 not in serie.episodeDict[1]
         assert serie.episodeDict[1] == [1, 3]
         # Cache was cleared
         download_service._anime_service._cached_list_missing.cache_clear.assert_called()
+        # Broadcast was sent so frontend gets real-time update
+        download_service._anime_service._broadcast_series_updated.assert_awaited_once_with(
+            "test-series"
+        )
         assert result is True
 
     @pytest.mark.asyncio
@@ -807,11 +894,20 @@ class TestRemoveEpisodeFromMissingList:
 
         # Mock DB calls
         mock_db_session = AsyncMock()
-        mock_delete = AsyncMock(return_value=True)
+        
+        # Mock series returned by get_by_key
+        mock_series = MagicMock()
+        mock_series.id = 1
+        
+        # Mock episode returned by get_by_episode
+        mock_episode = MagicMock()
+        mock_episode.id = 100
 
         with patch(
             "src.server.database.connection.get_db_session"
         ) as mock_get_db, patch(
+            "src.server.database.service.AnimeSeriesService"
+        ) as mock_series_svc, patch(
             "src.server.database.service.EpisodeService"
         ) as mock_ep_svc:
             mock_get_db.return_value.__aenter__ = AsyncMock(
@@ -820,7 +916,15 @@ class TestRemoveEpisodeFromMissingList:
             mock_get_db.return_value.__aexit__ = AsyncMock(
                 return_value=False
             )
-            mock_ep_svc.delete_by_series_and_episode = mock_delete
+            
+            # Mock get_by_key to return series
+            mock_series_svc.get_by_key = AsyncMock(return_value=mock_series)
+            
+            # Mock get_by_episode to return episode
+            mock_ep_svc.get_by_episode = AsyncMock(return_value=mock_episode)
+            
+            # Mock mark_downloaded to succeed
+            mock_ep_svc.mark_downloaded = AsyncMock(return_value=mock_episode)
 
             # Process the download
             item = download_service._pending_queue.popleft()

tests/unit/test_enhanced_provider.py

@@ -929,7 +929,7 @@ class TestFfmpegHlsOptions:
 
         captured_opts = {}
 
-        def capture_ytdl_download(ydl_opts, link):
+        def capture_ytdl_download(self, temp_path, ydl_opts, link):
             captured_opts.update(ydl_opts)
             with open(temp_path, "wb") as f:
                 f.write(b"fake-video-data")
@@ -961,3 +961,53 @@ class TestFfmpegHlsOptions:
         assert captured_opts.get("hls_use_mpegts") is True, (
             f"Expected hls_use_mpegts=True, got {captured_opts.get('hls_use_mpegts')}"
         )
+
+
+class TestHlsUrlDetection:
+    """Test HLS URL detection patterns."""
+
+    def test_voe_hls_pattern_extracts_hls_url(self):
+        """HLS_PATTERN should extract HLS URL from VOE embedded player HTML."""
+        import re
+        from src.core.providers.streaming.voe import HLS_PATTERN
+
+        html_with_hls = """
+        var playerConfig = {
+            'hls': 'aHR0cHM6Ly92b2Uuc3YvZS9hYmMuaGxtMTNobG0xNm0zNDU2Nzg5MGE0MzIxLm0zdTg=',
+            'source': 'direct_mp4_url'
+        };
+        """
+        match = HLS_PATTERN.search(html_with_hls)
+        assert match is not None
+        assert match.group("hls") == "aHR0cHM6Ly92b2Uuc3YvZS9hYmMuaGxtMTNobG0xNm0zNDU2Nzg5MGE0MzIxLm0zdTg="
+
+    def test_voe_hls_pattern_returns_none_when_no_hls(self):
+        """HLS_PATTERN should return None when no HLS URL in HTML."""
+        import re
+        from src.core.providers.streaming.voe import HLS_PATTERN
+
+        html_no_hls = """
+        var playerConfig = {
+            'source': 'https://direct.example.com/video.mp4'
+        };
+        """
+        match = HLS_PATTERN.search(html_no_hls)
+        assert match is None
+
+    def test_hls_url_detection_in_provider_flow(self, enhanced_loader, tmp_path):
+        """Provider should detect and handle HLS URLs from VOE extractor."""
+        import re
+        from src.core.providers.streaming.voe import HLS_PATTERN
+
+        # Simulate VOE returning an HLS URL (base64 encoded .m3u8)
+        encoded_hls = "aHR0cHM6Ly9leGFtcGxlLmNvbS92aWRlby5tM3U4"
+        expected_hls = "https://example.com/video.m3u8"
+
+        html = f"var playerConfig = {{'hls': '{encoded_hls}'}};"
+
+        # Verify pattern correctly decodes to an m3u8 URL
+        match = HLS_PATTERN.search(html)
+        assert match is not None
+        decoded = match.group("hls")
+        # Note: this is just the base64 encoding of the URL, not actual decoding in pattern
+        assert decoded == encoded_hls

tests/unit/test_nfo_minimal_fallback.py

@@ -0,0 +1,240 @@
+"""Unit tests for minimal NFO creation when TMDB fails.
+
+Tests the fallback behavior when TMDB lookup fails and we need to create
+a minimal NFO file just to track the series.
+"""
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from src.core.services.nfo_service import NFOService
+
+
+@pytest.fixture
+def nfo_service(tmp_path):
+    """Create NFO service with test directory.
+    
+    Note: anime_directory is set to tmp_path directly (not tmp_path / "anime")
+    because tmp_path already represents the test anime directory.
+    """
+    service = NFOService(
+        tmdb_api_key="test_api_key",
+        anime_directory=str(tmp_path),
+        image_size="w500",
+        auto_create=True
+    )
+    return service
+
+
+class TestCreateMinimalNFO:
+    """Test minimal NFO creation."""
+
+    @pytest.mark.asyncio
+    async def test_create_minimal_nfo_basic(self, nfo_service, tmp_path):
+        """Test creating minimal NFO with just title."""
+        # Setup - anime_directory is already tmp_path
+        serie_folder = "Test Series"
+        
+        # Create minimal NFO
+        nfo_path = await nfo_service.create_minimal_nfo(
+            serie_name="Test Series",
+            serie_folder=serie_folder
+        )
+        
+        # Verify
+        assert nfo_path.exists()
+        assert nfo_path.name == "tvshow.nfo"
+        
+        content = nfo_path.read_text(encoding="utf-8")
+        assert "<title>Test Series</title>" in content
+        assert "No metadata available" in content
+
+    @pytest.mark.asyncio
+    async def test_create_minimal_nfo_with_year(self, nfo_service, tmp_path):
+        """Test creating minimal NFO with year."""
+        # Setup - anime_directory is already tmp_path
+        serie_folder = "Test Series (2024)"
+        
+        # Create minimal NFO with explicit year
+        nfo_path = await nfo_service.create_minimal_nfo(
+            serie_name="Test Series",
+            serie_folder=serie_folder,
+            year=2024
+        )
+        
+        # Verify
+        assert nfo_path.exists()
+        content = nfo_path.read_text(encoding="utf-8")
+        assert "<title>Test Series</title>" in content
+        assert "<year>2024</year>" in content
+
+    @pytest.mark.asyncio
+    async def test_create_minimal_nfo_extracts_year_from_name(self, nfo_service, tmp_path):
+        """Test that year is extracted from series name format (YYYY)."""
+        # Setup - anime_directory is already tmp_path
+        serie_folder = "Test Series (2024)"
+        
+        # Create with name that has year
+        nfo_path = await nfo_service.create_minimal_nfo(
+            serie_name="Test Series (2024)",
+            serie_folder=serie_folder
+        )
+        
+        # Verify year was extracted
+        assert nfo_path.exists()
+        content = nfo_path.read_text(encoding="utf-8")
+        assert "<title>Test Series</title>" in content
+        assert "<year>2024</year>" in content
+
+    @pytest.mark.asyncio
+    async def test_create_minimal_nfo_creates_folder_if_missing(self, nfo_service, tmp_path):
+        """Test that folder is created if it doesn't exist."""
+        # Setup - anime_directory is tmp_path itself
+        serie_folder = "New Series"
+        
+        # Folder should not exist yet (under anime_directory which is tmp_path)
+        folder_path = tmp_path / serie_folder
+        assert not folder_path.exists()
+        
+        # Create minimal NFO
+        nfo_path = await nfo_service.create_minimal_nfo(
+            serie_name="New Series",
+            serie_folder=serie_folder
+        )
+        
+        # Verify folder and file were created
+        assert folder_path.exists()
+        assert nfo_path.exists()
+
+    @pytest.mark.asyncio
+    async def test_create_minimal_nfo_xml_is_valid(self, nfo_service, tmp_path):
+        """Test that generated XML is valid."""
+        # Create minimal NFO (anime_directory is already tmp_path)
+        nfo_path = await nfo_service.create_minimal_nfo(
+            serie_name="Test Anime",
+            serie_folder="Test Anime",
+            year=2020
+        )
+        
+        # Verify XML is valid
+        from lxml import etree
+        content = nfo_path.read_text(encoding="utf-8")
+        
+        # Should parse without errors
+        tree = etree.fromstring(content.encode("utf-8"))
+        assert tree is not None
+        assert tree.tag == "tvshow"
+        
+        # Check title element
+        title = tree.find("title")
+        assert title is not None
+        assert title.text == "Test Anime"
+
+    @pytest.mark.asyncio
+    async def test_create_minimal_nfo_no_tmdb_id(self, nfo_service, tmp_path):
+        """Test that minimal NFO has no TMDB ID."""
+        # Create minimal NFO (anime_directory is already tmp_path)
+        nfo_path = await nfo_service.create_minimal_nfo(
+            serie_name="Unknown Series",
+            serie_folder="Unknown Series",
+            year=1999
+        )
+        
+        # Verify no TMDB ID
+        content = nfo_path.read_text(encoding="utf-8")
+        assert "<tmdbid>" not in content
+        assert "uniqueid" not in content
+
+    @pytest.mark.asyncio
+    async def test_create_minimal_nfo_has_plot_explanation(self, nfo_service, tmp_path):
+        """Test that minimal NFO contains explanation in plot."""
+        # Create minimal NFO (anime_directory is already tmp_path)
+        nfo_path = await nfo_service.create_minimal_nfo(
+            serie_name="Mysterious Anime",
+            serie_folder="Mysterious Anime"
+        )
+        
+        # Verify plot explains why metadata is missing
+        content = nfo_path.read_text(encoding="utf-8")
+        assert "TMDB lookup failed" in content
+        assert "Mysterious Anime" in content
+
+
+class TestCreateMinimalNFOIntegration:
+    """Integration tests for minimal NFO with TMDB failure scenarios."""
+
+    @pytest.mark.asyncio
+    async def test_fallback_on_tmdb_search_failure(self, nfo_service, tmp_path):
+        """Test that minimal NFO is created when TMDB search fails."""
+        # Mock TMDB client to raise error
+        nfo_service.tmdb_client.search_tv_show = AsyncMock(
+            side_effect=Exception("TMDB API Error")
+        )
+        
+        # Try to create full NFO (should fail and fallback to minimal)
+        # We test the fallback method directly
+        nfo_path = await nfo_service.create_minimal_nfo(
+            serie_name="Failed Series",
+            serie_folder="Failed Series",
+            year=2021
+        )
+        
+        # Verify
+        assert nfo_path.exists()
+        content = nfo_path.read_text(encoding="utf-8")
+        assert "<title>Failed Series</title>" in content
+        assert "<year>2021</year>" in content
+
+    @pytest.mark.asyncio
+    async def test_minimal_nfo_allows_series_tracking(self, nfo_service, tmp_path):
+        """Test that minimal NFO allows series to be tracked."""
+        # anime_directory is already tmp_path
+        serie_folder = "Untracked Series"
+        
+        # Create minimal NFO
+        nfo_path = await nfo_service.create_minimal_nfo(
+            serie_name="Untracked Series",
+            serie_folder=serie_folder,
+            year=2018
+        )
+        
+        # Verify NFO exists (series can be tracked)
+        assert nfo_service.has_nfo(serie_folder) is True
+        
+        # Verify minimal content
+        content = nfo_path.read_text(encoding="utf-8")
+        assert "<title>Untracked Series</title>" in content
+
+
+class TestMinimalNFOContent:
+    """Test content of minimal NFO files."""
+
+    @pytest.mark.asyncio
+    async def test_minimal_nfo_contains_required_elements(self, nfo_service, tmp_path):
+        """Test that minimal NFO has title and plot."""
+        nfo_path = await nfo_service.create_minimal_nfo(
+            serie_name="Minimal Test",
+            serie_folder="Minimal Test"
+        )
+        
+        content = nfo_path.read_text(encoding="utf-8")
+        
+        # Must have title
+        assert "<title>Minimal Test</title>" in content
+        # Must have plot explaining situation
+        assert "plot" in content.lower()
+        assert "No metadata available" in content
+
+    @pytest.mark.asyncio
+    async def test_minimal_nfo_xml_declaration(self, nfo_service, tmp_path):
+        """Test that NFO has proper XML declaration."""
+        nfo_path = await nfo_service.create_minimal_nfo(
+            serie_name="XML Test",
+            serie_folder="XML Test"
+        )
+        
+        content = nfo_path.read_text(encoding="utf-8")
+        
+        # Should have XML declaration
+        assert content.startswith('<?xml version="1.0" encoding="UTF-8"')

tests/unit/test_queue_repository.py

@@ -70,6 +70,8 @@ def _make_db_item(
     completed_at: datetime | None = None,
     error_message: str | None = None,
     download_url: str | None = None,
+    status: str = "pending",
+    retry_count: int = 0,
 ):
     """Build a fake DB DownloadQueueItem."""
     episode = MagicMock()
@@ -91,6 +93,8 @@ def _make_db_item(
     db_item.completed_at = completed_at
     db_item.error_message = error_message
     db_item.download_url = download_url
+    db_item.status = status
+    db_item.retry_count = retry_count
     return db_item
 
 

tests/unit/test_scheduler_service.py

@@ -559,3 +559,38 @@ class TestStartupRecovery:
                 info_calls = [str(c) for c in mock_logger.info.call_args_list]
                 assert any("next_run" in c for c in info_calls)
 
+
+# ---------------------------------------------------------------------------
+# 12.8  ensure_started() – idempotent startup
+# ---------------------------------------------------------------------------
+
+class TestEnsureStarted:
+    @pytest.mark.asyncio
+    async def test_ensure_started_when_not_running(
+        self, scheduler_service, mock_config_service
+    ):
+        """ensure_started() calls start() when scheduler is not running."""
+        # Mock start method
+        scheduler_service.start = AsyncMock()
+
+        # Call ensure_started
+        await scheduler_service.ensure_started()
+
+        # Verify start() was called exactly once
+        scheduler_service.start.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_ensure_started_when_already_running(self, scheduler_service):
+        """ensure_started() returns immediately when already running (idempotent)."""
+        # Set up as already running
+        scheduler_service._is_running = True
+
+        # Mock start method
+        scheduler_service.start = AsyncMock()
+
+        # Call ensure_started
+        await scheduler_service.ensure_started()
+
+        # Verify start() was NOT called
+        scheduler_service.start.assert_not_called()
+