Merge remote-tracking branch 'origin/v3' into codex/manual-source-classification

# Conflicts:
#	docs/architecture-overview.md
#	docs/architecture/optimization-checklist.md
#	tests/fixtures/architecture/dependency-baseline.json

app/agent/llm/helper.py

@@ -7,6 +7,7 @@ import time
 from functools import wraps
 from typing import TYPE_CHECKING, Any, List, Optional
 from urllib.parse import urlsplit
+from uuid import uuid4
 
 from langchain_core.messages import AIMessage, AIMessageChunk
 
@@ -881,11 +882,17 @@ class LLMHelper:
             default_headers: dict[str, str] | None,
             model_kwargs: dict[str, Any],
     ) -> tuple[dict[str, str] | None, dict[str, Any]]:
-        """为 OpenAI 与 xAI 官方端点构造稳定提示词缓存路由参数。"""
+        """为官方端点构造稳定会话及提示词缓存路由参数。"""
         cache_key = str(prompt_cache_key or "").strip()
         headers = dict(default_headers or {})
         kwargs = dict(model_kwargs)
         provider_name = str(provider or "").strip().lower()
+        if cls._matches_endpoint_host(base_url, "opencode.ai"):
+            # 主对话沿用脱敏缓存键；独立测试、摘要调用的标识在模型实例内复用。
+            headers["x-opencode-session"] = cache_key or f"moviepilot-{uuid4().hex}"
+            if not any(key.lower() == "user-agent" for key in headers):
+                headers["User-Agent"] = "MoviePilot"
+            return headers, kwargs
         if not cache_key:
             return headers or None, kwargs
 

app/agent/policy/mcp.py

@@ -396,6 +396,8 @@ FIELD_DESCRIPTIONS = {
     "sample_rate": "Recorded audio sample rate in hertz.",
     "save_path": "Configured downloader-side save path for the download or subscription.",
     "scrape": "Generate metadata and images after manual transfer.",
+    "search_interval": "Scheduled search interval in whole hours (1-8760); null uses the system interval.",
+    "last_search": "Read-only UTC timestamp of the most recent subscription search attempt.",
     "search_imdbid": "Use IMDb identity during subscription search when set to 1.",
     "season": "Season number used by the media, search, subscription, or transfer operation.",
     "seasons": "Season-number expression recorded in history.",

app/agent/policy/resources/api_mcp_schema.json

@@ -1706,12 +1706,7 @@
         "tvdb",
         "musicbrainz",
         "theaudiodb",
-        "doubanmusic",
+        "doubanmusic"
-        "bilibili",
-        "mangguodiscover",
-        "migu",
-        "tencentvideodiscover",
-        "iqiyidiscover"
       ],
       "pattern": "^[a-z][a-z0-9._-]{0,63}$",
       "title": "MediaSource",
@@ -2862,6 +2857,18 @@
           "description": "Number of episodes still missing from the subscription.",
           "title": "Lack Episode"
         },
