refactor(scheduler): drop separate scheduler.db in favour of MemoryJobStore

Scheduler used a separate SQLite file (scheduler.db) only to persist one
cron job. This was originally required because APScheduler's
SQLAlchemyJobStore is sync-only, creating an async/sync driver conflict
when accessing the same file.

The job is rebuilt from config.json on every startup regardless
(replace_existing=True), so the persisted state only served misfire
detection. Moved misfire detection into the app layer by querying
system_settings.last_scan_timestamp on startup: if the last scan is
>23h but <25h ago, an immediate rescan is triggered.

Change summary:
- Remove SQLAlchemyJobStore; use default MemoryJobStore instead
- Add _check_missed_run() that reads last_scan_timestamp from aniworld.db
- Update docs/DEVELOPMENT.md scheduler troubleshooting section
- Update the scheduler unit test that verified SQLAlchemyJobStore

Docker/VERSION

@@ -1 +1 @@
-v1.1.17
+v1.1.18

docs/DEVELOPMENT.md

@@ -162,24 +162,37 @@ await client.close()  # May not be called if exception raised earlier
 
 ### Scheduler Persistence and Recovery
 
-APScheduler stores jobs in `data/scheduler.db` (SQLite) so they survive process restarts:
+The scheduler uses APScheduler's in-memory job store. Jobs are reconstructed from `config.json` on every startup — no separate database is needed.
 
 ```python
-from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
-
-jobstores = {
-    "default": SQLAlchemyJobStore(url="sqlite:///./data/scheduler.db"),
-}
-scheduler = AsyncIOScheduler(jobstores=jobstores)
+# Jobs are built from config on startup — no persistence DB required
+scheduler = AsyncIOScheduler()  # default MemoryJobStore
+scheduler.add_job(..., replace_existing=True)
 ```
 
-**Grace period:** `misfire_grace_time=3600` (1 hour). If server is down at scheduled time and restarts within 1 hour, missed job runs automatically via APScheduler coalesce behavior.
+**Startup misfire recovery:** On `start()`, the scheduler checks `system_settings.last_scan_timestamp` in `aniworld.db`. If the last scan is overdue (>23h but <25h ago), an immediate rescan is triggered. This replaces APScheduler's built-in misfire handling which required a separate SQLite database.
 
-**Startup recovery:** On `start()`, scheduler loads persisted jobs from DB. APScheduler handles missed jobs internally when `coalesce=True`.
+**Grace period:** If the server was down for more than 25 hours, no automatic recovery occurs to avoid surprise rescans after long downtime.
 
 **Health endpoint:** `GET /health` returns `scheduler_next_run` and `scheduler_last_run` for external monitors (Uptime Kuma, Prometheus, etc.).
 
-**If server is down >1 hour:** No automatic recovery. Manual trigger via `POST /api/scheduler/trigger-rescan` or wait for next scheduled run.
+**If server is down too long:** Manual trigger via `POST /api/scheduler/trigger-rescan` or wait for next scheduled run.
+
+### Database Session Management
+
+`get_async_session_factory()` returns a **new AsyncSession instance** directly (not a factory). The function name is historical — callers receive the session immediately:
+
+```python
+# Correct usage:
+db = get_async_session_factory()  # db IS the session
+await db.execute(...)
+await db.commit()
+await db.close()
+```
+
+Do NOT call the result again with `()` — that tries to call an `AsyncSession` object, causing `'AsyncSession' object is not callable`.
+
+For context manager usage, prefer `get_db_session()` (auto-commits) or `get_transactional_session()` (manual commit).
 
 ### Health Check Endpoints
 
@@ -241,30 +254,27 @@ DNS checks are warnings because failures can be transient. anime_directory error
 #### Scheduler missed a run
 
 1. Server was down at scheduled time (03:00 UTC by default).
