feat(4.0-P2.5): 下载操作按实例路由，重命名逐实例运行，不可用实例跳过并通知

- DownloadClient(instance_id) 按实例缓存客户端，凭据闩锁按实例隔离；DownloaderPool 在一次操作内惰性进入多个实例
- 新种子投递到规则的实例，规则未指定时用订阅的，再退回默认实例；种子行记录实际实例
- 规则 / 订阅 / 电影的 downloader_id 为空表示跟随默认实例（修改未发布的迁移 v26）
- rename_all 逐实例重命名，事件带真实 downloader_id；版本替换事务按实例过滤；不可用实例只在首次不可用时发 DownloaderUnavailableEvent
- 删除规则按种子所在的各实例删除；规则换实例时旧种子留在原位置
- /downloader/torrents 汇总所有实例并标注实例，暂停 / 恢复 / 删除 / 打标按 downloader_id 路由，新增 /downloader/instances

backend/src/ab_sdk/events.py

@@ -166,12 +166,13 @@ class DownloaderUnavailableEvent(SystemEvent):
 
     host: str
     reason: str  # unreachable | credentials | banned
+    instance_id: str = "default"  # 下载器实例 id（plugins.instances）
 
     def dedup_key(self) -> str | None:
-        return f"downloader:{self.host}"
+        return f"downloader:{self.instance_id}"
 
     def payload(self) -> dict:
-        return {"host": self.host, "reason": self.reason}
+        return {"host": self.host, "reason": self.reason, "instance": self.instance_id}
 
     def describe(self) -> tuple[str, str]:
         details = {
@@ -180,7 +181,10 @@ class DownloaderUnavailableEvent(SystemEvent):
             "unreachable": "无法连接下载器，请检查地址、端口和网络。",
         }
         detail = details.get(self.reason, details["unreachable"])
-        return ("下载器连接异常", f"下载器：{self.host}\n{detail}")
+        return (
+            "下载器连接异常",
+            f"下载器：{self.instance_id}（{self.host}）\n{detail}",
+        )
 
 
 @dataclass(frozen=True, slots=True)

backend/src/module/api/bangumi.py

@@ -6,8 +6,8 @@ from pydantic import BaseModel
 
 from module.conf import settings
 from module.database import Database, get_db
-from module.downloader import DownloadClient
-from module.manager import Renamer, TorrentManager
+from module.manager import TorrentManager
+from module.manager.renamer import rename_all
 from module.models import APIResponse, Bangumi, BangumiUpdate, ResponseModel, Torrent
 from module.parser.analyser.offset_detector import detect_offset_mismatch
 from module.parser.analyser.tmdb_parser import tmdb_parser
@@ -376,9 +376,7 @@ async def dismiss_review(bangumi_id: int):
 
 async def _trigger_rename() -> None:
     """Run a rename pass so applied offsets take effect immediately."""
-    async with DownloadClient() as client:
-        renamer = Renamer(client)
-        await renamer.rename()
+    await rename_all()
 
 
 # Registered before /apply-offset/{bangumi_id} so "many" is never captured as an id.

backend/src/module/api/downloader.py

@@ -4,13 +4,15 @@ import re
 from fastapi import APIRouter, Depends, HTTPException
 from pydantic import BaseModel
 
+from module.conf import settings
 from module.database import Database
 from module.database.bangumi import (
     build_save_path_index,
     match_bangumi_in_list,
     normalize_save_path,
 )
-from module.downloader import DownloadClient
+from module.downloader import DownloadClient, downloader_ids, list_torrents
+from module.models.config import DOWNLOADER_POINT
 from module.security.api import get_current_user
 
 logger = logging.getLogger(__name__)
@@ -20,10 +22,11 @@ router = APIRouter(prefix="/downloader", tags=["downloader"])
 
 class TorrentHashesRequest(BaseModel):
     hashes: list[str]
+    # 种子所在的下载器实例；为空时为默认实例
+    downloader_id: str | None = None
 
 
-class TorrentDeleteRequest(BaseModel):
-    hashes: list[str]
+class TorrentDeleteRequest(TorrentHashesRequest):
     delete_files: bool = False
 
 
@@ -32,12 +35,37 @@ class TorrentTagRequest(BaseModel):
 
     hash: str
     bangumi_id: int
+    downloader_id: str | None = None
+
+
+def _client(downloader_id: str | None) -> DownloadClient:
+    if downloader_id is not None and downloader_id not in downloader_ids():
+        raise HTTPException(
+            status_code=404, detail=f"Downloader instance {downloader_id} not found"
+        )
+    return DownloadClient(downloader_id)
+
+
+@router.get("/instances", dependencies=[Depends(get_current_user)])
+async def get_instances():
+    """下载器实例列表（供规则 / 订阅选择下载器）。"""
+    return {
+        "default": settings.plugins.slots.downloader,
+        "instances": [
+            {"id": i.id, "provider": i.provider}
+            for i in settings.plugins.instances
+            if i.point == DOWNLOADER_POINT
+        ],
+    }
 
 
 @router.get("/torrents", dependencies=[Depends(get_current_user)])
 async def get_torrents():
-    async with DownloadClient() as client:
-        return await client.get_torrent_info(category="Bangumi", status_filter=None)
+    """所有下载器实例的种子（每条带 downloader_id）；不可用的实例跳过。"""
+    torrents = await list_torrents()
+    if torrents is None:
+        raise HTTPException(status_code=503, detail="No downloader is reachable")
+    return torrents
 
 
 @router.get("/rename-conflicts", dependencies=[Depends(get_current_user)])
@@ -79,7 +107,7 @@ async def retry_rename_conflict(operation_id: int):
 @router.post("/torrents/pause", dependencies=[Depends(get_current_user)])
 async def pause_torrents(req: TorrentHashesRequest):
     hashes = "|".join(req.hashes)
-    async with DownloadClient() as client:
+    async with _client(req.downloader_id) as client:
         await client.pause_torrent(hashes)
     return {"msg_en": "Torrents paused", "msg_zh": "种子已暂停"}
 
@@ -87,7 +115,7 @@ async def pause_torrents(req: TorrentHashesRequest):
 @router.post("/torrents/resume", dependencies=[Depends(get_current_user)])
 async def resume_torrents(req: TorrentHashesRequest):
     hashes = "|".join(req.hashes)
-    async with DownloadClient() as client:
+    async with _client(req.downloader_id) as client:
         await client.resume_torrent(hashes)
     return {"msg_en": "Torrents resumed", "msg_zh": "种子已恢复"}
 
@@ -95,7 +123,7 @@ async def resume_torrents(req: TorrentHashesRequest):
 @router.post("/torrents/delete", dependencies=[Depends(get_current_user)])
 async def delete_torrents(req: TorrentDeleteRequest):
     hashes = "|".join(req.hashes)
-    async with DownloadClient() as client:
+    async with _client(req.downloader_id) as client:
         ok = await client.delete_torrent(hashes, delete_files=req.delete_files)
     if not ok:
         return {
@@ -124,7 +152,7 @@ async def tag_torrent(req: TorrentTagRequest):
             }
 
     tag = f"ab:{req.bangumi_id}"
