diff --git a/backend/src/module/api/config.py b/backend/src/module/api/config.py
index 49dd1645..3e271923 100644
--- a/backend/src/module/api/config.py
+++ b/backend/src/module/api/config.py
@@ -1,3 +1,4 @@
+import asyncio
 import logging
 
 from fastapi import APIRouter, Depends
@@ -70,7 +71,8 @@ async def update_config(config: Config, ctx: AppContext = Depends(get_context)):
     """Persist and reload configuration from the supplied payload."""
     try:
         config_dict = _restore_masked(config.dict(), settings.dict())
-        settings.save(config_dict=config_dict)
+        # settings.save() does synchronous file I/O; keep it off the event loop.
+        await asyncio.to_thread(settings.save, config_dict=config_dict)
         # reload_settings reloads from disk, resets the shared HTTP client,
         # rebuilds notifications, and re-applies the RSS/rename loops.
         await ctx.reload_settings()
diff --git a/backend/src/module/api/log.py b/backend/src/module/api/log.py
index f857d6e4..83a7f3eb 100644
--- a/backend/src/module/api/log.py
+++ b/backend/src/module/api/log.py
@@ -1,3 +1,5 @@
+import asyncio
+
 from fastapi import APIRouter, Depends, HTTPException, Response, status
 from fastapi.responses import JSONResponse
 
@@ -11,23 +13,30 @@ router = APIRouter(prefix="/log", tags=["log"])
 _TAIL_BYTES = 512 * 1024  # 512 KB
 
 
+def _read_log_tail() -> bytes:
+    """同步读取日志文件的最后 512 KB，供 asyncio.to_thread 在线程池中执行。"""
+    with open(LOG_PATH, "rb") as f:
+        f.seek(0, 2)
+        size = f.tell()
+        if size > _TAIL_BYTES:
+            f.seek(-_TAIL_BYTES, 2)
+            data = f.read()
+            # Drop first partial line
+            idx = data.find(b"\n")
+            if idx != -1:
+                data = data[idx + 1 :]
+        else:
+            f.seek(0)
+            data = f.read()
+        return data
+
+
 @router.get("", response_model=str, dependencies=[Depends(get_current_user)])
 async def get_log():
     if LOG_PATH.exists():
-        with open(LOG_PATH, "rb") as f:
-            f.seek(0, 2)
-            size = f.tell()
-            if size > _TAIL_BYTES:
-                f.seek(-_TAIL_BYTES, 2)
-                data = f.read()
-                # Drop first partial line
-                idx = data.find(b"\n")
-                if idx != -1:
-                    data = data[idx + 1 :]
-            else:
-                f.seek(0)
-                data = f.read()
-            return Response(data, media_type="text/plain")
+        # Up to 512 KB of sync file I/O; keep it off the event loop.
+        data = await asyncio.to_thread(_read_log_tail)
+        return Response(data, media_type="text/plain")
     else:
         return Response("Log file not found", status_code=404)
 
diff --git a/backend/src/module/api/setup.py b/backend/src/module/api/setup.py
index 9818b775..f0cc8c7b 100644
--- a/backend/src/module/api/setup.py
+++ b/backend/src/module/api/setup.py
@@ -1,3 +1,4 @@
+import asyncio
 import ipaddress
 import logging
 import socket
@@ -5,16 +6,19 @@ from pathlib import Path
 from urllib.parse import urlparse
 
 import httpx
-from fastapi import APIRouter, HTTPException, Request
+from fastapi import APIRouter, Depends, HTTPException, Request
 from pydantic import BaseModel, Field
 
 from module.conf import VERSION, settings
+from module.core import AppContext
 from module.models import Config, ResponseModel
 from module.models.config import NotificationProvider as ProviderConfig
 from module.network import RequestContent
 from module.notification import PROVIDER_REGISTRY
 from module.security.jwt import get_password_hash, verify_password
 
+from .deps import get_context
+
 logger = logging.getLogger(__name__)
 
 router = APIRouter(prefix="/setup", tags=["setup"])
@@ -321,7 +325,11 @@ async def test_notification(req: TestNotificationRequest):
 
 
 @router.post("/complete", response_model=ResponseModel)
-async def complete_setup(req: SetupCompleteRequest, request: Request):
+async def complete_setup(
+    req: SetupCompleteRequest,
+    request: Request,
+    ctx: AppContext = Depends(get_context),
+):
     """Save all wizard configuration and mark setup as complete."""
     _require_setup_needed()
     await _require_default_admin_or_authenticated(request)
@@ -361,10 +369,12 @@ async def complete_setup(req: SetupCompleteRequest, request: Request):
                 ],
             }
 
-        settings.save(config_dict)
-        # Reload settings in-place
-        config_obj = Config.parse_obj(config_dict)
-        settings.__dict__.update(config_obj.__dict__)
+        # settings.save() does synchronous file I/O; keep it off the event loop.
+        await asyncio.to_thread(settings.save, config_dict)
+        # Route through the AppContext reload path so the shared httpx client,
+        # notifier, and scheduler are rebuilt against the new config too — not
+        # just the in-memory settings dict.
+        await ctx.reload_settings()
 
         # 3. Add RSS feed if provided
         if req.rss_url:
diff --git a/backend/src/module/core/context.py b/backend/src/module/core/context.py
index f0f80ebf..85c36ae2 100644
--- a/backend/src/module/core/context.py
+++ b/backend/src/module/core/context.py
@@ -16,7 +16,11 @@ from module.downloader.download_client import shutdown as downloader_shutdown
 from module.models import ResponseModel
 from module.network.request_url import reset_shared_client
 from module.notification import NotificationManager
+from module.parser.analyser.mikan_parser import reset_cache as reset_mikan_cache
+from module.parser.analyser.tmdb_parser import reset_cache as reset_tmdb_cache
+from module.parser.title_parser import reset_cache as reset_openai_parser
 from module.rss import RSSAnalyser
