refactor(search): centralize keyword resolution

app/chain/search/media.py

@@ -3,7 +3,19 @@
 import asyncio
 import random
 import time
-from typing import Any, AsyncIterator, Dict, List, Optional, Tuple, cast
+from dataclasses import dataclass
+from typing import (
+    Any,
+    AsyncIterator,
+    Awaitable,
+    Callable,
+    Dict,
+    Generator,
+    List,
+    Optional,
+    Tuple,
+    cast,
+)
 
 from app.chain.media import MediaChain
 from app.chain.search.contract import _SearchOwnerBase as _SearchOwnerBase
@@ -76,6 +88,98 @@ def _should_stop_keyword_search(
     return not search_multiple_name and bool(torrents)
 
 
+@dataclass(frozen=True, slots=True)
+class _KeywordSearchRequest:
+    """描述共享关键字状态机交给真实 I/O 边界的一次请求。"""
+
+    keyword: str
+    search_count: int
+
+
+@dataclass(frozen=True, slots=True)
+class _KeywordSearchResult:
+    """冻结关键字状态机聚合结果及其提前停止决策。"""
+
+    torrents: List[TorrentInfo]
+    stopped_early: bool
+
+
+_KeywordSearchResolution = Generator[
+    _KeywordSearchRequest, List[TorrentInfo], _KeywordSearchResult
+]
+
+
+def _keyword_search_resolution(
+    keywords: List[str],
+    search_multiple_name: bool,
+) -> _KeywordSearchResolution:
+    """统一推进关键字顺序、结果聚合和首个有效结果短路。"""
+    torrents: List[TorrentInfo] = []
+    for search_count, keyword in enumerate(keywords):
+        torrents.extend(
+            (yield _KeywordSearchRequest(
+                keyword=keyword,
+                search_count=search_count,
+            ))
+        )
+        if _should_stop_keyword_search(search_multiple_name, torrents):
+            return _KeywordSearchResult(torrents=torrents, stopped_early=True)
+    return _KeywordSearchResult(torrents=torrents, stopped_early=False)
+
+
+def _run_keyword_search_sync(
+    resolution: _KeywordSearchResolution,
+    execute: Callable[[_KeywordSearchRequest], List[TorrentInfo]],
+) -> _KeywordSearchResult:
+    """通过同步 provider 驱动共享关键字状态机。"""
+    try:
+        request = next(resolution)
+    except StopIteration as outcome:
+        return cast(_KeywordSearchResult, outcome.value)
+    while True:
+        try:
+            request = resolution.send(execute(request))
+        except StopIteration as outcome:
+            return cast(_KeywordSearchResult, outcome.value)
+
+
+async def _run_keyword_search_async(
+    resolution: _KeywordSearchResolution,
+    execute: Callable[[_KeywordSearchRequest], Awaitable[List[TorrentInfo]]],
+) -> _KeywordSearchResult:
+    """通过异步 provider 驱动共享关键字状态机。"""
+    try:
+        request = next(resolution)
+    except StopIteration as outcome:
+        return cast(_KeywordSearchResult, outcome.value)
+    while True:
+        try:
+            request = resolution.send(await execute(request))
+        except StopIteration as outcome:
+            return cast(_KeywordSearchResult, outcome.value)
+
+
+def _build_result_params(
+    torrents: List[TorrentInfo],
+    mediainfo: MediaInfo,
+    keyword: Optional[str],
+    rule_groups: Optional[List[str]],
+    season_episodes: Optional[Dict[int, List[int]]],
+    custom_words: Optional[List[str]],
+    filter_params: Optional[Dict[str, str]],
+) -> Dict[str, Any]:
+    """构造同步、异步和流式结果解析共享的参数快照。"""
+    return {
+        "torrents": torrents,
+        "mediainfo": mediainfo,
+        "keyword": keyword,
+        "rule_groups": rule_groups,
+        "season_episodes": season_episodes,
+        "custom_words": custom_words,
+        "filter_params": filter_params,
+    }
+
+
 def _build_candidate_contexts(
     mediainfo: MediaInfo,
     torrents: List[TorrentInfo],
@@ -291,46 +395,45 @@ class SearchMediaOwner(_SearchOwnerBase):
         # 准备搜索参数
         season_episodes, keywords = self._prepare_params(mediainfo=mediainfo, keyword=keyword, no_exists=no_exists)
 
-        # 站点搜索结果
-        torrents: List[TorrentInfo] = []
-        # 站点搜索次数
-        search_count = 0
-
-        # 多关键字执行搜索
-        for search_word in keywords:
-            # 强制休眠 1-10 秒
-            if search_count > 0:
-                logger.info(f"已搜索 {search_count} 次，强制休眠 1-10 秒 ...")
+        def execute_search(request: _KeywordSearchRequest) -> List[TorrentInfo]:
+            """执行共享状态机请求的同步站点搜索。"""
+            if request.search_count > 0:
+                logger.info(
+                    f"已搜索 {request.search_count} 次，强制休眠 1-10 秒 ..."
+                )
                 time.sleep(random.randint(1, 10))
-
-            # 搜索站点
-            results = (
-                self._SearchChain__search_all_sites(mediainfo=mediainfo, keyword=search_word, sites=sites, area=area)
+            return (
+                self._SearchChain__search_all_sites(
+                    mediainfo=mediainfo,
+                    keyword=request.keyword,
+                    sites=sites,
+                    area=area,
+                )
                 or []
             )
-            # 合并结果
 
-            search_count += 1
-            torrents.extend(results)
-
-            # 有结果则停止
-            if _should_stop_keyword_search(
-                self.runtime_config.search_multiple_name, torrents
-            ):
-                logger.info(f"共搜索到 {len(torrents)} 个资源，停止搜索")
-                break
+        outcome = _run_keyword_search_sync(
+            _keyword_search_resolution(
+                keywords, self.runtime_config.search_multiple_name
+            ),
+            execute_search,
+        )
+        if outcome.stopped_early:
+            logger.info(f"共搜索到 {len(outcome.torrents)} 个资源，停止搜索")
 
         # 处理结果
         return cast(
             List[Context],
             self._parse_result(
-                torrents=torrents,
-                mediainfo=mediainfo,
-                keyword=keyword,
-                rule_groups=rule_groups,
-                season_episodes=season_episodes,
-                custom_words=custom_words,
-                filter_params=filter_params,
+                **_build_result_params(
+                    torrents=outcome.torrents,
+                    mediainfo=mediainfo,
+                    keyword=keyword,
+                    rule_groups=rule_groups,
+                    season_episodes=season_episodes,
+                    custom_words=custom_words,
+                    filter_params=filter_params,
+                )
             ),
         )
 
@@ -392,44 +495,48 @@ class SearchMediaOwner(_SearchOwnerBase):
         # 准备搜索参数
         season_episodes, keywords = self._prepare_params(mediainfo=mediainfo, keyword=keyword, no_exists=no_exists)
 
-        # 站点搜索结果
-        torrents: List[TorrentInfo] = []
-        # 站点搜索次数
-        search_count = 0
-
-        # 多关键字执行搜索
-        for search_word in keywords:
-            # 强制休眠 1-10 秒
-            if search_count > 0:
-                logger.info(f"已搜索 {search_count} 次，强制休眠 1-10 秒 ...")
+        async def execute_search(
+            request: _KeywordSearchRequest,
+        ) -> List[TorrentInfo]:
+            """执行共享状态机请求的异步站点搜索。"""
+            if request.search_count > 0:
+                logger.info(
+                    f"已搜索 {request.search_count} 次，强制休眠 1-10 秒 ..."
+                )
                 await asyncio.sleep(random.randint(1, 10))
-            # 搜索站点
-            torrents.extend(
+            return (
                 await self._SearchChain__async_search_all_sites(
-                    mediainfo=mediainfo, keyword=search_word, sites=sites, area=area
+                    mediainfo=mediainfo,
+                    keyword=request.keyword,
+                    sites=sites,
+                    area=area,
                 )
                 or []
             )
-            search_count += 1
-            # 未开启多名称搜索时，有结果则停止
-            if _should_stop_keyword_search(
-                self.runtime_config.search_multiple_name, torrents
-            ):
-                logger.info(f"共搜索到 {len(torrents)} 个资源，停止搜索")
-                break
+
+        outcome = await _run_keyword_search_async(
+            _keyword_search_resolution(
+                keywords, self.runtime_config.search_multiple_name
+            ),
+            execute_search,
+        )
+        if outcome.stopped_early:
+            logger.info(f"共搜索到 {len(outcome.torrents)} 个资源，停止搜索")
 
         # 处理结果
         return cast(
             List[Context],
             await run_in_threadpool(
                 self._parse_result,
-                torrents=torrents,
-                mediainfo=mediainfo,
-                keyword=keyword,
-                rule_groups=rule_groups,
-                season_episodes=season_episodes,
-                custom_words=custom_words,
-                filter_params=filter_params,
+                **_build_result_params(
+                    torrents=outcome.torrents,
+                    mediainfo=mediainfo,
+                    keyword=keyword,
+                    rule_groups=rule_groups,
+                    season_episodes=season_episodes,
+                    custom_words=custom_words,
+                    filter_params=filter_params,
+                ),
             ),
         )
 
@@ -482,53 +589,69 @@ class SearchMediaOwner(_SearchOwnerBase):
         # 准备搜索参数
         season_episodes, keywords = self._prepare_params(mediainfo=mediainfo, keyword=keyword, no_exists=no_exists)
 
-        torrents: List[TorrentInfo] = []
         candidate_contexts: List[Context] = []
-        search_count = 0
-
-        for search_word in keywords:
-            if search_count > 0:
-                logger.info(f"已搜索 {search_count} 次，强制休眠 1-10 秒 ...")
-                await asyncio.sleep(random.randint(1, 10))
-
-            async for event in self._SearchChain__async_search_all_sites_stream(
-                mediainfo=mediainfo, keyword=search_word, sites=sites, area=area
-            ):
-                result = event.pop("items", []) or []
-                torrents.extend(result)
-                batch_contexts = _build_candidate_contexts(mediainfo, result)
-                candidate_contexts.extend(batch_contexts)
-                yield {
-                    **event,
-                    "type": "append",
-                    "stage": "searching",
-                    "items": [cast(Any, context).to_dict() for context in batch_contexts],
-                    "total_items": len(candidate_contexts),
-                }
-
-            search_count += 1
-            if _should_stop_keyword_search(
-                self.runtime_config.search_multiple_name, torrents
-            ):
-                logger.info(f"共搜索到 {len(torrents)} 个资源，停止搜索")
-                break
+        resolution = _keyword_search_resolution(
+            keywords, self.runtime_config.search_multiple_name
+        )
+        outcome = _KeywordSearchResult(torrents=[], stopped_early=False)
+        try:
+            request = next(resolution)
+        except StopIteration as completed:
+            outcome = cast(_KeywordSearchResult, completed.value)
+        else:
+            while True:
+                if request.search_count > 0:
+                    logger.info(
+                        f"已搜索 {request.search_count} 次，强制休眠 1-10 秒 ..."
+                    )
+                    await asyncio.sleep(random.randint(1, 10))
+                search_results: List[TorrentInfo] = []
+                async for event in self._SearchChain__async_search_all_sites_stream(
+                    mediainfo=mediainfo,
+                    keyword=request.keyword,
+                    sites=sites,
+                    area=area,
+                ):
+                    result = event.pop("items", []) or []
+                    search_results.extend(result)
+                    batch_contexts = _build_candidate_contexts(mediainfo, result)
+                    candidate_contexts.extend(batch_contexts)
+                    yield {
+                        **event,
+                        "type": "append",
+                        "stage": "searching",
+                        "items": [
+                            cast(Any, context).to_dict()
+                            for context in batch_contexts
+                        ],
+                        "total_items": len(candidate_contexts),
+                    }
+                try:
+                    request = resolution.send(search_results)
+                except StopIteration as completed:
+                    outcome = cast(_KeywordSearchResult, completed.value)
+                    break
+        if outcome.stopped_early:
+            logger.info(f"共搜索到 {len(outcome.torrents)} 个资源，停止搜索")
 
         yield {
             "type": "progress",
             "stage": "filtering",
             "value": 98,
-            "text": f"正在过滤匹配 {len(torrents)} 个候选资源 ...",
+            "text": f"正在过滤匹配 {len(outcome.torrents)} 个候选资源 ...",
         }
 
         contexts = await run_in_threadpool(
             self._parse_result,
-            torrents=torrents,
-            mediainfo=mediainfo,
-            keyword=keyword,
-            rule_groups=rule_groups,
-            season_episodes=season_episodes,
-            custom_words=custom_words,
-            filter_params=filter_params,
+            **_build_result_params(
+                torrents=outcome.torrents,
+                mediainfo=mediainfo,
+                keyword=keyword,
+                rule_groups=rule_groups,
+                season_episodes=season_episodes,
+                custom_words=custom_words,
+                filter_params=filter_params,
+            ),
         )
         final_items = [context.to_dict() for context in contexts]
         yield {

app/chain/search/title.py

@@ -1,5 +1,6 @@
 """标题搜索入口与标题候选过滤 owner。"""
 
+from dataclasses import dataclass
 from typing import Any, AsyncIterator, Callable, Dict, List, Optional, Tuple, cast
 
 from app.application.configuration import (
@@ -57,6 +58,46 @@ def _build_title_contexts(
     ]
 
 
+@dataclass(frozen=True, slots=True)
+class _TitleSearchResult:
+    """封装标题搜索候选解析结果及需要记录的失败原因。"""
+
+    contexts: List[Context]
+    warning: Optional[str] = None
+
+
+def _resolve_title_search_result(
+    title: str,
+    torrents: List[TorrentInfo],
+    rule_groups: Optional[List[str]],
+    mtype: Optional[MediaType],
+    filter_torrents: Callable[..., List[TorrentInfo]],
+    build_meta: Callable[[TorrentInfo, Optional[MediaType]], Any],
+) -> _TitleSearchResult:
+    """统一标题候选为空、过滤和上下文投影的结果决策。"""
+    if not torrents:
+        return _TitleSearchResult(
+            contexts=[],
+            warning=f"{title} 未搜索到资源",
+        )
+    filtered_torrents = filter_torrents(
+        torrents=torrents,
+        rule_groups=rule_groups,
+    )
+    if not filtered_torrents:
+        return _TitleSearchResult(
+            contexts=[],
+            warning=f"{title} 没有符合过滤规则的资源",
+        )
+    return _TitleSearchResult(
+        contexts=_build_title_contexts(
+            filtered_torrents,
+            mtype,
+            build_meta,
+        )
+    )
+
+
 class SearchTitleOwner(_SearchOwnerBase):
     """标题搜索入口与标题候选过滤 owner。"""
 
@@ -92,25 +133,21 @@ class SearchTitleOwner(_SearchOwnerBase):
         else:
             logger.info(f"开始浏览资源，站点：{sites} ...")
         # 搜索
-        torrents = self._SearchChain__search_all_sites(**search_params) or []
-        if not torrents:
-            logger.warning(f"{title} 未搜索到资源")
-            return []
-        torrents = self._filter_title_search_torrents(
-            torrents=torrents,
+        result = _resolve_title_search_result(
+            title=title,
+            torrents=self._SearchChain__search_all_sites(**search_params) or [],
             rule_groups=rule_groups,
+            mtype=mtype,
+            filter_torrents=self._filter_title_search_torrents,
+            build_meta=self._build_title_search_meta,
         )
-        if not torrents:
-            logger.warning(f"{title} 没有符合过滤规则的资源")
+        if result.warning:
+            logger.warning(result.warning)
             return []
-        # 组装上下文
-        contexts = _build_title_contexts(
-            torrents, mtype, self._build_title_search_meta
-        )
         # 保存到本地文件
         if cache_local:
-            self._save_results(contexts)
-        return contexts
+            self._save_results(result.contexts)
+        return result.contexts
 
     async def async_search_by_title(
         self,
@@ -144,26 +181,38 @@ class SearchTitleOwner(_SearchOwnerBase):
         else:
             logger.info(f"开始浏览资源，站点：{sites} ...")
         # 搜索
-        torrents = await self._SearchChain__async_search_all_sites(**search_params) or []
-        if not torrents:
-            logger.warning(f"{title} 未搜索到资源")
-            return []
-        torrents = await run_in_threadpool(
-            self._filter_title_search_torrents,
-            torrents=torrents,
-            rule_groups=rule_groups,
+        torrents = (
+            await self._SearchChain__async_search_all_sites(**search_params) or []
         )
         if not torrents:
-            logger.warning(f"{title} 没有符合过滤规则的资源")
+            result = _resolve_title_search_result(
+                title=title,
+                torrents=torrents,
+                rule_groups=rule_groups,
+                mtype=mtype,
+                filter_torrents=self._filter_title_search_torrents,
+                build_meta=self._build_title_search_meta,
+            )
+        else:
+            result = cast(
+                _TitleSearchResult,
+                await run_in_threadpool(
+                    _resolve_title_search_result,
+                    title=title,
+                    torrents=torrents,
+                    rule_groups=rule_groups,
+                    mtype=mtype,
+                    filter_torrents=self._filter_title_search_torrents,
+                    build_meta=self._build_title_search_meta,
+                ),
+            )
+        if result.warning:
+            logger.warning(result.warning)
             return []
-        # 组装上下文
-        contexts = _build_title_contexts(
-            torrents, mtype, self._build_title_search_meta
-        )
         # 保存到本地文件
         if cache_local:
-            await self._async_save_results(contexts)
-        return contexts
+            await self._async_save_results(result.contexts)
+        return result.contexts
 
     async def async_search_by_title_stream(
         self,

tests/test_search_sync_async_parity.py

@@ -6,6 +6,7 @@ from types import SimpleNamespace
 from typing import Any
 
 from app.chain.search import media as media_module
+from app.chain.search import title as title_module
 from app.chain.search.facade import SearchChain
 from app.domain.context import MediaInfo, TorrentInfo
 from app.domain.metainfo import MetaInfo
@@ -177,6 +178,177 @@ def test_media_process_sync_async_share_keyword_stop_decision(monkeypatch):
     assert sync_result == async_result == [found]
 
 
+def test_keyword_resolution_searches_all_names_when_enabled():
+    """开启多名称搜索时共享状态机应完整执行并稳定聚合各关键字结果。"""
+    first = _torrent("First Result")
+    third = _torrent("Third Result")
+    expected = {
+        "first": [first],
+        "second": [],
+        "third": [third],
+    }
+    sync_keywords: list[str] = []
+    async_keywords: list[str] = []
+
+    def execute_sync(request):
+        """记录同步驱动顺序并返回关键字结果。"""
+        sync_keywords.append(request.keyword)
+        return expected[request.keyword]
+
+    async def execute_async(request):
+        """记录异步驱动顺序并返回关键字结果。"""
+        async_keywords.append(request.keyword)
+        return expected[request.keyword]
+
+    keywords = ["first", "second", "third"]
+    sync_result = media_module._run_keyword_search_sync(
+        media_module._keyword_search_resolution(keywords, search_multiple_name=True),
+        execute_sync,
+    )
+    async_result = asyncio.run(
+        media_module._run_keyword_search_async(
+            media_module._keyword_search_resolution(
+                keywords, search_multiple_name=True
+            ),
+            execute_async,
+        )
+    )
+
+    assert sync_keywords == async_keywords == keywords
+    assert sync_result == async_result
+    assert sync_result.torrents == [first, third]
+    assert sync_result.stopped_early is False
+
+
+def test_media_process_stream_shares_keyword_order_and_stop_decision(monkeypatch):
+    """流入口应复用双入口的关键字顺序、退避次数和首结果短路决策。"""
+    chain = _chain()
+    target = _media()
+    found = _torrent("测试电影 2026 1080p")
+    stream_keywords: list[str] = []
+    sleeps: list[int] = []
+    parsed: list[list[TorrentInfo]] = []
+
+    class FakeMediaChain:
+        """为流处理提供无需外部服务的附加信息结果。"""
+
+        async def async_supplement_media_info(self, mediainfo):
+            """返回异步附加信息。"""
+            return mediainfo
+
+    async def search_stream(**kwargs):
+        """第二个关键字输出资源，供共享状态机执行短路。"""
+        keyword = kwargs["keyword"]
+        stream_keywords.append(keyword)
+        yield {
+            "site_id": len(stream_keywords),
+            "items": [found] if keyword == "second" else [],
+        }
+
+    async def fake_async_sleep(delay):
+        """记录流关键字退避但不实际等待。"""
+        sleeps.append(delay)
+
+    def parse_result(**kwargs):
+        """记录流最终过滤前的资源顺序并返回空上下文。"""
+        parsed.append(list(kwargs["torrents"]))
+        return []
+
+    async def collect_events():
+        """完整消费媒体搜索流。"""
+        return [
+            event
+            async for event in chain.async_process_stream(mediainfo=target)
+        ]
+
+    monkeypatch.setattr(media_module, "MediaChain", FakeMediaChain)
+    monkeypatch.setattr(media_module.random, "randint", lambda _start, _end: 1)
+    monkeypatch.setattr(media_module.asyncio, "sleep", fake_async_sleep)
+    chain.runtime_config = SimpleNamespace(search_multiple_name=False)
+    chain._copy_media_input = deepcopy
+    chain._prepare_params = lambda **_kwargs: (None, ["first", "second", "third"])
+    chain._SearchChain__async_search_all_sites_stream = search_stream
+    chain._parse_result = parse_result
+
+    events = asyncio.run(collect_events())
+
+    assert stream_keywords == ["first", "second"]
+    assert sleeps == [1]
+    assert parsed == [[found]]
+    assert [event["type"] for event in events] == [
+        "append",
+        "append",
+        "progress",
+        "replace",
+        "done",
+    ]
+    assert events[-1]["candidate_items"] == 1
+
+
+def test_media_process_empty_keyword_plan_skips_all_provider_io(monkeypatch):
+    """空关键字计划应由共享状态机直接完成，且三入口均不触发 provider。"""
+    chain = _chain()
+    target = _media()
+    parsed: list[list[TorrentInfo]] = []
+
+    class FakeMediaChain:
+        """为三入口提供稳定的媒体附加信息。"""
+
+        def supplement_media_info(self, mediainfo):
+            """返回同步附加信息。"""
+            return mediainfo
+
+        async def async_supplement_media_info(self, mediainfo):
+            """返回异步附加信息。"""
+            return mediainfo
+
+    def unexpected_provider(**_kwargs):
+        """标记空计划错误触发了同步 provider。"""
+        raise AssertionError("空关键字计划不应触发同步 provider")
+
+    async def unexpected_async_provider(**_kwargs):
+        """标记空计划错误触发了异步 provider。"""
+        raise AssertionError("空关键字计划不应触发异步 provider")
+
+    async def unexpected_stream_provider(**_kwargs):
+        """标记空计划错误触发了流式 provider。"""
+        yield {"items": []}
+        raise AssertionError("空关键字计划不应触发流式 provider")
+
+    def parse_result(**kwargs):
+        """记录三入口交付给结果解析边界的空资源。"""
+        parsed.append(list(kwargs["torrents"]))
+        return []
+
+    async def collect_events():
+        """完整消费空关键字媒体搜索流。"""
+        return [
+            event
+            async for event in chain.async_process_stream(mediainfo=target)
+        ]
+
+    monkeypatch.setattr(media_module, "MediaChain", FakeMediaChain)
+    chain.runtime_config = SimpleNamespace(search_multiple_name=False)
+    chain._copy_media_input = deepcopy
+    chain._prepare_params = lambda **_kwargs: (None, [])
+    chain._SearchChain__search_all_sites = unexpected_provider
+    chain._SearchChain__async_search_all_sites = unexpected_async_provider
+    chain._SearchChain__async_search_all_sites_stream = unexpected_stream_provider
+    chain._parse_result = parse_result
+
+    assert chain.process(target) == []
+    assert asyncio.run(chain.async_process(target)) == []
+    events = asyncio.run(collect_events())
+
+    assert parsed == [[], [], []]
+    assert [event["type"] for event in events] == [
+        "progress",
+        "replace",
+        "done",
+    ]
+    assert events[-1]["candidate_items"] == 0
+
+
 def test_id_search_stream_failure_preserves_order_and_skips_result_cache(
     monkeypatch,
 ):
@@ -283,6 +455,73 @@ def test_title_search_sync_async_share_plan_filter_and_projection():
     ] == [f"mapped:{keep.title}"]
 
 
+def test_title_search_sync_async_share_empty_result_decisions(monkeypatch):
+    """标题双入口应共享无候选与过滤为空的警告，并跳过结果缓存。"""
+    chain = _chain()
+    candidate = _torrent("Movie 2026 1080p")
+    provider_result: list[TorrentInfo] = []
+    warnings: list[str] = []
+    filter_calls: list[list[TorrentInfo]] = []
+    saved_results: list[list[Any]] = []
+
+    def search_all_sites(**_kwargs):
+        """返回当前场景的同步候选快照。"""
+        return list(provider_result)
+
+    async def async_search_all_sites(**_kwargs):
+        """返回当前场景的异步候选快照。"""
+        return list(provider_result)
+
+    def filter_torrents(torrents, rule_groups=None):
+        """记录过滤输入并模拟候选全部被规则排除。"""
+        assert rule_groups == ["quality"]
+        filter_calls.append(list(torrents))
+        return []
+
+    async def save_params(**_kwargs):
+        """模拟异步请求参数缓存。"""
+
+    async def save_results(contexts):
+        """记录不应发生的异步结果缓存。"""
+        saved_results.append(list(contexts))
+
+    monkeypatch.setattr(title_module.logger, "warning", warnings.append)
+    chain._SearchChain__search_all_sites = search_all_sites
+    chain._SearchChain__async_search_all_sites = async_search_all_sites
+    chain._filter_title_search_torrents = filter_torrents
+    chain._build_title_search_meta = lambda torrent, _mtype: MetaInfo(
+        title=torrent.title
+    )
+    chain.cancel_ai_recommend = lambda: None
+    chain.save_last_search_params = lambda **_kwargs: None
+    chain.async_save_last_search_params = save_params
+    chain._save_results = lambda contexts: saved_results.append(list(contexts))
+    chain._async_save_results = save_results
+
+    for torrents, warning in (
+        ([], "Movie 未搜索到资源"),
+        ([candidate], "Movie 没有符合过滤规则的资源"),
+    ):
+        provider_result[:] = torrents
+        warning_count = len(warnings)
+        assert chain.search_by_title(
+            title="Movie",
+            cache_local=True,
+            rule_groups=["quality"],
+        ) == []
+        assert asyncio.run(
+            chain.async_search_by_title(
+                title="Movie",
+                cache_local=True,
+                rule_groups=["quality"],
+            )
+        ) == []
+        assert warnings[warning_count:] == [warning, warning]
+
+    assert filter_calls == [[candidate], [candidate]]
+    assert saved_results == []
+
+
 def test_title_stream_preserves_provider_and_result_order():
     """标题流应按 provider 批次顺序追加，并以相同顺序发布累计完成结果。"""
     chain = _chain()