-2. Check `data/scheduler.db` exists — if not, jobs are not persisted.
-3. If server was down >1 hour, missed job is dropped (misfire window exceeded).
+2. On restart, the scheduler checks `last_scan_timestamp` — if overdue by 23-25h, it triggers immediately.
+3. If server was down >25 hours, missed job is skipped to avoid surprise rescans.
 4. Trigger manually: `POST /api/scheduler/trigger-rescan`
 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:**
+1. **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
 
+2. **Verify the job is registered:**
+   ```
+   grep "Scheduler started with cron trigger" fastapi_app.log
+   ```
+
 3. **Verify APScheduler events in logs:**
    ```
    grep "apscheduler.executors.default" fastapi_app.log

docs/NFO_GUIDE.md

@@ -246,6 +246,7 @@ NFO files are created in the anime directory:
     <genre>Action</genre>
     <genre>Sci-Fi & Fantasy</genre>
     <uniqueid type="tmdb">1429</uniqueid>
+    <tmdbid>1429</tmdbid>
     <thumb aspect="poster">https://image.tmdb.org/t/p/w500/...</thumb>
     <fanart>
         <thumb>https://image.tmdb.org/t/p/original/...</thumb>
@@ -253,6 +254,13 @@ NFO files are created in the anime directory:
 </tvshow>
 ```
 