+from module.searcher.searcher import reset_cache as reset_poster_cache
 from module.update import (
     cache_image,
     data_migration,
@@ -247,8 +251,12 @@ class AppContext:
         it inline used to block both the lifespan and the /start,/restart API
         handlers for that long. It now runs as a supervised background task
         tracked on ``self._start_task`` so ``stop()`` can cancel it cleanly.
+
+        Reloads settings through ``reload_settings()`` (the single reload entry
+        point) rather than calling ``settings.load()`` directly; the scheduler
+        is not yet running at this point so its restart branch is a no-op.
         """
-        settings.load()
+        await self.reload_settings()
         if self._start_task is None or self._start_task.done():
             self._start_task = asyncio.create_task(self._run_start_tasks())
             self._start_task.add_done_callback(self._supervise_start_task)
@@ -331,9 +339,23 @@ class AppContext:
     # ------------------------------------------------------------------ reload
 
     async def reload_settings(self) -> None:
-        """Single choreography for a config update: reload + re-apply live state."""
-        settings.load()
+        """Single choreography for a config update: reload + re-apply live state.
+
+        This is the ONLY place settings are reloaded from disk — callers must
+        route through here (not a bare ``settings.load()``) so the HTTP
+        client, notifier, endpoint-keyed caches, and scheduler all stay in
+        sync with the new config.
+        """
+        # settings.load() does synchronous file I/O; keep it off the event loop.
+        await asyncio.to_thread(settings.load)
         await reset_shared_client()
+        # Endpoint-keyed caches (TMDB/Mikan/poster lookups, the OpenAI parser
+        # singleton) must be dropped too, or a changed tmdb_base_url/bgm_base_url
+        # keeps serving results cached from the old endpoint.
+        reset_tmdb_cache()
+        reset_mikan_cache()
+        reset_poster_cache()
+        reset_openai_parser()
         self.notifier.rebuild()
         if self.scheduler.running:
             await self.scheduler.stop_all()
diff --git a/backend/src/module/database/bangumi.py b/backend/src/module/database/bangumi.py
index f8b0afff..0dc807ee 100644
--- a/backend/src/module/database/bangumi.py
+++ b/backend/src/module/database/bangumi.py
@@ -100,6 +100,32 @@ def build_save_path_index(bangumi_list: list[Bangumi]) -> dict[str, Bangumi]:
     return index
 
 
+def _find_semantic_match(data: Bangumi, candidates: list[Bangumi]) -> Optional[Bangumi]:
+    """在已加载的候选列表中查找语义重复项（同一部番剧、命名规则不同）。
+
+    被 ``find_semantic_duplicate``（单条、逐条查库）和 ``add_all``（批量、
+    候选一次性查库后在内存中匹配）共用，匹配规则只维护一处。
+    """
+    for candidate in candidates:
+        is_exact_duplicate = (
+            candidate.title_raw == data.title_raw
+            and candidate.group_name == data.group_name
+        )
+        if is_exact_duplicate:
+            continue
+
+        is_semantic_match = (
+            candidate.dpi == data.dpi
+            and candidate.subtitle == data.subtitle
+            and candidate.source == data.source
+            and _groups_are_similar(candidate.group_name, data.group_name)
+        )
+        if is_semantic_match:
+            return candidate
+
+    return None
+
+
 def match_bangumi_in_list(
     torrent_name: str, bangumi_list: list[Bangumi]
 ) -> Optional[Bangumi]:
@@ -149,31 +175,16 @@ class BangumiDatabase:
         result = await self.session.execute(statement)
         candidates = result.scalars().all()
 
-        for candidate in candidates:
-            is_exact_duplicate = (
-                candidate.title_raw == data.title_raw
-                and candidate.group_name == data.group_name
+        match = _find_semantic_match(data, candidates)
+        if match:
+            logger.debug(
+                "[Database] Found semantic duplicate: '%s' matches "
+                "existing '%s' (official: %s)",
+                data.title_raw,
+                match.title_raw,
+                data.official_title,
             )
-            if is_exact_duplicate:
-                continue
-
-            is_semantic_match = (
-                candidate.dpi == data.dpi
-                and candidate.subtitle == data.subtitle
-                and candidate.source == data.source
-                and _groups_are_similar(candidate.group_name, data.group_name)
-            )
-            if is_semantic_match:
-                logger.debug(
-                    "[Database] Found semantic duplicate: '%s' matches "
-                    "existing '%s' (official: %s)",
-                    data.title_raw,
-                    candidate.title_raw,
-                    data.official_title,
-                )
-                return candidate
-
-        return None
+        return match
 
     async def add_title_alias(
         self, bangumi_id: int, new_title_raw: str, auto_commit: bool = True
@@ -279,11 +290,33 @@ class BangumiDatabase:
         # Filter out exact duplicates
         to_add = [d for d in datas if (d.title_raw, d.group_name) not in existing]
 
+        # Batch query: load all semantic-duplicate candidates (grouped by
+        # official_title) in one SELECT, instead of one find_semantic_duplicate()
+        # query per candidate. Safe because no rows are inserted mid-loop —
+        # add_title_alias() only mutates existing rows and the new bangumi are
+        # only added to the session after this loop finishes.
+        official_titles = {d.official_title for d in to_add}
+        candidates_by_title: dict[str, list[Bangumi]] = {}
+        if official_titles:
+            statement = select(Bangumi).where(
+                and_(
+                    Bangumi.official_title.in_(official_titles),
+                    Bangumi.deleted == false(),
+                )
+            )
+            result = await self.session.execute(statement)
+            for candidate in result.scalars().all():
+                candidates_by_title.setdefault(candidate.official_title, []).append(
+                    candidate
+                )
+
         # Check for semantic duplicates and add as aliases
         semantic_merged = 0
         really_to_add = []
         for d in to_add:
-            semantic_match = await self.find_semantic_duplicate(d)
+            semantic_match = _find_semantic_match(
+                d, candidates_by_title.get(d.official_title, [])
+            )
             if semantic_match:
                 # Add as alias instead of creating new entry (defer commit)
                 await self.add_title_alias(
diff --git a/backend/src/module/notification/providers/telegram.py b/backend/src/module/notification/providers/telegram.py
index 916d3137..f9fdee86 100644
--- a/backend/src/module/notification/providers/telegram.py
+++ b/backend/src/module/notification/providers/telegram.py
@@ -33,7 +33,7 @@ class TelegramProvider(NotificationProvider):
             "disable_notification": True,
         }
 
-        photo = load_image(notification.poster_path)
+        photo = await load_image(notification.poster_path)
         if photo:
             resp = await self.post_files(self.photo_url, data, files={"photo": photo})
         else:
diff --git a/backend/src/module/parser/analyser/mikan_parser.py b/backend/src/module/parser/analyser/mikan_parser.py
index e8eb47f4..86dd25b2 100644
--- a/backend/src/module/parser/analyser/mikan_parser.py
+++ b/backend/src/module/parser/analyser/mikan_parser.py
@@ -16,6 +16,12 @@ _MIKAN_CACHE_MAX = 512
 _mikan_cache: "OrderedDict[str, tuple[str, str]]" = OrderedDict()
 
 
+def reset_cache() -> None:
+    """清空 Mikan 主页解析缓存。配置重载后必须调用，否则会继续返回旧配置下
+    缓存的结果。"""
+    _mikan_cache.clear()
+
+
 def _cache_result(homepage: str, result: tuple[str, str]) -> tuple[str, str]:
     if len(_mikan_cache) >= _MIKAN_CACHE_MAX:
         _mikan_cache.popitem(last=False)
@@ -44,12 +50,15 @@ async def mikan_parser(homepage: str):
             img = await req.get_content(f"https://{root_path}{poster_path}")
             suffix = poster_path.split(".")[-1]
             # img can be None if the poster download failed; don't crash on it.
-            poster_link = save_image(img, suffix) if img else ""
+            poster_link = await save_image(img, suffix) if img else ""
             return _cache_result(homepage, (poster_link, official_title))
         return _cache_result(homepage, ("", official_title))
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     import asyncio
-    homepage = "https://mikanani.me/Home/Episode/c89b3c6f0c1c0567a618f5288b853823c87a9862"
+
+    homepage = (
+        "https://mikanani.me/Home/Episode/c89b3c6f0c1c0567a618f5288b853823c87a9862"
+    )
     print(asyncio.run(mikan_parser(homepage)))
diff --git a/backend/src/module/parser/analyser/tmdb_parser.py b/backend/src/module/parser/analyser/tmdb_parser.py
index 7645d1aa..bd6fd3cb 100644
--- a/backend/src/module/parser/analyser/tmdb_parser.py
+++ b/backend/src/module/parser/analyser/tmdb_parser.py
@@ -23,6 +23,12 @@ _TMDB_CACHE_MAX = 512
 _tmdb_cache: OrderedDict[str, "TMDBInfo | None"] = OrderedDict()
 
 
+def reset_cache() -> None:
+    """清空 TMDB 查询缓存。配置重载（如 tmdb_base_url 变更）后必须调用，否则会
+    继续返回旧接口地址下缓存的结果。"""
+    _tmdb_cache.clear()
+
+
 @dataclass
 class TMDBInfo:
     id: int
@@ -299,7 +305,7 @@ async def tmdb_parser(title, language, test: bool = False) -> TMDBInfo | None:
                         f"https://image.tmdb.org/t/p/w780{poster_path}"
                     )
                     # img is None if the poster download failed; don't crash on it.
-                    poster_link = save_image(img, "jpg") if img else None
+                    poster_link = await save_image(img, "jpg") if img else None
                 else:
                     poster_link = "https://image.tmdb.org/t/p/w780" + poster_path
             else:
diff --git a/backend/src/module/parser/title_parser.py b/backend/src/module/parser/title_parser.py
index 35059b05..34124c3d 100644
--- a/backend/src/module/parser/title_parser.py
+++ b/backend/src/module/parser/title_parser.py
@@ -27,6 +27,14 @@ def _get_openai_parser(kwargs: dict) -> OpenAIParser:
     return _openai_parser
 
 
+def reset_cache() -> None:
+    """清空 OpenAI 解析器单例。配置重载后必须调用，否则会继续使用旧配置
+    （如 api_base/api_key）构建的客户端。"""
+    global _openai_parser, _openai_parser_kwargs
+    _openai_parser = None
+    _openai_parser_kwargs = None
+
+
 class TitleParser:
     def __init__(self):
         pass
diff --git a/backend/src/module/searcher/searcher.py b/backend/src/module/searcher/searcher.py
index bc7bcd2b..8926eaa2 100644
--- a/backend/src/module/searcher/searcher.py
+++ b/backend/src/module/searcher/searcher.py
@@ -1,5 +1,6 @@
 import json
 import logging
+from collections import OrderedDict
 from typing import TypeAlias
 
 from module.models import Bangumi, RSSItem, Torrent
@@ -22,8 +23,17 @@ SEARCH_KEY = [
 
 BangumiJSON: TypeAlias = str
 
-# Cache for TMDB poster lookups by official_title
-_poster_cache: dict[str, str | None] = {}
+# Cache for TMDB poster lookups by official_title. Bounded (LRU-ish,
+# oldest-evicted) like _tmdb_cache/_mikan_cache — was previously a plain dict
+# and grew unbounded for the life of the process.
+_POSTER_CACHE_MAX = 512
+_poster_cache: "OrderedDict[str, str | None]" = OrderedDict()
+
+
+def reset_cache() -> None:
+    """清空 TMDB 海报查询缓存。配置重载（如 tmdb_base_url 变更）后必须调用，
+    否则会继续返回旧接口地址下缓存的结果。"""
+    _poster_cache.clear()
 
 
 class SearchTorrent:
@@ -37,18 +47,21 @@ class SearchTorrent:
     async def _fetch_tmdb_poster(self, title: str) -> str | None:
         """Fetch poster from TMDB if not in cache."""
         if title in _poster_cache:
+            _poster_cache.move_to_end(title)
             return _poster_cache[title]
 
+        poster_link = None
         try:
             tmdb_info = await tmdb_parser(title, "zh", test=True)
             if tmdb_info and tmdb_info.poster_link:
-                _poster_cache[title] = tmdb_info.poster_link
-                return tmdb_info.poster_link
+                poster_link = tmdb_info.poster_link
         except Exception as e:
             logger.debug("[Searcher] Failed to fetch TMDB poster for %s: %s", title, e)
 
-        _poster_cache[title] = None
-        return None
+        if len(_poster_cache) >= _POSTER_CACHE_MAX:
+            _poster_cache.popitem(last=False)
+        _poster_cache[title] = poster_link
+        return poster_link
 
     async def analyse_keyword(
         self, keywords: list[str], site: str = "mikan", limit: int = 100
diff --git a/backend/src/module/update/cross_version.py b/backend/src/module/update/cross_version.py
index 125478d2..f1b2ad9b 100644
--- a/backend/src/module/update/cross_version.py
+++ b/backend/src/module/update/cross_version.py
@@ -63,7 +63,7 @@ async def cache_image():
                         # Hash local path
                         img = await req.get_content(bangumi.poster_link)
                         suffix = bangumi.poster_link.split(".")[-1]
-                        img_path = save_image(img, suffix)
+                        img_path = await save_image(img, suffix)
                         if img_path:
                             bangumi.poster_link = img_path
                     except Exception as e:
diff --git a/backend/src/module/utils/cache_image.py b/backend/src/module/utils/cache_image.py
index 96915a79..5f3e0bb4 100644
--- a/backend/src/module/utils/cache_image.py
+++ b/backend/src/module/utils/cache_image.py
@@ -1,21 +1,32 @@
+import asyncio
 import hashlib
 
 
-def save_image(img, suffix):
+async def save_image(img: bytes | None, suffix: str) -> str | None:
+    """保存图片到本地缓存。文件写入是同步阻塞的，放到线程池中执行，避免
+    在 RSS/通知等异步热路径上阻塞事件循环。"""
     if img is None:
         # Fetching the poster failed upstream; skip caching instead of
         # crashing on hashlib.md5(None).
         return None
     img_hash = hashlib.md5(img).hexdigest()[0:8]
     image_path = f"data/posters/{img_hash}.{suffix}"
-    with open(image_path, "wb") as f:
-        f.write(img)
+
+    def _write() -> None:
+        with open(image_path, "wb") as f:
+            f.write(img)
+
+    await asyncio.to_thread(_write)
     return f"posters/{img_hash}.{suffix}"
 
 
-def load_image(img_path):
-    if img_path:
+async def load_image(img_path: str | None) -> bytes | None:
+    """读取缓存图片。文件读取是同步阻塞的，放到线程池中执行。"""
+    if not img_path:
+        return None
+
+    def _read() -> bytes:
         with open(f"data/{img_path}", "rb") as f:
             return f.read()
-    else:
-        return None
+
+    return await asyncio.to_thread(_read)
diff --git a/backend/src/test/test_cache_image.py b/backend/src/test/test_cache_image.py
new file mode 100644
index 00000000..9b0ddaf2
--- /dev/null
+++ b/backend/src/test/test_cache_image.py
@@ -0,0 +1,88 @@
+"""Tests for module.utils.cache_image: async, off-event-loop file I/O.
+
+save_image/load_image do real file I/O and are awaited from the RSS hot path
+and telegram sends; they must not block the event loop. Tests chdir into a
+tmp_path so they never touch the real backend/data/ directory.
+"""
+
+import asyncio
+
+from module.utils.cache_image import load_image, save_image
+
+
+class TestSaveImage:
+    async def test_save_image_returns_none_when_img_is_none(self):
+        """Fetching the poster failed upstream: skip caching, no filesystem I/O."""
+        result = await save_image(None, "jpg")
+        assert result is None
+
+    async def test_save_image_writes_file_and_returns_relative_path(
+        self, tmp_path, monkeypatch
+    ):
+        monkeypatch.chdir(tmp_path)
+        (tmp_path / "data" / "posters").mkdir(parents=True)
+
+        result = await save_image(b"fake-image-bytes", "jpg")
+
+        assert result is not None
+        assert result.startswith("posters/")
+        assert result.endswith(".jpg")
+        assert (tmp_path / "data" / result).read_bytes() == b"fake-image-bytes"
+
+    async def test_save_image_offloads_write_to_a_thread(self, tmp_path, monkeypatch):
+        """The write must go through asyncio.to_thread, not run inline on the
+        event loop (regression test for the blocking-I/O fix)."""
+        monkeypatch.chdir(tmp_path)
+        (tmp_path / "data" / "posters").mkdir(parents=True)
+
+        real_to_thread = asyncio.to_thread
+        calls = []
+
+        async def spy_to_thread(func, /, *args, **kwargs):
+            calls.append(func)
+            return await real_to_thread(func, *args, **kwargs)
+
+        monkeypatch.setattr("module.utils.cache_image.asyncio.to_thread", spy_to_thread)
+
+        await save_image(b"payload", "png")
+
+        assert len(calls) == 1
+
+
+class TestLoadImage:
+    async def test_load_image_returns_none_for_falsy_path(self):
+        assert await load_image(None) is None
+        assert await load_image("") is None
+
+    async def test_load_image_reads_file_relative_to_data_dir(
+        self, tmp_path, monkeypatch
+    ):
+        monkeypatch.chdir(tmp_path)
+        posters_dir = tmp_path / "data" / "posters"
+        posters_dir.mkdir(parents=True)
+        (posters_dir / "abc123.jpg").write_bytes(b"cached-poster-bytes")
+
+        result = await load_image("posters/abc123.jpg")
+
+        assert result == b"cached-poster-bytes"
+
+    async def test_load_image_offloads_read_to_a_thread(self, tmp_path, monkeypatch):
+        """The read must go through asyncio.to_thread, not run inline on the
+        event loop (regression test for the blocking-I/O fix)."""
+        monkeypatch.chdir(tmp_path)
+        posters_dir = tmp_path / "data" / "posters"
+        posters_dir.mkdir(parents=True)
+        (posters_dir / "abc123.jpg").write_bytes(b"cached-poster-bytes")
+
+        real_to_thread = asyncio.to_thread
+        calls = []
+
+        async def spy_to_thread(func, /, *args, **kwargs):
+            calls.append(func)
+            return await real_to_thread(func, *args, **kwargs)
+
+        monkeypatch.setattr("module.utils.cache_image.asyncio.to_thread", spy_to_thread)
+
+        await load_image("posters/abc123.jpg")
+
+        assert len(calls) == 1
diff --git a/backend/src/test/test_context.py b/backend/src/test/test_context.py
index ba7149c8..81ee3493 100644
--- a/backend/src/test/test_context.py
+++ b/backend/src/test/test_context.py
@@ -17,6 +17,10 @@ def ctx():
     ctx.scheduler = MagicMock()
     ctx.scheduler.stop_all = AsyncMock()
     ctx.scheduler.start_all = MagicMock()
+    # A freshly built context has not started its scheduler yet; set this
+    # explicitly since a bare MagicMock attribute is truthy by default and
+    # would make reload_settings() think a restart is needed.
+    ctx.scheduler.running = False
     return ctx
 
 
@@ -113,6 +117,7 @@ class TestStartTasks:
     async def test_start_tasks_waits_then_starts_scheduler(self, ctx):
         with (
             patch("module.core.context.settings"),
+            patch("module.core.context.reset_shared_client", new=AsyncMock()),
             patch(
                 "module.core.context.Checker.check_downloader",
                 new=AsyncMock(return_value=True),
@@ -129,6 +134,19 @@ class TestStartTasks:
         assert isinstance(resp, ResponseModel)
         assert resp.status is True
 
+    async def test_start_tasks_routes_through_reload_settings(self, ctx):
+        """start_tasks() must go through reload_settings() (the single reload
+        entry point) instead of calling settings.load() directly."""
+        ctx.reload_settings = AsyncMock()
+        with patch(
+            "module.core.context.Checker.check_downloader",
+            new=AsyncMock(return_value=True),
+        ):
+            await ctx.start_tasks()
+            await ctx._start_task
+
+        ctx.reload_settings.assert_awaited_once()
+
 
 # ---------------------------------------------------------------------------
 # stop()
@@ -200,3 +218,23 @@ class TestReloadSettings:
         ctx.notifier.rebuild.assert_called_once()
         ctx.scheduler.stop_all.assert_not_awaited()
         ctx.scheduler.start_all.assert_not_called()
+
+    async def test_reload_resets_endpoint_keyed_caches(self, ctx):
+        """A changed tmdb_base_url/bgm_base_url must not keep serving results
+        cached from the old endpoint — reload_settings() must drop the TMDB,
+        Mikan, and poster caches plus the OpenAI parser singleton."""
+        ctx.scheduler.running = False
+        with (
+            patch("module.core.context.settings"),
+            patch("module.core.context.reset_shared_client", new=AsyncMock()),
+            patch("module.core.context.reset_tmdb_cache") as mock_tmdb,
+            patch("module.core.context.reset_mikan_cache") as mock_mikan,
+            patch("module.core.context.reset_poster_cache") as mock_poster,
+            patch("module.core.context.reset_openai_parser") as mock_openai,
+        ):
+            await ctx.reload_settings()
+
+        mock_tmdb.assert_called_once()
+        mock_mikan.assert_called_once()
+        mock_poster.assert_called_once()
+        mock_openai.assert_called_once()
diff --git a/backend/src/test/test_database.py b/backend/src/test/test_database.py
index b90f080d..57106181 100644
--- a/backend/src/test/test_database.py
+++ b/backend/src/test/test_database.py
@@ -9,16 +9,18 @@ from module.models import Bangumi, RSSItem, Torrent
 async def _ensure_bangumi(session, bangumi_id: int):
     """确保 bangumi 表中存在指定 id 的记录，满足外键约束。"""
     if await session.get(Bangumi, bangumi_id) is None:
-        session.add(Bangumi(
-            id=bangumi_id,
-            official_title=f"Stub Anime {bangumi_id}",
-            title_raw=f"Stub {bangumi_id}",
-            group_name="TestGroup",
-            dpi="1080p",
-            source="Web",
-            subtitle="CHT",
-            rss_link=f"stub_{bangumi_id}",
-        ))
+        session.add(
+            Bangumi(
+                id=bangumi_id,
+                official_title=f"Stub Anime {bangumi_id}",
+                title_raw=f"Stub {bangumi_id}",
+                group_name="TestGroup",
+                dpi="1080p",
+                source="Web",
+                subtitle="CHT",
+                rss_link=f"stub_{bangumi_id}",
+            )
+        )
         await session.commit()
 
 
@@ -445,6 +447,102 @@ async def test_add_with_semantic_duplicate_creates_alias(db_session):
     assert "Frieren Beyond Journey's End" in aliases
 
 
+async def test_add_all_semantic_duplicates_across_batch_all_merged_as_aliases(
+    db_session,
+):
+    """add_all() must catch semantic duplicates for every item in a batch, not
+    just the first (regression test for the N+1 batching fix)."""
+    db = BangumiDatabase(db_session)
+
+    for i in range(5):
+        await db.add(
+            Bangumi(
+                official_title=f"Anime {i}",
+                title_raw=f"Raw {i}",
+                group_name="GroupA",
+                dpi="1080p",
+                source="Web",
+                subtitle="CHT",
+                rss_link=f"rss{i}",
+            )
+        )
+
+    batch = [
+        Bangumi(
+            official_title=f"Anime {i}",
+            title_raw=f"Raw {i} New",
+            group_name="GroupA&GroupB",  # naming changed mid-season
+            dpi="1080p",
+            source="Web",
+            subtitle="CHT",
+            rss_link=f"rss{i}-new",
+        )
+        for i in range(5)
+    ]
+
+    added = await db.add_all(batch)
+
+    assert added == 0  # every item merged as an alias, no new rows
+    assert len(await db.search_all()) == 5
+    for i in range(5):
+        original = await db.search_official_title(f"Anime {i}")
+        aliases = json.loads(original.title_aliases) if original.title_aliases else []
+        assert f"Raw {i} New" in aliases
+
+
+async def test_add_all_semantic_duplicate_lookup_uses_batched_query(
+    db_session, db_engine
+):
+    """add_all() loads semantic-duplicate candidates in one batched SELECT
+    instead of one find_semantic_duplicate() query per candidate."""
+    from sqlalchemy import event
+
+    db = BangumiDatabase(db_session)
+
+    for i in range(5):
+        await db.add(
+            Bangumi(
+                official_title=f"Anime {i}",
+                title_raw=f"Raw {i}",
+                group_name="GroupA",
+                dpi="1080p",
+                source="Web",
+                subtitle="CHT",
+                rss_link=f"rss{i}",
+            )
+        )
+
+    batch = [
+        Bangumi(
+            official_title=f"Anime {i}",
+            title_raw=f"Raw {i} New",
+            group_name="GroupA&GroupB",
+            dpi="1080p",
+            source="Web",
+            subtitle="CHT",
+            rss_link=f"rss{i}-new",
+        )
+        for i in range(5)
+    ]
+
+    select_statements = []
+
+    def _record(conn, cursor, statement, parameters, context, executemany):
+        if statement.strip().upper().startswith("SELECT") and "bangumi" in statement:
+            select_statements.append(statement)
+
+    event.listen(db_engine.sync_engine, "before_cursor_execute", _record)
+    try:
+        await db.add_all(batch)
+    finally:
+        event.remove(db_engine.sync_engine, "before_cursor_execute", _record)
+
+    # Exactly one SELECT for the exact-duplicate check plus one for the
+    # semantic-duplicate candidates: O(1), not one per batch item (would be
+    # 1 + 5 = 6 before the fix).
+    assert len(select_statements) == 2
+
+
 class TestDeleteByBangumiId:
     """Tests for TorrentDatabase.delete_by_bangumi_id."""
 
@@ -452,7 +550,11 @@ class TestDeleteByBangumiId:
         db = TorrentDatabase(db_session)
         await _ensure_bangumi(db_session, 10)
         for i in range(3):
-            await db.add(Torrent(name=f"torrent_{i}", url=f"https://example.com/{i}", bangumi_id=10))
+            await db.add(
+                Torrent(
+                    name=f"torrent_{i}", url=f"https://example.com/{i}", bangumi_id=10
+                )
+            )
         assert len(await db.search_all()) == 3
 
         count = await db.delete_by_bangumi_id(10)
@@ -463,8 +565,12 @@ class TestDeleteByBangumiId:
         db = TorrentDatabase(db_session)
         await _ensure_bangumi(db_session, 20)
         await _ensure_bangumi(db_session, 30)
-        await db.add(Torrent(name="keep", url="https://example.com/keep", bangumi_id=20))
-        await db.add(Torrent(name="delete", url="https://example.com/delete", bangumi_id=30))
+        await db.add(
+            Torrent(name="keep", url="https://example.com/keep", bangumi_id=20)
+        )
+        await db.add(
+            Torrent(name="delete", url="https://example.com/delete", bangumi_id=30)
+        )
 
         count = await db.delete_by_bangumi_id(30)
         assert count == 1
@@ -475,7 +581,9 @@ class TestDeleteByBangumiId:
     async def test_no_match_returns_zero(self, db_session):
         db = TorrentDatabase(db_session)
         await _ensure_bangumi(db_session, 5)
-        await db.add(Torrent(name="unrelated", url="https://example.com/1", bangumi_id=5))
+        await db.add(
+            Torrent(name="unrelated", url="https://example.com/1", bangumi_id=5)
+        )
 
         count = await db.delete_by_bangumi_id(999)
         assert count == 0
@@ -484,8 +592,12 @@ class TestDeleteByBangumiId:
     async def test_skips_null_bangumi_id(self, db_session):
         db = TorrentDatabase(db_session)
         await _ensure_bangumi(db_session, 7)
-        await db.add(Torrent(name="orphan", url="https://example.com/orphan", bangumi_id=None))
-        await db.add(Torrent(name="target", url="https://example.com/target", bangumi_id=7))
+        await db.add(
+            Torrent(name="orphan", url="https://example.com/orphan", bangumi_id=None)
+        )
+        await db.add(
+            Torrent(name="target", url="https://example.com/target", bangumi_id=7)
+        )
 
         count = await db.delete_by_bangumi_id(7)
         assert count == 1
@@ -609,10 +721,16 @@ class TestRSSDeleteWithTorrents:
         rss = RSSItem(url="https://mikanani.me/RSS/test", name="Test RSS")
         await rss_db.add(rss)
 
-        await torrent_db.add(Torrent(name="ep01", url="https://example.com/1", rss_id=rss.id))
-        await torrent_db.add(Torrent(name="ep02", url="https://example.com/2", rss_id=rss.id))
+        await torrent_db.add(
+            Torrent(name="ep01", url="https://example.com/1", rss_id=rss.id)
+        )
+        await torrent_db.add(
+            Torrent(name="ep02", url="https://example.com/2", rss_id=rss.id)
+        )
         # 不关联此 RSS 的 torrent
-        await torrent_db.add(Torrent(name="other", url="https://example.com/3", rss_id=None))
+        await torrent_db.add(
+            Torrent(name="other", url="https://example.com/3", rss_id=None)
+        )
 
         assert len(await torrent_db.search_rss(rss.id)) == 2
 
@@ -647,7 +765,9 @@ class TestRSSDeleteWithTorrents:
 
         for i in range(5):
             await torrent_db.add(
-                Torrent(name=f"ep{i:02d}", url=f"https://example.com/{i}", rss_id=rss.id)
+                Torrent(
+                    name=f"ep{i:02d}", url=f"https://example.com/{i}", rss_id=rss.id
+                )
             )
 
         assert len(await torrent_db.search_rss(rss.id)) == 5
@@ -668,10 +788,16 @@ class TestRSSDeleteWithTorrents:
         await rss_db.add(rss1)
         await rss_db.add(rss2)
 
-        await torrent_db.add(Torrent(name="t1", url="https://example.com/1", rss_id=rss1.id))
-        await torrent_db.add(Torrent(name="t2", url="https://example.com/2", rss_id=rss2.id))
+        await torrent_db.add(
+            Torrent(name="t1", url="https://example.com/1", rss_id=rss1.id)
+        )
+        await torrent_db.add(
+            Torrent(name="t2", url="https://example.com/2", rss_id=rss2.id)
+        )
         # rss_id 为 None 的 torrent 应不受影响
-        await torrent_db.add(Torrent(name="orphan", url="https://example.com/3", rss_id=None))
+        await torrent_db.add(
+            Torrent(name="orphan", url="https://example.com/3", rss_id=None)
+        )
 
         await rss_db.delete_all()
 
diff --git a/backend/src/test/test_mikan_parser.py b/backend/src/test/test_mikan_parser.py
new file mode 100644
index 00000000..ed7bd9a8
--- /dev/null
+++ b/backend/src/test/test_mikan_parser.py
@@ -0,0 +1,23 @@
+"""Tests for the Mikan homepage parser's module-level cache."""
+
+import importlib
+
+# `module.parser.analyser.__init__` re-exports the `mikan_parser` function under
+# the same name as this submodule, shadowing the submodule on the package
+# object — so `import module.parser.analyser.mikan_parser as x` would resolve
+# to the function, not the module. Go through importlib to get the module.
+mikan_parser_module = importlib.import_module("module.parser.analyser.mikan_parser")
+
+
+def test_reset_cache_clears_mikan_cache():
+    """reset_cache() must drop all cached homepage lookups (called on config
+    reload so a changed endpoint stops serving stale results)."""
+    mikan_parser_module._mikan_cache["https://mikanani.me/Home/Episode/stale"] = (
+        "",
+        "Stale Title",
+    )
+    assert len(mikan_parser_module._mikan_cache) > 0
+
+    mikan_parser_module.reset_cache()
+
+    assert len(mikan_parser_module._mikan_cache) == 0
diff --git a/backend/src/test/test_searcher.py b/backend/src/test/test_searcher.py
index b6094a7f..199038d4 100644
--- a/backend/src/test/test_searcher.py
+++ b/backend/src/test/test_searcher.py
@@ -1,6 +1,6 @@
 """Tests for search providers: URL construction, keyword handling."""
 
-from unittest.mock import patch
+from unittest.mock import AsyncMock, patch
 
 import pytest
 
@@ -92,9 +92,12 @@ class TestSpecialUrl:
             subtitle="CHT",
         )
 
