fix(subscribe): keep scheduled searches progressing across site retries

app/application/subscription/execution.py

@@ -82,6 +82,7 @@ class SubscriptionExecutionContext:
     cancel_requested: Optional[Callable[[], bool]] = None
     phase_changed: Optional[Callable[[str, Optional[int]], None]] = None
     download_started: bool = False
+    resuming_sites: bool = False
 
     def is_cancel_requested(self) -> bool:
         """判断调用入口是否请求在下一个安全边界退出。"""
@@ -211,8 +212,9 @@ class SubscriptionSearchRepository(Protocol):
         source: str,
         priority: int,
         available_at_by_subscription: Optional[Mapping[int, str]] = None,
+        refresh_pending: bool = False,
     ) -> SearchEnqueueResult:
-        """按订阅 ID 和各自到期时间建立或合并活动任务。"""
+        """建立或合并活动任务；新自动周期可恢复等待任务的完整搜索范围。"""
         ...
 
     async def async_enqueue(
@@ -222,8 +224,9 @@ class SubscriptionSearchRepository(Protocol):
         source: str,
         priority: int,
         available_at_by_subscription: Optional[Mapping[int, str]] = None,
+        refresh_pending: bool = False,
     ) -> SearchEnqueueResult:
-        """在异步会话中建立或合并活动任务。"""
+        """异步建立或合并活动任务，可为新周期刷新尚未运行的站点游标。"""
         ...
 
     def claim_next(self, *, owner: str, lease_seconds: int = 900) -> Optional[SearchTaskSnapshot]:

app/application/subscription/sitebudget.py

@@ -207,7 +207,8 @@ class SubscriptionSiteBudget:
             return claim
         if self._metrics:
             self._metrics.record_unavailable(site_id)
-        self._report_phase("waiting_site_budget", site_id)
+        # 单站点不可用不代表整条订阅已等待；其它站点可能仍在搜索。
+        # 由任务所有者在全部请求收口并持久化续跑时间后发布等待阶段。
         raise SubscriptionSiteBudgetUnavailable(
             site_id=site_id,
             retry_at=claim.retry_at,

app/chain/search/execution.py

@@ -7,6 +7,7 @@ from collections import Counter
 from dataclasses import dataclass, field
 from typing import Any, AsyncIterator, Callable, Dict, Generator, Iterator, List, Literal, Optional, cast
 
+from app.application.subscription.sitebudget import SubscriptionSiteBudget
 from app.chain.media import MediaChain
 from app.chain.search.contract import _SearchOwnerBase
 from app.chain.search.plan import SearchPlanOwner
@@ -52,6 +53,27 @@ class _SearchOutcome:
     recognition_failed: bool = False
 
 
+def _has_subscription_budget(owner: _SearchOwnerBase) -> bool:
+    """识别已启用站点预算的订阅搜索，普通搜索保留原有查询策略。"""
+    return isinstance(getattr(owner, "_subscription_site_budget", None), SubscriptionSiteBudget)
+
+
+def _keyword_delay(owner: _SearchOwnerBase, search_count: int) -> int:
+    """订阅已有逐站流控，不再叠加换词时的整轮随机等待。"""
+    if not search_count or _has_subscription_budget(owner):
+        return 0
+    return random.randint(1, 10)
+
+
+def _query_keywords(
+    owner: _SearchOwnerBase, plan: MediaSearchPlan, mediainfo: MediaInfo | MusicInfo, keywords: List[str],
+) -> List[str]:
+    """订阅按 IMDb ID 查询时只执行一轮，避免每个别名被投影成相同请求。"""
+    if _has_subscription_budget(owner) and plan.area == "imdbid" and getattr(mediainfo, "imdb_id", None):
+        return keywords[:1]
+    return keywords
+
+
 def _result_params(plan: MediaSearchPlan, mediainfo: MediaInfo | MusicInfo,
                    torrents: List[TorrentInfo], season_episodes: Any, counts: Counter[str]) -> Dict[str, Any]:
     """为每次完整过滤建立统一参数，计数只描述当前累计候选而不重复叠加。"""
@@ -86,7 +108,7 @@ def _search_resolution(owner: _SearchOwnerBase, plan: MediaSearchPlan) -> Genera
     contexts: List[Context] = []
     counts: Counter[str] = Counter()
     parsed = False
-    for index, keyword in enumerate(keywords):
+    for index, keyword in enumerate(_query_keywords(owner, plan, mediainfo, keywords)):
         batch = yield _SearchStep("search", {
             "mediainfo": mediainfo, "keyword": keyword, "sites": plan.sites, "area": plan.area,
         }, search_count=index)
@@ -133,8 +155,8 @@ class SearchExecutionOwner:
             elif step.kind == "supplement":
                 response = media_chain.supplement_media_info(**step.params)
             elif step.kind == "search":
-                if step.search_count:
-                    time.sleep(random.randint(1, 10))
+                if delay := _keyword_delay(owner, step.search_count):
+                    time.sleep(delay)
                 response = owner._SearchChain__search_all_sites(**step.params) or []
             else:
                 response = owner._parse_result(**step.params)
@@ -151,8 +173,8 @@ class SearchExecutionOwner:
     async def _provider_events(owner: _SearchOwnerBase, step: _SearchStep,
                                streaming: bool) -> AsyncIterator[Dict[str, Any]]:
         """只在 I/O 适配层区分普通请求和站点事件流。"""
-        if step.search_count:
-            await asyncio.sleep(random.randint(1, 10))
+        if delay := _keyword_delay(owner, step.search_count):
+            await asyncio.sleep(delay)
         if streaming:
             async for event in owner._SearchChain__async_search_all_sites_stream(**step.params):
                 yield event

app/chain/subscribe/search.py

@@ -385,6 +385,7 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase):
                 source,
                 subscription_ids,
             ),