+**Manual TMDB ID Override**: To skip TMDB search and use a specific ID directly, include `<tmdbid>YOUR_ID</tmdbid>` in the NFO. This is useful when:
+- TMDB search fails for your series (e.g., new or obscure anime)
+- You already know the correct TMDB ID
+- You want to avoid rate limiting from repeated searches
+
+Aniworld reads `<tmdbid>` element and `<uniqueid type="tmdb">` first. If found, it uses the ID directly instead of searching.
+
 ### 4.3 Episode NFO Format
 
 ```xml
@@ -629,6 +637,36 @@ Every poster check action is logged:
 4. Check network speed to TMDB servers
 5. Verify disk I/O performance
 
+### 6.7 TMDB Lookup Fails for My Series
+
+**Problem**: TMDB search fails with "No results found" for a valid series.
+
+**Solutions**:
+
+1. **Check if series exists on TMDB**: Visit https://www.themoviedb.org and search for your series
+2. **Use manual ID override**: Add TMDB ID directly to `tvshow.nfo`:
+   ```xml
+   <?xml version="1.0" encoding="UTF-8" standalone="yes"?>
+   <tvshow>
+       <title>Your Series Name</title>
+       <tmdbid>12345</tmdbid>
+       <uniqueid type="tmdb">12345</uniqueid>
+   </tvshow>
+   ```
+   Aniworld will use this ID directly instead of searching.
+
+3. **Try alternative titles**: Some anime have different titles (Japanese, romaji, English). If you have access to the folder, rename it to match the TMDB title.
+
+4. **Add to existing NFO**: If `tvshow.nfo` exists but has no TMDB ID, edit it to add:
+   ```xml
+   <tmdbid>YOUR_TMDB_ID</tmdbid>
+   ```
+   Then use the Update endpoint to refresh metadata.
+
+5. **Check for rate limiting**: If many lookups fail at once, you may be hitting TMDB rate limits. Wait and retry later.
+
+6. **Verify API key**: Ensure your TMDB API key is valid and has not exceeded usage limits.
+
 ---
 
 ## 7. Best Practices

package.json

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

src/core/SerieScanner.py

@@ -44,8 +44,13 @@ class SerieScanner:
         in keyDict and can be retrieved after scanning.
     
     Example:
+        # Synchronous context (CLI):
         scanner = SerieScanner("/path/to/anime", loader)
-        scanner.scan()
+        scanner.scan()  # asyncio.run() used internally when no event loop
+
+        # Asynchronous context (server/scheduler):
+        # scan() detects running event loop and uses create_task()
+        # internally, so no special handling needed by caller.
         # Results are in scanner.keyDict
 
         # With DB lookup fallback:
@@ -218,8 +223,7 @@ class SerieScanner:
         try:
             from src.server.database.connection import get_async_session_factory
 
-            session_factory = get_async_session_factory()
-            db = session_factory()
+            db = get_async_session_factory()
             try:
                 existing = await AnimeSeriesService.get_by_key(db, serie.key)
                 if existing:
@@ -433,7 +437,14 @@ class SerieScanner:
 
                         # Persist to database (async)
                         try:
-                            asyncio.run(self._persist_serie_to_db(serie))
+                            try:
+                                loop = asyncio.get_running_loop()
+                            except RuntimeError:
+                                # No running loop — safe to use asyncio.run()
+                                asyncio.run(self._persist_serie_to_db(serie))
+                            else:
+                                # Already in async context — schedule as task
+                                asyncio.create_task(self._persist_serie_to_db(serie))
                         except Exception as e:
                             logger.warning(
                                 "DB persistence failed for '%s', "

src/core/services/nfo_service.py

@@ -163,58 +163,89 @@ class NFOService:
             logger.info("Creating series folder: %s", folder_path)
             folder_path.mkdir(parents=True, exist_ok=True)
         
+        # Check for existing NFO with TMDB ID to skip search
+        nfo_path = folder_path / "tvshow.nfo"
+        existing_ids = None
+        if nfo_path.exists():
+            try:
+                existing_ids = self.parse_nfo_ids(nfo_path)
+                if existing_ids.get("tmdb_id"):
+                    logger.info(
+                        "Found existing TMDB ID %s in NFO, using directly",
+                        existing_ids["tmdb_id"]
+                    )
+            except Exception as e:
+                logger.debug("Could not parse existing NFO IDs: %s", e)
+        
         try:
             await self.tmdb_client._ensure_session()
             
-            # Search for TV show - try multiple strategies
-            tv_show, search_source = await self._search_with_fallback(
-                search_name, year, alt_titles
-            )
-            tv_id = tv_show["id"]
+            # Use existing TMDB ID if found, otherwise search
+            if existing_ids and existing_ids.get("tmdb_id"):
+                tv_id = existing_ids["tmdb_id"]
+                logger.info("Fetching details directly for TMDB ID: %s", tv_id)
+                details = await self.tmdb_client.get_tv_show_details(
+                    tv_id,
+                    append_to_response="credits,external_ids,images"
+                )
+                content_ratings = await self.tmdb_client.get_tv_show_content_ratings(tv_id)
+                tv_show = {"id": tv_id, "name": details.get("name", serie_name)}
+                search_source = "nfo_override"
+            else:
+                # Search for TV show - try multiple strategies
+                tv_show, search_source = await self._search_with_fallback(
+                    search_name, year, alt_titles
+                )
+                tv_id = tv_show["id"]
 
             logger.info("Found match: %s (ID: %s)", tv_show['name'], tv_id)
 
             # Get detailed information with multi-language image support
-            details = await self.tmdb_client.get_tv_show_details(
-                tv_id,
-                append_to_response="credits,external_ids,images"
-            )
+            # Skip if we already fetched details via nfo_override
+            if search_source != "nfo_override":
+                details = await self.tmdb_client.get_tv_show_details(
+                    tv_id,
+                    append_to_response="credits,external_ids,images"
+                )
 
-            # Get content ratings for FSK
-            content_ratings = await self.tmdb_client.get_tv_show_content_ratings(tv_id)
+                # Get content ratings for FSK
+                content_ratings = await self.tmdb_client.get_tv_show_content_ratings(tv_id)
 
-            # Enrich with fallback languages for empty overview/tagline
-            # Pass search result overview as last resort fallback
-            search_overview = tv_show.get("overview") or None
-            if not search_overview:
-                try:
-                    logger.debug(
-                        "No overview in German search result, trying en-US search fallback for: %s",
-                        search_name,
-                    )
-                    en_search_results = await self.tmdb_client.search_tv_show(
-                        search_name,
-                        language="en-US",
-                    )
-                    if en_search_results.get("results"):
-                        en_match = self._find_best_match(
-                            en_search_results["results"], search_name, year
+                # Enrich with fallback languages for empty overview/tagline
+                # Pass search result overview as last resort fallback
+                search_overview = tv_show.get("overview") or None
+                if not search_overview:
+                    try:
+                        logger.debug(
+                            "No overview in German search result, trying en-US search fallback for: %s",
+                            search_name,
                         )
-                        search_overview = en_match.get("overview") or None
-                        if search_overview:
-                            logger.info(
-                                "Using en-US search overview fallback for %s",
-                                search_name,
+                        en_search_results = await self.tmdb_client.search_tv_show(
+                            search_name,
+                            language="en-US",
+                        )
+                        if en_search_results.get("results"):
+                            en_match = self._find_best_match(
+                                en_search_results["results"], search_name, year
                             )
-                except (TMDBAPIError, Exception) as exc:
-                    logger.warning(
-                        "Failed en-US search fallback for overview: %s",
-                        exc,
-                    )
+                            search_overview = en_match.get("overview") or None
+                            if search_overview:
+                                logger.info(
+                                    "Using en-US search overview fallback for %s",
+                                    search_name,
+                                )
+                    except (TMDBAPIError, Exception) as exc:
+                        logger.warning(
+                            "Failed en-US search fallback for overview: %s",
+                            exc,
+                        )
 
-            details = await self._enrich_details_with_fallback(
-                details, search_overview=search_overview
-            )
+                details = await self._enrich_details_with_fallback(
+                    details, search_overview=search_overview
+                )
+            else:
+                # When using nfo_override, content_ratings already fetched
+                pass
 
             # Convert TMDB data to TVShowNFO model
             nfo_model = tmdb_to_nfo_model(
@@ -646,21 +677,45 @@ class NFOService:
                 {"query": normalized, "year": year, "lang": "de-DE", "desc": f"normalized:{normalized}"}
             )
         
+        # Strategy 6: Try search/multi for series indexed as movies
+        search_strategies.append(
+            {"query": primary_query, "year": year, "lang": "en-US", "desc": "multi_search", "use_multi": True}
+        )
+        
         last_error = None
         for strategy in search_strategies:
             query = strategy["query"]
             lang = strategy["lang"]
             desc = strategy["desc"]
+            use_multi = strategy.get("use_multi", False)
             
             try:
                 logger.debug(
                     "TMDB search attempt: query='%s', lang=%s, year=%s, strategy=%s",
                     query, lang, strategy["year"], desc
                 )
-                search_results = await self.tmdb_client.search_tv_show(
-                    query,
-                    language=lang
-                )
+                
+                # Use search/multi for multi_search strategy
+                if use_multi:
+                    search_results = await self.tmdb_client.search_multi(
+                        query,
+                        language=lang
+                    )
+                    # Filter for TV shows only
+                    if search_results.get("results"):
+                        tv_results = [
+                            r for r in search_results["results"]
+                            if r.get("media_type") == "tv"
+                        ]
+                        if tv_results:
+                            search_results["results"] = tv_results
+                        else:
+                            search_results["results"] = []
+                else:
+                    search_results = await self.tmdb_client.search_tv_show(
+                        query,
+                        language=lang
+                    )
                 
                 if search_results.get("results"):
                     # Apply year filter if we have one
@@ -784,6 +839,7 @@ class NFOService:
     async def close(self):
         """Clean up resources."""
         await self.tmdb_client.close()
+        await self.image_downloader.close()
     
     async def create_minimal_nfo(
         self,

src/core/services/series_manager_service.py

@@ -129,6 +129,9 @@ class SeriesManagerService:
         if not self.nfo_service:
             return
         
+        nfo_exists = False
+        ids = {}
+        
         try:
             folder_path = Path(self.anime_directory) / serie_folder
             nfo_path = folder_path / "tvshow.nfo"
@@ -195,22 +198,49 @@ class SeriesManagerService:
                 logger.info(
                     f"Creating NFO for '{serie_name}' ({serie_folder})"
                 )
-                await self.nfo_service.create_tvshow_nfo(
-                    serie_name=serie_name,
-                    serie_folder=serie_folder,
-                    year=year,
-                    download_poster=self.download_poster,
-                    download_logo=self.download_logo,
-                    download_fanart=self.download_fanart
-                )
-                logger.info("Successfully created NFO for '%s'", serie_name)
+                try:
+                    await self.nfo_service.create_tvshow_nfo(
+                        serie_name=serie_name,
+                        serie_folder=serie_folder,
+                        year=year,
+                        download_poster=self.download_poster,
+                        download_logo=self.download_logo,
+                        download_fanart=self.download_fanart
+                    )
+                    logger.info("Successfully created NFO for '%s'", serie_name)
+                except TMDBAPIError as create_error:
+                    # TMDB lookup failed, create minimal NFO to track the series
+                    logger.warning(
+                        "TMDB lookup failed for '%s', creating minimal NFO: %s",
+                        serie_name, create_error
+                    )
+                    try:
+                        await self.nfo_service.create_minimal_nfo(
+                            serie_name=serie_name,
+                            serie_folder=serie_folder,
+                            year=year
+                        )
+                        logger.info("Created minimal NFO for '%s'", serie_name)
+                    except Exception as minimal_error:
+                        logger.error(
+                            "Failed to create minimal NFO for '%s': %s",
+                            serie_name, minimal_error
+                        )
             elif nfo_exists:
                 logger.debug(
                     f"NFO exists for '{serie_name}', skipping download"
                 )
                     
         except TMDBAPIError as e:
-            logger.error("TMDB API error processing '%s': %s", serie_name, e)
+            # Only log at ERROR if no NFO exists and we have no IDs
+            # If NFO exists with IDs, this is just a lookup failure, log at DEBUG
+            if nfo_exists and (ids.get("tmdb_id") or ids.get("tvdb_id")):
+                logger.debug(
+                    "TMDB API lookup failed for '%s' (has NFO with IDs): %s",
+                    serie_name, e
+                )
+            else:
+                logger.error("TMDB API error processing '%s': %s", serie_name, e)
         except Exception as e:
             logger.error(
                 f"Unexpected error processing NFO for '{serie_name}': {e}",

src/server/services/scheduler_service.py

@@ -4,17 +4,16 @@ Uses APScheduler's AsyncIOScheduler with CronTrigger for precise
 cron-based scheduling.  The legacy interval-based loop has been removed
 in favour of the cron approach.
 
-Jobs are persisted to a SQLite database so they survive process restarts.
-On startup, if the last scheduled run was missed (server was down at the
-cron time), the job is triggered immediately within a grace period.
+Jobs are held in memory (no separate scheduler database).  On startup,
+if the last scan timestamp indicates a missed run (server was down at the
+scheduled cron time), a rescan is triggered immediately.
 """
 from __future__ import annotations
 
 import logging
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from typing import List, Optional
 
-from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from apscheduler.triggers.cron import CronTrigger
 
@@ -83,10 +82,9 @@ class SchedulerService:
             logger.error("Failed to load scheduler configuration: %s", exc)
             raise SchedulerServiceError(f"Failed to load config: {exc}") from exc
 
-        jobstores = {
-            "default": SQLAlchemyJobStore(url="sqlite:///./data/scheduler.db"),
-        }
-        self._scheduler = AsyncIOScheduler(jobstores=jobstores)
+        # Use in-memory job store — no separate scheduler.db needed.
+        # Jobs are reconstructed from config on every startup.
+        self._scheduler = AsyncIOScheduler()
 
         if not self._config.enabled:
             logger.info("Scheduler is disabled in configuration — not adding jobs")
@@ -125,10 +123,7 @@ class SchedulerService:
         self._scheduler.start()
         self._is_running = True
 
-        # 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()
+        # Log next scheduled run for visibility.
         job = self._scheduler.get_job(_JOB_ID)
         if job:
             next_run = job.next_run_time
@@ -137,6 +132,11 @@ class SchedulerService:
                 next_run.isoformat() if next_run else None,
             )
 
+        # Startup misfire recovery: check if the last scan was missed while
+        # the server was down.  If overdue by more than one interval but within
+        # the grace period, trigger an immediate rescan.
+        await self._check_missed_run()
+
     async def stop(self) -> None:
         """Stop the APScheduler gracefully."""
         logger.info("SchedulerService.stop() called")
@@ -303,6 +303,67 @@ class SchedulerService:
         )
         return trigger
 
+    async def _check_missed_run(self) -> None:
+        """Check if a scheduled rescan was missed while the server was down.
+
+        Compares system_settings.last_scan_timestamp against the expected
+        schedule.  If the last scan is overdue (more than 24h ago for a daily
+        schedule) but within the grace period, triggers an immediate rescan.
+        """
+        if not self._config or not self._config.enabled:
+            return
+        if not self._config.schedule_days:
+            return
+
+        try:
+            from src.server.database.connection import (  # noqa: PLC0415
+                get_db_session,
+            )
+            from src.server.database.system_settings_service import (  # noqa: PLC0415
+                SystemSettingsService,
+            )
+
+            async with get_db_session() as db:
+                settings = await SystemSettingsService.get_or_create(db)
+                last_scan = settings.last_scan_timestamp
+
+            if last_scan is None:
+                # Never scanned before — trigger immediately
+                logger.info("No previous scan recorded — triggering immediate rescan")
+                await self._perform_rescan()
+                return
+
+            # Ensure timezone-aware comparison
+            if last_scan.tzinfo is None:
+                last_scan = last_scan.replace(tzinfo=timezone.utc)
+
+            now = datetime.now(timezone.utc)
+            elapsed = now - last_scan
+
+            # If last scan was more than 24h + grace period ago, don't trigger
+            # (avoids surprise rescans after long downtime).
+            max_overdue = timedelta(hours=24, seconds=_MISFIRE_GRACE_SECONDS)
+            # If last scan was more than ~25h ago, skip (too stale)
+            if elapsed > max_overdue:
+                logger.info(
+                    "Last scan was %s ago (> %s) — skipping missed-run recovery",
+                    elapsed,
+                    max_overdue,
+                )
+                return
+
+            # Check if a run should have happened between last_scan and now.
+            # Simple heuristic: if elapsed > 24h, we missed at least one daily run.
+            if elapsed > timedelta(hours=23):
+                logger.info(
+                    "Missed scheduled rescan detected (last scan %s ago) — triggering now",
+                    elapsed,
+                )
+                await self._perform_rescan()
+
+        except Exception as exc:  # pylint: disable=broad-exception-caught
+            logger.warning("Missed-run check failed (non-fatal): %s", exc)
+
     async def _broadcast(self, event_type: str, data: dict) -> None:
         """Broadcast a WebSocket event to all connected clients."""
         try:

tests/unit/test_nfo_service.py

@@ -1809,3 +1809,107 @@ class TestNegativeCache:
         assert "expired_key" not in tmdb_client._negative_cache
         assert "valid_key" in tmdb_client._negative_cache
 
+
+class TestNFOIDOverride:
+    """Tests for manual TMDB ID override via NFO."""
+
+    @pytest.mark.asyncio
+    async def test_create_tvshow_nfo_uses_existing_tmdb_id(self, nfo_service, tmp_path, mock_tmdb_data, mock_content_ratings_de):
+        """Test that existing TMDB ID in NFO skips search."""
+        # Create series folder with existing NFO containing TMDB ID
+        series_folder = tmp_path / "Attack on Titan"
+        series_folder.mkdir()
+        nfo_path = series_folder / "tvshow.nfo"
+        nfo_path.write_text("""<?xml version="1.0" encoding="UTF-8"?>
+<tvshow>
+    <title>Attack on Titan</title>
+    <tmdbid>1429</tmdbid>
+</tvshow>
+""", encoding="utf-8")
+
+        with patch.object(nfo_service.tmdb_client, 'get_tv_show_details', new_callable=AsyncMock) as mock_details, \
+             patch.object(nfo_service.tmdb_client, 'get_tv_show_content_ratings', new_callable=AsyncMock) as mock_ratings, \
+             patch.object(nfo_service, '_download_media_files', new_callable=AsyncMock), \
+             patch.object(nfo_service.tmdb_client, '_ensure_session', new_callable=AsyncMock):
+
+            mock_details.return_value = mock_tmdb_data
+            mock_ratings.return_value = mock_content_ratings_de
+
+            nfo_path_result = await nfo_service.create_tvshow_nfo(
+                "Attack on Titan",
+                "Attack on Titan",
+                download_poster=False, download_logo=False, download_fanart=False
+            )
+
+            # Verify NFO was created
+            assert nfo_path_result.exists()
+
+            # Verify get_tv_show_details was called directly with the ID (no search)
+            mock_details.assert_called_once_with(1429, append_to_response="credits,external_ids,images")
+
+            # Verify search was NOT called
+            # (we can check by verifying no search_tv_show mock was set up)
+
+    @pytest.mark.asyncio
+    async def test_create_tvshow_nfo_searches_when_no_tmdb_id(self, nfo_service, tmp_path, mock_tmdb_data, mock_content_ratings_de):
+        """Test that search is used when NFO has no TMDB ID."""
+        # Create series folder without existing NFO
+        series_folder = tmp_path / "Test Anime"
+        series_folder.mkdir()
+
+        with patch.object(nfo_service.tmdb_client, 'search_tv_show', new_callable=AsyncMock) as mock_search, \
+             patch.object(nfo_service.tmdb_client, 'get_tv_show_details', new_callable=AsyncMock) as mock_details, \
+             patch.object(nfo_service.tmdb_client, 'get_tv_show_content_ratings', new_callable=AsyncMock) as mock_ratings, \
+             patch.object(nfo_service, '_download_media_files', new_callable=AsyncMock), \
+             patch.object(nfo_service.tmdb_client, '_ensure_session', new_callable=AsyncMock):
+
+            mock_search.return_value = {
+                "results": [{
+                    "id": 1429,
+                    "name": "Test Anime",
+                    "first_air_date": "2024-01-01",
+                    "overview": "Test overview"
+                }]
+            }
+            mock_details.return_value = mock_tmdb_data
+            mock_ratings.return_value = mock_content_ratings_de
+
+            nfo_path = await nfo_service.create_tvshow_nfo(
+                "Test Anime",
+                "Test Anime",
+                download_poster=False, download_logo=False, download_fanart=False
+            )
+
+            # Verify search was called
+            mock_search.assert_called()
+
+
+class TestSearchMultiStrategy:
+    """Tests for search/multi fallback strategy."""
+
+    @pytest.mark.asyncio
+    async def test_search_multi_strategy_used_as_fallback(self, nfo_service, mock_tmdb_data):
+        """Test that search/multi is tried after regular search fails."""
+        mock_search = AsyncMock()
+        mock_multi = AsyncMock()
+
+        # First: regular search fails
+        # Second: multi search returns TV result
+        mock_search.return_value = {"results": []}
+        mock_multi.return_value = {
+            "results": [
+                {"media_type": "movie", "id": 123},
+                {"media_type": "tv", "id": 456, "name": "Found Show", "first_air_date": "2024-01-01"}
+            ]
+        }
+
+        with patch.object(nfo_service.tmdb_client, 'search_tv_show', mock_search), \
+             patch.object(nfo_service.tmdb_client, 'search_multi', mock_multi):
+
+            result, source = await nfo_service._search_with_fallback(
+                "Unknown Show", 2024, None
+            )
+
+            assert result["id"] == 456
+            assert source == "multi_search"
+

tests/unit/test_scheduler_service.py

@@ -489,12 +489,12 @@ class TestSingletonHelpers:
 
 
 # ---------------------------------------------------------------------------
-# 12.12  Persistent job store — SQLAlchemyJobStore passed to AsyncIOScheduler
+# 12.12  In-memory job store — no separate scheduler.db needed
 # ---------------------------------------------------------------------------
 
-class TestPersistentJobStore:
+class TestInMemoryJobStore:
     @pytest.mark.asyncio
-    async def test_start_creates_scheduler_with_sqlalchemy_jobstore(
+    async def test_start_creates_scheduler_without_jobstore_arg(
         self, scheduler_service, mock_config_service
     ):
         with patch(
@@ -508,10 +508,9 @@ class TestPersistentJobStore:
 
             MockScheduler.assert_called_once()
             call_kwargs = MockScheduler.call_args
-            jobstores = call_kwargs[1]["jobstores"]
-            assert "default" in jobstores
-            # Verify it's a SQLAlchemyJobStore (class check via module name)
-            assert "sqlalchemy" in type(jobstores["default"]).__module__
+            # No jobstores argument — uses default MemoryJobStore
+            if call_kwargs[1]:
+                assert "jobstores" not in call_kwargs[1]
 
     @pytest.mark.asyncio
     async def test_job_options_include_misfire_grace_and_coalesce(

tests/unit/test_serie_scanner_db_writes.py

@@ -209,7 +209,7 @@ class TestPersistSerieToDbErrorHandling:
 
         with patch(
             "src.server.database.connection.get_async_session_factory",
-            return_value=mock_factory
+            return_value=mock_session
         ):
             with patch(
                 "src.server.database.service.AnimeSeriesService.get_by_key",

tests/unit/test_tmdb_client.py

@@ -444,6 +444,77 @@ class TestTMDBClientSessionLeak:
             "Unexpected warning about unclosed session"
 
 
+class TestTMDBClientLifecycleIntegration:
+    """Integration tests for TMDBClient lifecycle management."""
+    
+    @pytest.mark.asyncio
+    async def test_context_manager_no_resource_warning(self, caplog):
+        """Test async with TMDBClient produces no ResourceWarning."""
+        import logging
+        import warnings
+        caplog.set_level(logging.WARNING)
+        
+        # Use context manager properly - should not leak
+        async with TMDBClient(api_key="test_key") as client:
+            await client._ensure_session()
+            assert client.session is not None
+        
+        # Session should be closed after context exit
+        assert client.session is None or client.session.closed
+    
+    @pytest.mark.asyncio
+    async def test_exception_safety_during_api_call(self, caplog):
+        """Test session is closed even when exception raised during API call."""
+        import logging
+        caplog.set_level(logging.WARNING)
+        
+        close_called = False
+        
+        class TrackingSession:
+            def __init__(self):
+                self.closed = False
+            
+            async def close(self):
+                nonlocal close_called
+                close_called = True
+                self.closed = True
+            
+            async def get(self, url, **kwargs):
+                raise TMDBAPIError("Simulated API failure")
+        
+        client = TMDBClient(api_key="test_key")
+        client.session = TrackingSession()
+        
+        # Exception during context should still close session
+        with pytest.raises(TMDBAPIError):
+            async with client:
+                raise TMDBAPIError("Simulated API failure")
+        
+        assert close_called, "Session was not closed after API exception"
+    
+    @pytest.mark.asyncio
+    async def test_reuse_session_across_multiple_requests(self, caplog):
+        """Test session is reused across multiple requests without leaks."""
+        import logging
+        caplog.set_level(logging.WARNING)
+        
+        client = TMDBClient(api_key="test_key")
+        
+        async with client as c:
+            # First request
+            await c._ensure_session()
+            session1 = c.session
+            
+            # Second request should reuse same session
+            await c._ensure_session()
+            session2 = c.session
+            
+            assert session1 is session2, "Session should be reused"
+        
+        # After context exit, session should be closed
+        assert client.session is None or client.session.closed
+
+
 class TestTMDBClientConnectorClosed:
     """Test handling of 'Connector is closed' errors."""