-    async with DownloadClient() as client:
+    async with _client(req.downloader_id) as client:
         await client.add_tag(req.hash, tag)
 
     return {
@@ -150,42 +178,45 @@ async def auto_tag_torrents():
         bangumi_list = await db.bangumi.search_all()
     save_path_index = build_save_path_index(bangumi_list)
 
-    async with DownloadClient() as client:
-        # Get all Bangumi torrents
-        torrents = await client.get_torrent_info(category="Bangumi", status_filter=None)
+    for instance_id in downloader_ids():
+        async with DownloadClient(instance_id) as client:
+            # Get all Bangumi torrents
+            torrents = await client.get_torrent_info(
+                category="Bangumi", status_filter=None
+            )
 
-        for torrent in torrents:
-            torrent_hash = torrent["hash"]
-            torrent_name = torrent["name"]
-            save_path = torrent["save_path"]
-            tags = torrent.get("tags", "")
+            for torrent in torrents:
+                torrent_hash = torrent["hash"]
+                torrent_name = torrent["name"]
+                save_path = torrent["save_path"]
+                tags = torrent.get("tags", "")
 
-            # Skip if already has an ab:<id> link tag。必须精确匹配数字 id：
-            # ab:renamed（处理完成标记）等同前缀标签不代表已关联番剧
-            if re.search(r"ab:\d+", tags):
-                continue
+                # Skip if already has an ab:<id> link tag。必须精确匹配数字 id：
+                # ab:renamed（处理完成标记）等同前缀标签不代表已关联番剧
+                if re.search(r"ab:\d+", tags):
+                    continue
 
-            # First try by torrent name, then fall back to save_path
-            bangumi = match_bangumi_in_list(torrent_name, bangumi_list)
-            if not bangumi:
-                bangumi = save_path_index.get(normalize_save_path(save_path))
+                # First try by torrent name, then fall back to save_path
+                bangumi = match_bangumi_in_list(torrent_name, bangumi_list)
+                if not bangumi:
+                    bangumi = save_path_index.get(normalize_save_path(save_path))
 
-            if bangumi and not bangumi.deleted:
-                tag = f"ab:{bangumi.id}"
-                await client.add_tag(torrent_hash, tag)
-                tagged_count += 1
-                logger.info(
-                    f"Tagged '{torrent_name[:50]}...' with {tag} "
-                    f"(matched: {bangumi.official_title})"
-                )
-            else:
-                unmatched.append(
-                    {
-                        "hash": torrent_hash,
-                        "name": torrent_name,
-                        "save_path": save_path,
-                    }
-                )
+                if bangumi and not bangumi.deleted:
+                    tag = f"ab:{bangumi.id}"
+                    await client.add_tag(torrent_hash, tag)
+                    tagged_count += 1
+                    logger.info(
+                        f"Tagged '{torrent_name[:50]}...' with {tag} "
+                        f"(matched: {bangumi.official_title})"
+                    )
+                else:
+                    unmatched.append(
+                        {
+                            "hash": torrent_hash,
+                            "name": torrent_name,
+                            "save_path": save_path,
+                        }
+                    )
 
     return {
         "status": True,

backend/src/module/api/events.py

@@ -22,7 +22,7 @@ from module.api.log import _read_log_tail
 from module.conf import LOG_PATH, VERSION
 from module.core import AppContext
 from module.database import Database
-from module.downloader import DownloadClient
+from module.downloader import list_torrents
 from module.notification.inbox import InboxChanged, inbox_revision
 from module.plugin.host import get_bus
 from module.security.api import get_current_user
@@ -54,25 +54,14 @@ def _status_payload(ctx: AppContext) -> dict:
     }
 
 
-async def _fetch_torrents() -> list[dict]:
-    async with DownloadClient() as client:
-        return await client.get_torrent_info(category="Bangumi", status_filter=None)
-
-
 async def _downloader_payload() -> list[dict] | None:
-    """获取种子列表；下载器未配置、不可达或超时时返回 None。
+    """获取所有下载器实例的种子列表；全部实例不可达或超时时返回 None。
 
-    asyncio.wait_for 超时会取消内部任务并等待其退出，慢查询不会泄漏；
-    None 会以 null 推送给前端，作为显式的"下载器不可用"信号。
+    超时按实例计算，一个实例不可用不影响其它实例的种子；asyncio.wait_for
+    超时会取消内部任务并等待其退出，慢查询不会泄漏。None 会以 null 推送给
+    前端，作为显式的"下载器不可用"信号。
     """
-    try:
-        return await asyncio.wait_for(_fetch_torrents(), _DOWNLOADER_TIMEOUT_SECONDS)
-    except asyncio.TimeoutError:
-        logger.debug("SSE: downloader status fetch timed out")
-        return None
-    except Exception:
-        logger.debug("SSE: downloader status unavailable", exc_info=True)
-        return None
+    return await list_torrents(timeout=_DOWNLOADER_TIMEOUT_SECONDS)
 
 
 async def _notification_payload() -> dict:

backend/src/module/api/rss.py

@@ -2,7 +2,7 @@ from fastapi import APIRouter, Depends
 from fastapi.responses import JSONResponse
 
 from module.database import Database, get_db
-from module.downloader import DownloadClient
+from module.downloader import DownloadClient, DownloaderPool, resolve_downloader_id
 from module.manager import SeasonCollector
 from module.models import APIResponse, Bangumi, Movie, RSSItem, RSSUpdate, Torrent
 from module.rss import RSSAnalyser, RSSEngine
@@ -140,9 +140,9 @@ async def update_rss(
     dependencies=[Depends(get_current_user)],
 )
 async def refresh_all(db: Database = Depends(get_db)):
-    async with DownloadClient() as client:
+    async with DownloaderPool() as downloaders:
         engine = RSSEngine(db)
-        await engine.refresh_rss(client)
+        await engine.refresh_rss(downloaders)
     return JSONResponse(
         status_code=200,
         content={
@@ -158,9 +158,9 @@ async def refresh_all(db: Database = Depends(get_db)):
     dependencies=[Depends(get_current_user)],
 )
 async def refresh_rss(rss_id: int, db: Database = Depends(get_db)):
-    async with DownloadClient() as client:
+    async with DownloaderPool() as downloaders:
         engine = RSSEngine(db)
-        await engine.refresh_rss(client, rss_id)
+        await engine.refresh_rss(downloaders, rss_id)
     return JSONResponse(
         status_code=200,
         content={"msg_en": "Refresh RSS successfully.", "msg_zh": "刷新 RSS 成功。"},
@@ -200,7 +200,7 @@ async def analysis(rss: RSSItem):
     "/collect", response_model=APIResponse, dependencies=[Depends(get_current_user)]
 )
 async def download_collection(data: Bangumi):
-    async with DownloadClient() as client:
+    async with DownloadClient(resolve_downloader_id(data.downloader_id)) as client:
         collector = SeasonCollector(client)
         resp = await collector.collect_season(data, data.rss_link)
         return u_response(resp)

backend/src/module/core/context.py

@@ -271,7 +271,9 @@ class AppContext:
         try:
             await self.notifier.send_event(
                 DownloaderUnavailableEvent(
-                    host=self.settings.downloader.host, reason=reason
+                    host=self.settings.downloader.host,
+                    reason=reason,
+                    instance_id=self.settings.plugins.slots.downloader,
                 )
             )
         except Exception:

backend/src/module/core/loops.py

@@ -9,8 +9,9 @@ import logging
 
 from module.conf import settings
 from module.database import Database
-from module.downloader import DownloadClient
-from module.manager import Renamer, TorrentManager, eps_complete
+from module.downloader import DownloaderPool
+from module.manager import TorrentManager, eps_complete
+from module.manager.renamer import rename_all
 from module.notification import NotificationManager, UpdateAvailableEvent
 from module.rss import RSSAnalyser, RSSEngine
 from module.update.updater import updater
@@ -22,7 +23,7 @@ logger = logging.getLogger(__name__)
 
 async def rss_tick(analyser: RSSAnalyser, notifier: NotificationManager) -> None:
     """Analyse aggregate RSS feeds and refresh the RSS engine once."""
-    async with DownloadClient() as client:
+    async with DownloaderPool() as downloaders:
         async with Database() as db:
             engine = RSSEngine(db)
             # Analyse RSS
@@ -46,7 +47,7 @@ async def rss_tick(analyser: RSSAnalyser, notifier: NotificationManager) -> None
                             exc_info=True,
                         )
             # Run RSS Engine
-            events = await engine.refresh_rss(client)
+            events = await engine.refresh_rss(downloaders)
     if events:
         # 站内通知中心要求事件总是送达 notifier（send_event 内部先落库，
         # 再按 settings.notification.enable 决定是否外发）。
@@ -61,11 +62,8 @@ async def rss_tick(analyser: RSSAnalyser, notifier: NotificationManager) -> None
 
 
 async def rename_tick(notifier: NotificationManager) -> None:
-    """Rename completed downloads and notify via the shared notifier."""
-    async with DownloadClient() as client:
-        renamer = Renamer(client)
-        renamed_info = await renamer.rename()
-        rename_events = list(renamer.events)
+    """Rename completed downloads on every downloader instance and notify."""
+    renamed_info, rename_events = await rename_all()
     if rename_events:
         await asyncio.gather(*[notifier.send_event(event) for event in rename_events])
     if settings.notification.enable and renamed_info:

backend/src/module/database/migrations.py

@@ -100,8 +100,14 @@ class Migration:
         return self.statements
 
 
-# 存量行归属默认下载器实例（plugins.instances 中 id 为 default 的实例）
-_DOWNLOADER_ID_TABLES = ("bangumi", "movie", "rssitem", "torrent")
+# 规则与订阅的 downloader_id 为空表示「用默认实例」，存量行保持为空；
+# 种子记录实际所在的实例，存量种子都在 3.3 的下载器，即实例 default
+_DOWNLOADER_ID_COLUMNS = {
+    "bangumi": "downloader_id VARCHAR",
+    "movie": "downloader_id VARCHAR",
+    "rssitem": "downloader_id VARCHAR",
+    "torrent": "downloader_id VARCHAR DEFAULT 'default'",
+}
 
 
 def _has_downloader_id(table: str) -> AppliedCheck:
@@ -817,17 +823,13 @@ MIGRATIONS: tuple[Migration, ...] = (
         26,
         "add downloader_id to bangumi, movie, rssitem and torrent",
         tuple(
-            f"ALTER TABLE {table} ADD COLUMN downloader_id VARCHAR DEFAULT 'default'"
-            for table in _DOWNLOADER_ID_TABLES
+            f"ALTER TABLE {table} ADD COLUMN {column}"
+            for table, column in _DOWNLOADER_ID_COLUMNS.items()
         ),
-        all_checks(*(_has_downloader_id(table) for table in _DOWNLOADER_ID_TABLES)),
+        all_checks(*(_has_downloader_id(table) for table in _DOWNLOADER_ID_COLUMNS)),
         tuple(
-            (
-                f"ALTER TABLE {table} ADD COLUMN downloader_id VARCHAR "
-                "DEFAULT 'default'",
-                _has_downloader_id(table),
-            )
-            for table in _DOWNLOADER_ID_TABLES
+            (f"ALTER TABLE {table} ADD COLUMN {column}", _has_downloader_id(table))
+            for table, column in _DOWNLOADER_ID_COLUMNS.items()
         ),
     ),
 )

backend/src/module/database/rename_operation.py

@@ -356,10 +356,14 @@ class RenameOperationDatabase:
         )
         return list(result.scalars().all())
 
-    async def list_active_replacements(self, limit: int = 100) -> list[RenameOperation]:
+    async def list_active_replacements(
+        self, downloader_type: str, limit: int = 100
+    ) -> list[RenameOperation]:
+        """某个下载器实例（``downloader_type``）上未完成的版本替换事务。"""
         result = await self.session.execute(
             select(RenameOperation)
             .where(
+                RenameOperation.downloader_type == downloader_type,
                 RenameOperation.kind == "replacement",
                 col(RenameOperation.state).in_(
                     ("planned", "old_staged", "new_promoted", "old_removed")

backend/src/module/downloader/__init__.py

@@ -6,15 +6,26 @@ from .base import (
     RenameOutcome,
     RenameResult,
 )
-from .download_client import DownloadClient, shutdown
+from .download_client import (
+    DownloadClient,
+    DownloaderPool,
+    downloader_ids,
+    list_torrents,
+    resolve_downloader_id,
+    shutdown,
+)
 
 __all__ = [
     "AddResult",
     "CoreDownloaderClient",
     "DownloadClient",
+    "DownloaderPool",
     "DownloaderCapabilities",
     "DownloaderClient",
     "RenameOutcome",
     "RenameResult",
+    "downloader_ids",
+    "list_torrents",
+    "resolve_downloader_id",
     "shutdown",
 ]

backend/src/module/downloader/download_client.py

@@ -1,12 +1,14 @@
 import asyncio
 import logging
 from collections import defaultdict
+from contextlib import AsyncExitStack
 from urllib.parse import urlparse
 
 from ab_sdk import points
 from ab_sdk.ingest import AddRequest
 from module.conf import settings
 from module.models import Bangumi, Torrent
+from module.models.config import DOWNLOADER_POINT, DownloaderInstance
 from module.network import RequestContent
 from module.plugin import host as plugin_host
 from module.plugin.views import bangumi_info, torrent_info
@@ -116,24 +118,44 @@ async def _apply_adding_hooks(
 # `_bookkeeping_lock` serializes reads/writes of this shared state across the
 # awaits in ``__aenter__``/``__aexit__``.
 # ---------------------------------------------------------------------------
-_client_cache: tuple[tuple, DownloaderClient] | None = None
+# 实例 id → (连接设置 key, 具体客户端)。每个下载器实例各缓存一个客户端。
+_client_cache: dict[str, tuple[tuple, DownloaderClient]] = {}
 _stale_clients: list[DownloaderClient] = []
 _active_holders: dict[int, int] = {}
 _pending_close: set[int] = set()
 _bookkeeping_lock = asyncio.Lock()
-# 凭据被服务端明确拒绝后的闩锁：记录失败时的连接设置 key。命中时 enter 直接
-# 失败、不再发 login POST——每个 tick 重试一次登录，约 5 次即触发 qB 的
+# 凭据被服务端明确拒绝后的闩锁：实例 id → 失败时的连接设置 key。命中时 enter
+# 直接失败、不再发 login POST——每个 tick 重试一次登录，约 5 次即触发 qB 的
 # WebUI IP ban。设置变更（key 不同）自然解锁；同值重存经
 # clear_credential_latch()（AppContext.reload_settings）解锁。
-_credential_failed_key: tuple | None = None
+_credential_failed: dict[str, tuple] = {}
 
 # Warn at most once per (client type, operation) when a backend cannot perform
 # an operation, so aria2 users are not spammed every rename cycle.
 _warned_unsupported: set[tuple[str, str]] = set()
 
 
-def _settings_key() -> tuple:
-    d = settings.downloader
+def downloader_ids() -> list[str]:
+    """所有下载器实例 id（配置顺序）。"""
+    return [i.id for i in settings.plugins.instances if i.point == DOWNLOADER_POINT]
+
+
+def resolve_downloader_id(*candidates: str | None) -> str:
+    """取第一个指向已配置实例的候选 id；都为空时用默认实例（slots.downloader）。
+
+    指向已删除实例的候选记录告警后跳过，不让一条规则卡住整轮。
+    """
+    known = downloader_ids()
+    for candidate in candidates:
+        if not candidate:
+            continue
+        if candidate in known:
+            return candidate
+        logger.warning("Downloader instance %r no longer exists; skipped", candidate)
+    return settings.plugins.slots.downloader
+
+
+def _settings_key(d: DownloaderInstance) -> tuple:
     # 带上 Provider 登记项：插件重载后登记项换新，旧客户端随之退役
     entry = plugin_host.get_registry().providers(points.DOWNLOADER).get(d.type)
     return (d.type, d.host, d.username, d.password, d.ssl, entry)
@@ -141,8 +163,8 @@ def _settings_key() -> tuple:
 
 def _reset_client_cache() -> None:
     """Drop the cached/stale concrete clients and refcount state (used by tests)."""
-    global _client_cache, _stale_clients, _active_holders, _pending_close
-    _client_cache = None
+    global _stale_clients, _active_holders, _pending_close
+    _client_cache.clear()
     _stale_clients = []
     _active_holders = {}
     _pending_close = set()
@@ -150,9 +172,8 @@ def _reset_client_cache() -> None:
 
 
 def clear_credential_latch() -> None:
-    """解除凭据失败闩锁（配置保存后调用，允许用户重试同值凭据）。"""
-    global _credential_failed_key
-    _credential_failed_key = None
+    """解除所有实例的凭据失败闩锁（配置保存后调用，允许用户重试同值凭据）。"""
+    _credential_failed.clear()
 
 
 async def _close_client(client) -> None:
@@ -163,15 +184,13 @@ async def _close_client(client) -> None:
 
 
 async def shutdown() -> None:
-    """Log out and close the cached concrete client.
+    """Log out and close every cached concrete client.
 
     Invoked by the composition root (`AppContext`) on application shutdown.
     """
-    global _client_cache, _stale_clients
-    clients = list(_stale_clients)
-    if _client_cache is not None:
-        clients.append(_client_cache[1])
-    _client_cache = None
+    global _stale_clients
+    clients = list(_stale_clients) + [c for _, c in _client_cache.values()]
+    _client_cache.clear()
     _stale_clients = []
     for client in clients:
         await _close_client(client)
@@ -187,30 +206,37 @@ class DownloadClient:
     down by :func:`shutdown`.
     """
 
-    def __init__(self):
-        global _client_cache
-        key = _settings_key()
+    def __init__(self, instance_id: str | None = None):
+        # None：默认实例（slots.downloader）
+        self.instance: DownloaderInstance = (
+            settings.downloader
+            if instance_id is None
+            else settings.downloader_instance(instance_id)
+        )
+        self.instance_id = self.instance.id
+        key = _settings_key(self.instance)
         self.client: DownloaderClient
         self._cache_key = key
-        if _client_cache is not None and _client_cache[0] == key:
-            self.client = _client_cache[1]
+        cached = _client_cache.get(self.instance_id)
+        if cached is not None and cached[0] == key:
+            self.client = cached[1]
         else:
-            if _client_cache is not None:
+            if cached is not None:
                 # Settings changed: retire the previous client, close it later.
                 # 用列表累积——连续两次改设置（期间没有 enter）不得把第一个
                 # 被撤下的客户端顶掉，否则它的连接池泄漏到进程结束。
-                _stale_clients.append(_client_cache[1])
-            self.client = self.__getClient()
-            _client_cache = (key, self.client)
+                _stale_clients.append(cached[1])
+            self.client = self.__getClient(self.instance)
+            _client_cache[self.instance_id] = (key, self.client)
         self.authed = False
 
     @staticmethod
-    def __getClient() -> DownloaderClient:
-        """按 ``downloader.type`` 从扩展注册表取下载器工厂并实例化。
+    def __getClient(instance: DownloaderInstance) -> DownloaderClient:
+        """按实例的 Provider id 从扩展注册表取下载器工厂并实例化。
 
         内置的 qbittorrent / aria2 / mock 与插件提供的下载器走同一条路径。
         """
-        downloader_type = settings.downloader.type
+        downloader_type = instance.type
         factory = plugin_host.provider(points.DOWNLOADER, downloader_type)
         if factory is None:
             logger.error("Unsupported downloader type: %s", downloader_type)
@@ -218,10 +244,10 @@ class DownloadClient:
         if downloader_type == "mock":
             logger.debug("Using MockDownloader for local development")
         conn = DownloaderConnection(
-            host=settings.downloader.host,
-            username=settings.downloader.username,
-            password=settings.downloader.password,
-            ssl=settings.downloader.ssl,
+            host=instance.host,
+            username=instance.username,
+            password=instance.password,
+            ssl=instance.ssl,
         )
         # 只实现 CoreDownloaderClient 的后端（如 aria2 没有 qB 的 RSS 规则）
         # 由 _supports() 按 capabilities 跳过不支持的操作
@@ -251,11 +277,7 @@ class DownloadClient:
         return False
 
     async def __aenter__(self):
-        global _client_cache, _credential_failed_key
-        if (
-            _credential_failed_key is not None
-            and _credential_failed_key == self._cache_key
-        ):
+        if _credential_failed.get(self.instance_id) == self._cache_key:
             # 凭据上次已被服务端明确拒绝且设置未变：不再发 login POST，
             # 避免逐 tick 累积到 qB 的 IP ban。checker/等待循环读的是
             # 具体客户端上的失败原因，这里补齐。
@@ -297,15 +319,16 @@ class DownloadClient:
                 raise
             if not self.authed:
                 if self.last_auth_error == "credentials":
-                    _credential_failed_key = self._cache_key
+                    _credential_failed[self.instance_id] = self._cache_key
                 # Release our slot and close the concrete client's connection
                 # pool now or it leaks on every failed connect (#1043) --
                 # unless another already-entered block still holds it, in
                 # which case defer to its __aexit__ instead of yanking the
                 # pool out from under it.
                 async with _bookkeeping_lock:
-                    if _client_cache is not None and _client_cache[1] is self.client:
-                        _client_cache = None
+                    cached = _client_cache.get(self.instance_id)
+                    if cached is not None and cached[1] is self.client:
+                        del _client_cache[self.instance_id]
                     _pending_close.add(id(self.client))
                 await self._release_holder()
                 raise ConnectionError("Download client authentication failed")
@@ -441,7 +464,10 @@ class DownloadClient:
         调用方据此决定是否重试或发送失败通知。
         """
         if not bangumi.save_path:
-            bangumi.save_path = gen_save_path(bangumi)
+            bangumi.save_path = gen_save_path(bangumi, self.instance.path)
+        # 种子行记录实际投递的实例，之后的重命名 / 删除按它路由
+        for t in torrent if isinstance(torrent, list) else [torrent]:
+            t.downloader_id = self.instance_id
         torrent_url: str | list[str] | None
         async with RequestContent() as req:
             if isinstance(torrent, list):
@@ -513,3 +539,67 @@ class DownloadClient:
             return
         await self.client.add_tag(torrent_hash, tag)
         logger.debug("Added tag '%s' to torrent %s...", tag, torrent_hash[:8])
+
+
+class DownloaderPool:
+    """一次操作内按实例 id 惰性进入 :class:`DownloadClient`，退出时全部释放。
+
+    进入失败的实例在本池内记住、直接抛 ``ConnectionError``，不再重复登录
+    （一轮 RSS 可能有很多种子指向同一个不可用实例）。
+    """
+
+    def __init__(self) -> None:
+        self._stack = AsyncExitStack()
+        self._clients: dict[str, DownloadClient] = {}
+        self._failed: set[str] = set()
+
+    async def __aenter__(self) -> "DownloaderPool":
+        return self
+
+    async def __aexit__(self, *_: object) -> None:
+        await self._stack.aclose()
+
+    async def get(self, instance_id: str) -> DownloadClient:
+        if instance_id in self._failed:
+            raise ConnectionError(f"Downloader instance {instance_id} is unavailable")
+        if instance_id not in self._clients:
+            try:
+                self._clients[instance_id] = await self._stack.enter_async_context(
+                    DownloadClient(instance_id)
+                )
+            except Exception as e:
+                self._failed.add(instance_id)
+                raise ConnectionError(
+                    f"Downloader instance {instance_id} is unavailable: {e}"
+                ) from e
+        return self._clients[instance_id]
+
+
+async def list_torrents(
+    status_filter: str | None = None, timeout: float | None = None
+) -> list[dict] | None:
+    """所有下载器实例中 Bangumi 分类的种子，每条带 ``downloader_id``。
+
+    不可用（或超过 ``timeout`` 秒）的实例跳过；全部实例都不可用时返回 None。
+    """
+
+    async def one(instance_id: str) -> list[dict]:
+        async with DownloadClient(instance_id) as client:
+            torrents = await client.get_torrent_info(
+                category="Bangumi", status_filter=status_filter
+            )
+        return [{**t, "downloader_id": instance_id} for t in torrents]
+
+    ids = downloader_ids()
+    results = await asyncio.gather(
+        *(asyncio.wait_for(one(i), timeout) for i in ids), return_exceptions=True
+    )
+    listed: list[dict] = []
+    reachable = False
+    for instance_id, result in zip(ids, results):
+        if isinstance(result, BaseException):
+            logger.debug("Downloader instance %s unavailable: %s", instance_id, result)
+            continue
+        reachable = True
+        listed.extend(result)
+    return listed if reachable else None

backend/src/module/downloader/path.py

@@ -70,12 +70,15 @@ def check_files(files: list[dict]):
     return media_list, subtitle_list
 
 
-def path_to_bangumi(save_path: PathLike[str] | str, torrent_name: str = ""):
+def path_to_bangumi(
+    save_path: PathLike[str] | str, torrent_name: str = "", root: str | None = None
+):
+    """``root`` 为种子所在下载器实例的下载目录，缺省为默认实例的。"""
     # Use PureWindowsPath regardless of the host AB runs on: it accepts
     # both "\" and "/" separators, so a qBittorrent-on-Windows save_path
     # reaching a Linux AB still splits into segments correctly (#1016).
     save_parts = PureWindowsPath(save_path).parts
-    download_parts = PureWindowsPath(settings.downloader.path).parts
+    download_parts = PureWindowsPath(root or settings.downloader.path).parts
     # Get bangumi name and season
     bangumi_name = ""
     season = 1
@@ -107,9 +110,13 @@ def _media_folder(data: Bangumi | BangumiUpdate | Movie | MovieUpdate) -> str:
     return "Unknown Bangumi"
 
 
-def gen_save_path(data: Bangumi | BangumiUpdate | Movie | MovieUpdate) -> str:
+def gen_save_path(
+    data: Bangumi | BangumiUpdate | Movie | MovieUpdate, root: str | None = None
+) -> str:
     """Generate save path for a bangumi.
 
+    ``root`` 为目标下载器实例的下载目录，缺省为默认实例的。
+
     The save path uses the adjusted season number (season + season_offset)
     so files are saved directly to the correct season folder.
 
@@ -121,7 +128,7 @@ def gen_save_path(data: Bangumi | BangumiUpdate | Movie | MovieUpdate) -> str:
     episode_type = getattr(data, "episode_type", "episode")
     if isinstance(data, (Movie, MovieUpdate)) or episode_type == "movie":
         # 电影/剧场版：Title (Year)/Title (Year).ext，不建 Season 子目录
-        return str(Path(settings.downloader.path) / folder)
+        return str(Path(root or settings.downloader.path) / folder)
     # Apply season_offset to get the adjusted season number for the folder
     adjusted_season = data.season + getattr(data, "season_offset", 0)
     # 季号下限：普通剧集最小为 1——偏移到 Season 0 会被 Plex/Jellyfin 当作
@@ -132,7 +139,9 @@ def gen_save_path(data: Bangumi | BangumiUpdate | Movie | MovieUpdate) -> str:
         logger.warning(
             f"Season offset would result in invalid season for {data.official_title}, using original season"
         )
-    save_path = Path(settings.downloader.path) / folder / f"Season {adjusted_season}"
+    save_path = (
+        Path(root or settings.downloader.path) / folder / f"Season {adjusted_season}"
+    )
     return str(save_path)
 
 

backend/src/module/manager/collector.py

@@ -2,7 +2,12 @@ import logging
 
 from module.database import Database
 from module.database.bangumi import release_fits_bangumi
-from module.downloader import AddResult, DownloadClient
+from module.downloader import (
+    AddResult,
+    DownloadClient,
+    DownloaderPool,
+    resolve_downloader_id,
+)
 from module.models import Bangumi, ResponseModel
 from module.network import RequestContent
 from module.rss import RSSEngine
@@ -129,10 +134,12 @@ async def eps_complete():
         datas = await db.bangumi.not_complete()
         if datas:
             logger.info("Start collecting full season...")
-            async with DownloadClient() as client:
-                collector = SeasonCollector(client)
+            async with DownloaderPool() as downloaders:
                 for data in datas:
                     if not data.eps_collect:
-                        await collector.collect_season(data)
+                        client = await downloaders.get(
+                            resolve_downloader_id(data.downloader_id)
+                        )
+                        await SeasonCollector(client).collect_season(data)
                     data.eps_collect = True
             await db.bangumi.update_all(datas)

backend/src/module/manager/renamer.py

@@ -21,10 +21,19 @@ from module.database.bangumi import (
     match_bangumi_in_list,
     normalize_save_path,
 )
-from module.downloader import DownloadClient, RenameOutcome, RenameResult
+from module.downloader import (
+    DownloadClient,
+    RenameOutcome,
+    RenameResult,
+    downloader_ids,
+)
 from module.downloader.path import check_files, is_ep, path_to_bangumi
 from module.models import EpisodeFile, Notification, SubtitleFile
-from module.notification import RenameSkippedEvent, SystemEvent
+from module.notification import (
+    DownloaderUnavailableEvent,
+    RenameSkippedEvent,
+    SystemEvent,
+)
 from module.parser import TitleParser
 from module.plugin import host as plugin_host
 from module.plugin.registry import ProviderEntry
@@ -44,8 +53,8 @@ from .revision_saga import (
 
 logger = logging.getLogger(__name__)
 
-# 下载器 id：多下载器（P2.5）之前只有一个
-_DOWNLOADER_ID = "default"
+# 上一轮不可用的下载器实例：只在「可用 → 不可用」时通知一次
+_unavailable: set[str] = set()
 # 已记录过「未登记」日志的重命名方式（每个进程只记一次，重新登记后清除）
 _missing_methods: set[str] = set()
 # 已发布过的 TorrentOrganized（种子 hash → 最终文件）。未打 ab:renamed 标签的
@@ -589,7 +598,7 @@ class Renamer(RevisionSaga):
                     old_path=_downloader_path(save_path, old),
                     new_path=_downloader_path(save_path, new),
                     file_kind=kind,
-                    downloader_id=_DOWNLOADER_ID,
+                    downloader_id=self.client.instance_id,
                 )
             )
         reason = self._skipped.pop(torrent_hash, None)