-        with patch("module.searcher.provider.SEARCH_CONFIG", {
-            "mikan": "https://mikanani.me/RSS/Search?searchstr=%s",
-        }):
+        with patch(
+            "module.searcher.provider.SEARCH_CONFIG",
+            {
+                "mikan": "https://mikanani.me/RSS/Search?searchstr=%s",
+            },
+        ):
             result = SearchTorrent.special_url(bangumi, "mikan")
 
         assert isinstance(result, RSSItem)
@@ -116,10 +119,59 @@ class TestSpecialUrl:
             subtitle=None,
         )
 
-        with patch("module.searcher.provider.SEARCH_CONFIG", {
-            "mikan": "https://mikanani.me/RSS/Search?searchstr=%s",
-        }):
+        with patch(
+            "module.searcher.provider.SEARCH_CONFIG",
+            {
+                "mikan": "https://mikanani.me/RSS/Search?searchstr=%s",
+            },
+        ):
             result = SearchTorrent.special_url(bangumi, "mikan")
 
         # Only title_raw should be in the URL
         assert "Test" in result.url
+
+
+# ---------------------------------------------------------------------------
+# _poster_cache: bounded LRU + reset_cache()
+# ---------------------------------------------------------------------------
+
+
+class TestPosterCache:
+    @pytest.fixture(autouse=True)
+    def _clean_poster_cache(self):
+        """Isolate the module-level poster cache between tests."""
+        from module.searcher import searcher as searcher_module
+
+        searcher_module.reset_cache()
+        yield
+        searcher_module.reset_cache()
+
+    def test_reset_cache_clears_poster_cache(self):
+        from module.searcher import searcher as searcher_module
+
+        searcher_module._poster_cache["Test Anime"] = "http://example.com/p.jpg"
+        assert len(searcher_module._poster_cache) > 0
+
+        searcher_module.reset_cache()
+
+        assert len(searcher_module._poster_cache) == 0
+
+    async def test_poster_cache_evicts_oldest_when_full(self, monkeypatch):
+        """_poster_cache is bounded (LRU-ish) like _tmdb_cache/_mikan_cache,
+        instead of growing without limit for the life of the process."""
+        from module.searcher import searcher as searcher_module
+        from module.searcher.searcher import SearchTorrent
+
+        monkeypatch.setattr(searcher_module, "_POSTER_CACHE_MAX", 3)
+        torrent = SearchTorrent()
+
+        with patch(
+            "module.searcher.searcher.tmdb_parser", new=AsyncMock(return_value=None)
+        ):
+            for i in range(4):
+                await torrent._fetch_tmdb_poster(f"Title {i}")
+
+        assert len(searcher_module._poster_cache) == 3
+        # The oldest entry ("Title 0") was evicted; the rest remain.
+        assert "Title 0" not in searcher_module._poster_cache
+        assert "Title 3" in searcher_module._poster_cache
diff --git a/backend/src/test/test_setup.py b/backend/src/test/test_setup.py
index d342a21a..57f18e0b 100644
--- a/backend/src/test/test_setup.py
+++ b/backend/src/test/test_setup.py
@@ -1,20 +1,31 @@
 from pathlib import Path