+        "last_search": {
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "description": "Read-only UTC timestamp of the most recent subscription search attempt.",
+          "title": "Last Search"
+        },
         "last_update": {
           "anyOf": [
             {
@@ -3057,6 +3064,20 @@
           "description": "Use IMDb identity during subscription search when set to 1.",
           "title": "Search Imdbid"
         },
+        "search_interval": {
+          "anyOf": [
+            {
+              "maximum": 8760.0,
+              "minimum": 1.0,
+              "type": "integer"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "description": "Scheduled search interval in whole hours (1-8760); null uses the system interval.",
+          "title": "Search Interval"
+        },
         "season": {
           "anyOf": [
             {

app/api/endpoints/media.py

@@ -64,11 +64,6 @@ _BUILTIN_MEDIA_SOURCES = (
         media_source=MediaSource.DoubanMusic,
         media_types=[MediaType.MUSIC],
     ),
-    _SchemaMediaSourceInfo(name="哔哩哔哩", media_source=MediaSource.Bilibili),
-    _SchemaMediaSourceInfo(name="芒果TV", media_source=MediaSource.MangoTV),
-    _SchemaMediaSourceInfo(name="咪咕视频", media_source=MediaSource.MiguVideo),
-    _SchemaMediaSourceInfo(name="腾讯视频", media_source=MediaSource.TencentVideo),
-    _SchemaMediaSourceInfo(name="爱奇艺", media_source=MediaSource.Iqiyi),
 )
 
 

app/application/subscription/contract.py

@@ -21,6 +21,8 @@ _SUBSCRIPTION_FIELDS = frozenset(
         "year",
         "type",
         "keyword",
+        "search_interval",
+        "last_search",
         "media_source",
         "media_id",
         "music_type",
@@ -83,6 +85,7 @@ _SUBSCRIPTION_HISTORY_FIELDS = (_SUBSCRIPTION_FIELDS - {
     "note",
     "state",
     "last_update",
+    "last_search",
     "downloader",
     "manual_total_episode",
 }) | _CLASSIFICATION_HISTORY_FIELDS
@@ -153,6 +156,7 @@ class SubscriptionSnapshot:
     year: Optional[str] = None
     type: Optional[str] = None
     keyword: Optional[str] = None
+    search_interval: Optional[int] = None
     media_source: Optional[MediaSource] = None
     media_id: Optional[str] = None
     music_type: Optional[str] = None
@@ -179,6 +183,7 @@ class SubscriptionSnapshot:
     note: Optional[builtins.list[int]] = None
     state: str = "N"
     last_update: Optional[str] = None
+    last_search: Optional[str] = None
     date: Optional[str] = None
     username: Optional[str] = None
     sites: Optional[builtins.list[int]] = None
@@ -219,6 +224,7 @@ class SubscriptionHistorySnapshot:
     year: Optional[str] = None
     type: Optional[str] = None
     keyword: Optional[str] = None
+    search_interval: Optional[int] = None
     media_source: Optional[MediaSource] = None
     media_id: Optional[str] = None
     music_type: Optional[str] = None

app/application/subscription/execution.py

@@ -119,7 +119,8 @@ def raise_subscription_site_budget_deferral(
         return
     retry_at = min(deferrals, key=lambda item: item.retry_at).retry_at
     site_ids = tuple(dict.fromkeys(item.site_id for item in deferrals))
-    raise SubscriptionSearchDeferred(retry_at=retry_at, site_ids=site_ids)
+    wait_reason = "cooldown" if all(item.wait_reason == "cooldown" for item in deferrals) else "busy"
+    raise SubscriptionSearchDeferred(retry_at=retry_at, site_ids=site_ids, wait_reason=wait_reason)
 
 
 def handle_subscription_search_deferred(
@@ -135,7 +136,8 @@ def handle_subscription_search_deferred(
         lease_token=lease_token,
         available_at=deferred.retry_at,
         phase="waiting_site_budget",
-        message="站点暂时忙，系统会自动继续搜索",
+        message=str(deferred),
+        pending_site_ids=deferred.site_ids,
     )
     if requeued:
         record("requeued", "site_budget_deferred")
@@ -184,6 +186,7 @@ class SearchTaskSnapshot:
     finished_at: Optional[str] = None
     last_error: Optional[str] = None
     current_site_id: Optional[int] = None
+    pending_site_ids: Optional[tuple[int, ...]] = None
 
 
 @dataclass(frozen=True, slots=True)
@@ -265,8 +268,9 @@ class SubscriptionSearchRepository(Protocol):
         available_at: str,
         phase: str = "waiting_site_budget",
         message: Optional[str] = None,
+        pending_site_ids: Optional[tuple[int, ...]] = None,
     ) -> bool:
-        """把临时不可执行任务退回队列，并保留用户可理解的等待原因。"""
+        """延后任务并保存等待原因与待搜站点；省略站点时保留已有游标。"""
         ...
 
     def is_cancel_requested(self, task_id: str) -> bool:

app/application/subscription/query.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from collections.abc import Mapping
+from datetime import datetime, timedelta
 from typing import Optional
 
 from app.application.subscription.contract import (
@@ -253,3 +254,25 @@ class SubscriptionQueryService:
         return any(
             subscribe.type == MediaType.MUSIC.value for subscribe in self._repository.list(searchable_states) or []
         )
+
+
+def subscription_search_due(
+    subscribe: SubscriptionSnapshot,
+    default_interval: int,
+    now: datetime,
+) -> bool:
+    """按最近一次主动搜索计算定时搜索到期；旧记录以本地创建时间起算。"""
+    if subscribe.state not in {"R", "P"}:
+        return False
+    previous = subscribe.last_search or subscribe.date
+    if not previous:
+        return True
+    try:
+        started = datetime.fromisoformat(previous)
+    except ValueError:
+        return True
+    # 旧创建时间没有时区，保持其本地时间语义；新搜索时间始终携带 UTC 时区。
+    if started.tzinfo is None:
+        started = started.astimezone()
+    interval = subscribe.search_interval or default_interval
+    return now >= started + timedelta(hours=interval)

app/application/subscription/sitebudget.py

@@ -21,14 +21,17 @@ class SubscriptionSiteBudgetDeferral:
 
     site_id: int
     retry_at: str
+    wait_reason: Optional[str] = None
 
 
 class SubscriptionSearchDeferred(RuntimeError):
     """表示订阅搜索未失败，而是应在站点预算可用后重新入队。"""
 
-    def __init__(self, *, retry_at: str, site_ids: tuple[int, ...]) -> None:
+    def __init__(
+        self, *, retry_at: str, site_ids: tuple[int, ...], wait_reason: Optional[str] = None,
+    ) -> None:
         """保存队列恢复所需的时间和站点，避免把临时等待写成错误。"""
-        super().__init__("站点暂时忙，系统会自动继续搜索")
+        super().__init__("站点冷却中" if wait_reason == "cooldown" else "等待站点")
         self.retry_at = retry_at
         self.site_ids = site_ids
 
@@ -176,8 +179,10 @@ class SubscriptionSiteBudget:
         clock: Callable[[], datetime] = _utc_now,
         phase_changed: Optional[Callable[[str, Optional[int]], None]] = None,
         metrics: Optional[SubscriptionSiteBudgetMetrics] = None,
+        pending_site_ids: Optional[tuple[int, ...]] = None,
     ) -> None:
-        """保存持久化端口及可注入的时钟和阶段回调。"""
+        """保存站点预算及恢复范围；首次搜索的 pending_site_ids 为 None。"""
+        self.pending_site_ids = pending_site_ids
         self._repository = repository
         self._owner = owner
         self._cancelled = cancelled

app/chain/search/provider.py

@@ -262,6 +262,7 @@ class _SearchProviderSyncOwner(_SearchOwnerBase):
                 SubscriptionSiteBudgetDeferral(
                     site_id=error.site_id,
                     retry_at=error.retry_at,
+                    wait_reason=error.wait_reason,
                 )
             )
             logger.debug(str(error))
@@ -429,11 +430,16 @@ class _SearchProviderSyncOwner(_SearchOwnerBase):
         area: Optional[str] = "title",
         mtype: Optional[MediaType] = None,
     ) -> Optional[List[TorrentInfo]]:
-        """通过共享线程 owner 按站点顺序翻页并汇总同步 provider 结果。"""
+        """共享线程按站点翻页；恢复搜索仅查询待完成且仍启用的站点。"""
         indexer_sites = self._sync_indexers(sites)
         media_type = self._torrent_type(mediainfo, mtype)
         search_keyword = self._torrent_keyword(keyword, mediainfo, area)
-        plugin_results = self.search_plugin_torrents(
+        budget = getattr(self, "_subscription_site_budget", None)
+        pending_site_ids = budget.pending_site_ids if isinstance(budget, SubscriptionSiteBudget) else None
+        # 已完成的站点和插件源不再重复查询；仍按当前启用配置过滤被移除的站点。
+        if pending_site_ids is not None:
+            indexer_sites = [site for site in indexer_sites if site.get("id") in pending_site_ids]
+        plugin_results = [] if pending_site_ids is not None else self.search_plugin_torrents(
             keyword=search_keyword,
             mtype=media_type,
             page=page,

app/chain/subscribe/search.py

@@ -1,5 +1,6 @@
 """订阅主动搜索编排"""
 
+import asyncio
 import random
 import time
 from datetime import datetime, timedelta, timezone
@@ -28,7 +29,7 @@ from app.application.subscription.observability import (
     batch_progress_text,
     inline_search_result,
 )
-from app.application.subscription.query import SubscriptionQueryService
+from app.application.subscription.query import SubscriptionQueryService, subscription_search_due
 from app.application.subscription.sitebudget import (
     SubscriptionSearchCancelled,
     SubscriptionSearchDeferred,
@@ -46,8 +47,10 @@ from app.domain.context import (
     MusicInfo,
 )
 from app.domain.meta.metabase import MetaBase
+from app.runtime.execution import await_task_to_terminal
 from app.runtime.log import logger
 from app.runtime.stop import runtime_stop_state
+from app.runtime.tasks import get_task_registry
 from app.schemas.types import (
     MediaType,
     SystemConfigKey,
@@ -92,19 +95,14 @@ def _search_task_available_at(
     *,
     now: Optional[datetime] = None,
 ) -> dict[int, str]:
-    """把兜底搜索的随机节奏持久化为逐订阅到期时间。"""
+    """整批仅抖动一次；请求节奏由站点限流控制，不随订阅数量累计空等。"""
     ordered_ids = tuple(dict.fromkeys(subscription_ids))
     if not ordered_ids:
         return {}
     cursor = now or datetime.now(timezone.utc)
     if source == "fallback":
         cursor += timedelta(seconds=random.randint(0, 60))
-    available_at: dict[int, str] = {}
+    return dict.fromkeys(ordered_ids, cursor.isoformat(timespec="seconds"))
-    for position, subscription_id in enumerate(ordered_ids):
-        if source == "fallback" and position:
-            cursor += timedelta(seconds=random.randint(60, 300))
-        available_at[subscription_id] = cursor.isoformat(timespec="seconds")
-    return available_at
 
 
 class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase):
@@ -160,11 +158,12 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase):
         manual: Optional[bool] = False,
         progress_callback: Optional[Callable[..., None]] = None,
         sids: Optional[tuple[int, ...]] = None,
+        scheduled_interval: Optional[int] = None,
     ) -> Optional[str]:
         """
         执行订阅搜索。
 
-        保持定时任务、API 和插件使用的公开签名，搜索实现委托给内部执行阶段。
+        scheduled_interval 仅供定时调度传入系统间隔；手动和指定目标搜索不受周期限制。
         """
         return self._execute_search(
             sid=sid,
@@ -172,6 +171,7 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase):
             state=state,
             manual=manual,
             progress_callback=progress_callback,
+            scheduled_interval=scheduled_interval,
         )
 
     def _execute_search(
@@ -181,6 +181,7 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase):
         manual: Optional[bool] = False,
         progress_callback: Optional[Callable[..., None]] = None,
         sids: Optional[tuple[int, ...]] = None,
+        scheduled_interval: Optional[int] = None,
     ) -> Optional[str]:
         """
         订阅搜索
@@ -202,6 +203,7 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase):
                 state=state,
                 manual=manual,
                 progress_callback=progress_callback,
+                scheduled_interval=scheduled_interval,
             )
         self._execute_inline_search(
             sid=sid,
@@ -209,6 +211,7 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase):
             state=state,
             manual=manual,
             progress_callback=progress_callback,
+            scheduled_interval=scheduled_interval,
         )
         return None
 
@@ -219,6 +222,7 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase):
         state: Optional[str],
         manual: Optional[bool],
         progress_callback: Optional[Callable[..., None]],
+        scheduled_interval: Optional[int] = None,
     ) -> None:
         """在独立 Search 通道内按订阅准入执行兼容搜索。"""
         lock_acquired = self._acquire_run_lock("search", progress_callback)
@@ -228,7 +232,10 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase):
         processed = []
         summary: Optional[SearchExecutionSummary] = None
         try:
-            subscribes = self._load_search_subscriptions(sid=sid, sids=sids, state=state)
+            subscribes = self._load_search_subscriptions(
+                sid=sid, sids=sids, state=state,
+                scheduled_interval=None if manual else scheduled_interval,
+            )
             total = len(subscribes)
             source, _priority = _search_source_and_priority(
                 sid=sid,
@@ -352,9 +359,17 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase):
         state: Optional[str],
         manual: Optional[bool],
         progress_callback: Optional[Callable[..., None]],
-    ) -> str:
+        scheduled_interval: Optional[int] = None,
+    ) -> Optional[str]:
         """将搜索转为持久任务并在无 Match 长锁的短租约中串行消费。"""
-        subscribes = self._load_search_subscriptions(sid=sid, sids=sids, state=state)
+        subscribes = self._load_search_subscriptions(
+            sid=sid, sids=sids, state=state,
+            scheduled_interval=None if manual else scheduled_interval,
+        )
+        if scheduled_interval is not None and not subscribes:
+            if progress_callback:
+                progress_callback(value=100, text="暂无到期订阅，无需搜索")
+            return None
         source, priority = _search_source_and_priority(
             sid=sid,
             sids=sids,
@@ -501,6 +516,27 @@ class _SubscribeSearchQueueOwner(_SubscribeSearchQueueCoordinator):
         )
         return runner.execute()
 
+    async def async_resume_search_queue(
+        self,
+        progress_callback: Optional[Callable[..., None]] = None,
+        limit: int = 50,
+        manual_sids: Optional[tuple[int, ...]] = None,
+    ) -> None:
+        """托管恢复协程及手工反馈，等待同步消费者真实结束后才释放运行所有权。"""
+        task = get_task_registry().create_sync(
+            self.resume_search_queue,
+            owner="subscription.search_queue",
+            progress_callback=progress_callback,
+            limit=limit,
+            manual_sids=manual_sids,
+        )
+        try:
+            await asyncio.shield(task)
+        except asyncio.CancelledError:
+            # 同步线程不可强制取消；真实收尾前不能让下一次轮询取得消费者所有权。
+            await await_task_to_terminal(task)
+            raise
+
     def resume_search_queue(
         self,
         progress_callback: Optional[Callable[..., None]] = None,
@@ -573,6 +609,7 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner):
         sid: Optional[int],
         sids: Optional[tuple[int, ...]],
         state: Optional[str],
+        scheduled_interval: Optional[int] = None,
     ) -> list[SubscriptionSnapshot]:
         """按单条、指定批次或状态读取本轮搜索订阅。"""
         repository = self.subscription_repository
@@ -581,7 +618,11 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner):
             return [subscribe] if subscribe else []
         if sids is not None:
             return [item for current_id in sids if (item := repository.get(current_id)) is not None]
-        return cast(list[SubscriptionSnapshot], repository.list(self.get_states_for_search(state or "N")))
+        subscribes = cast(list[SubscriptionSnapshot], repository.list(self.get_states_for_search(state or "N")))
+        if scheduled_interval is None:
+            return subscribes
+        now = datetime.now(timezone.utc)
+        return [item for item in subscribes if subscription_search_due(item, scheduled_interval, now)]
 
     @staticmethod
     def _recent_subscription_retry_at(
@@ -682,6 +723,11 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner):
     ) -> Optional[SubscriptionSnapshot]:
         """处理单个订阅，并返回下载后重新读取的状态快照。"""
         _ensure_execution_active(execution_context)
+        subscribe = self._SubscribeChain__apply_subscribe_update(
+            subscribe,
+            {"last_search": datetime.now(timezone.utc).isoformat(timespec="seconds")},
+            scene="search",
+        )
         logger.debug(f"开始搜索订阅，标题：{subscribe.name} ...")
         target = prepare_search_target(
             self, subscribe, MediaChain(), partial(_ensure_execution_active, execution_context),

app/chain/subscribe/searchtask.py

@@ -209,6 +209,7 @@ class SubscriptionSearchTaskRunner:
                     stop_state=self.stop_state,
                     phase_changed=phase_changed,
                     metrics=self.summary.site_metrics,
+                    pending_site_ids=self.task.pending_site_ids,
                 )
             )
             current = self.process_subscription(

app/db/adapters/subscription.py

@@ -77,6 +77,7 @@ def _project_subscription(record: Subscribe) -> SubscriptionSnapshot:
         year=record.year,
         type=record.type,
         keyword=record.keyword,
+        search_interval=record.search_interval,
         media_source=_media_source(record.media_source),
         media_id=record.media_id,
         music_type=record.music_type,
@@ -103,6 +104,7 @@ def _project_subscription(record: Subscribe) -> SubscriptionSnapshot:
         note=cast(Optional[builtins.list[int]], record.note),
         state=record.state,
         last_update=record.last_update,
+        last_search=record.last_search,
         date=record.date,
         username=record.username,
         sites=cast(Optional[builtins.list[int]], record.sites),
@@ -134,6 +136,7 @@ def _project_history(record: SubscribeHistory) -> SubscriptionHistorySnapshot:
         year=record.year,
         type=record.type,
         keyword=record.keyword,
+        search_interval=record.search_interval,
         media_source=_media_source(record.media_source),
         media_id=record.media_id,
         music_type=record.music_type,

app/db/adapters/subscriptionsearch.py

@@ -67,6 +67,7 @@ def _task(record: SubscriptionSearchTask) -> SearchTaskSnapshot:
         finished_at=record.finished_at,
         last_error=record.last_error,
         current_site_id=record.current_site_id,
+        pending_site_ids=tuple(record.pending_site_ids) if record.pending_site_ids is not None else None,
     )
 
 
@@ -248,8 +249,9 @@ class TransactionalSubscriptionSearchRepository:
         available_at: str,
         phase: str = "waiting_site_budget",
         message: Optional[str] = None,
+        pending_site_ids: Optional[tuple[int, ...]] = None,
     ) -> bool:
-        """按指定时间和可见原因重新排队任务。"""
+        """重新排队并保存待搜站点；未提供站点时保留原游标。"""
         return self._write(
             lambda repository: repository.defer_task(
                 task_id=task_id,
@@ -257,6 +259,7 @@ class TransactionalSubscriptionSearchRepository:
                 available_at=available_at,
                 phase=phase,
                 message=message,
+                pending_site_ids=pending_site_ids,
             )
         )
 

app/db/models/subscribe.py

@@ -22,6 +22,8 @@ class Subscribe(Base):
     year: Mapped[Optional[str]] = mapped_column(String)
     # 类型
     type: Mapped[Optional[str]] = mapped_column(String)
+    # 自定义定时搜索间隔（小时）；空值跟随系统
+    search_interval: Mapped[Optional[int]] = mapped_column(Integer)
     # 搜索关键字
     keyword: Mapped[Optional[str]] = mapped_column(String)
     media_source: Mapped[Optional[str]] = mapped_column(String, index=True)
@@ -72,6 +74,8 @@ class Subscribe(Base):
     note: Mapped[Optional[Any]] = mapped_column(JSON)
     # 状态：N-新建 R-订阅中 P-待定 S-暂停
     state: Mapped[str] = mapped_column(String, nullable=False, index=True, default="N")
+    # 最近一次主动搜索开始时间，使用带时区的 UTC 时间
+    last_search: Mapped[Optional[str]] = mapped_column(String)
     # 最后更新时间
     last_update: Mapped[Optional[str]] = mapped_column(String)
     # 创建时间

app/db/models/subscribehistory.py

@@ -20,6 +20,8 @@ class SubscribeHistory(Base):
     year: Mapped[Optional[str]] = mapped_column(String)
     # 类型
     type: Mapped[Optional[str]] = mapped_column(String)
+    # 自定义定时搜索间隔（小时）；空值跟随系统
+    search_interval: Mapped[Optional[int]] = mapped_column(Integer)
     # 搜索关键字
     keyword: Mapped[Optional[str]] = mapped_column(String)
     media_source: Mapped[Optional[str]] = mapped_column(String, index=True)

app/db/models/subscriptionsearch.py

@@ -2,7 +2,7 @@
 
 from typing import Optional
 
-from sqlalchemy import Index, Integer, String, Text, UniqueConstraint
+from sqlalchemy import JSON, Index, Integer, String, Text, UniqueConstraint
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.db.base import Base, get_id_column
@@ -48,6 +48,8 @@ class SubscriptionSearchTask(Base):
     state: Mapped[str] = mapped_column(String(32), nullable=False, default="queued")
     phase: Mapped[str] = mapped_column(String(32), nullable=False, default="queued")
     current_site_id: Mapped[Optional[int]] = mapped_column(Integer)
+    # None 表示首次完整搜索；重试只保留尚未完成的站点，跨重启不重复请求成功站点。
+    pending_site_ids: Mapped[Optional[list[int]]] = mapped_column(JSON(none_as_null=True))
     attempt_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
     cancel_requested: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
     lease_owner: Mapped[Optional[str]] = mapped_column(String(128))

app/db/oper/subscriptionsearch.py

@@ -108,6 +108,11 @@ class SubscriptionSearchOper(DbOper):
                             (promote_queued_task, None),
                             else_=SubscriptionSearchTask.last_error,
                         ),
+                        # 用户重新指定搜索时按当前站点配置重搜；自动周期合并保留恢复游标。
+                        pending_site_ids=case(
+                            (and_(SubscriptionSearchTask.state == "queued", source in {"manual", "targeted"}), None),
+                            else_=SubscriptionSearchTask.pending_site_ids,
+                        ),
                         available_at=case(
                             (
                                 or_(
@@ -138,7 +143,7 @@ class SubscriptionSearchOper(DbOper):
         return batch, created, coalesced, tuple(dict.fromkeys(active_batch_ids))
 
     def claim_next(self, *, owner: str, lease_seconds: int) -> Optional[SubscriptionSearchTask]:
-        """使用 CAS 认领最高优先级任务，过期 running 任务可被恢复。"""
+        """CAS 认领任务并清除旧等待提示，过期 running 任务保留站点游标恢复。"""
         if not isinstance(self._db, Session):
             raise RuntimeError("订阅搜索认领需要调用方提供同步 Session")
         now = utc_now_text()
@@ -209,6 +214,7 @@ class SubscriptionSearchOper(DbOper):
                 .values(
                     state="running",
                     phase="matching",
+                    last_error=None,
                     current_site_id=None,
                     lease_owner=owner,
                     lease_token=lease_token,
@@ -403,8 +409,9 @@ class SubscriptionSearchOper(DbOper):
         available_at: str,
         phase: str,
         message: Optional[str],
+        pending_site_ids: Optional[tuple[int, ...]] = None,
     ) -> bool:
-        """释放当前租约并在指定时间后按可见原因恢复同一任务。"""
+        """释放租约并保存重试站点；普通准入等待保留已有站点游标。"""
         if not isinstance(self._db, Session):
             raise RuntimeError("订阅搜索延后需要调用方提供同步 Session")
         task = self._db.execute(
@@ -443,6 +450,7 @@ class SubscriptionSearchOper(DbOper):
                 updated_at=now,
                 finished_at=None,
                 last_error=message,
+                pending_site_ids=list(pending_site_ids) if pending_site_ids is not None else task.pending_site_ids,
             ),
             execution_options={"synchronize_session": False},
         )

app/domain/classification/sources.py

@@ -18,15 +18,8 @@ FIXTURE_CLASSIFICATION_SOURCES: Final[tuple[str, ...]] = (
 )
 """首版分类体系具备真实标准投影 fixture 的内置来源顺序。"""
 
-BUILTIN_CLASSIFICATION_SOURCES: Final[tuple[str, ...]] = (
+BUILTIN_CLASSIFICATION_SOURCES: Final[tuple[str, ...]] = FIXTURE_CLASSIFICATION_SOURCES
-    *FIXTURE_CLASSIFICATION_SOURCES,
+"""仅保留宿主模块实现的来源；插件来源由启用插件声明，不预占标识。"""
-    MediaSource.Bilibili.value,
-    MediaSource.MangoTV.value,
-    MediaSource.MiguVideo.value,
-    MediaSource.TencentVideo.value,
-    MediaSource.Iqiyi.value,
-)
-"""媒体来源 API 当前暴露的全部内置来源顺序。"""
 
 STANDARD_CLASSIFICATION_FIELD_IDS: Final[tuple[str, ...]] = (
     "identity.media_source",
@@ -155,11 +148,6 @@ _SOURCE_FIELD_SUPPORT: Final[dict[str, dict[str, ClassificationSourceSupport]]]
         "music.genres": _PARTIAL,
         "music.tags": _PARTIAL,
     },
-    MediaSource.Bilibili.value: {},
-    MediaSource.MangoTV.value: {},
-    MediaSource.MiguVideo.value: {},
-    MediaSource.TencentVideo.value: {},
-    MediaSource.Iqiyi.value: {},
 }
 
 

app/locales/en-US.json

@@ -103,6 +103,7 @@
     }
   },
   "messages": {
+    "暂无到期订阅，无需搜索": "No subscriptions are due for search",
     "调用工具失败": "Tool call failed",
     "无效的媒体来源": "Invalid media source",
     "该媒体来源不支持此音乐接口": "This media source is not supported by this music endpoint",

app/locales/zh-TW.json

@@ -103,6 +103,7 @@
     }
   },
   "messages": {
+    "暂无到期订阅，无需搜索": "暫無到期訂閱，無需搜尋",
     "调用工具失败": "調用工具失敗",
     "媒体来源和媒体 ID 必须同时提供": "媒體來源和媒體 ID 必須同時提供",
     "media_source 和 media_id 必须同时提供": "media_source 和 media_id 必須同時提供",

app/scheduler/catalog.py

@@ -36,11 +36,14 @@ class _MediaServerSchedule(TypedDict):
     interval: int
 
 
-def _subscription_search_job_specs(services: SchedulerServices) -> tuple[JobSpec, ...]:
+def _subscription_search_job_specs(
+    services: SchedulerServices, search_interval: int = 24,
+) -> tuple[JobSpec, ...]:
     """构造订阅搜索、新增搜索与持久队列恢复任务目录。"""
     return (
         JobSpec(
-            "subscribe_search", "订阅搜索补全", services.search_subscribe, "subscription", kwargs={"state": "R"}
+            "subscribe_search", "订阅搜索补全", services.search_subscribe, "subscription",
+            kwargs={"state": "R", "scheduled_interval": search_interval}
         ),
         JobSpec(
             "new_subscribe_search",
@@ -147,16 +150,20 @@ class SchedulerCatalogOwner(_SchedulerOwnerBase):
             replace_existing=True,
         )
 
+    def _poll_subscription_search_queue(self) -> None:
+        """仅在消费者空闲时唤醒托管协程，轮询不等待搜索也不重复报告重入。"""
+        if not self._is_job_active("subscribe_search_queue"):
+            self.start("subscribe_search_queue")
+
     def _register_subscription_search_queue_job(self, config: SchedulerRuntimeConfig) -> None:
-        """注册短周期持久搜索队列恢复任务。"""
+        """注册轻量轮询；搜索协程的真实生命周期由 Scheduler 持有。"""
         self._scheduler.add_job(
-            self.start,
+            self._poll_subscription_search_queue,
             "interval",
             id="subscribe_search_queue",
             name="恢复订阅搜索队列",
             seconds=10,
             next_run_time=datetime.now(pytz.timezone(config.timezone)) + timedelta(seconds=5),
-            kwargs={"job_id": "subscribe_search_queue"},
         )
 
     def _initialize_catalog(self, config: SchedulerRuntimeConfig) -> None:
@@ -168,7 +175,7 @@ class SchedulerCatalogOwner(_SchedulerOwnerBase):
                 JobSpec("cookiecloud", "同步CookieCloud站点", services.sync_cookies, "site"),
                 JobSpec("mediaserver_sync", "同步媒体服务器", services.sync_mediaserver, "mediaserver"),
                 JobSpec("subscribe_tmdb", "订阅元数据更新", services.check_subscribe, "subscription"),
-                *_subscription_search_job_specs(services),
+                *_subscription_search_job_specs(services, config.subscribe_search_interval),
                 JobSpec("subscribe_refresh", "订阅刷新", services.refresh_subscribe, "subscription"),
                 JobSpec("subscribe_follow", "关注的订阅分享", services.follow_subscribe, "subscription"),
                 JobSpec(
@@ -298,14 +305,14 @@ class SchedulerCatalogOwner(_SchedulerOwnerBase):
             kwargs={"job_id": "subscribe_tmdb"},
         )
 
-        # 订阅状态每隔24小时搜索一次
+        # 每五分钟检查逐条订阅到期时间；实际搜索仍受系统或独立周期约束。
         if config.subscribe_search:
             self._scheduler.add_job(
                 self.start,
                 "interval",
                 id="subscribe_search",
                 name="订阅搜索补全",
-                hours=config.subscribe_search_interval,
+                minutes=5,
                 kwargs={"job_id": "subscribe_search"},
             )
 

app/schemas/event.py

@@ -900,22 +900,9 @@ class DiscoverMediaSource(BaseModel):
         if media_source and not mediaid_prefix:
             normalized["mediaid_prefix"] = str(media_source)
         elif mediaid_prefix and not media_source:
-            normalized["media_source"] = cls._media_source_from_prefix(
+            normalized["media_source"] = MediaSource(str(mediaid_prefix))
-                str(mediaid_prefix)
-            )
         return normalized
 
-    @staticmethod
-    def _media_source_from_prefix(mediaid_prefix: str) -> MediaSource:
-        """将旧插件前缀映射为内置或插件扩展媒体来源。"""
-        aliases = {
-            "mangguo": MediaSource.MangoTV,
-            "tencentvideo": MediaSource.TencentVideo,
-        }
-        if mediaid_prefix in aliases:
-            return aliases[mediaid_prefix]
-        return MediaSource(mediaid_prefix)
-
 
 class MediaSourceInfo(BaseModel):
     """

app/schemas/media.py

@@ -28,15 +28,6 @@ MEDIA_SOURCE_ALIASES = {
     "audio_db": MediaSource.TheAudioDB,
     "doubanmusic": MediaSource.DoubanMusic,
     "douban_music": MediaSource.DoubanMusic,
-    "bilibili": MediaSource.Bilibili,
-    "mangguodiscover": MediaSource.MangoTV,
-    "mango_tv": MediaSource.MangoTV,
-    "migu": MediaSource.MiguVideo,
-    "migu_video": MediaSource.MiguVideo,
-    "tencentvideodiscover": MediaSource.TencentVideo,
-    "tencent_video": MediaSource.TencentVideo,
-    "iqiyi": MediaSource.Iqiyi,
-    "iqiyidiscover": MediaSource.Iqiyi,
 }
 
 MEDIA_SOURCE_PREFIXES = {
@@ -49,11 +40,6 @@ MEDIA_SOURCE_PREFIXES = {
     MediaSource.MusicBrainz: "musicbrainz",
     MediaSource.TheAudioDB: "theaudiodb",
     MediaSource.DoubanMusic: "doubanmusic",
-    MediaSource.Bilibili: "bilibili",
-    MediaSource.MangoTV: "mangguodiscover",
-    MediaSource.MiguVideo: "migu",
-    MediaSource.TencentVideo: "tencentvideodiscover",
-    MediaSource.Iqiyi: "iqiyidiscover",
 }
 
 

app/schemas/subscribe.py

@@ -111,7 +111,7 @@ class Subscribe(OptionalMediaIdentityMixin, BaseModel):
         "note", "state", "last_update", "username", "current_priority", "episode_priority", "date",
         "current_audio_format", "current_bitrate", "current_bit_depth", "current_sample_rate",
         "classification_rule_id", "classification_policy_revision", "classification_source",
-        "execution_status",
+        "execution_status", "last_search",
     })
 
     id: Optional[int] = None
@@ -121,6 +121,10 @@ class Subscribe(OptionalMediaIdentityMixin, BaseModel):
     year: Optional[str] = None
     # 订阅类型 电影/电视剧
     type: Optional[str] = None
+    # 定时搜索间隔（小时）；空值跟随系统设置
+    search_interval: Optional[int] = Field(default=None, ge=1, le=8760)
+    # 最近一次主动搜索开始时间，仅供读取
+    last_search: Optional[str] = None
     # 搜索关键字
     keyword: Optional[str] = None
     media_source: Optional[MediaSource] = None
@@ -262,7 +266,7 @@ class Subscribe(OptionalMediaIdentityMixin, BaseModel):
             data = dict(data)
             for key, value in list(data.items()):
                 if isinstance(value, str) and value == "":
-                    if key in {"media_source", "media_id"} or key in cls.CLEARABLE_FILTER_FIELDS:
+                    if key in {"media_source", "media_id", "search_interval"} or key in cls.CLEARABLE_FILTER_FIELDS:
                         data[key] = None
                     else:
                         data.pop(key)

app/schemas/types.py

@@ -85,10 +85,6 @@ _MEDIA_SOURCE_VALUE_ALIASES = {
     "tmdb": "themoviedb",
     "audio_db": "theaudiodb",
     "douban_music": "doubanmusic",
-    "mango_tv": "mangguodiscover",
-    "migu_video": "migu",
-    "tencent_video": "tencentvideodiscover",
-    "iqiyi": "iqiyidiscover",
 }
 
 
@@ -104,11 +100,6 @@ class MediaSource(str, Enum):
     MusicBrainz = "musicbrainz"
     TheAudioDB = "theaudiodb"
     DoubanMusic = "doubanmusic"
-    Bilibili = "bilibili"
-    MangoTV = "mangguodiscover"
-    MiguVideo = "migu"
-    TencentVideo = "tencentvideodiscover"
-    Iqiyi = "iqiyidiscover"
 
     def __str__(self) -> str:
         """返回可直接用于 API 和数据库的规范值。"""

app/startup/initializers/scheduler.py

@@ -61,7 +61,7 @@ def configure_scheduler_services() -> None:
             sync_mediaserver=mediaserver_chain.sync,
             check_subscribe=subscribe_chain.check_and_reconcile,
             search_subscribe=subscribe_chain.search,
-            resume_subscribe_search=subscribe_chain.resume_search_queue,
+            resume_subscribe_search=subscribe_chain.async_resume_search_queue,
             refresh_subscribe=subscribe_chain.refresh,
             follow_subscribe=subscribe_chain.follow,
             process_transfer=transfer_chain.process,

database/versions/a9c3e7f1b5d8_3_0_31.py

@@ -0,0 +1,48 @@
+"""3.0.31 保存订阅搜索重试站点并解除存量批次的累计空等。"""
+
+# Alembic 的 op 是运行期代理。
+# pylint: disable=no-member
+
+from datetime import datetime, timezone
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "a9c3e7f1b5d8"
+down_revision = "f8a2c6e9b1d4"
+branch_labels = None
+depends_on = None
+
+_TABLE = "subscriptionsearchtask"
+
+
+def upgrade() -> None:
+    """增加可恢复站点游标，只提前尚未执行的自动任务，不绕过冷却和用户停止。"""
+    inspector = sa.inspect(op.get_bind())
+    if _TABLE not in inspector.get_table_names():
+        return
+    columns = {column["name"] for column in inspector.get_columns(_TABLE)}
+    if "pending_site_ids" in columns:
+        return
+    op.add_column(_TABLE, sa.Column("pending_site_ids", sa.JSON(none_as_null=True), nullable=True))
+    tasks = sa.table(
+        _TABLE,
+        sa.column("source"), sa.column("state"), sa.column("phase"),
+        sa.column("attempt_count"), sa.column("cancel_requested"), sa.column("available_at"),
+    )
+    now = datetime.now(timezone.utc).isoformat(timespec="seconds")
+    op.execute(
+        tasks.update().where(
+            tasks.c.source == "fallback", tasks.c.state == "queued", tasks.c.phase == "queued",
+            tasks.c.attempt_count == 0, tasks.c.cancel_requested == 0, tasks.c.available_at > now,
+        ).values(available_at=now)
+    )
+
+
+def downgrade() -> None:
+    """移除站点游标；恢复旧版本后未完成任务仍可完整重试。"""
+    inspector = sa.inspect(op.get_bind())
+    if _TABLE in inspector.get_table_names():
+        columns = {column["name"] for column in inspector.get_columns(_TABLE)}
+        if "pending_site_ids" in columns:
+            op.drop_column(_TABLE, "pending_site_ids")

database/versions/f8a2c6e9b1d4_3_0_30.py

@@ -0,0 +1,27 @@
+"""3.0.30 增加逐订阅搜索周期和持久搜索时间。"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "f8a2c6e9b1d4"
+down_revision = "e7f3a9c1d5b2"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    """兼容首次建表和已有数据库，旧订阅默认跟随系统周期。"""
+    inspector = sa.inspect(op.get_bind())
+    for table in ("subscribe", "subscribehistory"):
+        columns = {column["name"] for column in inspector.get_columns(table)}
+        if "search_interval" not in columns:
+            op.add_column(table, sa.Column("search_interval", sa.Integer(), nullable=True))
+        if table == "subscribe" and "last_search" not in columns:
+            op.add_column(table, sa.Column("last_search", sa.String(), nullable=True))
+
+
+def downgrade() -> None:
+    """移除新增周期字段，保留原有订阅和历史数据。"""
+    op.drop_column("subscribe", "last_search")
+    for table in ("subscribe", "subscribehistory"):
+        op.drop_column(table, "search_interval")

docker/Dockerfile

@@ -304,6 +304,8 @@ COPY --from=uv /uv /usr/local/bin/uv
 
 # 浏览器运行依赖
 RUN playwright install-deps chromium \
+    && apt-get update \
+    && apt-get install -y --no-install-recommends libde265-0 \
     && apt-get autoremove -y \
     && apt-get clean \
     && rm -rf \

docs/architecture-overview.md

@@ -755,7 +755,7 @@ flowchart LR
 | 指标 | 当前值 |
 |---|---:|
 | Python 模块 | 983 |
-| 内部导入边 | 8,338 |
+| 内部导入边 | 8,340 |
 | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） |
 | Application / Chain 具体 Adapter 直连 | 0 / 0 |
 | Direct egress | 53（债务已清零，53 条精确 containment） |

docs/architecture/optimization-checklist.md

@@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
 
 | 指标 | 当前值 | 解释 |
 |---|---:|---|
-| 宿主 Python 模块 / 内部依赖边 | 983 / 8,338 | `dependency-baseline.json` 当前快照；分类离线词表与下载资源分类新增模块及其受控依赖 |
+| 宿主 Python 模块 / 内部依赖边 | 983 / 8,340 | `dependency-baseline.json` 当前快照；分类离线词表、下载资源分类与订阅搜索运行时任务新增模块及其受控依赖 |
 | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 |
 | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 |
 | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |

docs/mcp-api.md

@@ -265,7 +265,9 @@ FastAPI 的 HTTP 异常和参数校验异常统一使用 `message`，不再返
 
 #### 媒体识别 / 整理
 
-媒体识别、搜索和手动整理统一使用 `media_source` + `media_id` 表示媒体主身份。内置来源通过 `MediaSource` 提供 `themoviedb`、`douban`、`bangumi`、`anilist`、`imdb`、`tvdb`、`musicbrainz`、`theaudiodb`、`doubanmusic`、`bilibili`、`mangguodiscover`、`migu` 和 `tencentvideodiscover` 等常量；该列表不是插件来源白名单，插件可以注册符合 OpenAPI 格式约束的稳定扩展标识。`media_id` 是该来源的原生 ID，不添加 `tmdb:` 等前缀。需要精确身份时两个字段必须同时提供，不能只传其中一个。
+媒体识别、搜索和手动整理统一使用 `media_source` + `media_id` 表示媒体主身份。内置来源通过 `MediaSource` 提供 `themoviedb`、`douban`、`bangumi`、`anilist`、`imdb`、`tvdb`、`musicbrainz`、`theaudiodb`、`doubanmusic` 九个具有宿主模块实现的常量；该列表不是插件来源白名单，插件可以注册符合 OpenAPI 格式约束的稳定扩展标识。`media_id` 是该来源的原生 ID，不添加 `tmdb:` 等前缀。需要精确身份时两个字段必须同时提供，不能只传其中一个。
+
+媒体来源列表 `/api/v1/media/source` 仅预置上述九个来源，其余来源由启用插件注册后提供。哔哩哔哩、芒果 TV、咪咕视频、腾讯视频、爱奇艺不再占用内置来源标识，宿主也不再转换这些插件来源的旧别名；调用方应使用插件声明的准确来源 ID。
 
 影视自动识别在未指定来源时只使用 TMDB，未命中时不会继续查询其它影视源。音乐路径识别严格按 AcoustID 音频指纹、文件标签、文件名三级依次执行；指纹或标签直接提供 MusicBrainz Recording ID 时，会直接查询 MusicBrainz 详情，标签和文件名标题识别也只使用 MusicBrainz。其它元数据源仅在手动操作通过请求级 `media_source`，或通过完整的 `media_source` + `media_id` 精确指定时使用，不修改系统默认值，也不会跨来源兜底。`MediaInfo` 响应仍可能包含 `tmdb_id`、`douban_id`、`bangumi_id`、`anilist_id` 等跨源映射辅助字段，但这些字段不是通用请求入口。明确归属 `/tmdb`、`/douban`、`/bangumi`、`/anilist` 的接口，以及固定使用 TMDB 的剧集组和排期接口，仍可按其单数据源契约接收原生 ID。
 
@@ -429,6 +431,24 @@ TMDB 缓存查询响应的 `data` 包含 `count`、`recognized`、`unrecognized`
 旧版未包含这些证据的派生缓存在升级后重新建立，不影响下载历史或订阅数据。
 名称确认规则更新时同样重建旧派生缓存，避免艺术家前后缀误截断的旧结果继续命中。
 
+### 单条订阅搜索周期
+
+`POST /api/v1/subscribe/` 和 `PUT /api/v1/subscribe/` 支持 `search_interval`：
+取值为 1–8760 的整数小时数，`null` 表示跟随系统的 `SUBSCRIBE_SEARCH_INTERVAL`。
+更新时省略该字段保留原值，显式传入 `null` 恢复系统周期。默认订阅规则同样可保存此字段。
+
+仅在启用 `SUBSCRIBE_SEARCH` 时执行定时搜索，每五分钟检查到期订阅，再通过原有搜索队列和站点限流执行；
+实际开始时间可能因站点忙而延后。`last_search` 为系统维护的 UTC 搜索尝试开始时间，重启后仍有效，公共写接口忽略它。
+旧订阅尚无搜索时间时，以添加时间计算到期。新订阅首次搜索、手动搜索和 RSS 刷新不受周期过滤影响；
+手动主动搜索会更新最近搜索时间。电影、电视剧和音乐均支持独立周期。
+
+自动批次只在启动时随机错峰 0–60 秒，不再按订阅数量累加分钟级等待；同站点访问间隔、唯一在途租约和错误冷却继续生效。
+站点暂不可用时，任务保存未完成站点并按 `next_run_at` 恢复，不重复查询已完成站点或插件源；队列和站点游标可跨重启恢复。
+`waiting_site_budget` 表示可恢复等待，`error` 中的“等待站点”或“站点冷却中”是原因提示，不表示搜索失败；重新执行时清除旧提示。
+卡片应按 `state` / `phase` 展示简短标签，原因放入详情提示。恢复调度每 10 秒检查空闲消费者，长搜索由调度器持续托管。
+实际吞吐仍受站点访问限制和网络响应时间影响，配置周期不是全部站点必须完成的截止时间。
+
+
 ### 插件补充接口
 
 **GET** `/api/v1/plugin/history/{plugin_id}`

skills/database-operation/SKILL.md

@@ -228,13 +228,13 @@ python scripts/mp-db.py write "UPDATE subscribe SET state = 'S' WHERE id = 123"
 - Purpose: Stores active movie, TV, or music subscriptions, filters, progress, and download targets.
 - Useful queries: Inspecting state, missing episodes/tracks, quality rules, site scope, and match progress.
 - Write boundary: Create, update, search, or delete through the subscription API to preserve state-machine consistency.
-- Columns: `id`, `name`, `year`, `type`, `keyword`, `media_source`, `media_id`, `music_type`, `total_tracks`, `season`, `poster`, `backdrop`, `vote`, `description`, `filter`, `include`, `exclude`, `quality`, `resolution`, `effect`, `audio_quality`, `audio_format`, `min_bitrate`, `min_bit_depth`, `min_sample_rate`, `total_episode`, `start_episode`, `lack_episode`, `note`, `state`, `last_update`, `date`, `username`, `sites`, `downloader`, `best_version`, `best_version_full`, `current_priority`, `current_audio_format`, `current_bitrate`, `current_bit_depth`, `current_sample_rate`, `episode_priority`, `save_path`, `search_imdbid`, `manual_total_episode`, `custom_words`, `media_category_id`, `media_category`, `filter_groups`, `episode_group`
+- Columns: `id`, `name`, `year`, `type`, `search_interval`, `keyword`, `media_source`, `media_id`, `music_type`, `total_tracks`, `season`, `poster`, `backdrop`, `vote`, `description`, `filter`, `include`, `exclude`, `quality`, `resolution`, `effect`, `audio_quality`, `audio_format`, `min_bitrate`, `min_bit_depth`, `min_sample_rate`, `total_episode`, `start_episode`, `lack_episode`, `note`, `state`, `last_search`, `last_update`, `date`, `username`, `sites`, `downloader`, `best_version`, `best_version_full`, `current_priority`, `current_audio_format`, `current_bitrate`, `current_bit_depth`, `current_sample_rate`, `episode_priority`, `save_path`, `search_imdbid`, `manual_total_episode`, `custom_words`, `media_category_id`, `media_category`, `filter_groups`, `episode_group`
 
 ### `subscribehistory`
 - Purpose: Stores snapshots of completed or archived subscriptions and their final filter state.
 - Useful queries: Auditing historical subscriptions, media identity, completion criteria, and filter configuration.
 - Write boundary: Generated by subscription completion and archival; restore or delete through its business API.
-- Columns: `id`, `name`, `year`, `type`, `keyword`, `media_source`, `media_id`, `music_type`, `total_tracks`, `season`, `poster`, `backdrop`, `vote`, `description`, `filter`, `include`, `exclude`, `quality`, `resolution`, `effect`, `audio_quality`, `audio_format`, `min_bitrate`, `min_bit_depth`, `min_sample_rate`, `total_episode`, `start_episode`, `date`, `username`, `sites`, `best_version`, `best_version_full`, `current_priority`, `current_audio_format`, `current_bitrate`, `current_bit_depth`, `current_sample_rate`, `episode_priority`, `save_path`, `search_imdbid`, `custom_words`, `media_category_id`, `media_category`, `classification_rule_id`, `classification_policy_revision`, `classification_source`, `filter_groups`, `episode_group`
+- Columns: `id`, `name`, `year`, `type`, `search_interval`, `keyword`, `media_source`, `media_id`, `music_type`, `total_tracks`, `season`, `poster`, `backdrop`, `vote`, `description`, `filter`, `include`, `exclude`, `quality`, `resolution`, `effect`, `audio_quality`, `audio_format`, `min_bitrate`, `min_bit_depth`, `min_sample_rate`, `total_episode`, `start_episode`, `date`, `username`, `sites`, `best_version`, `best_version_full`, `current_priority`, `current_audio_format`, `current_bitrate`, `current_bit_depth`, `current_sample_rate`, `episode_priority`, `save_path`, `search_imdbid`, `custom_words`, `media_category_id`, `media_category`, `classification_rule_id`, `classification_policy_revision`, `classification_source`, `filter_groups`, `episode_group`
 
 ### `subscriptionsearchbatch`
 - Purpose: Stores durable subscription search batches, source, aggregate state, counts, and cancellation requests.

tests/fixtures/architecture/dependency-baseline.json

@@ -1074,8 +1074,8 @@
       "runtime_only": true
     }
   },
-  "edge_count": 8338,
+  "edge_count": 8340,
-  "edge_sha256": "181d760e7da33f12022ca86f906617a11e1d672d95e7e3fd4c3bc324a9d9dd63",
+  "edge_sha256": "b3092e1b2d83dd6b9356e1107e034c1a5b65f2246a5677f7457687a37aeba4cb",
   "edges": [
     "app -> app.foundation",
     "app -> app.foundation.environment",
@@ -4927,8 +4927,10 @@
     "app.chain.subscribe.search -> app.domain.meta",
     "app.chain.subscribe.search -> app.domain.meta.metabase",
     "app.chain.subscribe.search -> app.runtime",
+    "app.chain.subscribe.search -> app.runtime.execution",
     "app.chain.subscribe.search -> app.runtime.log",
     "app.chain.subscribe.search -> app.runtime.stop",
+    "app.chain.subscribe.search -> app.runtime.tasks",
     "app.chain.subscribe.search -> app.schemas",
     "app.chain.subscribe.search -> app.schemas.types",
     "app.chain.subscribe.searchtask -> app.application",

tests/test_database_backup_scheduler.py

@@ -148,6 +148,29 @@ def test_scheduled_backup_uses_registered_database_governance(monkeypatch) -> No
     governance.create_backup.assert_called_once_with()
 
 
+@pytest.mark.parametrize("enabled", [False, True])
+def test_subscription_search_scans_due_items_only_when_enabled(monkeypatch, enabled) -> None:
+    """系统开关控制到期扫描，五分钟扫描节奏与实际搜索间隔分别传递。"""
+    scheduler = _scheduler()
+    scheduler._services = Mock()
+    background_scheduler = Mock()
+    monkeypatch.setattr(scheduler_catalog, "BackgroundScheduler", lambda **_kwargs: background_scheduler)
+    monkeypatch.setattr(scheduler_catalog, "get_plugin_manager", lambda: Mock())
+    monkeypatch.setattr(scheduler_catalog, "get_mediaserver_configs", lambda **_kwargs: [])
+    monkeypatch.setattr(scheduler, "init_workflow_jobs", lambda: None)
+    monkeypatch.setattr(scheduler, "init_agent_task_jobs", lambda: None)
+    monkeypatch.setattr(scheduler, "init_plugin_jobs", lambda: None)
+
+    scheduler._initialize_catalog(_config(subscribe_search=enabled, subscribe_search_interval=48))
+
+    calls = [call for call in background_scheduler.add_job.call_args_list
+             if call.kwargs.get("id") == "subscribe_search"]
+    assert bool(calls) is enabled
+    if enabled:
+        assert calls[0].kwargs["minutes"] == 5
+        assert scheduler._jobs["subscribe_search"]["kwargs"]["scheduled_interval"] == 48
+
+
 def test_scheduler_database_dependencies_are_explicit_module_imports() -> None:
     scheduler_root = Path(__file__).parents[1] / "app" / "scheduler"
     trees = [ast.parse(path.read_text(encoding="utf-8")) for path in scheduler_root.glob("*.py")]

tests/test_explicit_response_models.py

@@ -158,8 +158,8 @@ def test_collection_json_schemas_define_items_or_tuple_members():
     assert crew_schema["items"]["$ref"].endswith("/TmdbEpisodeCrew")
 
 
-def test_discover_media_source_keeps_legacy_prefix_compatible():
+def test_discover_media_source_preserves_plugin_identifiers():
-    """发现源应兼容旧插件前缀，并同时输出规范媒体来源。"""
+    """发现源按插件标识补齐身份字段，不转换插件特定前缀。"""
     legacy = DiscoverMediaSource(
         name="哔哩哔哩",
         mediaid_prefix="bilibili",
@@ -167,7 +167,7 @@ def test_discover_media_source_keeps_legacy_prefix_compatible():
     )
     current = DiscoverMediaSource(
         name="腾讯视频",
-        media_source=MediaSource.TencentVideo,
+        media_source=MediaSource("tencentvideodiscover"),
         api_path="plugin/TencentVideoDiscover/discover",
     )
     historical_alias = DiscoverMediaSource(
@@ -181,9 +181,9 @@ def test_discover_media_source_keeps_legacy_prefix_compatible():
         api_path="plugin/AcmeVideo/discover",
     )
 
-    assert legacy.media_source is MediaSource.Bilibili
+    assert legacy.media_source == MediaSource("bilibili")
     assert legacy.model_dump(mode="json")["mediaid_prefix"] == "bilibili"
-    assert current.mediaid_prefix == MediaSource.TencentVideo.value
+    assert current.mediaid_prefix == "tencentvideodiscover"
-    assert historical_alias.media_source is MediaSource.MangoTV
+    assert historical_alias.media_source == MediaSource("mangguo")
     assert plugin_source.media_source == MediaSource("acme.video")
     assert plugin_source.mediaid_prefix == "acme.video"

tests/test_llm_opencode_session.py

@@ -0,0 +1,101 @@
+"""OpenCode 官方端点的会话路由请求头回归测试。"""
+
+import asyncio
+from unittest.mock import AsyncMock
+
+import httpx
+import pytest
+
+from app.agent.llm import helper
+from app.agent.llm.helper import LLMHelper
+
+
+@pytest.mark.parametrize("base_url", [
+    "https://opencode.ai/zen/v1",
+    "https://opencode.ai/zen/go/v1",
+])
+@pytest.mark.parametrize("cache_key", ["moviepilot-agent-private-hash", None])
+def test_opencode_model_sends_stable_session_on_every_request(monkeypatch, base_url, cache_key):
+    """主对话与独立调用经过真实 SDK 后仍携带稳定标识，工具绑定不丢失请求头。"""
+    requests = []
+
+    def respond(request):
+        """在本地捕获 SDK 请求并返回最小聊天响应，禁止真实出站。"""
+        requests.append(request)
+        return httpx.Response(200, json={
+            "id": "chatcmpl-test",
+            "object": "chat.completion",
+            "created": 0,
+            "model": "test-model",
+            "choices": [{"index": 0, "message": {"role": "assistant", "content": "OK"},
+                         "finish_reason": "stop"}],
+        })
+
+    transport = httpx.MockTransport(respond)
+    with httpx.Client(transport=transport) as client:
+        async def invoke():
+            """使用离线运行时创建模型，并验证同步与异步请求使用同一标识。"""
+            async with httpx.AsyncClient(transport=transport) as async_client:
+                monkeypatch.setattr(
+                    helper, "_build_httpx_client",
+                    lambda _proxy, **kwargs: async_client if kwargs.get("async_client") else client,
+                )
+                runtime = AsyncMock()
+                runtime.resolve_runtime.return_value = {
+                    "runtime": "openai_compatible", "model_id": "test-model",
+                    "api_key": "test-key", "base_url": base_url,
+                }
+                model = await LLMHelper.get_llm(
+                    provider="opencode", model="test-model", user_agent="",
+                    use_proxy=False, api_protocol="chat_completions", web_search_mode="disabled",
+                    prompt_cache_key=cache_key, provider_runtime=runtime,
+                )
+                bound = model.bind_tools([{
+                    "type": "function", "function": {"name": "example", "description": "测试工具",
+                        "parameters": {"type": "object", "properties": {}}},
+                }])
+                assert bound.invoke("hello").content == "OK"
+                assert (await bound.ainvoke("again")).content == "OK"
+
+        asyncio.run(invoke())
+
+    assert len(requests) == 2
+    session = requests[0].headers["x-opencode-session"]
+    assert session
+    assert requests[1].headers["x-opencode-session"] == session
+    if cache_key:
+        assert session == cache_key
+    assert requests[0].headers["user-agent"] == "MoviePilot"
+
+
+@pytest.mark.parametrize("base_url", [
+    "https://opencode.ai/zen/go/v1", "https://OPENCODE.AI/zen/v1",
+])
+def test_opencode_independent_models_have_distinct_sessions(base_url):
+    """没有对话标识的独立模型互不共用会话，并保留自定义 UA 和模型参数。"""
+    original = {"user-agent": "custom-client/1.0", "X-Test": "value"}
+    options = {"extra_body": {"test": True}}
+    headers = [LLMHelper._build_openai_prompt_cache_options(
+        provider="custom", base_url=base_url, use_responses_api=True,
+        prompt_cache_key=None, default_headers=original, model_kwargs=options,
+    ) for _ in range(2)]
+    assert headers[0][0]["x-opencode-session"] != headers[1][0]["x-opencode-session"]
+    assert headers[0][0]["user-agent"] == "custom-client/1.0"
+    assert "User-Agent" not in headers[0][0]
+    assert headers[0][0]["X-Test"] == "value"
+    assert headers[0][1] == options
+    assert "x-opencode-session" not in original
+
+
+@pytest.mark.parametrize("base_url", [
+    "https://opencode.ai.example/zen/go/v1", "https://proxy.example/opencode.ai",
+    "https://opencode.ai@proxy.example/v1", "https://[invalid", None,
+])
+def test_opencode_headers_do_not_leak_to_other_hosts(base_url):
+    """供应商名称与路径不能代替官方主机校验，兼容端点保持原有参数。"""
+    headers, kwargs = LLMHelper._build_openai_prompt_cache_options(
+        provider="opencode", base_url=base_url, use_responses_api=False,
+        prompt_cache_key="private-cache-key", default_headers=None, model_kwargs={},
+    )
+    assert headers is None
+    assert kwargs == {}

tests/test_media_classification_catalog.py

@@ -98,20 +98,16 @@ def test_catalog_separates_retired_fields_from_new_rule_options() -> None:
     assert retired_fields[0].replacement_field == "media.countries"
 
 
-def test_discover_only_builtin_sources_are_explicitly_unavailable() -> None:
+def test_plugin_sources_have_no_builtin_capability_declarations() -> None:
-    """仅提供发现入口的内置来源不能被误认为可形成分类事实。"""
+    """无宿主模块的来源不得预占插件标识或预设不可用的字段能力。"""
     discover_sources = {
-        MediaSource.Bilibili.value,
+        "bilibili", "mangguodiscover", "migu", "tencentvideodiscover", "iqiyidiscover",
-        MediaSource.MangoTV.value,
-        MediaSource.MiguVideo.value,
-        MediaSource.TencentVideo.value,
-        MediaSource.Iqiyi.value,
     }
 
-    for media_source in discover_sources:
+    assert discover_sources.isdisjoint(BUILTIN_CLASSIFICATION_SOURCES)
-        assert set(builtin_source_field_support(media_source).values()) == {
+    assert discover_sources.isdisjoint(source.value for source in MediaSource)
-            "unavailable"
+    for field in get_standard_classification_fields():
-        }
+        assert discover_sources.isdisjoint(field.source_support)
 
 
 def test_catalog_dictionary_values_match_normalized_facts() -> None:

tests/test_media_classification_plugin_extensions.py

@@ -306,6 +306,61 @@ def test_registry_rejects_builtin_duplicate_and_cross_plugin_sources() -> None:
     assert registry.sources(OTHER_PLUGIN_ID) == []
 
 
+@pytest.mark.parametrize(
+    "media_source",
+    ["bilibili", "mangguodiscover", "migu", "tencentvideodiscover", "iqiyidiscover"],
+)
+def test_plugin_owned_sources_follow_registration_lifecycle(
+    media_source: str,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """无内置模块的来源由插件注册，API 和分类字段随插件启停增删。"""
+    from app.api.endpoints.media import _registered_media_sources
+    from app.application.classification.catalog import build_classification_field_catalog
+    from app.domain.classification.sources import BUILTIN_CLASSIFICATION_SOURCES
+
+    registry = _registry()
+    monkeypatch.setattr(
+        "app.application.plugin.runtime.get_plugin_manager",
+        lambda: SimpleNamespace(get_media_sources=registry.sources),
+    )
+    source_id = media_source
+    field_id = f"extensions.{source_id}.region_group"
+    baseline_sources = _registered_media_sources()
+    assert {source.media_source.value for source in baseline_sources} == set(
+        BUILTIN_CLASSIFICATION_SOURCES
+    )
+    assert media_source not in {source.media_source for source in baseline_sources}
+    assert all(
+        source_id not in field.source_support
+        for field in build_classification_field_catalog()
+    )
+
+    registry.replace(PLUGIN_ID, [_source(
+        source_id=source_id,
+        fields=[_field("region_group", source_id=source_id)],
+    )])
+    sources = _registered_media_sources()
+    assert [source for source in sources if source.media_source == media_source] == [
+        MediaSourceInfo.model_validate(registry.sources()[0])
+    ]
+    fields = {field.id: field for field in build_classification_field_catalog(registry.fields())}
+    assert fields[field_id].source_support == {source_id: "extension"}
+    media = MediaInfo(
+        media_source=media_source,
+        media_id="native-1",
+        type=MediaType.MOVIE,
+        classification_facts={field_id: "亚洲"},
+    )
+    assert registry.facts(media) == {source_id: {"region_group": "亚洲"}}
+
+    registry.remove(PLUGIN_ID)
+    assert _registered_media_sources() == baseline_sources
+    assert field_id not in {
+        field.id for field in build_classification_field_catalog(registry.fields())
+    }
+
+
 @pytest.mark.parametrize(
     ("field", "error_text"),
     [

tests/test_media_source_routing.py

@@ -26,13 +26,14 @@ def test_generic_source_id_resolves_as_fixed_enum() -> None:
 def test_iqiyi_source_routes_through_unified_identity() -> None:
     """爱奇艺探索来源应支持选择解析、身份解析和媒体键构造。"""
     assert parse_media_source_selection("iqiyi,iqiyidiscover") == (
-        MediaSource.Iqiyi,
+        MediaSource("iqiyi"), MediaSource("iqiyidiscover"),
     )
     assert resolve_media_identity(
-        media_source="iqiyi",
+        media_source="iqiyidiscover",
         media_id="album-1",
-    ) == (MediaSource.Iqiyi, "album-1")
+    ) == (MediaSource("iqiyidiscover"), "album-1")
-    assert build_media_key("iqiyi", "album-1") == "iqiyidiscover:album-1"
+    assert build_media_key("iqiyidiscover", "album-1") == "iqiyidiscover:album-1"
+    assert build_media_key("iqiyi", "album-1") == "iqiyi:album-1"
 
 
 def test_plugin_source_is_preserved_as_dynamic_enum() -> None:

tests/test_release_supply_chain.py

@@ -126,6 +126,10 @@ def test_base_image_uses_refreshable_tag_and_apt_does_not_upgrade_in_place() ->
     assert "ARG MOVIEPILOT_PYTHON_VERSION" in free_threaded_stage
     assert 'uv python install --no-bin "${MOVIEPILOT_PYTHON_VERSION}t"' in free_threaded_stage
     assert "apt-get upgrade" not in dockerfile
+    assert (
+        "apt-get install -y --no-install-recommends libde265-0"
+        in dockerfile
+    )
     assert "\n    openssl \\\n" in dockerfile
     assert "\n    util-linux \\\n" in dockerfile
 

tests/test_scheduler_lifecycle.py

@@ -849,3 +849,46 @@ def test_cancelled_cross_thread_proxy_waits_for_target_loop_cleanup(
     assert scheduler._registry.is_active("cancel-before-start") is False
     assert scheduler._registry.handles() == ()
     assert not any("was never awaited" in str(item.message) for item in captured)
+
+
+@pytest.mark.anyio
+async def test_subscription_queue_poll_returns_while_owned_search_is_running(monkeypatch) -> None:
+    """长搜索不占用轮询调用，重复轮询不重入且取消必须等待同步 worker 结束。"""
+    from app.chain.subscribe.facade import SubscribeChain
+
+    _patch_progress(monkeypatch)
+    chain = object.__new__(SubscribeChain)
+    started = threading.Event()
+    release = threading.Event()
+    calls = []
+
+    def consume(**kwargs):
+        """模拟慢搜索并捕获手工搜索参数，禁止真实业务调用。"""
+        calls.append(kwargs)
+        started.set()
+        assert release.wait(timeout=5)
+
+    chain.resume_search_queue = consume
+    job_id = "subscribe_search_queue"
+    scheduler = _scheduler(job_id, chain.async_resume_search_queue)
+    scheduler._jobs[job_id]["kwargs"] = {"limit": 2, "manual_sids": (1, 2)}
+    scheduler._poll_subscription_search_queue()
+    try:
+        assert await asyncio.to_thread(started.wait, 2)
+        assert scheduler._is_job_active(job_id)
+        for _ in range(3):
+            scheduler._poll_subscription_search_queue()
+        assert len(calls) == 1
+        assert calls[0]["limit"] == 2
+        assert calls[0]["manual_sids"] == (1, 2)
+        handles = scheduler._registry.handles()
+        assert handles
+        stopping = asyncio.create_task(scheduler.stop_async())
+        await asyncio.sleep(0)
+        assert not stopping.done()
+        assert scheduler._is_job_active(job_id)
+    finally:
+        release.set()
+    await asyncio.wait_for(stopping, timeout=3)
+    assert not scheduler._is_job_active(job_id)
+    assert scheduler._registry.handles() == ()

tests/test_search_media_sources.py

@@ -7,7 +7,7 @@ import pytest
 from app.api.endpoints import media as media_endpoint
 from app.api.endpoints import search as search_endpoint
 from app.chain.subscribe import create as subscribe_create
-from app.chain.subscribe import SubscribeChain
+from app.chain.subscribe.facade import SubscribeChain
 from app.domain.context import MediaInfo
 from app.schemas.types import MediaSource, MediaType
 from app.schemas.media import normalize_media_source
@@ -21,11 +21,16 @@ def test_media_source_normalization_accepts_plugin_source() -> None:
     assert normalize_media_source("plugin source:invalid") is None
 
 
-def test_iqiyi_media_source_aliases_are_normalized() -> None:
+@pytest.mark.parametrize("source_id", [
-    """爱奇艺探索来源的历史前缀和规范前缀应归一到同一媒体来源。"""
+    "bilibili", "mangguodiscover", "migu", "tencentvideodiscover", "iqiyidiscover",
-    assert normalize_media_source("iqiyi") is MediaSource.Iqiyi
+    "mango_tv", "migu_video", "tencent_video", "iqiyi",
-    assert normalize_media_source("iqiyidiscover") is MediaSource.Iqiyi
+])
-    assert MediaSource("iqiyi") is MediaSource.Iqiyi
+def test_plugin_media_sources_preserve_exact_identifiers(source_id: str) -> None:
+    """插件来源不再属于内置枚举，也不替插件转换来源别名。"""
+    source = normalize_media_source(source_id)
+    assert source is MediaSource(source_id)
+    assert source.value == source_id
+    assert source not in list(MediaSource)
 
 
 def test_resolve_anilist_search_params_preserves_identity() -> None:

tests/test_subscribe_search_state.py

@@ -125,8 +125,9 @@ def test_new_subscribe_search_marks_state_after_attempt(monkeypatch) -> None:
         chain.search(state="N", manual=False)
 
     media_chain.recognize_media.assert_called_once()
-    assert len(_SubscribeOper.updates) == 1
+    assert len(_SubscribeOper.updates) == 2
-    subscribe_id, subscription_patch = _SubscribeOper.updates[0]
+    assert "last_search" in _SubscribeOper.updates[0][1].to_payload()
+    subscribe_id, subscription_patch = _SubscribeOper.updates[1]
     assert subscribe_id == 31
     assert subscription_patch == SubscriptionPatch({"state": "R"})
 
@@ -150,6 +151,7 @@ def test_targeted_batch_searches_all_ids_without_state_scan(monkeypatch) -> None
     with patch.object(subscribe_search, "MediaChain", return_value=media_chain):
         chain = object.__new__(SubscribeChain)
         chain.subscription_repository = subscribe_oper
+        monkeypatch.setattr(chain, "_SubscribeChain__apply_subscribe_update", lambda sub, *_args, **_kwargs: sub)
         chain.search(sids=(31, 32), state=None, manual=False)
 
     assert [item.args for item in subscribe_oper.get.call_args_list] == [

tests/test_subscription_schedule.py

@@ -0,0 +1,136 @@
+"""逐订阅搜索周期的到期、写入和数据库迁移合同。"""
+
+import importlib
+from dataclasses import replace
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+import sqlalchemy as sa
+from alembic.migration import MigrationContext
+from alembic.operations import Operations
+from pydantic import ValidationError
+
+from app.application.subscription.contract import SubscriptionSnapshot
+from app.application.subscription.query import subscription_search_due
+from app.chain.subscribe.facade import SubscribeChain
+from app.db.adapters.subscription import TransactionalSubscriptionRepository
+from app.db.models.subscribe import Subscribe
+from app.db.session import SessionFactory, async_session_scope
+from app.scheduler.catalog import _subscription_search_job_specs
+from app.schemas.subscribe import Subscribe as SubscribeSchema
+
+NOW = datetime(2026, 9, 8, 12, tzinfo=timezone.utc)
+
+
+@pytest.mark.parametrize("interval,elapsed,due", [
+    (None, 23, False), (None, 24, True), (1, 0.99, False), (1, 1, True),
+    (48, 24, False), (48, 48, True), (1, -1, False),
+])
+def test_search_due_uses_custom_or_system_interval(interval, elapsed, due):
+    """独立周期可短于或长于系统周期，并在精确到期时放行。"""
+    subscribe = SubscriptionSnapshot(
+        id=1, name="周期测试", state="R", search_interval=interval,
+        last_search=(NOW - timedelta(hours=elapsed)).isoformat(),
+    )
+    assert subscription_search_due(subscribe, 24, NOW) is due
+    assert subscription_search_due(replace(subscribe, state="P"), 24, NOW) is due
+    assert not subscription_search_due(replace(subscribe, state="S"), 24, NOW)
+    assert not subscription_search_due(replace(subscribe, state="N"), 24, NOW)
+
+
+def test_legacy_creation_time_and_missing_timestamp():
+    """旧本地时间按本地时区解析，缺失或损坏时间允许恢复搜索。"""
+    recent = (NOW - timedelta(hours=1)).astimezone().strftime("%Y-%m-%d %H:%M:%S")
+    subscribe = SubscriptionSnapshot(id=1, name="旧订阅", state="R", date=recent)
+    assert not subscription_search_due(subscribe, 24, NOW)
+    for value in (None, "invalid"):
+        assert subscription_search_due(replace(subscribe, date=value), 24, NOW)
+
+
+def test_targeted_and_manual_selection_bypasses_schedule():
+    """自动调度仅选到期记录，手动及指定目标保持原有立即搜索语义。"""
+    recent = datetime.now(timezone.utc).isoformat()
+    subscribe = SubscriptionSnapshot(id=1, name="测试", state="R", last_search=recent)
+    chain = object.__new__(SubscribeChain)
+    chain.subscription_repository = SimpleNamespace(
+        list=Mock(return_value=[subscribe]), get=Mock(return_value=subscribe),
+    )
+    assert chain._load_search_subscriptions(None, None, "R", scheduled_interval=24) == []
+    assert chain._load_search_subscriptions(None, None, "R") == [subscribe]
+    assert chain._load_search_subscriptions(1, None, "R", scheduled_interval=24) == [subscribe]
+    assert chain._load_search_subscriptions(None, (1,), "R", scheduled_interval=24) == [subscribe]
+
+
+def test_scheduled_scan_does_not_create_empty_batches():
+    """尚未到期的订阅不产生空批次，也不触发队列消费。"""
+    subscribe = SubscriptionSnapshot(
+        id=1, name="测试", state="R", last_search=datetime.now(timezone.utc).isoformat(),
+    )
+    chain = object.__new__(SubscribeChain)
+    chain.subscription_repository = SimpleNamespace(list=Mock(return_value=[subscribe]))
+    chain.subscription_search_repository = Mock()
+    progress = Mock()
+    assert chain.search(state="R", scheduled_interval=24, progress_callback=progress) is None
+    chain.subscription_search_repository.enqueue.assert_not_called()
+    chain.subscription_search_repository.claim_next.assert_not_called()
+    assert progress.call_args.kwargs["value"] == 100
+
+
+@pytest.mark.parametrize("interval", [0, -1, 1.5, 8761])
+def test_invalid_search_interval_is_rejected(interval):
+    """接口拒绝零值、负数、小数和超出范围的独立周期。"""
+    with pytest.raises(ValidationError):
+        SubscribeSchema(search_interval=interval)
+
+
+def test_public_write_can_clear_interval_but_cannot_forge_search_time():
+    """恢复系统周期必须保留 null，客户端不能覆盖内部搜索时钟。"""
+    for value in (None, ""):
+        payload = SubscribeSchema(search_interval=value, last_search=NOW.isoformat())
+        assert payload.to_public_write_payload(exclude_unset=True) == {"search_interval": None}
+    assert SubscribeSchema().to_public_write_payload(exclude_unset=True) == {}
+
+
+def test_schedule_persists_across_repository_recreation(db):
+    """周期和搜索时间在 Session 关闭及仓储重建后仍可用于到期判断。"""
+    row = db.add(Subscribe(
+        name="持久周期", state="R", search_interval=48, last_search=NOW.isoformat(),
+    ))
+    for _ in range(2):
+        repository = TransactionalSubscriptionRepository(
+            sync_session=SessionFactory, async_session=async_session_scope,
+        )
+        snapshot = repository.get(row.id)
+        assert snapshot.search_interval == 48
+        assert snapshot.last_search == NOW.isoformat()
+        assert not subscription_search_due(snapshot, 24, NOW + timedelta(hours=24))
+
+
+def test_scheduler_passes_system_interval_only_to_periodic_search():
+    """调度目录传递可热重载的系统间隔，新增订阅保持首次立即搜索。"""
+    specs = {spec.job_id: spec for spec in _subscription_search_job_specs(Mock(), 12)}
+    assert specs["subscribe_search"].kwargs == {"state": "R", "scheduled_interval": 12}
+    assert specs["new_subscribe_search"].kwargs == {"state": "N"}
+
+
+def test_schedule_migration_preserves_existing_rows_and_is_reversible(monkeypatch):
+    """旧数据默认跟随系统，重复升级与回滚不损坏已有订阅。"""
+    migration = importlib.import_module("database.versions.f8a2c6e9b1d4_3_0_30")
+    engine = sa.create_engine("sqlite://")
+    with engine.begin() as connection:
+        for table in ("subscribe", "subscribehistory"):
+            connection.execute(sa.text(f"CREATE TABLE {table} (id INTEGER PRIMARY KEY, name TEXT)"))
+            connection.execute(sa.text(f"INSERT INTO {table} VALUES (1, '旧订阅')"))
+        monkeypatch.setattr(migration, "op", Operations(MigrationContext.configure(connection)))
+        migration.upgrade()
+        migration.upgrade()
+        for table in ("subscribe", "subscribehistory"):
+            row = connection.execute(sa.text(f"SELECT name, search_interval FROM {table}")).one()
+            assert row == ("旧订阅", None)
+        assert connection.execute(sa.text("SELECT last_search FROM subscribe")).scalar_one() is None
+        migration.downgrade()
+        migration.upgrade()
+        assert connection.execute(sa.text("SELECT name FROM subscribe")).scalar_one() == "旧订阅"
+    engine.dispose()

tests/test_subscription_search_cursor_migration.py

@@ -0,0 +1,43 @@
+"""搜索站点游标与旧排期升级回归。"""
+
+import importlib
+from datetime import datetime, timedelta, timezone
+
+import sqlalchemy as sa
+from alembic.migration import MigrationContext
+from alembic.operations import Operations
+
+
+def test_search_cursor_migration_preserves_cooldown_and_running_tasks(monkeypatch):
+    """只提前未开始的自动排期，冷却、停止、新增保护期和在途任务保持原状。"""
+    migration = importlib.import_module("database.versions.a9c3e7f1b5d8_3_0_31")
+    engine = sa.create_engine("sqlite://")
+    metadata = sa.MetaData()
+    tasks = sa.Table(
+        "subscriptionsearchtask", metadata,
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("source", sa.String()), sa.Column("state", sa.String()),
+        sa.Column("phase", sa.String()), sa.Column("attempt_count", sa.Integer()),
+        sa.Column("cancel_requested", sa.Integer()), sa.Column("available_at", sa.String()),
+    )
+    future = (datetime.now(timezone.utc) + timedelta(days=2)).isoformat(timespec="seconds")
+    with engine.begin() as connection:
+        metadata.create_all(connection)
+        rows = [dict(id=index, source="fallback", state="queued", phase="queued", attempt_count=0,
+                     cancel_requested=0, available_at=future) for index in range(5)]
+        rows[1].update(phase="waiting_site_budget", attempt_count=1)
+        rows[2].update(cancel_requested=1)
+        rows[3].update(source="new", phase="scheduled")
+        rows[4].update(state="running", attempt_count=1)
+        connection.execute(tasks.insert(), rows)
+        monkeypatch.setattr(migration, "op", Operations(MigrationContext.configure(connection)))
+        migration.upgrade()
+        migration.upgrade()
+        stored = connection.execute(sa.select(tasks.c.id, tasks.c.available_at).order_by(tasks.c.id)).all()
+        assert stored[0].available_at < future
+        assert all(row.available_at == future for row in stored[1:])
+        assert "pending_site_ids" in {col["name"] for col in sa.inspect(connection).get_columns(tasks.name)}
+        migration.downgrade()
+        assert "pending_site_ids" not in {col["name"] for col in sa.inspect(connection).get_columns(tasks.name)}
+        assert connection.scalar(sa.select(sa.func.count()).select_from(tasks)) == 5
+    engine.dispose()

tests/test_subscription_search_governance.py

@@ -100,25 +100,15 @@ def _make_tasks_ready(monkeypatch) -> None:
     )
 
 
-def test_fallback_task_schedule_staggers_each_subscription(monkeypatch):
+def test_fallback_task_schedule_jitters_once_for_large_batch(monkeypatch):
-    """兜底批次首条抖动后，每条后续订阅都按独立随机间隔到期。"""
+    """千条订阅共享一次启动抖动，不能再累计成数日空等。"""
     now = datetime(2026, 9, 3, 1, 2, 3, tzinfo=timezone.utc)
-    delays = iter((12, 60, 300))
+    jitter = Mock(return_value=12)
-    monkeypatch.setattr(
+    monkeypatch.setattr("app.chain.subscribe.search.random.randint", jitter)
-        "app.chain.subscribe.search.random.randint",
+    schedule = _search_task_available_at("fallback", tuple(range(1000)), now=now)
-        lambda _low, _high: next(delays),
+    assert len(schedule) == 1000
-    )
+    assert set(schedule.values()) == {(now + timedelta(seconds=12)).isoformat(timespec="seconds")}
-
+    jitter.assert_called_once_with(0, 60)
-    schedule = _search_task_available_at(
-        "fallback",
-        (1, 2, 3),
-        now=now,
-    )
-
-    available = [datetime.fromisoformat(schedule[subscribe_id]) for subscribe_id in (1, 2, 3)]
-    assert (available[0] - now).total_seconds() == 12
-    assert (available[1] - available[0]).total_seconds() == 60
-    assert (available[2] - available[1]).total_seconds() == 300
 
 
 def test_inline_fallback_search_preserves_site_pressure_stagger(tmp_path, monkeypatch):
@@ -349,7 +339,8 @@ def test_site_budget_conflict_requeues_task_without_batch_failure(tmp_path, monk
         assert task.state == "queued"
         assert task.phase == "waiting_site_budget"
         assert task.available_at == retry_at
-        assert task.last_error == "站点暂时忙，系统会自动继续搜索"
+        assert task.last_error == "等待站点"
+        assert task.pending_site_ids == [31]
     assert chain.subscription_search_repository.claim_next(owner="worker-after-retry") is None
 
 

tests/test_subscription_search_queue.py

@@ -362,3 +362,54 @@ def test_search_queue_keeps_manual_work_ahead_of_aged_fallback(tmp_path):
 
     assert claimed.subscription_id == 9
     assert claimed.source == "manual"
+
+
+def test_retry_sites_survive_reopen_and_admission_wait_without_stale_error(tmp_path):
+    """站点游标跨重建仓储和准入等待保留，真正恢复时清除上轮等待提示。"""
+    repository, engine = _repository(tmp_path)
+    repository.enqueue(subscription_ids=(701,), source="fallback", priority=10)
+    first = repository.claim_next(owner="before-restart")
+    ready_at = (datetime.now(timezone.utc) - timedelta(seconds=1)).isoformat(timespec="seconds")
+    assert repository.defer_task(
+        task_id=first.task_id, lease_token=first.lease_token, available_at=ready_at,
+        message="站点冷却中", pending_site_ids=(8, 9),
+    )
+    reopened = TransactionalSubscriptionSearchRepository(sessionmaker(bind=engine))
+    resumed = reopened.claim_next(owner="after-restart")
+    assert resumed.task_id == first.task_id
+    assert resumed.pending_site_ids == (8, 9)
+    assert resumed.state == "running"
+    assert resumed.last_error is None
+    assert reopened.defer_task(
+        task_id=resumed.task_id, lease_token=resumed.lease_token, available_at=ready_at,
+        phase="waiting_subscription", message="等待任务",
+    )
+    continued = reopened.claim_next(owner="after-match")
+    assert continued.pending_site_ids == (8, 9)
+    assert continued.last_error is None
+    assert reopened.defer_task(
+        task_id=continued.task_id, lease_token=first.lease_token, available_at=ready_at,
+        pending_site_ids=(99,),
+    ) is False
+    assert reopened.finish_task(task_id=continued.task_id, lease_token=continued.lease_token, state="completed")
+    engine.dispose()
+
+
+def test_manual_search_restarts_full_scope_but_automatic_merge_keeps_cursor(tmp_path):
+    """自动调度合并保留进度，用户主动重搜可包含新配置的站点。"""
+    repository, engine = _repository(tmp_path)
+    repository.enqueue(subscription_ids=(702,), source="fallback", priority=10)
+    first = repository.claim_next(owner="worker")
+    ready_at = (datetime.now(timezone.utc) - timedelta(seconds=1)).isoformat(timespec="seconds")
+    assert repository.defer_task(
+        task_id=first.task_id, lease_token=first.lease_token, available_at=ready_at,
+        pending_site_ids=(9,),
+    )
+    repository.enqueue(subscription_ids=(702,), source="fallback", priority=10)
+    with Session(engine) as session:
+        assert session.scalar(select(SubscriptionSearchTask.pending_site_ids)) == [9]
+    repository.enqueue(subscription_ids=(702,), source="manual", priority=100)
+    resumed = repository.claim_next(owner="manual")
+    assert resumed.pending_site_ids is None
+    assert resumed.source == "manual"
+    engine.dispose()

tests/test_subscription_site_budget.py

@@ -549,3 +549,48 @@ def test_search_provider_logs_site_budget_release_failure(monkeypatch):
     assert result == ["torrent"]
     assert metrics.snapshot().release_failure_count == 1
     assert errors == ["订阅站点预算释放失败: site_id=14 site=Stale"]
+
+
+def test_retry_provider_only_searches_pending_enabled_sites(monkeypatch):
+    """恢复时不重搜成功站点和插件源，也不重新启用已经移除的站点。"""
+    from unittest.mock import Mock
+
+    from app.chain.search.provider import SearchProviderOwner
+
+    chain = object.__new__(SearchChain)
+    chain.configure_subscription_site_budget(SubscriptionSiteBudget(
+        repository=_WaitingRepository(), owner="retry", cancelled=lambda: False,
+        stop_state=ProcessStopState(), pending_site_ids=(2, 3),
+    ))
+    chain._sync_indexers = lambda _sites: [{"id": 1}, {"id": 2}]
+    chain._torrent_type = lambda *_args: None
+    chain._torrent_keyword = lambda *_args: "movie"
+    chain._build_search_pages = lambda _page: [0]
+    chain.search_plugin_torrents = Mock(return_value=[])
+    captured = []
+
+    def collect(**kwargs):
+        """只观察 provider 选择，禁止真实站点和插件调用。"""
+        captured.extend(site["id"] for site in kwargs["indexer_sites"])
+        return {}
+
+    monkeypatch.setattr(SearchProviderOwner, "_collect_sync_site_results", lambda _self, **kwargs: collect(**kwargs))
+    monkeypatch.setattr("app.chain.search.provider.ProgressHelper", Mock())
+    SearchProviderOwner._search_all_sites(chain, keyword="movie", sites=[1, 2])
+    assert captured == [2]
+    chain.search_plugin_torrents.assert_not_called()
+
+
+def test_deferral_reports_cooldown_separately_from_busy():
+    """冷却提示不伪装成站点占用，并保留所有未完成站点。"""
+    from app.application.subscription.execution import raise_subscription_site_budget_deferral
+    from app.application.subscription.sitebudget import SubscriptionSearchDeferred, SubscriptionSiteBudgetDeferral
+
+    deferred_sites = (
+        SubscriptionSiteBudgetDeferral(site_id=1, retry_at="2026-09-08T10:00:00+00:00", wait_reason="cooldown"),
+        SubscriptionSiteBudgetDeferral(site_id=2, retry_at="2026-09-08T11:00:00+00:00", wait_reason="cooldown"),
+    )
+    with pytest.raises(SubscriptionSearchDeferred, match="站点冷却中") as caught:
+        raise_subscription_site_budget_deferral(deferred_sites, None)
+    assert caught.value.site_ids == (1, 2)
+    assert caught.value.retry_at == deferred_sites[0].retry_at