@@ -624,7 +633,7 @@ class Renamer(RevisionSaga):
                 torrent_hash=torrent_hash,
                 bangumi_id=bangumi_id,
                 files=files,
-                downloader_id=_DOWNLOADER_ID,
+                downloader_id=self.client.instance_id,
             )
         )
 
@@ -651,6 +660,10 @@ class Renamer(RevisionSaga):
                 # Collect all hashes for batch query
                 hashes = [info["hash"] for info in torrents_info]
                 torrent_records = await db.torrent.search_by_qb_hashes(hashes)
+                # 同一 hash 在多个实例都有记录时，以本实例的记录为准（排在后面覆盖）
+                torrent_records.sort(
+                    key=lambda r: r.downloader_id == self.client.instance_id
+                )
                 hash_to_bangumi_id = {
                     r.qb_hash: r.bangumi_id for r in torrent_records if r.bangumi_id
                 }
@@ -758,7 +771,9 @@ class Renamer(RevisionSaga):
             info_by_hash.setdefault(info["hash"], info)
         all_infos = list(info_by_hash.values())
         async with Database() as db:
-            active_replacements = await db.rename_operation.list_active_replacements()
+            active_replacements = await db.rename_operation.list_active_replacements(
+                self._downloader_type()
+            )
         active_replacement_ids = {
             operation.new_task_id for operation in active_replacements
         }