-from unittest.mock import AsyncMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 from fastapi.testclient import TestClient
 
+from module.api.deps import get_context
 from module.api.setup import SENTINEL_PATH, router
 
 
 @pytest.fixture
-def client():
+def mock_ctx():
+    """Mock AppContext whose reload_settings is an awaitable no-op."""
+    ctx = MagicMock()
+    ctx.reload_settings = AsyncMock()
+    return ctx
+
+
+@pytest.fixture
+def client(mock_ctx):
     """Create a test client for the FastAPI app."""
     from fastapi import FastAPI
 
     app = FastAPI()
     app.include_router(router, prefix="/api/v1")
-    return TestClient(app)
+    app.dependency_overrides[get_context] = lambda: mock_ctx
+    yield TestClient(app)
+    app.dependency_overrides.clear()
 
 
 @pytest.fixture
@@ -293,6 +304,56 @@ class TestRequestValidation:
         assert response.status_code == 422
 
 
+class TestSetupComplete:
+    """Issue: setup wizard must route config reload through AppContext, not
+    mutate settings.__dict__ directly (the shared httpx client, notifier, and
+    scheduler must be rebuilt after first-run setup)."""
+
+    @staticmethod
+    def _payload():
+        return {
+            "username": "testuser",
+            "password": "testpassword123",
+            "downloader_type": "qbittorrent",
+            "downloader_host": "localhost:8080",
+            "downloader_username": "admin",
+            "downloader_password": "admin",
+            "downloader_path": "/downloads",
+            "downloader_ssl": False,
+            "rss_url": "",
+            "rss_name": "",
+            "notification_enable": False,
+            "notification_type": "telegram",
+            "notification_token": "",
+            "notification_chat_id": "",
+        }
+
+    def test_complete_routes_through_ctx_reload_settings(
+        self, client, mock_first_run, mock_ctx
+    ):
+        """A successful /setup/complete call saves the config and awaits
+        ctx.reload_settings(), instead of poking settings.__dict__ directly."""
+        from fastapi import HTTPException
+
+        with patch("module.database.Database") as mock_db_cls:
+            db_instance = AsyncMock()
+            # No admin account yet: _require_default_admin_or_authenticated()
+            # treats this as a fresh install and lets setup proceed.
+            db_instance.user.get_user = AsyncMock(
+                side_effect=HTTPException(status_code=404, detail="not found")
+            )
+            db_instance.user.update_user = AsyncMock(return_value=True)
+            mock_db_cls.return_value.__aenter__ = AsyncMock(return_value=db_instance)
+            mock_db_cls.return_value.__aexit__ = AsyncMock(return_value=False)
+
+            response = client.post("/api/v1/setup/complete", json=self._payload())
+
+        assert response.status_code == 200
+        data = response.json()
+        assert data["status"] is True
+        mock_ctx.reload_settings.assert_awaited_once()
+
+
 class TestSentinelPath:
     def test_sentinel_path_is_in_config_dir(self):
         assert str(SENTINEL_PATH) == "config/.setup_complete"