+            refresh_pending=scheduled_interval is not None,
         )
         total = len(subscribes)
         summary = SearchExecutionSummary(
@@ -723,11 +724,13 @@ 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",
-        )
+        # 站点补查属于同一轮搜索，不能反复推迟下一轮健康站点的完整检查。
+        if execution_context is None or not execution_context.resuming_sites:
+            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

@@ -190,6 +190,7 @@ class SubscriptionSearchTaskRunner:
             task_id=task_id,
             cancel_requested=lambda: cancelled() or self.stop_state.is_system_stopped,
             phase_changed=phase_changed,
+            resuming_sites=self.task.pending_site_ids is not None,
         )
         current: Optional[SubscriptionSnapshot] = subscribe
         try:

app/db/adapters/subscriptionsearch.py

@@ -78,6 +78,7 @@ def _enqueue_result(
     source: str,
     priority: int,
     available_at_by_subscription: Optional[Mapping[int, str]],
+    refresh_pending: bool,
 ) -> SearchEnqueueResult:
     """在当前事务中创建搜索批次并投影返回结果。"""
     record, created, coalesced, active_batch_ids = repository.enqueue(
@@ -85,6 +86,7 @@ def _enqueue_result(
         source=source,
         priority=priority,
         available_at_by_subscription=available_at_by_subscription,
+        refresh_pending=refresh_pending,
     )
     return SearchEnqueueResult(
         batch=_batch(record),
@@ -148,6 +150,7 @@ class TransactionalSubscriptionSearchRepository:
         source: str,
         priority: int,
         available_at_by_subscription: Optional[Mapping[int, str]] = None,
+        refresh_pending: bool = False,
     ) -> SearchEnqueueResult:
         """创建批次并返回 single-flight 合并计数。"""
         return self._write(
@@ -157,6 +160,7 @@ class TransactionalSubscriptionSearchRepository:
                 source=source,
                 priority=priority,
                 available_at_by_subscription=available_at_by_subscription,
+                refresh_pending=refresh_pending,
             )
         )
 
@@ -167,6 +171,7 @@ class TransactionalSubscriptionSearchRepository:
         source: str,
         priority: int,
         available_at_by_subscription: Optional[Mapping[int, str]] = None,
+        refresh_pending: bool = False,
     ) -> SearchEnqueueResult:
         """在短异步事务中创建批次并返回合并计数。"""
         return await self._async_write(
@@ -176,6 +181,7 @@ class TransactionalSubscriptionSearchRepository:
                 source=source,
                 priority=priority,
                 available_at_by_subscription=available_at_by_subscription,
+                refresh_pending=refresh_pending,
             )
         )
 

app/db/oper/subscriptionsearch.py

@@ -31,6 +31,7 @@ class SubscriptionSearchOper(DbOper):
         source: str,
         priority: int,
         available_at_by_subscription: Optional[Mapping[int, str]],
+        refresh_pending: bool = False,
     ) -> tuple[SubscriptionSearchBatch, int, int, tuple[str, ...]]:
         """创建批次，并以活动键合并同一订阅的重叠搜索入口。"""
         if not isinstance(self._db, Session):
@@ -87,6 +88,11 @@ class SubscriptionSearchOper(DbOper):
                     SubscriptionSearchTask.priority < priority,
                     SubscriptionSearchTask.state == "queued",
                 )
+                refresh_queued_task = and_(
+                    refresh_pending,
+                    SubscriptionSearchTask.state == "queued",
+                    SubscriptionSearchTask.pending_site_ids.is_not(None),
+                )
                 execute_dml(
                     self._db,
                     update(SubscriptionSearchTask)
@@ -101,16 +107,19 @@ class SubscriptionSearchOper(DbOper):
                             else_=SubscriptionSearchTask.priority,
                         ),
                         phase=case(
-                            (promote_queued_task, "queued"),
+                            (or_(promote_queued_task, refresh_queued_task), "queued"),
                             else_=SubscriptionSearchTask.phase,
                         ),
                         last_error=case(
-                            (promote_queued_task, None),
+                            (or_(promote_queued_task, refresh_queued_task), None),
                             else_=SubscriptionSearchTask.last_error,
                         ),