@@ -808,7 +823,9 @@ class Renamer(RevisionSaga):
                 )
                 continue
             media_list, subtitle_list = check_files(files)
-            bangumi_name, season = path_to_bangumi(save_path, torrent_name)
+            bangumi_name, season = path_to_bangumi(
+                save_path, torrent_name, self.client.instance.path
+            )
             episode_offset, episode_type = offset_map[torrent_hash]
             kwargs = {
                 "torrent_name": torrent_name,
@@ -875,3 +892,38 @@ class Renamer(RevisionSaga):
             )
         logger.debug("Rename process finished.")
         return renamed_info
+
+
+async def rename_all() -> tuple[list[Notification], list[SystemEvent]]:
+    """逐个下载器实例运行一轮重命名。
+
+    进不去的实例（连不上、凭据被拒、Provider 未登记）跳过，不影响其它实例；
+    它从可用变为不可用时产生一条 :class:`DownloaderUnavailableEvent`。
+    """
+    renamed: list[Notification] = []
+    events: list[SystemEvent] = []
+    for instance_id in downloader_ids():
+        client: DownloadClient | None = None
+        try:
+            client = DownloadClient(instance_id)
+            await client.__aenter__()
+        except Exception as e:
+            logger.warning("Downloader %s unavailable, skipped: %s", instance_id, e)
+            if instance_id not in _unavailable:
+                _unavailable.add(instance_id)
+                events.append(
+                    DownloaderUnavailableEvent(
+                        host=settings.downloader_instance(instance_id).host,
+                        reason=(client and client.last_auth_error) or "unreachable",
+                        instance_id=instance_id,
+                    )
+                )
+            continue
+        _unavailable.discard(instance_id)
+        try:
+            renamer = Renamer(client)
+            renamed += await renamer.rename()
+            events += renamer.events
+        finally:
+            await client.__aexit__(None, None, None)
+    return renamed, events

backend/src/module/manager/revision_saga.py

@@ -167,13 +167,14 @@ class RevisionSaga:
         return value <= datetime.now(timezone.utc)
 
     def _downloader_type(self) -> str:
-        configured = settings.downloader.type
+        instance = self.client.instance
+        configured = instance.type
         downloader_type = (
             configured
             if isinstance(configured, str)
             else type(self.client.client).__name__.lower()
         )
-        host = settings.downloader.host
+        host = instance.host
         if not isinstance(host, str) or not host:
             return downloader_type
         instance_hash = hashlib.sha256(host.strip().lower().encode()).hexdigest()[:12]