@@ -379,9 +440,7 @@ class TestTestDownloaderHardening:
         from unittest.mock import MagicMock
 
         get_resp = MagicMock(text="qBittorrent WebUI")
-        login_resp = MagicMock(
-            status_code=200, text="<html><body>portal</body></html>"
-        )
+        login_resp = MagicMock(status_code=200, text="<html><body>portal</body></html>")
         cls_patch = self._mock_client(get_resp=get_resp, login_resp=login_resp)
         try:
             response = self._post(client)
diff --git a/backend/src/test/test_title_parser.py b/backend/src/test/test_title_parser.py
index 1fc792fa..656b2118 100644
--- a/backend/src/test/test_title_parser.py
+++ b/backend/src/test/test_title_parser.py
@@ -1,6 +1,7 @@
 import pytest
 
 from module.conf import settings
+from module.parser import title_parser as title_parser_module
 from module.parser.title_parser import TitleParser
 
 
@@ -26,3 +27,20 @@ class TestTitleParser:
         assert result.dpi == "1080P"
         assert result.season == 1
         assert result.subtitle == "GB_JP"
+
+
+class TestOpenAIParserResetCache:
+    """reset_cache() must drop the lazy OpenAI-parser singleton on config
+    reload, or a changed api_key/api_base keeps using the old client."""
+
+    def teardown_method(self):
+        title_parser_module.reset_cache()
+
+    def test_reset_cache_clears_singleton(self):
+        title_parser_module._openai_parser = object()
+        title_parser_module._openai_parser_kwargs = {"api_key": "old"}
+
+        title_parser_module.reset_cache()
+
+        assert title_parser_module._openai_parser is None
+        assert title_parser_module._openai_parser_kwargs is None
diff --git a/backend/src/test/test_tmdb.py b/backend/src/test/test_tmdb.py
index e81805e4..00bf3fa3 100644
--- a/backend/src/test/test_tmdb.py
+++ b/backend/src/test/test_tmdb.py
@@ -70,6 +70,17 @@ async def test_tmdb_parser(mocker):
     assert tmdb_info.last_season == bangumi_season
 
 
+def test_reset_cache_clears_tmdb_cache():
+    """reset_cache() must drop all cached lookups (called on config reload so
+    a changed tmdb_base_url stops serving results from the old endpoint)."""
+    tmdb_parser_module._tmdb_cache["stale-key"] = None
+    assert len(tmdb_parser_module._tmdb_cache) > 0
+
+    tmdb_parser_module.reset_cache()
+
+    assert len(tmdb_parser_module._tmdb_cache) == 0
+
+
 @pytest.mark.live
 @pytest.mark.skipif(
     not os.environ.get("RUN_LIVE_TMDB_TESTS"),