-                        # 用户重新指定搜索时按当前站点配置重搜；自动周期合并保留恢复游标。
+                        # 用户重搜和已到期的新周期恢复完整范围；普通合并仍保留补查游标。
                         pending_site_ids=case(
-                            (and_(SubscriptionSearchTask.state == "queued", source in {"manual", "targeted"}), None),
+                            (or_(
+                                and_(SubscriptionSearchTask.state == "queued", source in {"manual", "targeted"}),
+                                refresh_queued_task,
+                            ), None),
                             else_=SubscriptionSearchTask.pending_site_ids,
                         ),
                         available_at=case(

docs/architecture/subscription-search-review.md

@@ -0,0 +1,104 @@
+# V3 订阅自动搜索机制审查
+
+审查日期：2026-09-09。后端基线 `ef978d606`，前端基线 `e5de69f4`。
+结论来自当前源码和隔离测试；没有连接用户部署、采集生产数据库或测量真实站点响应速度。
+
+## 当前链路
+
+```mermaid
+flowchart TD
+  A[每五分钟检查到期订阅] --> B[按订阅合并持久任务]
+  M[手动或指定订阅搜索] --> B
+  B --> C[全局单消费者逐条处理订阅]
+  C --> D[刷新媒体信息和缺失范围]
+  D --> E[按关键词轮次并行查询站点]
+  E --> F[站点唯一在途请求与错误冷却]
+  F --> G[收齐当前关键词的站点结果]
+  G --> H[过滤排序和下载选择]
+  H --> I{还有待搜索站点且未开始下载}
+  I -->|是| J[保存剩余站点和最早恢复时间]
+  J --> K[每十秒检查可恢复队列]
+  K --> C
+  I -->|否| L[收口本次任务]
+```
+
+- 自动周期由 `subscription_search_due()` 使用 `search_interval` / 系统间隔与 `last_search` 判断。
+- `active_key` 合并同一订阅的重复请求；订阅准入阻止 Search 和 Match 同时修改同一订阅。
+- 站点预算是非阻塞认领。忙碌站点通常十秒后再试，错误冷却按类别指数退避。
+- 成功站点不追加冷却；站点配置的访问间隔仍在 provider 中执行。
+- 未完成站点通过 `pending_site_ids` 保存，普通续跑不会重新访问已经完成的站点或插件搜索源。
+- 正常定时入口持久来源为 `fallback`，新增入口为 `new`；前端已经隐藏这两类后台状态。持续可见的等待还应核对是否来自手工、指定搜索或旧状态。
+
+## 已确认的问题与本次修正
+
+| 问题 | 触发与影响 | 本次修正 |
+| --- | --- | --- |
+| 续跑刷新整个订阅的搜索时间 | 只补查坏站点也写 `last_search`，正常站点下一轮到期时间不断后移 | 上下文明确区分剩余站点续跑，只在完整搜索轮次开始时更新搜索时间 |
+| 新周期合并仍只查旧剩余站点 | 活动键长期存在，自动入队保留旧游标；正常站点的新内容得不到新一轮查询 | 只有经过到期筛选的新周期，才把尚未运行的续跑任务恢复为当前完整站点范围；普通恢复保留游标，在途任务不改写 |
+| 单站点状态覆盖整体状态 | 一个并行请求遇到冷却，立即把任务写成等待，其他站点其实仍在搜索 | 等待由任务完成本轮处理并持久化恢复时间后统一发布 |
+| 订阅重复承担整轮随机等待 | 关键词切换前再等待一至十秒，与已有站点预算和访问间隔叠加 | 仅订阅预算路径去掉额外随机等待，站点实际访问间隔继续执行 |
+| IMDb 按别名重复发相同查询 | 多个名称最终都投影成同一个 IMDb ID，无结果时重复查询 | 订阅使用有效 IMDb ID 搜索时只执行一个关键词轮次；缺少 ID 时继续名称回退 |
+| 等待和失败提示缺乏预期与操作 | 只有黄色“等待站点”，未使用已有 `next_run_at`；原始错误直接显示或依赖悬停 | 中性等待提示、可触摸详情和预计恢复时间；失败明细折叠，并提供重试与有权限的站点检查入口 |
+
+例如：A 站正常、B 站持续故障。首次已搜 A、只剩 B 时，后续补查 B 不应推迟 A 的下一次正常搜索。到用户配置的下一周期，任务重新覆盖当前启用站点；B 仍遵守自己的冷却，A 可以查找新发布的资源。
+
+本次不缩短站点错误冷却、不增加同站并发，也不改变缺集、整季包和洗版优先级。剩余站点仍保留恢复机会，不通过丢弃任务来制造“已完成”。
+
+## 仍影响吞吐的机制及后续优先级
+
+### 1. 按站点调度可用工作，而非长期占住整条订阅消费者
+
+当前一个订阅的慢站点会阻止消费者开始下一条订阅；单条订阅内部虽已并行，但整个队列仍串行。
+下一阶段应先建立可恢复的站点子工作及结果检查点，由站点就绪时间调度请求，避免所有订阅排在慢站后面。
+
+需要保留三个边界：每站唯一在途请求、每订阅提交下载时的独占权、下载器副作用后的事实对账。
+不能仅把全局锁删除或简单开启多个整订阅 worker；那会同时增加媒体识别、媒体库查询和下载器竞争。
+
+### 2. 完善请求事实，再调整错误退避
+
+当前部分 Spider 将 HTTP 失败或超时压成布尔错误，Indexer 只能得到通用 `error`，因此预期的超时、登录失效、403、429分类并不总能生效。
+需要在传输与站点适配边界保留结构化结果，并用真实返回形状的隔离测试覆盖整个调用链，再考虑分类退避和恢复探测。
+
+建议临时网络错误静默退避；登录失效等可操作问题集中到站点层展示一次，给 Cookie 更新或连通性测试入口。
+不要从一段错误字符串可靠性不足地推断“必须用户处理”，也不要把每个订阅对同一坏站的失败重复推送。
+
+### 3. 把恢复检查点细化到实际请求
+
+当前游标只有站点 ID，没有关键词和页码。某站点已经返回第一页、第二页发生冲突时，恢复会从该站第一页开始；成功来源事实也只保存在单次预算对象中。
+后续应保存关键词、页码、来源结果及已处理候选身份，防止多页恢复重复工作，并让跨多次恢复的最终结果准确表达“部分完成”。
+
+### 4. 控制慢站长尾，但保留择优语义
+
+`FIRST_COMPLETED` 只是及时回收完成请求；当前关键词仍需等待所有站点和续页，随后才统一过滤和择优下载。
+快站结果提前可见与提前下载是两个不同合同。可以先渐进展示计数和进度；提前下载必须证明剩余来源不会改变整季包、优先级或洗版选择，否则会产生额外下载。
+
+应先测量每站请求耗时、线程池排队时间、限流等待时间与关键词轮次数，再决定请求超时和慢站策略。
+当前同步流控 `sleep` 发生在站点租约内，后续可改为可取消的就绪调度；不能在线程未退出时释放其租约并让新请求接管。
+
+## 用户反馈原则
+
+| 情况 | 主界面 | 详情或操作 |
+| --- | --- | --- |
+| 正常自动检查、暂时无资源 | 维持正常订阅状态 | 无需用户处理 |
+| 手工搜索暂时遇到站点繁忙或冷却 | 中性“稍后继续” | 预计继续时间，说明系统自动处理 |
+| 部分站点失败、其它来源成功 | 不升级为整个订阅失败 | 站点诊断或折叠明细 |
+| 本轮所有实际来源失败 | 简洁说明本轮未完成 | 重试搜索；管理权限用户可检查站点 |
+| 用户需要修改凭据或配置 | 后续通过结构化错误识别，集中提示 | 直接进入对应站点处理，避免重复消息 |
+
+预计继续时间表示最早可尝试时间，不保证该时刻完成搜索。现有取消接口是批次级，不能把它包装为“只停止这个订阅”。
+
+## 验证口径
+
+- 使用独立 SQLite、模拟站点返回和可控时间，验证恢复任务的时间语义、到期范围刷新、在途保护和并行状态投影。
+- 记录实际 provider 调用与随机等待调用，验证 IMDb 去重及仅订阅移除随机等待；站点间隔和普通搜索行为需要维持回归覆盖。
+- 前端验证后台静默、等待到搜索的恢复、预计时间、失败明细折叠、重试操作和站点管理权限，并检查窄屏触摸入口。
+- 全量通过与真实部署变快不是同一结论。没有生产耗时样本时，只报告已消除的请求/等待次数，不给未经测量的整体加速倍数。
+
+### 本次本地验证结果
+
+- 后端执行 `uv run --locked --no-sync python tests/run.py`，首轮三个分片通过，第四片出现一次受控规模验收失败。排查复现了原验收脚本的线程协调竞态：租约已占用但模拟请求尚未开始时，正确的拒绝被误判。修复试验同步并增加确定性回归，保持全部通过条件和反向变异检查；最终重跑第四片后，四分片合计 **9042 passed、9 skipped**。
+- 架构策略测试 **165 passed**，基线与 CLI 测试 **45 passed**；事件策略、host 快照、mypy、两代复杂度、并发、异步阻塞、后台任务、服务定位、Ruff / mypy ratchet 和三次启动采样通过。依赖快照只新增搜索执行对既有 Application 站点预算的合法依赖。
+- 前端卡片、列表和搜索详情组件 **117 项测试通过**，包括同一任务持续更新、任务切换关闭详情，以及旧批次消失后新批次不会自动弹窗。
+- 375px 手机与 1280px 桌面使用真实 Vuetify 组件和隔离状态预览，验证等待、到期、失败折叠、触摸展开、重试事件和站点权限；未连接真实后端或调用外站。
+- 随后尝试将修改前端连接开发配置指定的本机 `localhost:3001` 进行真实账号验证，但健康检查和前端代理请求均超时，登录页未就绪，未进入账号登录。实际部署的订阅状态与恢复时效仍未验证。
+- 本地验证使用独立 worktree；前端未执行本地生产 build，生产环境整体搜索耗时未测量。提交与发布结果以仓库最终提交对应的 CI 和 Release 为准。

docs/rules/05-architecture.md

@@ -602,6 +602,10 @@ own the remaining focused capabilities. Startup imports the subscription share P
 directly from `notify.py`; canonical callers import identity projection directly from
 `identity.py`. The retired `app/chain/subscribe.py` monolith must not return, package
 owners must not be re-exported, and `_music` must not import its concrete chain owner.
+`app.chain.search.execution` reads the Application-owned `SubscriptionSiteBudget`
+contract to apply subscription-only keyword scheduling and IMDb query deduplication;
+the site budget and provider retain request admission, per-site intervals and cooldown
+ownership. Ordinary resource search retains its existing keyword policy.
 A concrete chain that exposes slash-command interaction inherits
 `InteractionChainMixin`, injects its handler class via `_interaction_handler_type` and
 implements only `_interaction_handler`; it must not re-export application-layer

docs/subscribe-lifecycle.md

@@ -80,6 +80,19 @@ MoviePilot 当前以媒体类型和洗版方式组合出常用订阅模式。普
 
 ## 维护边界
 
+### 自动搜索与站点恢复
+
+`last_search` 记录完整搜索轮次的开始时间，供订阅独立周期或系统周期判断下一次搜索是否到期。
+因站点暂时不可用而仅补查剩余站点时，不刷新此时间，避免坏站点持续推迟正常站点的新一轮搜索。
+同一轮恢复保留待搜索站点；经过到期筛选的新周期会把尚未运行的恢复任务重新覆盖当前启用站点。
+正在执行的任务保持原范围，由后续周期调度接续，不改变已开始的下载副作用。
+
+站点等待不代表搜索失败。其它站点仍在执行时保持搜索状态；本轮结束并保存续跑时间后才进入等待。
+订阅搜索的同站唯一在途预算、站点访问间隔和错误冷却始终生效。机制审查及后续优化边界见
+[订阅搜索机制审查](architecture/subscription-search-review.md)。
+
+### 事实写入
+
 - 公共创建与普通订阅编辑只负责目标和配置变更，不直接维护下载事实和进度摘要；Agent 或维护工具需要修正运行事实时，应使用受控的内部工具或领域入口。
 - 主程序下载链路产生的电视剧下载必须同时维护 `note` 和 `episode_priority`。
 - backfill 入口只接收调用方确认的外部存在事实，不主动扫描媒体库。

scripts/validation/subscription_governance_scale.py

@@ -423,6 +423,9 @@ def _run_site_pressure_case(
             """同步启动一个 owner，并保存 wrapper 的结果或异常。"""
             try:
                 start_barrier.wait(timeout=_SITE_PRESSURE_SYNC_TIMEOUT)
+                # 租约取得不代表请求已经开始；竞争方必须在真实请求在途时验证拒绝。
+                if owner == owners[1] and not boundary.first_started.wait(timeout=_SITE_PRESSURE_SYNC_TIMEOUT):
+                    raise TimeoutError("首个站点请求未进入受控请求边界")
                 with result_lock:
                     owner_invocations[owner] += 1
                 result = chains[owner]._search_site_torrents_with_budget(  # pylint: disable=protected-access

tests/fixtures/architecture/dependency-baseline.json

@@ -1074,8 +1074,8 @@
       "runtime_only": true
     }
   },
-  "edge_count": 8377,
-  "edge_sha256": "b0f844fa3ef29873f1b18130fc514d79fa75ebea6add4e20d6d6047b703b8152",
+  "edge_count": 8380,
+  "edge_sha256": "85f2c071926270095a086e1358127133b8201742690c6694c2b75d504a304e26",
   "edges": [
     "app -> app.foundation",
     "app -> app.foundation.environment",
@@ -4469,6 +4469,9 @@
     "app.chain.search.cache -> app.schemas.types",
     "app.chain.search.contract -> app.chain",
     "app.chain.search.contract -> app.chain.base",
+    "app.chain.search.execution -> app.application",
+    "app.chain.search.execution -> app.application.subscription",
+    "app.chain.search.execution -> app.application.subscription.sitebudget",
     "app.chain.search.execution -> app.chain",
     "app.chain.search.execution -> app.chain.media",
     "app.chain.search.execution -> app.chain.search",

tests/test_subscribe_search_state.py

@@ -1,15 +1,16 @@
 import threading
 from contextlib import contextmanager
 from dataclasses import replace
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from types import SimpleNamespace
 from unittest.mock import Mock, patch
 
 import pytest
 
 from app.application.subscription.contract import SubscriptionPatch, SubscriptionSnapshot
-from app.application.subscription.execution import SubscriptionExecutionAdmission
+from app.application.subscription.execution import SubscriptionExecutionAdmission, SubscriptionExecutionContext
 from app.application.subscription.mutation import SubscriptionMutation
+from app.application.subscription.query import subscription_search_due
 from app.chain.subscribe import search as subscribe_search
 from app.chain.subscribe.facade import SubscribeChain
 from app.schemas.types import MediaType
@@ -132,6 +133,36 @@ def test_new_subscribe_search_marks_state_after_attempt(monkeypatch) -> None:
     assert subscription_patch == SubscriptionPatch({"state": "R"})
 
 
+@pytest.mark.parametrize("resuming_sites", [False, True])
+def test_site_resume_preserves_the_full_search_schedule(monkeypatch, resuming_sites) -> None:
+    """补查坏站点不推迟健康站点的新周期，完整搜索才更新周期起点。"""
+    now = datetime.now(timezone.utc)
+    subscribe = replace(
+        _new_subscribe(datetime.now() - timedelta(days=2)), state="R",
+        last_search=(now - timedelta(hours=25)).isoformat(timespec="seconds"),
+    )
+    admission = SubscriptionExecutionAdmission()
+    lease = admission.try_acquire(subscription_id=subscribe.id, operation="search", ttl_seconds=60)
+    context = SubscriptionExecutionContext(
+        lease=lease, admission=admission, resuming_sites=resuming_sites,
+    )
+    chain = object.__new__(SubscribeChain)
+    monkeypatch.setattr(subscribe_search, "prepare_search_target", Mock(return_value=None))
+    monkeypatch.setattr(subscribe_search, "MediaChain", Mock())
+    update = Mock(side_effect=lambda item, payload, **_kwargs: replace(item, **payload))
+    monkeypatch.setattr(chain, "_SubscribeChain__apply_subscribe_update", update)
+
+    result = chain._process_search_subscription(subscribe, Mock(), execution_context=context)
+
+    assert subscription_search_due(result, 24, now) is resuming_sites
+    if resuming_sites:
+        update.assert_not_called()
+        assert result.last_search == subscribe.last_search
+    else:
+        assert result.last_search != subscribe.last_search
+    assert admission.release(lease)
+
+
 def test_targeted_batch_searches_all_ids_without_state_scan(monkeypatch) -> None:
     """用户归属订阅批次只按指定 ID 顺序读取，不扩大为全局状态搜索。"""
     first = _new_subscribe(datetime.now() - timedelta(minutes=2))

tests/test_subscription_governance_scale.py

@@ -1,5 +1,6 @@
 """订阅执行治理最终受控规模门禁。"""
 
+import json
 import threading
 
 import pytest
@@ -23,7 +24,7 @@ def test_subscription_governance_controlled_scale_matrix() -> None:
     result = run_acceptance()
 
     assert result["schema_version"] == 5
-    assert result["passed"] is True
+    assert result["passed"] is True, json.dumps(result, ensure_ascii=False, indent=2)
     assert all(result["gates"].values())
     assert result["method"]["match_entrypoint"] == "SubscribeChain.match"
     assert result["method"]["download_selection"] == "DownloadChain.batch_download"
@@ -99,6 +100,53 @@ def test_subscription_governance_controlled_scale_matrix() -> None:
     )
 
 
+def test_scale_pressure_waits_for_transport_after_first_lease(monkeypatch, tmp_path):
+    """首个租约取得后延迟进入请求，也必须在请求真正进行时验证并发拒绝。"""
+    coordinator = threading.current_thread()
+    contender_observed = threading.Event()
+    first_lease = threading.Event()
+
+    class RequestStartedEvent(threading.Event):
+        """让租约持有者知道另一 owner 已等待真实请求开始。"""
+
+        def wait(self, timeout=None):
+            """排除主协调线程的等待，仅由请求竞争方放行首个租约。"""
+            if threading.current_thread() is not coordinator:
+                contender_observed.set()
+            return super().wait(timeout)
+
+    class DelayedRequestBoundary(scale._ScaleSiteRequestBoundary):
+        """区分正确的在途等待与过早发生的预算拒绝。"""
+
+        def __init__(self, *, site_id):
+            """替换请求开始事件，控制首个租约到网络边界之间的调度。"""
+            super().__init__(site_id=site_id)
+            self.first_started = RequestStartedEvent()
+
+        def record_budget_rejection(self):
+            """保留拒绝瞬间的在途数量，再解除首个租约的受控暂停。"""
+            super().record_budget_rejection()
+            contender_observed.set()
+
+    original_acquire = scale.SubscriptionSiteBudgetController.acquire
+
+    def acquire_before_transport(self, site_id):
+        """将首个合法租约暂停到竞争方已等待请求或提前遭到拒绝。"""
+        claim = original_acquire(self, site_id)
+        if not first_lease.is_set():
+            first_lease.set()
+            assert contender_observed.wait(timeout=scale._SITE_PRESSURE_SYNC_TIMEOUT)
+        return claim
+
+    monkeypatch.setattr(scale, "_ScaleSiteRequestBoundary", DelayedRequestBoundary)
+    monkeypatch.setattr(scale.SubscriptionSiteBudgetController, "acquire", acquire_before_transport)
+
+    result = _run_durable_governance(ScaleCase("delayed-request", 2, 1, 2, 1), tmp_path)
+
+    assert result["site_pressure_valid"] is True, json.dumps(result, ensure_ascii=False, indent=2)
+    assert result["site_observations"][0]["request_active_at_rejection"] == 1
+
+
 @pytest.mark.parametrize("mutation", ["bypass", "early_release", "allow_concurrent"])
 def test_scale_validator_rejects_fake_site_pressure(monkeypatch, tmp_path, mutation):
     """绕过 wrapper、提前释放或并发放行时，请求边界峰值必须暴露门禁失效。"""

tests/test_subscription_search_governance.py

@@ -21,6 +21,7 @@ from app.application.subscription.sitebudget import (
     SubscriptionSearchDeferred,
 )
 from app.chain.search.facade import SearchChain
+from app.chain.search.provider import SearchProviderOwner
 from app.chain.subscribe import search as subscribe_search
 from app.chain.subscribe.facade import SubscribeChain
 from app.chain.subscribe.search import _search_task_available_at
@@ -47,6 +48,7 @@ class _SubscriptionRepository:
     """为搜索队列返回稳定的多订阅快照。"""
 
     def __init__(self, subscribes: list[SubscriptionSnapshot]) -> None:
+        """保存各订阅的不可变测试快照。"""
         self._subscribes = {subscribe.id: subscribe for subscribe in subscribes}
 
     def list(self, _state: str = None) -> list[SubscriptionSnapshot]:
@@ -209,6 +211,61 @@ def test_fallback_queue_executes_without_match_global_lock(tmp_path, monkeypatch
     assert batch.failed_count == 0
 
 
+@pytest.mark.parametrize("scheduled_interval", [None, 24])
+def test_due_search_restores_healthy_sites_while_ordinary_resume_keeps_scope(
+    tmp_path, monkeypatch, scheduled_interval,
+):
+    """到期轮次恢复完整站点和插件源，普通合并仍只补查未完成的站点。"""
+    subscribe = replace(
+        _subscribe(91),
+        last_search=(datetime.now(timezone.utc) - timedelta(hours=25)).isoformat(timespec="seconds"),
+    )
+    chain = _chain(tmp_path, [subscribe])
+    _make_tasks_ready(monkeypatch)
+    queue = chain.subscription_search_repository
+    original = queue.enqueue(subscription_ids=(subscribe.id,), source="fallback", priority=10)
+    first = queue.claim_next(owner="initial")
+    assert queue.defer_task(
+        task_id=first.task_id, lease_token=first.lease_token,
+        available_at=(datetime.now(timezone.utc) + timedelta(hours=6)).isoformat(timespec="seconds"),
+        pending_site_ids=(9,),
+    )
+    searchchain = object.__new__(SearchChain)
+    searchchain._sync_indexers = lambda _sites: [{"id": 1}, {"id": 9}]
+    searchchain._torrent_type = lambda *_args: None
+    searchchain._torrent_keyword = lambda *_args: subscribe.name
+    searchchain._build_search_pages = lambda _page: [0]
+    searchchain.search_plugin_torrents = Mock(return_value=[])
+    scopes = []
+    searched_sites = []
+
+    def collect(_self, **kwargs):
+        """记录实际 provider 选择的站点，同时隔离外部请求。"""
+        searched_sites.extend(site["id"] for site in kwargs["indexer_sites"])
+        return {}
+
+    monkeypatch.setattr(SearchProviderOwner, "_collect_sync_site_results", collect)
+    monkeypatch.setattr("app.chain.search.provider.ProgressHelper", Mock())
+
+    def process(item, current_searchchain, *, execution_context):
+        """记录真实消费者为本轮 provider 注入的范围，禁止外部搜索。"""
+        scopes.append((
+            execution_context.resuming_sites,
+            execution_context.task_id,
+        ))
+        SearchProviderOwner._search_all_sites(current_searchchain, keyword=item.name, sites=[1, 9])
+        return item
+
+    monkeypatch.setattr(chain, "_process_search_subscription", process)
+    with patch("app.chain.subscribe.search.SearchChain", return_value=searchchain):
+        chain.search(state="R", scheduled_interval=scheduled_interval)
+
+    assert scopes == [(scheduled_interval is None, first.task_id)]
+    assert searched_sites == ([1, 9] if scheduled_interval else [9])
+    assert searchchain.search_plugin_torrents.call_count == (1 if scheduled_interval else 0)
+    assert queue.get_batch(original.batch.batch_id).state == "completed"
+
+
 def test_fallback_queue_continues_after_one_subscription_failure(tmp_path, monkeypatch):
     """单订阅异常不得中止批次后续任务，聚合终态必须暴露失败。"""
     subscribes = [_subscribe(3), _subscribe(4)]

tests/test_subscription_search_keywords.py

@@ -0,0 +1,83 @@
+"""订阅换词策略减少重复等待和 IMDb 请求，同时保留普通搜索行为。"""
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+
+from app.application.subscription.sitebudget import SubscriptionSiteBudget
+from app.chain.search import SearchChain, execution, media
+from app.domain.context import MediaInfo
+from app.schemas.types import MediaType
+
+
+@pytest.mark.parametrize("mode", ["sync", "async", "stream"])
+@pytest.mark.parametrize("subscription", [False, True])
+@pytest.mark.parametrize("area,imdb_id", [("imdbid", "tt1234567"), ("imdbid", None), ("title", "tt1234567")])
+def test_subscription_queries_skip_round_wait_and_duplicate_imdb_requests(
+    monkeypatch, mode, subscription, area, imdb_id,
+):
+    """真实执行循环在无命中时也不重复 IMDb 请求，缺 ID 和普通搜索仍完整换词。"""
+    chain = object.__new__(SearchChain)
+    chain.runtime_config = SimpleNamespace(search_multiple_name=True)
+    aliases = ["Main Title", "Original Title", "Another Title"]
+    target = MediaInfo(tmdb_id=1, title=aliases[0], names=aliases, imdb_id=imdb_id, type=MediaType.MOVIE)
+    if subscription:
+        chain.configure_subscription_site_budget(SubscriptionSiteBudget(
+            repository=Mock(), owner="test-keywords", cancelled=lambda: False,
+            stop_state=SimpleNamespace(is_system_stopped=False),
+        ))
+    queries = []
+    sleeps = []
+
+    def search(**kwargs):
+        """在站点 I/O 边界记录最终检索词，所有查询都返回空结果。"""
+        queries.append(chain._torrent_keyword(kwargs["keyword"], kwargs["mediainfo"], kwargs["area"]))
+        return []
+
+    async def async_search(**kwargs):
+        """异步查询使用同一站点边界替身。"""
+        return search(**kwargs)
+
+    async def stream(**kwargs):
+        """流式查询发布与同步查询相同的空页。"""
+        yield {"items": search(**kwargs)}
+
+    async def supplement(mediainfo):
+        """隔离外部媒体补充请求。"""
+        return mediainfo
+
+    async def sleep(delay):
+        """记录异步等待时间而不让测试实际睡眠。"""
+        sleeps.append(delay)
+
+    monkeypatch.setattr(media, "MediaChain", lambda: SimpleNamespace(
+        supplement_media_info=lambda mediainfo: mediainfo, async_supplement_media_info=supplement,
+    ))
+    monkeypatch.setattr(execution.random, "randint", lambda _start, _end: 5)
+    monkeypatch.setattr(execution.time, "sleep", sleeps.append)
+    monkeypatch.setattr(execution.asyncio, "sleep", sleep)
+    chain._prepare_params = lambda **_kwargs: (None, aliases)
+    chain._parse_result = Mock(return_value=[])
+    chain._SearchChain__search_all_sites = search
+    chain._SearchChain__async_search_all_sites = async_search
+    chain._SearchChain__async_search_all_sites_stream = stream
+    if mode == "sync":
+        result = chain.process(target, area=area)
+    elif mode == "async":
+        result = asyncio.run(chain.async_process(target, area=area))
+    else:
+        async def collect():
+            """从流式完成事件读取最终搜索结果。"""
+            return [event async for event in chain.async_process_stream(target, area=area)]
+
+        result = asyncio.run(collect())[-1]["contexts"]
+
+    if area == "imdbid" and imdb_id:
+        assert queries == [imdb_id] * (1 if subscription else len(aliases))
+    else:
+        assert queries == aliases
+    assert sleeps == ([] if subscription else [5] * (len(queries) - 1))
+    assert result == []
+    chain._parse_result.assert_called_once()

tests/test_subscription_search_queue.py

@@ -413,3 +413,42 @@ def test_manual_search_restarts_full_scope_but_automatic_merge_keeps_cursor(tmp_
     assert resumed.pending_site_ids is None
     assert resumed.source == "manual"
     engine.dispose()
+
+
+@pytest.mark.parametrize("running", [False, True])
+@pytest.mark.parametrize("source,priority", [("fallback", 10), ("new", 50), ("manual", 120)])
+def test_new_search_cycle_refreshes_only_waiting_site_scope(tmp_path, running, source, priority):
+    """新周期重搜等待任务的完整站点，不能覆盖在途游标或丢失原手动优先级。"""
+    repository, engine = _repository(tmp_path)
+    original = repository.enqueue(subscription_ids=(703,), source=source, priority=priority)
+    first = repository.claim_next(owner="initial")
+    ready_at = (datetime.now(timezone.utc) - timedelta(seconds=1)).isoformat(timespec="seconds")
+    later_at = (datetime.now(timezone.utc) + timedelta(hours=6)).isoformat(timespec="seconds")
+    assert repository.defer_task(
+        task_id=first.task_id, lease_token=first.lease_token,
+        available_at=ready_at if running else later_at,
+        message="站点冷却中", pending_site_ids=(9,),
+    )
+    if running:
+        assert repository.claim_next(owner="resuming").task_id == first.task_id
+    renewed = repository.enqueue(
+        subscription_ids=(703,), source="fallback", priority=10,
+        available_at_by_subscription={703: ready_at}, refresh_pending=True,
+    )
+
+    assert renewed.created_count == 0
+    assert renewed.coalesced_count == 1
+    assert renewed.active_batch_ids == (original.batch.batch_id,)
+    with Session(engine) as session:
+        task = session.scalar(select(SubscriptionSearchTask))
+        assert task.task_id == first.task_id
+        assert task.active_key == "subscription:703"
+        assert task.source == source
+        assert task.priority == priority
+        assert task.pending_site_ids == ([9] if running else None)
+        assert task.state == ("running" if running else "queued")
+        assert task.phase == ("matching" if running else "queued")
+        assert task.last_error is None
+        if not running:
+            assert task.available_at == ready_at
+    engine.dispose()

tests/test_subscription_site_budget.py

@@ -68,6 +68,35 @@ def test_site_budget_recovers_expired_inflight_lease(tmp_path):
     assert recovered.lease_token != first.lease_token
 
 
+def test_busy_site_does_not_hide_other_inflight_search(tmp_path):
+    """一个站点繁忙时，任务仍应展示其它已取得租约的站点正在搜索。"""
+    repository, engine = _repository(tmp_path)
+    busy_claim = repository.claim_site(site_id=1, owner="other-search", lease_seconds=900)
+    phases = []
+    budget = SubscriptionSiteBudget(
+        repository=repository,
+        owner="current-search",
+        cancelled=lambda: False,
+        stop_state=ProcessStopState(),
+        phase_changed=lambda phase, site_id: phases.append((phase, site_id)),
+    )
+    try:
+        searching_claim = budget.acquire(2)
+        with pytest.raises(SubscriptionSiteBudgetUnavailable) as deferred:
+            budget.acquire(1)
+        assert deferred.value.site_id == 1
+        assert phases[-1] == ("searching", 2)
+        assert budget.finish(searching_claim, SiteSearchObservation())
+        assert repository.finish_site(
+            site_id=1,
+            lease_token=busy_claim.lease_token,
+            outcome="success",
+            next_allowed_at=datetime.now(timezone.utc).isoformat(timespec="seconds"),
+        )
+    finally:
+        engine.dispose()
+
+
 def test_site_budget_applies_error_cooldown_and_gradual_success_recovery(tmp_path):
     """失败增加冷却计数，后续成功每次只恢复一级而非直接清零。"""
     repository, engine = _repository(tmp_path)