@@ -191,7 +192,9 @@ class RevisionSaga:
 
         incoming_id = self._parse_bangumi_id_from_tags(incoming.get("tags"))
         _, incoming_season = path_to_bangumi(
-            incoming.get("save_path", ""), incoming.get("name", "")
+            incoming.get("save_path", ""),
+            incoming.get("name", ""),
+            self.client.instance.path,
         )
         incoming_identity = parse_revision_identity(
             incoming.get("name", ""),
@@ -229,7 +232,9 @@ class RevisionSaga:
             ):
                 continue
             _, old_season = path_to_bangumi(
-                info.get("save_path", ""), info.get("name", "")
+                info.get("save_path", ""),
+                info.get("name", ""),
+                self.client.instance.path,
             )
             owners.append(
                 RevisionOwner(

backend/src/module/manager/torrent.py

@@ -3,7 +3,7 @@ import logging
 from module.conf import settings
 from module.database import Database
 from module.database.bangumi import normalize_save_path
-from module.downloader import DownloadClient
+from module.downloader import DownloadClient, downloader_ids, resolve_downloader_id
 from module.downloader.path import gen_save_path
 from module.downloader.rules import build_rss_rule
 from module.models import Bangumi, BangumiUpdate, ResponseModel
@@ -19,8 +19,10 @@ class TorrentManager:
         self.db = db
 
     @staticmethod
-    async def __match_torrents_list(data: Bangumi | BangumiUpdate) -> list:
-        async with DownloadClient() as client:
+    async def __match_torrents_list(
+        data: Bangumi | BangumiUpdate, instance_id: str | None = None
+    ) -> list:
+        async with DownloadClient(instance_id) as client:
             torrents = await client.get_torrent_info(status_filter=None)
         target_save_path = normalize_save_path(data.save_path)
         return [
@@ -29,16 +31,33 @@ class TorrentManager:
             if normalize_save_path(torrent.get("save_path")) == target_save_path
         ]
 
-    async def delete_torrents(self, data: Bangumi, client: DownloadClient):
-        hash_list = await self.__match_torrents_list(data)
-        if hash_list:
-            if not await client.delete_torrent(hash_list):
+    async def _torrent_instances(self, data: Bangumi) -> list[str]:
+        """番剧的种子所在的下载器实例：种子行记录的实例，加上规则当前的实例。
+
+        规则换过下载器时，旧种子仍在原实例上，删除要逐个实例处理。
+        """
+        rows = await self.db.torrent.search_by_bangumi_id(data.id)
+        ids = {r.downloader_id for r in rows}
+        ids.add(resolve_downloader_id(data.downloader_id))
+        return sorted(ids & set(downloader_ids()))
+
+    async def delete_torrents(self, data: Bangumi, instance_ids: list[str]):
+        found = False
+        for instance_id in instance_ids:
+            hash_list = await self.__match_torrents_list(data, instance_id)
+            if not hash_list:
+                continue
+            found = True
+            async with DownloadClient(instance_id) as client:
+                deleted = await client.delete_torrent(hash_list)
+            if not deleted:
                 return ResponseModel(
                     status_code=500,
                     status=False,
                     msg_en=f"Failed to delete torrents for {data.official_title}",
                     msg_zh=f"删除 {data.official_title} 种子失败",
                 )
+        if found:
             logger.info(f"Delete rule and torrents for {data.official_title}")
             return ResponseModel(
                 status_code=200,
@@ -83,6 +102,8 @@ class TorrentManager:
     async def delete_rule(self, _id: int | str, file: bool = False):
         data = await self.db.bangumi.search_id(int(_id))
         if isinstance(data, Bangumi):
+            # 种子行删除前先记下种子所在的实例
+            instance_ids = await self._torrent_instances(data) if file else []
             # Clean up torrent records so re-adding the same anime can re-download
             await self.db.torrent.delete_by_bangumi_id(int(_id))
             await self.db.bangumi.delete_one(int(_id))
@@ -92,16 +113,15 @@ class TorrentManager:
             if file:
                 # Only the file-cleanup path needs the downloader, so an
                 # unreachable downloader shouldn't block a DB-only delete.
-                async with DownloadClient() as client:
-                    torrent_message = await self.delete_torrents(data, client)
-                    if torrent_message.status_code == 500:
-                        return ResponseModel(
-                            status_code=500,
-                            status=False,
-                            msg_en=f"Deleted rule for {data.official_title}, "
-                            "but deleting its torrents failed.",
-                            msg_zh=f"已删除 {data.official_title} 规则，但删除种子失败。",
-                        )
+                torrent_message = await self.delete_torrents(data, instance_ids)
+                if torrent_message.status_code == 500:
+                    return ResponseModel(
+                        status_code=500,
+                        status=False,
+                        msg_en=f"Deleted rule for {data.official_title}, "
+                        "but deleting its torrents failed.",
+                        msg_zh=f"已删除 {data.official_title} 规则，但删除种子失败。",
+                    )
             logger.info(f"Delete rule for {data.official_title}")
             return ResponseModel(
                 status_code=200,
@@ -125,8 +145,9 @@ class TorrentManager:
             if file:
                 # Only the file-cleanup path needs the downloader, so an
                 # unreachable downloader shouldn't block a DB-only disable.
-                async with DownloadClient() as client:
-                    return await self.delete_torrents(data, client)
+                return await self.delete_torrents(
+                    data, await self._torrent_instances(data)
+                )
             logger.info(f"Disable rule for {data.official_title}")
             return ResponseModel(
                 status_code=200,
@@ -165,25 +186,29 @@ class TorrentManager:
     async def update_rule(self, bangumi_id, data: BangumiUpdate):
         old_data = await self.db.bangumi.search_id(bangumi_id)
         if old_data:
-            # Move torrent
-            match_list = await self.__match_torrents_list(old_data)
-            async with DownloadClient() as client:
-                new_path = gen_save_path(data)
-                old_path = old_data.save_path
+            old_instance = resolve_downloader_id(old_data.downloader_id)
+            new_instance = resolve_downloader_id(data.downloader_id)
+            new_path = gen_save_path(
+                data, settings.downloader_instance(new_instance).path
+            )
+            old_path = old_data.save_path
+            # 换了下载器实例：旧种子留在原实例原位置，之后的新种子进新实例
+            if new_instance == old_instance and new_path != old_path:
+                match_list = await self.__match_torrents_list(old_data, old_instance)
+                async with DownloadClient(old_instance) as client:
+                    # Move existing torrents to new location if path changed
+                    if match_list:
+                        await client.move_torrent(match_list, new_path)
+                        logger.info(f"Moved torrents from {old_path} to {new_path}")
 
-                # Move existing torrents to new location if path changed
-                if match_list and new_path != old_path:
-                    await client.move_torrent(match_list, new_path)
-                    logger.info(f"Moved torrents from {old_path} to {new_path}")
-
-                # Update qBittorrent RSS rule if save_path changed
-                if new_path != old_path and old_data.rule_name:
-                    # Recreate the rule with the new save_path
-                    rule = build_rss_rule(data, new_path)
-                    await client.set_rss_rule(old_data.rule_name, rule)
-                    logger.info(
-                        f"Updated RSS rule {old_data.rule_name} with new save_path"
-                    )
+                    # Update qBittorrent RSS rule if save_path changed
+                    if old_data.rule_name:
+                        # Recreate the rule with the new save_path
+                        rule = build_rss_rule(data, new_path)
+                        await client.set_rss_rule(old_data.rule_name, rule)
+                        logger.info(
+                            f"Updated RSS rule {old_data.rule_name} with new save_path"
+                        )
 
             data.save_path = new_path
             await self.db.bangumi.update(data, bangumi_id)

backend/src/module/mcp/tools.py

@@ -6,7 +6,7 @@ from mcp import types
 
 from module.conf import VERSION
 from module.database import Database
-from module.downloader import DownloadClient
+from module.downloader import DownloaderPool, list_torrents
 from module.manager import SeasonCollector, TorrentManager
 from module.models import Bangumi, BangumiUpdate, RSSItem
 from module.rss import RSSAnalyser, RSSEngine
@@ -301,10 +301,7 @@ async def _unsubscribe_anime(bangumi_id: int, delete: bool) -> dict:
 
 async def _list_downloads(status: str) -> list[dict]:
     status_filter = None if status == "all" else status
-    async with DownloadClient() as client:
-        torrents = await client.get_torrent_info(
-            status_filter=status_filter, category="Bangumi"
-        )
+    torrents = await list_torrents(status_filter) or []
     return [
         {
             "name": t.get("name", ""),
@@ -314,6 +311,7 @@ async def _list_downloads(status: str) -> list[dict]:
             "dlspeed": t.get("dlspeed", 0),
             "upspeed": t.get("upspeed", 0),
             "eta": t.get("eta", 0),
+            "downloader": t["downloader_id"],
         }
         for t in torrents
     ]
@@ -348,10 +346,10 @@ def _get_program_status() -> dict:
 
 
 async def _refresh_feeds() -> dict:
-    async with DownloadClient() as client:
+    async with DownloaderPool() as downloaders:
         async with Database() as db:
             engine = RSSEngine(db)
-            await engine.refresh_rss(client)
+            await engine.refresh_rss(downloaders)
     return {"status": True, "message": "RSS feeds refreshed successfully"}
 
 

backend/src/module/models/bangumi.py

@@ -61,8 +61,9 @@ class Bangumi(SQLModel, table=True):
     episode_type: str = Field(
         default="episode", alias="episode_type", title="剧集类型"
     )  # "episode" | "movie" | "special"
-    downloader_id: str = Field(
-        default="default", alias="downloader_id", title="下载器实例"
+    # 为空时用默认下载器实例（plugins.slots.downloader）
+    downloader_id: Optional[str] = Field(
+        default=None, alias="downloader_id", title="下载器实例"
     )
 
 
@@ -87,6 +88,9 @@ class BangumiUpdate(SQLModel):
     added: bool = Field(default=False, alias="added", title="是否已添加")
     rule_name: Optional[str] = Field(alias="rule_name", title="番剧规则名")
     save_path: Optional[str] = Field(alias="save_path", title="番剧保存路径")
+    downloader_id: Optional[str] = Field(
+        default=None, alias="downloader_id", title="下载器实例"
+    )
     deleted: bool = Field(False, alias="deleted", title="是否已删除")
     archived: bool = Field(default=False, alias="archived", title="是否已归档")
     air_weekday: Optional[int] = Field(

backend/src/module/models/config.py

@@ -466,10 +466,15 @@ class Config(BaseModel):
 
     def downloader_instance(self, instance_id: str) -> DownloaderInstance:
         instance = next(
-            i
-            for i in self.plugins.instances
-            if i.id == instance_id and i.point == DOWNLOADER_POINT
+            (
+                i
+                for i in self.plugins.instances
+                if i.id == instance_id and i.point == DOWNLOADER_POINT
+            ),
+            None,
         )
+        if instance is None:
+            raise KeyError(f"unknown downloader instance {instance_id!r}")
         options = DownloaderOptions.model_validate(instance.options)
         return DownloaderInstance(
             id=instance.id,

backend/src/module/models/movie.py

@@ -25,8 +25,9 @@ class Movie(SQLModel, table=True):
     save_path: str | None = Field(default=None, alias="save_path", title="保存路径")
     rule_name: str | None = Field(default=None, alias="rule_name", title="规则名")
     filter: str = Field(default="", alias="filter", title="过滤器")
-    downloader_id: str = Field(
-        default="default", alias="downloader_id", title="下载器实例"
+    # 为空时用默认下载器实例
+    downloader_id: str | None = Field(
+        default=None, alias="downloader_id", title="下载器实例"
     )
 
 
@@ -42,3 +43,4 @@ class MovieUpdate(SQLModel):
     rss_link: str | None = None
     save_path: str | None = None
     filter: str | None = None
+    downloader_id: str | None = None

backend/src/module/models/rss.py

@@ -13,7 +13,8 @@ class RSSItem(SQLModel, table=True):
     connection_status: Optional[str] = Field(None, alias="connection_status")
     last_checked_at: Optional[str] = Field(None, alias="last_checked_at")
     last_error: Optional[str] = Field(None, alias="last_error")
-    downloader_id: str = Field("default", alias="downloader_id")
+    # 由此订阅新建的规则继承；为空时用默认下载器实例
+    downloader_id: Optional[str] = Field(None, alias="downloader_id")
 
 
 class RSSUpdate(SQLModel):
@@ -22,3 +23,4 @@ class RSSUpdate(SQLModel):
     aggregate: Optional[bool] = Field(True, alias="aggregate")
     parser: Optional[str] = Field("mikan", alias="parser")
     enabled: Optional[bool] = Field(True, alias="enabled")
+    downloader_id: Optional[str] = Field(None, alias="downloader_id")

backend/src/module/rss/analyser.py

@@ -70,6 +70,7 @@ class RSSAnalyser:
                     movie=result, rss=rss, torrent=torrent
                 )
                 result.rss_link = rss.url
+                result.downloader_id = rss.downloader_id
                 seen_identities.add(identity)
                 new_movies.append(result)
                 logger.info(f"New movie found: {result.official_title}")
@@ -78,6 +79,7 @@ class RSSAnalyser:
                     bangumi=result, rss=rss, torrent=torrent
                 )
                 result.rss_link = rss.url
+                result.downloader_id = rss.downloader_id
                 if not full_parse:
                     return [result], new_movies
                 seen_identities.add(identity)
@@ -98,6 +100,7 @@ class RSSAnalyser:
                     fetch_poster=fetch_poster,
                 )
                 result.rss_link = rss.url
+                result.downloader_id = rss.downloader_id
             else:
                 await self.official_title_parser(
                     bangumi=result,
@@ -106,6 +109,7 @@ class RSSAnalyser:
                     fetch_poster=fetch_poster,
                 )
                 result.rss_link = rss.url
+                result.downloader_id = rss.downloader_id
             return result
         return None
 

backend/src/module/rss/engine.py

@@ -14,7 +14,12 @@ from module.database.bangumi import (
     match_bangumi_in_list,
     release_fits_bangumi,
 )
-from module.downloader import AddResult, DownloadClient
+from module.downloader import (
+    AddResult,
+    DownloadClient,
+    DownloaderPool,
+    resolve_downloader_id,
+)
 from module.models import Bangumi, ResponseModel, RSSItem, Torrent
 from module.network import RequestContent
 from module.notification.events import (
@@ -323,14 +328,17 @@ class RSSEngine:
         return skip_ids
 
     async def refresh_rss(
-        self, client: DownloadClient, rss_id: Optional[int] = None
+        self, downloaders: DownloaderPool, rss_id: Optional[int] = None
     ) -> list[SystemEvent]:
-        """Refresh feeds with one parser engine for the complete workflow."""
+        """Refresh feeds with one parser engine for the complete workflow.
+
+        匹配到的种子投递到规则的下载器实例，规则未指定时用订阅的，再退回默认实例。
+        """
         with parser_engine_snapshot():
-            return await self._refresh_rss(client, rss_id)
+            return await self._refresh_rss(downloaders, rss_id)
 
     async def _refresh_rss(
-        self, client: DownloadClient, rss_id: Optional[int] = None
+        self, downloaders: DownloaderPool, rss_id: Optional[int] = None
     ) -> list[SystemEvent]:
         # Get All RSS Items
         if not rss_id:
@@ -438,6 +446,17 @@ class RSSEngine:
                             matched_data.official_title,
                         )
                         continue
+                    try:
+                        client = await downloaders.get(
+                            resolve_downloader_id(
+                                matched_data.downloader_id, rss_item.downloader_id
+                            )
+                        )
+                    except ConnectionError as e:
+                        # 实例不可用：不入库，下一轮重试；不可用由重命名轮次通知
+                        logger.warning("Skip %s: %s", torrent.name, e)
+                        failed_ids.add(id(torrent))
+                        continue
                     result = await client.add_torrent(torrent, matched_data)
                     if result is AddResult.FAILED:
                         # 投递失败：不入库（check_new 按 URL 去重，入库就
@@ -480,7 +499,9 @@ class RSSEngine:
                     t for t in torrents if await self.plugin_accepts(t, bangumi)
                 ]
             if torrents:
-                async with DownloadClient() as client:
+                async with DownloadClient(
+                    resolve_downloader_id(bangumi.downloader_id)
+                ) as client:
                     result = await client.add_torrent(torrents, bangumi)
                     if result is AddResult.FAILED:
                         return ResponseModel(

backend/src/test/factories.py

@@ -2,6 +2,7 @@
 
 from datetime import datetime, timezone
 
+from module.downloader import DownloaderPool
 from module.models import Bangumi, RSSItem, Torrent
 from module.models.config import Config
 from module.models.passkey import Passkey
@@ -85,3 +86,14 @@ def make_passkey(**overrides) -> Passkey:
     )
     defaults.update(overrides)
     return Passkey(**defaults)
+
+
+class SingleClientPool(DownloaderPool):
+    """所有实例都返回同一个（模拟）客户端的下载器池。"""
+
+    def __init__(self, client) -> None:
+        super().__init__()
+        self.client = client
+
+    async def get(self, instance_id: str):
+        return self.client

backend/src/test/test_api_bangumi_extended.py

@@ -297,79 +297,64 @@ class TestApplyOffset:
         """POST /bangumi/apply-offset/{id} applies offset and reruns the renamer."""
         mock_db = MagicMock()
         mock_db.bangumi.apply_offset = AsyncMock(return_value=True)
-        mock_renamer = AsyncMock()
+        mock_rename_all = AsyncMock(return_value=([], []))
         with patch("module.api.bangumi.Database", _async_db_mock(mock_db)):
-            with patch("module.api.bangumi.Renamer", return_value=mock_renamer):
-                with patch("module.api.bangumi.DownloadClient") as MockClient:
-                    MockClient.return_value.__aenter__ = AsyncMock(
-                        return_value=MagicMock()
-                    )
-                    MockClient.return_value.__aexit__ = AsyncMock(return_value=False)
-                    response = authed_client.post("/api/v1/bangumi/apply-offset/1")
+            with patch("module.api.bangumi.rename_all", mock_rename_all):
+                response = authed_client.post("/api/v1/bangumi/apply-offset/1")
 
         assert response.status_code == 200
         data = response.json()
         assert data["status"] is True
         mock_db.bangumi.apply_offset.assert_awaited_once_with(1)
-        mock_renamer.rename.assert_awaited_once()
+        mock_rename_all.assert_awaited_once()
 
     def test_apply_offset_not_found_skips_rename(self, authed_client):
         """POST /bangumi/apply-offset/{id} with a non-existent bangumi does not rename."""
         mock_db = MagicMock()
         mock_db.bangumi.apply_offset = AsyncMock(return_value=False)
         with patch("module.api.bangumi.Database", _async_db_mock(mock_db)):
-            with patch("module.api.bangumi.Renamer") as MockRenamer:
+            with patch("module.api.bangumi.rename_all") as mock_rename_all:
                 response = authed_client.post("/api/v1/bangumi/apply-offset/999")
 
         assert response.status_code == 404
-        MockRenamer.assert_not_called()
+        mock_rename_all.assert_not_called()
 
     def test_apply_offset_many_success(self, authed_client):
         """POST /bangumi/apply-offset/many applies offsets to every id and renames once."""
         mock_db = MagicMock()
         mock_db.bangumi.apply_offset = AsyncMock(side_effect=[True, True])
-        mock_renamer = AsyncMock()
+        mock_rename_all = AsyncMock(return_value=([], []))
         with patch("module.api.bangumi.Database", _async_db_mock(mock_db)):
-            with patch("module.api.bangumi.Renamer", return_value=mock_renamer):
-                with patch("module.api.bangumi.DownloadClient") as MockClient:
-                    MockClient.return_value.__aenter__ = AsyncMock(
-                        return_value=MagicMock()
-                    )
-                    MockClient.return_value.__aexit__ = AsyncMock(return_value=False)
-                    response = authed_client.post(
-                        "/api/v1/bangumi/apply-offset/many", json=[1, 2]
-                    )
+            with patch("module.api.bangumi.rename_all", mock_rename_all):
+                response = authed_client.post(
+                    "/api/v1/bangumi/apply-offset/many", json=[1, 2]
+                )
 
         assert response.status_code == 200
         assert mock_db.bangumi.apply_offset.await_count == 2
-        mock_renamer.rename.assert_awaited_once()
+        mock_rename_all.assert_awaited_once()
 
     def test_apply_offset_many_partial_failure(self, authed_client):
         """POST /bangumi/apply-offset/many reports a partial failure and still renames."""
         mock_db = MagicMock()
         mock_db.bangumi.apply_offset = AsyncMock(side_effect=[True, False])
-        mock_renamer = AsyncMock()
+        mock_rename_all = AsyncMock(return_value=([], []))
         with patch("module.api.bangumi.Database", _async_db_mock(mock_db)):
-            with patch("module.api.bangumi.Renamer", return_value=mock_renamer):
-                with patch("module.api.bangumi.DownloadClient") as MockClient:
-                    MockClient.return_value.__aenter__ = AsyncMock(
-                        return_value=MagicMock()
-                    )
-                    MockClient.return_value.__aexit__ = AsyncMock(return_value=False)
-                    response = authed_client.post(
-                        "/api/v1/bangumi/apply-offset/many", json=[1, 2]
-                    )
+            with patch("module.api.bangumi.rename_all", mock_rename_all):
+                response = authed_client.post(
+                    "/api/v1/bangumi/apply-offset/many", json=[1, 2]
+                )
 
         assert response.status_code == 500
-        mock_renamer.rename.assert_awaited_once()
+        mock_rename_all.assert_awaited_once()
 
     def test_apply_offset_many_empty_list(self, authed_client):
         """POST /bangumi/apply-offset/many with an empty list is a 400, no DB/rename calls."""
-        with patch("module.api.bangumi.Renamer") as MockRenamer:
+        with patch("module.api.bangumi.rename_all") as mock_rename_all:
             response = authed_client.post("/api/v1/bangumi/apply-offset/many", json=[])
 
         assert response.status_code == 400
-        MockRenamer.assert_not_called()
+        mock_rename_all.assert_not_called()
 
     def test_apply_offset_auth_required(self, unauthed_client):
         """POST /bangumi/apply-offset/{id} requires authentication."""

backend/src/test/test_api_downloader.py

@@ -182,7 +182,7 @@ class TestOtherAuthRequired:
 class TestGetTorrents:
     def test_get_torrents_success(self, authed_client, mock_download_client):
         """GET /downloader/torrents returns list of torrents."""
-        with patch("module.api.downloader.DownloadClient") as MockClient:
+        with patch("module.downloader.download_client.DownloadClient") as MockClient:
             MockClient.return_value.__aenter__ = AsyncMock(
                 return_value=mock_download_client
             )
@@ -194,11 +194,12 @@ class TestGetTorrents:
         data = response.json()
         assert len(data) == 2
         assert data[0]["hash"] == "abc123"
+        assert data[0]["downloader_id"] == "default"
 
     def test_get_torrents_empty(self, authed_client, mock_download_client):
         """GET /downloader/torrents returns empty list when no torrents."""
         mock_download_client.get_torrent_info.return_value = []
-        with patch("module.api.downloader.DownloadClient") as MockClient:
+        with patch("module.downloader.download_client.DownloadClient") as MockClient:
             MockClient.return_value.__aenter__ = AsyncMock(
                 return_value=mock_download_client
             )

backend/src/test/test_api_events.py

@@ -84,10 +84,12 @@ class TestDownloaderPayload:
         mock_client.__aenter__ = AsyncMock(return_value=mock_client)
         mock_client.__aexit__ = AsyncMock(return_value=False)
 
-        with patch("module.api.events.DownloadClient", return_value=mock_client):
+        with patch(
+            "module.downloader.download_client.DownloadClient", return_value=mock_client
+        ):
             result = await _downloader_payload()
 
-        assert result == torrents
+        assert result == [{**torrents[0], "downloader_id": "default"}]
         mock_client.get_torrent_info.assert_called_once_with(
             category="Bangumi", status_filter=None
         )
@@ -99,7 +101,9 @@ class TestDownloaderPayload:
         mock_client.__aenter__ = AsyncMock(side_effect=RuntimeError("no downloader"))
         mock_client.__aexit__ = AsyncMock(return_value=False)
 
-        with patch("module.api.events.DownloadClient", return_value=mock_client):
+        with patch(
+            "module.downloader.download_client.DownloadClient", return_value=mock_client
+        ):
             result = await _downloader_payload()
 
         assert result is None
@@ -132,7 +136,10 @@ class TestDownloaderPayloadTimeout:
         mock_client = _make_hung_client(cancelled)
 
         with (
-            patch("module.api.events.DownloadClient", return_value=mock_client),
+            patch(
+                "module.downloader.download_client.DownloadClient",
+                return_value=mock_client,
+            ),
             patch("module.api.events._DOWNLOADER_TIMEOUT_SECONDS", 0.05),
         ):
             start = time.monotonic()
@@ -163,7 +170,10 @@ class TestEventGeneratorNonBlocking:
         mock_client = _make_hung_client(cancelled)
 
         with (
-            patch("module.api.events.DownloadClient", return_value=mock_client),
+            patch(
+                "module.downloader.download_client.DownloadClient",
+                return_value=mock_client,
+            ),
             patch("module.api.events._DOWNLOADER_TIMEOUT_SECONDS", 0.05),
             patch("module.api.events.LOG_PATH", tmp_path / "missing.log"),
         ):

backend/src/test/test_integration.py

@@ -6,7 +6,7 @@ from module.database import Database
 from module.downloader import AddResult
 from module.models import EpisodeFile, Torrent
 from module.rss.engine import RSSEngine
-from test.factories import make_bangumi, make_rss_item
+from test.factories import SingleClientPool, make_bangumi, make_rss_item
 
 # ---------------------------------------------------------------------------
 # RSS → Download Flow
@@ -58,7 +58,7 @@ class TestRssToDownloadFlow:
             mock_client.add_torrent = AsyncMock(return_value=AddResult.ADDED)
 
             # 5. Execute refresh_rss
-            await engine.refresh_rss(mock_client)
+            await engine.refresh_rss(SingleClientPool(mock_client))
 
         # 6. Verify: matched torrents were downloaded
         assert mock_client.add_torrent.call_count == 2
@@ -109,7 +109,7 @@ class TestRssToDownloadFlow:
             mock_get.return_value = torrents
             mock_client = AsyncMock()
             mock_client.add_torrent = AsyncMock(return_value=AddResult.ADDED)
-            await engine.refresh_rss(mock_client)
+            await engine.refresh_rss(SingleClientPool(mock_client))
 
         # Only 1080p should be downloaded (720p is filtered)
         assert mock_client.add_torrent.call_count == 1
@@ -149,7 +149,7 @@ class TestRssToDownloadFlow:
             mock_get.return_value = torrents
             mock_client = AsyncMock()
             mock_client.add_torrent = AsyncMock(return_value=AddResult.ADDED)
-            await engine.refresh_rss(mock_client)
+            await engine.refresh_rss(SingleClientPool(mock_client))
 
         # Only ep02 should be downloaded (ep01 already exists)
         assert mock_client.add_torrent.call_count == 1

backend/src/test/test_loops.py

@@ -99,7 +99,7 @@ class TestRssTick:
 
         with (
             patch(
-                "module.core.loops.DownloadClient",
+                "module.core.loops.DownloaderPool",
                 return_value=_async_cm(AsyncMock()),
             ),
             patch("module.core.loops.Database", return_value=_async_cm(mock_db)),
@@ -125,7 +125,7 @@ class TestRssTick:
 
         with (
             patch(
-                "module.core.loops.DownloadClient",
+                "module.core.loops.DownloaderPool",
                 return_value=_async_cm(AsyncMock()),
             ),
             patch("module.core.loops.Database", return_value=_async_cm(mock_db)),
@@ -144,6 +144,15 @@ class TestRssTick:
 # ---------------------------------------------------------------------------
 
 
+def _rename_all(renamer):
+    """rename_all 的替身：返回模拟 Renamer 的结果与事件。"""
+
+    async def run():
+        return await renamer.rename(), list(renamer.events)
+
+    return run
+
+
 class TestRenameTick:
     async def test_persists_and_dispatches_conflict_event_when_push_disabled(self):
         event = RenameConflictEvent(
@@ -158,11 +167,7 @@ class TestRenameTick:
         mock_renamer.events = [event]
 
         with (
-            patch(
-                "module.core.loops.DownloadClient",
-                return_value=_async_cm(AsyncMock()),
-            ),
-            patch("module.core.loops.Renamer", return_value=mock_renamer),
+            patch("module.core.loops.rename_all", _rename_all(mock_renamer)),
             patch("module.core.loops.settings") as mock_settings,
         ):
             mock_settings.notification.enable = False
@@ -180,11 +185,7 @@ class TestRenameTick:
         mock_renamer.rename = AsyncMock(return_value=[notify])
 
         with (
-            patch(
-                "module.core.loops.DownloadClient",
-                return_value=_async_cm(AsyncMock()),
-            ),
-            patch("module.core.loops.Renamer", return_value=mock_renamer),
+            patch("module.core.loops.rename_all", _rename_all(mock_renamer)),
             patch("module.core.loops.settings") as mock_settings,
         ):
             mock_settings.notification.enable = True
@@ -200,11 +201,7 @@ class TestRenameTick:
         mock_renamer.rename = AsyncMock(return_value=[notify])
 
         with (
-            patch(
-                "module.core.loops.DownloadClient",
-                return_value=_async_cm(AsyncMock()),
-            ),
-            patch("module.core.loops.Renamer", return_value=mock_renamer),
+            patch("module.core.loops.rename_all", _rename_all(mock_renamer)),
             patch("module.core.loops.settings") as mock_settings,
         ):
             mock_settings.notification.enable = False
@@ -232,11 +229,7 @@ class TestRenameTick:
         mock_renamer.rename = AsyncMock(return_value=notifications)
 
         with (
-            patch(
-                "module.core.loops.DownloadClient",
-                return_value=_async_cm(AsyncMock()),
-            ),
-            patch("module.core.loops.Renamer", return_value=mock_renamer),
+            patch("module.core.loops.rename_all", _rename_all(mock_renamer)),
             patch("module.core.loops.settings") as mock_settings,
         ):
             mock_settings.notification.enable = True

backend/src/test/test_mcp_tools.py

@@ -401,7 +401,9 @@ class TestDispatch:
         mock_client.__aenter__ = AsyncMock(return_value=mock_client)
         mock_client.__aexit__ = AsyncMock(return_value=False)
 
-        with patch("module.mcp.tools.DownloadClient", return_value=mock_client):
+        with patch(
+            "module.downloader.download_client.DownloadClient", return_value=mock_client
+        ):
             result = await _dispatch("list_downloads", {"status": "all"})
 
         mock_client.get_torrent_info.assert_called_once_with(
@@ -417,7 +419,9 @@ class TestDispatch:
         mock_client.__aenter__ = AsyncMock(return_value=mock_client)
         mock_client.__aexit__ = AsyncMock(return_value=False)
 
-        with patch("module.mcp.tools.DownloadClient", return_value=mock_client):
+        with patch(
+            "module.downloader.download_client.DownloadClient", return_value=mock_client
+        ):
             await _dispatch("list_downloads", {"status": "downloading"})
 
         mock_client.get_torrent_info.assert_called_once_with(
@@ -443,7 +447,9 @@ class TestDispatch:
         mock_client.__aenter__ = AsyncMock(return_value=mock_client)
         mock_client.__aexit__ = AsyncMock(return_value=False)
 
-        with patch("module.mcp.tools.DownloadClient", return_value=mock_client):
+        with patch(
+            "module.downloader.download_client.DownloadClient", return_value=mock_client
+        ):
             result = await _dispatch("list_downloads", {})
 
         expected_keys = {
@@ -454,6 +460,7 @@ class TestDispatch:
             "dlspeed",
             "upspeed",
             "eta",
+            "downloader",
         }
         assert set(result[0].keys()) == expected_keys
 
@@ -516,7 +523,7 @@ class TestDispatch:
         mock_engine.refresh_rss = AsyncMock(return_value=None)
 
         with (
-            patch("module.mcp.tools.DownloadClient", return_value=mock_client),
+            patch("module.mcp.tools.DownloaderPool", return_value=mock_client),
             patch("module.mcp.tools.RSSEngine", return_value=mock_engine),
         ):
             result = await _dispatch("refresh_feeds", {})

backend/src/test/test_migrations_module.py

@@ -705,8 +705,10 @@ class TestRunMigrations:
         assert prefix == "legacy_cccccccc"
 
     @pytest.mark.parametrize("preexisting", [None, "torrent"])
-    def test_v26_downloader_id_existing_rows_default(self, preexisting, monkeypatch):
-        """v26：存量番剧 / 电影 / 订阅 / 种子都归属默认下载器实例；
+    def test_v26_downloader_id_existing_rows_follow_default(
+        self, preexisting, monkeypatch
+    ):
+        """v26：存量种子归属实例 default；番剧 / 电影 / 订阅为空（跟随默认实例）。
         已有该列的表（如被 create_all 提前建出）跳过，其它表照常补列。"""
         engine = _make_v0_engine()
         with engine.begin() as conn:
@@ -728,12 +730,13 @@ class TestRunMigrations:
 
         run_migrations(engine)
 
+        expected = {"bangumi": None, "movie": None, "rssitem": None}
         with engine.connect() as conn:
             for table in ("bangumi", "movie", "rssitem", "torrent"):
                 values = conn.execute(
                     text(f"SELECT downloader_id FROM {table}")
                 ).scalars()
-                assert list(values) == ["default"], table
+                assert list(values) == [expected.get(table, "default")], table
 
     def test_is_idempotent(self):
         engine = _make_v0_engine()

backend/src/test/test_multi_downloader.py

@@ -0,0 +1,330 @@
+"""多下载器实例（P2.5）：按实例缓存客户端、按规则 / 订阅路由新种子、
+重命名逐实例运行、不可用实例跳过并通知一次。两个实例都用 mock 下载器。"""
+
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from ab_sdk.events import DownloaderUnavailableEvent, TorrentOrganized
+from module.conf import settings
+from module.database import Database
+from module.downloader import (
+    DownloadClient,
+    DownloaderPool,
+    list_torrents,
+    resolve_downloader_id,
+)
+from module.downloader.client.mock_downloader import MockDownloader
+from module.manager import renamer as renamer_module
+from module.manager.renamer import Renamer, rename_all
+from module.models import Torrent
+from module.models.config import PluginInstance, Plugins, Slots
+from module.rss.engine import RSSEngine
+from test.factories import make_bangumi, make_rss_item
+
+
+def _instance(instance_id: str, path: str) -> PluginInstance:
+    return PluginInstance(
+        id=instance_id,
+        point="downloader",
+        provider="mock",
+        options={"host": f"{instance_id}:8080", "path": path},
+    )
+
+
+@pytest.fixture
+def two_instances(monkeypatch):
+    """实例 a（默认）与 b，各自的下载目录不同。"""
+    plugins = Plugins(
+        instances=[_instance("a", "/a/Bangumi"), _instance("b", "/b/Bangumi")],
+        slots=Slots(downloader="a"),
+    )
+    monkeypatch.setattr(settings, "plugins", plugins)
+    monkeypatch.setattr(renamer_module, "_unavailable", set())
+    return plugins
+
+
+def _mock(instance_id: str) -> MockDownloader:
+    client = DownloadClient(instance_id).client
+    assert isinstance(client, MockDownloader)
+    return client
+
+
+def _completed(name: str, torrent_hash: str, save_path: str) -> dict:
+    return {
+        "hash": torrent_hash,
+        "name": name,
+        "save_path": save_path,
+        "category": "Bangumi",
+        "state": "completed",
+        "progress": 1.0,
+        "files": [{"name": f"{name}.mkv", "size": 100}],
+        "tags": "",
+    }
+
+
+class TestClientCache:
+    def test_download_client_two_instances_separate_cached_clients(self, two_instances):
+        a, b = _mock("a"), _mock("b")
+        assert a is not b
+        assert _mock("a") is a
+        assert DownloadClient().instance_id == "a"
+
+    async def test_download_client_credentials_rejected_latch_is_per_instance(
+        self, two_instances
+    ):
+        b = _mock("b")
+        b.auth = AsyncMock(return_value=False)  # type: ignore[method-assign]
+        b.last_auth_error = "credentials"  # type: ignore[attr-defined]
+        with pytest.raises(ConnectionError):
+            async with DownloadClient("b"):
+                pass
+        with pytest.raises(ConnectionError):
+            async with DownloadClient("b"):
+                pass
+        assert b.auth.await_count == 1  # 闩锁命中，不再登录
+        async with DownloadClient("a") as client:
+            assert client.authed
+
+
+@pytest.mark.parametrize(
+    "candidates, expected",
+    [
+        ((None, None), "a"),
+        (("", None), "a"),
+        (("b", None), "b"),
+        ((None, "b"), "b"),
+        (("a", "b"), "a"),
+        (("gone", "b"), "b"),
+        (("gone",), "a"),
+    ],
+)
+def test_resolve_downloader_id_candidates_first_known_else_default(
+    two_instances, candidates, expected
+):
+    assert resolve_downloader_id(*candidates) == expected
+
+
+class TestRssRouting:
+    @pytest.mark.parametrize(
+        "bangumi_instance, rss_instance, expected",
+        [("b", None, "b"), (None, "b", "b"), (None, None, "a"), ("a", "b", "a")],
+    )
+    async def test_refresh_rss_adds_to_rule_then_feed_then_default_instance(
+        self, two_instances, db_engine, bangumi_instance, rss_instance, expected
+    ):
+        engine = RSSEngine(Database(engine=db_engine))
+        await engine.db.rss.add(make_rss_item(enabled=True, downloader_id=rss_instance))
+        await engine.db.bangumi.add(
+            make_bangumi(
+                title_raw="Mushoku Tensei",
+                filter="",
+                save_path=None,
+                downloader_id=bangumi_instance,
+            )
+        )
+        torrent = Torrent(
+            name="[Sub] Mushoku Tensei - 12 [1080p].mkv",
+            url="magnet:?xt=urn:btih:12",
+        )
+        with patch.object(
+            RSSEngine, "_get_torrents", AsyncMock(return_value=[torrent])
+        ):
+            async with DownloaderPool() as pool:
+                await engine.refresh_rss(pool)
+
+        rows = await engine.db.torrent.search_all()
+        assert [(r.downloaded, r.downloader_id) for r in rows] == [(True, expected)]
+        other = "a" if expected == "b" else "b"
+        assert len(_mock(expected)._torrents) == 1
+        assert _mock(other)._torrents == {}
+        # 新规则的保存目录取目标实例的下载目录
+        (added,) = _mock(expected)._torrents.values()
+        assert added["save_path"].startswith(f"/{expected}/Bangumi/")
+
+    async def test_refresh_rss_unavailable_instance_torrent_not_persisted(
+        self, two_instances, db_engine
+    ):
+        engine = RSSEngine(Database(engine=db_engine))
+        await engine.db.rss.add(make_rss_item(enabled=True))
+        await engine.db.bangumi.add(
+            make_bangumi(title_raw="Mushoku Tensei", filter="", downloader_id="b")
+        )
+        _mock("b").auth = AsyncMock(return_value=False)  # type: ignore[method-assign]
+        torrent = Torrent(
+            name="[Sub] Mushoku Tensei - 12 [1080p].mkv",
+            url="magnet:?xt=urn:btih:12",
+        )
+        with patch.object(
+            RSSEngine, "_get_torrents", AsyncMock(return_value=[torrent])
+        ):
+            async with DownloaderPool() as pool:
+                events = await engine.refresh_rss(pool)
+
+        assert events == []
+        assert await engine.db.torrent.search_all() == []  # 下一轮重试
+
+
+class TestRenameAll:
+    async def test_rename_all_events_carry_instance_of_torrent(self, two_instances):
+        _mock("b")._torrents["hb"] = _completed(
+            "[Sub] Show - 01 [1080p]", "hb", "/b/Bangumi/Show/Season 1"
+        )
+        published: list[object] = []
+        with patch.object(renamer_module.plugin_host, "publish", published.append):
+            renamed, events = await rename_all()
+
+        organized = [e for e in published if isinstance(e, TorrentOrganized)]
+        assert [(e.torrent_hash, e.downloader_id) for e in organized] == [("hb", "b")]
+        assert events == []
+        assert [n.official_title for n in renamed] == ["Show"]
+
+    async def test_rename_all_unavailable_instance_skipped_and_notified_once(
+        self, two_instances
+    ):
+        _mock("a")._torrents["ha"] = _completed(
+            "[Sub] Show - 01 [1080p]", "ha", "/a/Bangumi/Show/Season 1"
+        )
+        _mock("b").auth = AsyncMock(return_value=False)  # type: ignore[method-assign]
+
+        first_renamed, first_events = await rename_all()
+        _, second_events = await rename_all()
+
+        assert [n.official_title for n in first_renamed] == ["Show"]
+        assert first_events == [
+            DownloaderUnavailableEvent(
+                host="b:8080", reason="unreachable", instance_id="b"
+            )
+        ]
+        assert second_events == []
+
+    async def test_rename_all_recovered_instance_notified_again_on_next_outage(
+        self, two_instances
+    ):
+        counts = []
+        for reachable in (False, True, False):
+            # 登录失败后缓存的客户端被丢弃，每轮给新取到的客户端打桩
+            _mock("b").auth = AsyncMock(return_value=reachable)  # type: ignore[method-assign]
+            counts.append(len((await rename_all())[1]))
+        assert counts == [1, 0, 1]
+
+    async def test_rename_replacement_of_other_instance_not_recovered(
+        self, two_instances
+    ):
+        """实例 a 的版本替换事务不能被实例 b 的重命名当成「新种子已消失」去恢复。"""
+        from module.models.rename_operation import RenameOperation
+
+        async with DownloadClient("a") as client_a:
+            a_type = Renamer(client_a)._downloader_type()
+        async with Database() as db:
+            db.add(
+                RenameOperation(
+                    kind="replacement",
+                    state="planned",
+                    downloader_type=a_type,
+                    new_task_id="only-on-a",
+                    save_path="/a/Bangumi/Show/Season 1",
+                    target_path="Show S01E01.mkv",
+                    source_path="x.mkv",
+                )
+            )
+            await db.commit()
+
+        with patch.object(
+            Renamer, "_recover_missing_replacement", AsyncMock()
+        ) as recover:
+            async with DownloadClient("b") as client_b:
+                await Renamer(client_b).rename()
+        recover.assert_not_awaited()
+
+
+async def test_list_torrents_one_instance_down_lists_the_other(two_instances):
+    _mock("a")._torrents["ha"] = _completed("A", "ha", "/a/Bangumi/A/Season 1")
+    _mock("b").auth = AsyncMock(return_value=False)  # type: ignore[method-assign]
+    listed = await list_torrents()
+    assert [(t["hash"], t["downloader_id"]) for t in listed or []] == [("ha", "a")]
+
+
+class TestRuleOnAnotherInstance:
+    """规则换到另一个实例后，旧种子留在原实例；删除规则时两边的种子都删。"""
+
+    async def _rule_with_torrent_on_a(self) -> int:
+        _mock("a")._torrents["old"] = _completed(
+            "Old", "old", "/a/Bangumi/Test Anime (2024)/Season 1"
+        )
+        async with Database() as db:
+            await db.bangumi.add(
+                make_bangumi(save_path="/a/Bangumi/Test Anime (2024)/Season 1")
+            )
+            await db.torrent.add(
+                Torrent(name="Old", url="u-old", bangumi_id=1, downloader_id="a")
+            )
+        return 1
+
+    async def test_update_rule_new_instance_keeps_old_torrents_in_place(
+        self, two_instances
+    ):
+        from module.manager import TorrentManager
+        from module.models import BangumiUpdate
+
+        bangumi_id = await self._rule_with_torrent_on_a()
+        async with Database() as db:
+            old = await db.bangumi.search_id(bangumi_id)
+            assert old is not None
+            update = BangumiUpdate(**old.model_dump(exclude={"id"}))
+            update.downloader_id = "b"
+            await TorrentManager(db).update_rule(bangumi_id, update)
+            new = await db.bangumi.search_id(bangumi_id)
+
+        assert new is not None and new.downloader_id == "b"
+        assert new.save_path == "/b/Bangumi/Test Anime (2024)/Season 1"
+        moved = _mock("a")._torrents["old"]["save_path"]
+        assert moved == "/a/Bangumi/Test Anime (2024)/Season 1"
+
+    async def test_delete_rule_deletes_torrents_on_every_instance(self, two_instances):
+        from module.manager import TorrentManager
+
+        bangumi_id = await self._rule_with_torrent_on_a()
+        _mock("b")._torrents["new"] = _completed(
+            "New", "new", "/a/Bangumi/Test Anime (2024)/Season 1"
+        )
+        async with Database() as db:
+            bangumi = await db.bangumi.search_id(bangumi_id)
+            assert bangumi is not None
+            bangumi.downloader_id = "b"
+            await db.bangumi.update(bangumi)
+            resp = await TorrentManager(db).delete_rule(bangumi_id, file=True)
+
+        assert resp.status
+        assert _mock("a")._torrents == {}
+        assert _mock("b")._torrents == {}
+
+
+class TestDownloaderApi:
+    def test_pause_routes_to_given_instance(self, two_instances, authed_client):
+        _mock("b")._torrents["hb"] = _completed("B", "hb", "/b/Bangumi/B")
+        resp = authed_client.post(
+            "/api/v1/downloader/torrents/pause",
+            json={"hashes": ["hb"], "downloader_id": "b"},
+        )
+        assert resp.status_code == 200
+        assert _mock("b")._torrents["hb"]["state"] == "paused"
+
+    def test_pause_unknown_instance_404(self, two_instances, authed_client):
+        resp = authed_client.post(
+            "/api/v1/downloader/torrents/pause",
+            json={"hashes": ["hb"], "downloader_id": "gone"},
+        )
+        assert resp.status_code == 404
+
+    def test_instances_lists_downloaders_and_default(
+        self, two_instances, authed_client
+    ):
+        resp = authed_client.get("/api/v1/downloader/instances")
+        assert resp.json() == {
+            "default": "a",
+            "instances": [
+                {"id": "a", "provider": "mock"},
+                {"id": "b", "provider": "mock"},
+            ],
+        }

backend/src/test/test_notification.py

@@ -561,9 +561,13 @@ class TestEventInboxContract:
         assert event.kind == "downloader_unavailable"
         assert event.severity == "error"
         assert event.once is False
-        # reason 不进 dedup_key：unreachable→credentials 的翻转合并为一条
-        assert event.dedup_key() == "downloader:http://qb:8080"
-        assert event.payload() == {"host": "http://qb:8080", "reason": "credentials"}
+        # 按实例去重，reason 不进 dedup_key：unreachable→credentials 的翻转合并为一条
+        assert event.dedup_key() == "downloader:default"
+        assert event.payload() == {
+            "host": "http://qb:8080",
+            "reason": "credentials",
+            "instance": "default",
+        }
         title, body = event.describe()
         assert "下载器" in title
         assert "密码" in body

backend/src/test/test_organize.py

@@ -79,6 +79,7 @@ def add_strategy(registry, provider_id: str, target_name, plugin_id: str = "ext"
 @pytest.fixture
 def renamer(mock_qb_client):
     with patch("module.downloader.download_client.settings") as mock_settings:
+        mock_settings.downloader.id = "default"
         mock_settings.downloader.type = "qbittorrent"
         mock_settings.downloader.host = "localhost:8080"
         mock_settings.downloader.path = "/downloads/Bangumi"

backend/src/test/test_plugin_ingest.py

@@ -31,7 +31,7 @@ from module.plugin.registry import HookEntry, ProviderEntry
 from module.plugin.runner import CircuitBreaker, HookRunner
 from module.rss.analyser import RSSAnalyser
 from module.rss.engine import RSSEngine
-from test.factories import make_bangumi, make_rss_item, make_torrent
+from test.factories import SingleClientPool, make_bangumi, make_rss_item, make_torrent
 
 TORRENT_NAME = "[Sub] Mushoku Tensei - 12 [1080p].mkv"
 
@@ -69,7 +69,7 @@ async def refresh_with(engine: RSSEngine, torrents: list[Torrent]):
     client.add_torrent = AsyncMock(return_value=AddResult.ADDED)
     with patch.object(RSSEngine, "_get_torrents", new_callable=AsyncMock) as get:
         get.return_value = torrents
-        await engine.refresh_rss(client)
+        await engine.refresh_rss(SingleClientPool(client))
     return client
 
 

backend/src/test/test_rss_engine_new.py

@@ -14,7 +14,7 @@ from module.notification.events import DownloadFailureEvent, RssFailureEvent
 from module.parser.analyser.selector import parse_configured_release_title
 from module.parser.analyser.tokenizer import ReleaseKind
 from module.rss.engine import RSSEngine
-from test.factories import make_bangumi, make_rss_item, make_torrent
+from test.factories import SingleClientPool, make_bangumi, make_rss_item, make_torrent
 
 
 @pytest_asyncio.fixture
@@ -281,7 +281,7 @@ class TestRefreshRss:
             client = AsyncMock()
             client.add_torrent = AsyncMock(return_value=AddResult.ADDED)
 
-            await rss_engine.refresh_rss(client)
+            await rss_engine.refresh_rss(SingleClientPool(client))
 
         # Verify download was attempted
         client.add_torrent.assert_called_once()
@@ -305,7 +305,7 @@ class TestRefreshRss:
         ) as mock_get:
             mock_get.return_value = [unmatched]
             client = AsyncMock()
-            await rss_engine.refresh_rss(client)
+            await rss_engine.refresh_rss(SingleClientPool(client))
 
         client.add_torrent.assert_not_called()
         all_torrents = await rss_engine.db.torrent.search_all()
@@ -341,7 +341,7 @@ class TestRefreshRss:
             mock_get.return_value = [matched, unmatched]
             client = AsyncMock()
             client.add_torrent = AsyncMock(return_value=AddResult.ADDED)
-            await rss_engine.refresh_rss(client)
+            await rss_engine.refresh_rss(SingleClientPool(client))
 
         all_torrents = await rss_engine.db.torrent.search_all()
         assert len(all_torrents) == 1
@@ -372,7 +372,7 @@ class TestRefreshRss:
             mock_get.return_value = [matched, unmatched]
             client = AsyncMock()
             client.add_torrent = AsyncMock(return_value=AddResult.ADDED)
-            await rss_engine.refresh_rss(client)
+            await rss_engine.refresh_rss(SingleClientPool(client))
 
         all_torrents = await rss_engine.db.torrent.search_all()
         assert len(all_torrents) == 2
@@ -390,7 +390,7 @@ class TestRefreshRss:
         ) as mock_get:
             mock_get.return_value = []
             client = AsyncMock()
-            await rss_engine.refresh_rss(client, rss_id=2)
+            await rss_engine.refresh_rss(SingleClientPool(client), rss_id=2)
 
         # Only called once (for rss_id=2)
         mock_get.assert_called_once()
@@ -401,7 +401,7 @@ class TestRefreshRss:
             RSSEngine, "_get_torrents", new_callable=AsyncMock
         ) as mock_get:
             client = AsyncMock()
-            await rss_engine.refresh_rss(client, rss_id=999)
+            await rss_engine.refresh_rss(SingleClientPool(client), rss_id=999)
 
         mock_get.assert_not_called()
 
@@ -426,7 +426,7 @@ class TestRefreshRssEvents:
         ) as mock_pull:
             mock_pull.return_value = ([], "connection refused")
             client = AsyncMock()
-            events = await rss_engine.refresh_rss(client)
+            events = await rss_engine.refresh_rss(SingleClientPool(client))
 
         assert len(events) == 1
         assert isinstance(events[0], RssFailureEvent)
@@ -443,8 +443,8 @@ class TestRefreshRssEvents:
         ) as mock_pull:
             mock_pull.return_value = ([], "connection refused")
             client = AsyncMock()
-            first_events = await rss_engine.refresh_rss(client)
-            second_events = await rss_engine.refresh_rss(client)
+            first_events = await rss_engine.refresh_rss(SingleClientPool(client))
+            second_events = await rss_engine.refresh_rss(SingleClientPool(client))
 
         assert len(first_events) == 1
         assert second_events == []
@@ -459,7 +459,7 @@ class TestRefreshRssEvents:
         ) as mock_pull:
             mock_pull.return_value = ([], None)
             client = AsyncMock()
-            events = await rss_engine.refresh_rss(client)
+            events = await rss_engine.refresh_rss(SingleClientPool(client))
 
         assert events == []
 
@@ -481,7 +481,7 @@ class TestRefreshRssEvents:
             client = AsyncMock()
             client.add_torrent = AsyncMock(return_value=AddResult.FAILED)
 
-            events = await rss_engine.refresh_rss(client)
+            events = await rss_engine.refresh_rss(SingleClientPool(client))
 
         assert len(events) == 1
         assert isinstance(events[0], DownloadFailureEvent)
@@ -512,7 +512,7 @@ class TestRefreshRssEvents:
             client = AsyncMock()
             client.add_torrent = AsyncMock(return_value=AddResult.ADDED)
 
-            events = await rss_engine.refresh_rss(client)
+            events = await rss_engine.refresh_rss(SingleClientPool(client))
 
         assert events == []
 
@@ -537,7 +537,7 @@ class TestRefreshRssEvents:
             client = AsyncMock()
             client.add_torrent = AsyncMock(return_value=AddResult.DUPLICATE)
 
-            events = await rss_engine.refresh_rss(client)
+            events = await rss_engine.refresh_rss(SingleClientPool(client))
 
         assert events == []
         all_torrents = await rss_engine.db.torrent.search_all()
@@ -573,14 +573,14 @@ class TestRefreshRssRetry:
             # Tick 1: transient failure
             mock_get.return_value = [feed_torrent()]
             client.add_torrent = AsyncMock(return_value=AddResult.FAILED)
-            await rss_engine.refresh_rss(client)
+            await rss_engine.refresh_rss(SingleClientPool(client))
             client.add_torrent.assert_called_once()
             assert await rss_engine.db.torrent.search_all() == []
 
             # Tick 2: same feed item is still there, add now succeeds
             mock_get.return_value = [feed_torrent()]
             client.add_torrent = AsyncMock(return_value=AddResult.ADDED)
-            await rss_engine.refresh_rss(client)
+            await rss_engine.refresh_rss(SingleClientPool(client))
             client.add_torrent.assert_called_once()
 
         all_torrents = await rss_engine.db.torrent.search_all()
@@ -610,9 +610,9 @@ class TestRefreshRssRetry:
             client.add_torrent = AsyncMock(return_value=AddResult.ADDED)
 
             mock_get.return_value = [feed_torrent()]
-            await rss_engine.refresh_rss(client)
+            await rss_engine.refresh_rss(SingleClientPool(client))
             mock_get.return_value = [feed_torrent()]
-            await rss_engine.refresh_rss(client)
+            await rss_engine.refresh_rss(SingleClientPool(client))
 
         client.add_torrent.assert_not_called()
         all_torrents = await rss_engine.db.torrent.search_all()
@@ -811,7 +811,7 @@ class TestRefreshRssConcurrency:
             rss_engine, "_pull_rss_with_status", side_effect=track_concurrency
         ):
             client = AsyncMock()
-            await rss_engine.refresh_rss(client)
+            await rss_engine.refresh_rss(SingleClientPool(client))
 
         assert max_active <= 5
 
@@ -855,7 +855,7 @@ class TestRefreshRssPerHostThrottle:
             ),
             patch("module.rss.engine.RSS_PER_HOST_DELAY", 0),
         ):
-            await rss_engine.refresh_rss(client)
+            await rss_engine.refresh_rss(SingleClientPool(client))
 
         assert max_active["nyaa.example"] == 1
         assert max_active["mikan.example"] == 1
@@ -878,7 +878,7 @@ class TestRefreshRssPerHostThrottle:
             ),
             patch("module.rss.engine.RSS_PER_HOST_DELAY", 0),
         ):
-            await rss_engine.refresh_rss(client)
+            await rss_engine.refresh_rss(SingleClientPool(client))
 
         for rss_id in (1, 2):
             item = await rss_engine.db.rss.search_id(rss_id)
@@ -1334,7 +1334,7 @@ class TestRefreshRssPreferenceDedup:
             client = AsyncMock()
             client.add_torrent = AsyncMock(return_value=AddResult.ADDED)
 
-            await rss_engine.refresh_rss(client)
+            await rss_engine.refresh_rss(SingleClientPool(client))
 
             parsed_after_refresh = parse_configured_release_title(torrent.name)
             assert parsed_after_refresh is not None
@@ -1365,7 +1365,7 @@ class TestRefreshRssPreferenceDedup:
             client = AsyncMock()
             client.add_torrent = AsyncMock(return_value=AddResult.ADDED)
 
-            await rss_engine.refresh_rss(client)
+            await rss_engine.refresh_rss(SingleClientPool(client))
 
         client.add_torrent.assert_called_once()
         called_torrent = client.add_torrent.call_args[0][0]
@@ -1400,6 +1400,6 @@ class TestRefreshRssPreferenceDedup:
             client = AsyncMock()
             client.add_torrent = AsyncMock(return_value=AddResult.ADDED)
 
-            await rss_engine.refresh_rss(client)
+            await rss_engine.refresh_rss(SingleClientPool(client))
 
         assert client.add_torrent.call_count == 2