refactor(search): unify sync async decisions

app/chain/search/media.py

@@ -3,7 +3,7 @@
 import asyncio
 import random
 import time
-from typing import Any, AsyncIterator, Dict, List, Optional, cast
+from typing import Any, AsyncIterator, Dict, List, Optional, Tuple, cast
 
 from app.chain.media import MediaChain
 from app.chain.search.contract import _SearchOwnerBase as _SearchOwnerBase
@@ -19,6 +19,80 @@ from app.schemas.types import (
 )
 
 
+def _build_id_search_params(
+    media_source: MediaSource,
+    media_id: str,
+    mtype: Optional[MediaType],
+    area: Optional[str],
+    season: Optional[int],
+    sites: Optional[List[int]],
+    music_type: Optional[str],
+) -> Tuple[Dict[str, Any], Dict[str, Any]]:
+    """构造同步、异步和流式 ID 搜索共享的识别与缓存参数。"""
+    recognition_params = {
+        "media_source": media_source,
+        "media_id": media_id,
+        "mtype": mtype,
+        "music_type": music_type,
+    }
+    cache_params = {
+        **recognition_params,
+        "area": area,
+        "season": season,
+        "sites": sites,
+    }
+    return recognition_params, cache_params
+
+
+def _build_missing_media_map(
+    mediainfo: MediaInfo,
+    season: Optional[int],
+) -> Optional[Dict[str, Dict[int, NotExistMediaInfo]]]:
+    """将指定季转换为统一的缺集搜索映射。"""
+    if season is None:
+        return None
+    resolved_source, resolved_id = resolve_media_identity(media=mediainfo)
+    return {
+        build_media_key(resolved_source, resolved_id): {
+            season: NotExistMediaInfo(episodes=[])
+        }
+    }
+
+
+def _normalize_media_search_input(mediainfo: MediaInfo) -> MediaInfo:
+    """归一化非 TMDB 输入标题，并保留调用方对象由外层复制的所有权约束。"""
+    if not mediainfo.tmdb_id:
+        meta = MetaInfo(title=mediainfo.title)
+        mediainfo.title = meta.name
+        mediainfo.season = cast(int, meta.begin_season)
+    return mediainfo
+
+
+def _should_stop_keyword_search(
+    search_multiple_name: bool,
+    torrents: List[TorrentInfo],
+) -> bool:
+    """统一判断首个有效关键字结果是否终止后续搜索。"""
+    return not search_multiple_name and bool(torrents)
+
+
+def _build_candidate_contexts(
+    mediainfo: MediaInfo,
+    torrents: List[TorrentInfo],
+) -> List[Context]:
+    """将流式站点候选映射为尚未精确过滤的搜索上下文。"""
+    return [
+        Context(
+            meta_info=MetaInfo(title=torrent.title, subtitle=torrent.description),
+            media_info=mediainfo,
+            torrent_info=torrent,
+            resource_source="search",
+            media_info_is_target=True,
+        )
+        for torrent in torrents
+    ]
+
+
 class SearchMediaOwner(_SearchOwnerBase):
     """精确媒体搜索与同步异步编排 owner。"""
 
@@ -44,31 +118,24 @@ class SearchMediaOwner(_SearchOwnerBase):
         :param sites: 站点ID列表
         :param cache_local: 是否缓存到本地
         """
-        if cache_local:
-            self.cancel_ai_recommend()
-            self.save_last_search_params(
-                media_source=media_source,
-                media_id=media_id,
-                mtype=mtype,
-                area=area,
-                season=season,
-                sites=sites,
-                music_type=music_type,
-            )
-        # 音乐统一在 MediaChain.recognize_media 内按固定来源路由
-        mediainfo = MediaChain().recognize_media(
+        recognition_params, cache_params = _build_id_search_params(
             media_source=media_source,
             media_id=media_id,
             mtype=mtype,
+            area=area,
+            season=season,
+            sites=sites,
             music_type=music_type,
         )
+        if cache_local:
+            self.cancel_ai_recommend()
+            self.save_last_search_params(**cache_params)
+        # 音乐统一在 MediaChain.recognize_media 内按固定来源路由
+        mediainfo = MediaChain().recognize_media(**recognition_params)
         if not mediainfo:
             logger.error(f"{self._build_search_keyword(media_source, media_id)} 媒体信息识别失败！")
             return []
-        no_exists: Optional[Dict[str, Dict[int, NotExistMediaInfo]]] = None
-        if season is not None:
-            resolved_source, resolved_id = resolve_media_identity(media=mediainfo)
-            no_exists = {build_media_key(resolved_source, resolved_id): {season: NotExistMediaInfo(episodes=[])}}
+        no_exists = _build_missing_media_map(mediainfo, season)
         results = self.process(mediainfo=mediainfo, sites=sites, area=area, no_exists=no_exists)
         # 保存到本地文件
         if cache_local:
@@ -97,31 +164,24 @@ class SearchMediaOwner(_SearchOwnerBase):
         :param sites: 站点ID列表
         :param cache_local: 是否缓存到本地
         """
-        if cache_local:
-            self.cancel_ai_recommend()
-            await self.async_save_last_search_params(
-                media_source=media_source,
-                media_id=media_id,
-                mtype=mtype,
-                area=area,
-                season=season,
-                sites=sites,
-                music_type=music_type,
-            )
-        # 音乐统一在 MediaChain.async_recognize_media 内按固定来源路由
-        mediainfo = await MediaChain().async_recognize_media(
+        recognition_params, cache_params = _build_id_search_params(
             media_source=media_source,
             media_id=media_id,
             mtype=mtype,
+            area=area,
+            season=season,
+            sites=sites,
             music_type=music_type,
         )
+        if cache_local:
+            self.cancel_ai_recommend()
+            await self.async_save_last_search_params(**cache_params)
+        # 音乐统一在 MediaChain.async_recognize_media 内按固定来源路由
+        mediainfo = await MediaChain().async_recognize_media(**recognition_params)
         if not mediainfo:
             logger.error(f"{self._build_search_keyword(media_source, media_id)} 媒体信息识别失败！")
             return []
-        no_exists: Optional[Dict[str, Dict[int, NotExistMediaInfo]]] = None
-        if season is not None:
-            resolved_source, resolved_id = resolve_media_identity(media=mediainfo)
-            no_exists = {build_media_key(resolved_source, resolved_id): {season: NotExistMediaInfo(episodes=[])}}
+        no_exists = _build_missing_media_map(mediainfo, season)
         results = await self.async_process(mediainfo=mediainfo, sites=sites, area=area, no_exists=no_exists)
         # 保存到本地文件
         if cache_local:
@@ -142,33 +202,26 @@ class SearchMediaOwner(_SearchOwnerBase):
         """
         根据数据源媒体 ID 渐进式搜索资源，先返回站点原始候选，再返回过滤匹配后的最终结果
         """
-        if cache_local:
-            self.cancel_ai_recommend()
-            await self.async_save_last_search_params(
-                media_source=media_source,
-                media_id=media_id,
-                mtype=mtype,
-                area=area,
-                season=season,
-                sites=sites,
-                music_type=music_type,
-            )
-        # 音乐统一在 MediaChain.async_recognize_media 内按固定来源路由
-        mediainfo = await MediaChain().async_recognize_media(
+        recognition_params, cache_params = _build_id_search_params(
             media_source=media_source,
             media_id=media_id,
             mtype=mtype,
+            area=area,
+            season=season,
+            sites=sites,
             music_type=music_type,
         )
+        if cache_local:
+            self.cancel_ai_recommend()
+            await self.async_save_last_search_params(**cache_params)
+        # 音乐统一在 MediaChain.async_recognize_media 内按固定来源路由
+        mediainfo = await MediaChain().async_recognize_media(**recognition_params)
         if not mediainfo:
             logger.error(f"{self._build_search_keyword(media_source, media_id)} 媒体信息识别失败！")
             yield {"type": "error", "success": False, "message": "媒体信息识别失败"}
             return
 
-        no_exists: Optional[Dict[str, Dict[int, NotExistMediaInfo]]] = None
-        if season is not None:
-            resolved_source, resolved_id = resolve_media_identity(media=mediainfo)
-            no_exists = {build_media_key(resolved_source, resolved_id): {season: NotExistMediaInfo(episodes=[])}}
+        no_exists = _build_missing_media_map(mediainfo, season)
 
         contexts: List[Context] = []
         async for event in self.async_process_stream(mediainfo=mediainfo, sites=sites, area=area, no_exists=no_exists):
@@ -215,12 +268,7 @@ class SearchMediaOwner(_SearchOwnerBase):
                 ),
             )
 
-        mediainfo = self._copy_media_input(mediainfo)
-        # 豆瓣标题处理
-        if not mediainfo.tmdb_id:
-            meta = MetaInfo(title=mediainfo.title)
-            mediainfo.title = meta.name
-            mediainfo.season = cast(int, meta.begin_season)
+        mediainfo = _normalize_media_search_input(self._copy_media_input(mediainfo))
         logger.info(f"开始搜索资源，关键词：{keyword or mediainfo.title} ...")
 
         # 补充媒体信息
@@ -266,7 +314,9 @@ class SearchMediaOwner(_SearchOwnerBase):
             torrents.extend(results)
 
             # 有结果则停止
-            if not self.runtime_config.search_multiple_name and torrents:
+            if _should_stop_keyword_search(
+                self.runtime_config.search_multiple_name, torrents
+            ):
                 logger.info(f"共搜索到 {len(torrents)} 个资源，停止搜索")
                 break
 
@@ -319,12 +369,7 @@ class SearchMediaOwner(_SearchOwnerBase):
                 ),
             )
 
-        mediainfo = self._copy_media_input(mediainfo)
-        # 豆瓣标题处理
-        if not mediainfo.tmdb_id:
-            meta = MetaInfo(title=mediainfo.title)
-            mediainfo.title = meta.name
-            mediainfo.season = cast(int, meta.begin_season)
+        mediainfo = _normalize_media_search_input(self._copy_media_input(mediainfo))
         logger.info(f"开始搜索资源，关键词：{keyword or mediainfo.title} ...")
 
         # 补充媒体信息
@@ -367,7 +412,9 @@ class SearchMediaOwner(_SearchOwnerBase):
             )
             search_count += 1
             # 未开启多名称搜索时，有结果则停止
-            if not self.runtime_config.search_multiple_name and torrents:
+            if _should_stop_keyword_search(
+                self.runtime_config.search_multiple_name, torrents
+            ):
                 logger.info(f"共搜索到 {len(torrents)} 个资源，停止搜索")
                 break
 
@@ -412,12 +459,7 @@ class SearchMediaOwner(_SearchOwnerBase):
                 yield event
             return
 
-        mediainfo = self._copy_media_input(mediainfo)
-        # 豆瓣标题处理
-        if not mediainfo.tmdb_id:
-            meta = MetaInfo(title=mediainfo.title)
-            mediainfo.title = meta.name
-            mediainfo.season = cast(int, meta.begin_season)
+        mediainfo = _normalize_media_search_input(self._copy_media_input(mediainfo))
         logger.info(f"开始渐进式搜索资源，关键词：{keyword or mediainfo.title} ...")
 
         # 补充媒体信息
@@ -454,16 +496,7 @@ class SearchMediaOwner(_SearchOwnerBase):
             ):
                 result = event.pop("items", []) or []
                 torrents.extend(result)
-                batch_contexts = [
-                    Context(
-                        meta_info=MetaInfo(title=torrent.title, subtitle=torrent.description),
-                        media_info=mediainfo,
-                        torrent_info=torrent,
-                        resource_source="search",
-                        media_info_is_target=True,
-                    )
-                    for torrent in result
-                ]
+                batch_contexts = _build_candidate_contexts(mediainfo, result)
                 candidate_contexts.extend(batch_contexts)
                 yield {
                     **event,
@@ -474,7 +507,9 @@ class SearchMediaOwner(_SearchOwnerBase):
                 }
 
             search_count += 1
-            if not self.runtime_config.search_multiple_name and torrents:
+            if _should_stop_keyword_search(
+                self.runtime_config.search_multiple_name, torrents
+            ):
                 logger.info(f"共搜索到 {len(torrents)} 个资源，停止搜索")
                 break
 

app/chain/search/title.py

@@ -1,6 +1,6 @@
 """标题搜索入口与标题候选过滤 owner。"""
 
-from typing import Any, AsyncIterator, Callable, Dict, List, Optional, cast
+from typing import Any, AsyncIterator, Callable, Dict, List, Optional, Tuple, cast
 
 from app.application.configuration import (
     get_configured_system_config,
@@ -17,6 +17,46 @@ from app.schemas.types import (
 )
 
 
+def _build_title_search_params(
+    title: str,
+    page: Optional[int],
+    sites: Optional[List[int]],
+    mtype: Optional[MediaType],
+    include_empty_type: bool = False,
+) -> Tuple[Dict[str, Any], Dict[str, Any]]:
+    """构造标题搜索共享的 provider 与缓存参数。"""
+    provider_params: Dict[str, Any] = {
+        "keyword": title,
+        "sites": sites,
+        "page": page,
+    }
+    if mtype is not None or include_empty_type:
+        provider_params["mtype"] = mtype
+    cache_params = {
+        "keyword": title,
+        "mtype": mtype,
+        "area": "title",
+        "sites": sites,
+    }
+    return provider_params, cache_params
+
+
+def _build_title_contexts(
+    torrents: List[TorrentInfo],
+    mtype: Optional[MediaType],
+    build_meta: Callable[[TorrentInfo, Optional[MediaType]], Any],
+) -> List[Context]:
+    """将过滤后的标题候选统一投影为搜索上下文。"""
+    return [
+        Context(
+            meta_info=build_meta(torrent, mtype),
+            torrent_info=torrent,
+            resource_source="search",
+        )
+        for torrent in torrents
+    ]
+
+
 class SearchTitleOwner(_SearchOwnerBase):
     """标题搜索入口与标题候选过滤 owner。"""
 
@@ -38,27 +78,21 @@ class SearchTitleOwner(_SearchOwnerBase):
         :param mtype: 限定站点资源分类
         :param rule_groups: 指定过滤规则组，为空时使用默认搜索过滤规则
         """
+        search_params, cache_params = _build_title_search_params(
+            title=title,
+            page=page,
+            sites=sites,
+            mtype=mtype,
+        )
         if cache_local:
             self.cancel_ai_recommend()
-            self.save_last_search_params(
-                keyword=title,
-                mtype=mtype,
-                area="title",
-                sites=sites,
-            )
+            self.save_last_search_params(**cache_params)
         if title:
             logger.info(f"开始搜索资源，关键词：{title} ...")
         else:
             logger.info(f"开始浏览资源，站点：{sites} ...")
         # 搜索
-        search_kwargs: Dict[str, Any] = {
-            "keyword": title,
-            "sites": sites,
-            "page": page,
-        }
-        if mtype is not None:
-            search_kwargs["mtype"] = mtype
-        torrents = self._SearchChain__search_all_sites(**search_kwargs) or []
+        torrents = self._SearchChain__search_all_sites(**search_params) or []
         if not torrents:
             logger.warning(f"{title} 未搜索到资源")
             return []
@@ -70,14 +104,9 @@ class SearchTitleOwner(_SearchOwnerBase):
             logger.warning(f"{title} 没有符合过滤规则的资源")
             return []
         # 组装上下文
-        contexts = [
-            Context(
-                meta_info=self._build_title_search_meta(torrent, mtype),
-                torrent_info=torrent,
-                resource_source="search",
-            )
-            for torrent in torrents
-        ]
+        contexts = _build_title_contexts(
+            torrents, mtype, self._build_title_search_meta
+        )
         # 保存到本地文件
         if cache_local:
             self._save_results(contexts)
@@ -101,27 +130,21 @@ class SearchTitleOwner(_SearchOwnerBase):
         :param mtype: 限定站点资源分类
         :param rule_groups: 指定过滤规则组，为空时使用默认搜索过滤规则
         """
+        search_params, cache_params = _build_title_search_params(
+            title=title,
+            page=page,
+            sites=sites,
+            mtype=mtype,
+        )
         if cache_local:
             self.cancel_ai_recommend()
-            await self.async_save_last_search_params(
-                keyword=title,
-                mtype=mtype,
-                area="title",
-                sites=sites,
-            )
+            await self.async_save_last_search_params(**cache_params)
         if title:
             logger.info(f"开始搜索资源，关键词：{title} ...")
         else:
             logger.info(f"开始浏览资源，站点：{sites} ...")
         # 搜索
-        search_kwargs: Dict[str, Any] = {
-            "keyword": title,
-            "sites": sites,
-            "page": page,
-        }
-        if mtype is not None:
-            search_kwargs["mtype"] = mtype
-        torrents = await self._SearchChain__async_search_all_sites(**search_kwargs) or []
+        torrents = await self._SearchChain__async_search_all_sites(**search_params) or []
         if not torrents:
             logger.warning(f"{title} 未搜索到资源")
             return []
@@ -134,14 +157,9 @@ class SearchTitleOwner(_SearchOwnerBase):
             logger.warning(f"{title} 没有符合过滤规则的资源")
             return []
         # 组装上下文
-        contexts = [
-            Context(
-                meta_info=self._build_title_search_meta(torrent, mtype),
-                torrent_info=torrent,
-                resource_source="search",
-            )
-            for torrent in torrents
-        ]
+        contexts = _build_title_contexts(
+            torrents, mtype, self._build_title_search_meta
+        )
         # 保存到本地文件
         if cache_local:
             await self._async_save_results(contexts)
@@ -159,14 +177,16 @@ class SearchTitleOwner(_SearchOwnerBase):
         """
         根据标题渐进式搜索资源，不识别媒体信息，按默认搜索过滤规则返回结果
         """
+        search_params, cache_params = _build_title_search_params(
+            title=title,
+            page=page,
+            sites=sites,
+            mtype=mtype,
+            include_empty_type=True,
+        )
         if cache_local:
             self.cancel_ai_recommend()
-            await self.async_save_last_search_params(
-                keyword=title,
-                mtype=mtype,
-                area="title",
-                sites=sites,
-            )
+            await self.async_save_last_search_params(**cache_params)
         if title:
             logger.info(f"开始渐进式搜索资源，关键词：{title} ...")
         else:
@@ -178,7 +198,7 @@ class SearchTitleOwner(_SearchOwnerBase):
         if rule_groups is None:
             rule_groups = get_configured_system_config().get(SystemConfigKey.SearchFilterRuleGroups) or []
         async for event in self._SearchChain__async_search_all_sites_stream(
-            keyword=title, sites=sites, page=page, mtype=mtype
+            **search_params
         ):
             result = event.pop("items", []) or []
             candidate_count += len(result)
@@ -187,14 +207,9 @@ class SearchTitleOwner(_SearchOwnerBase):
                 torrents=result,
                 rule_groups=rule_groups,
             )
-            batch_contexts = [
-                Context(
-                    meta_info=self._build_title_search_meta(torrent, mtype),
-                    torrent_info=torrent,
-                    resource_source="search",
-                )
-                for torrent in result
-            ]
+            batch_contexts = _build_title_contexts(
+                result, mtype, self._build_title_search_meta
+            )
             if batch_contexts:
                 contexts.extend(batch_contexts)
             yield {

tests/test_search_sync_async_parity.py

@@ -0,0 +1,324 @@
+"""搜索链同步、异步与流式入口的共享业务决策回归测试。"""
+
+import asyncio
+from copy import deepcopy
+from types import SimpleNamespace
+from typing import Any
+
+from app.chain.search import media as media_module
+from app.chain.search.facade import SearchChain
+from app.domain.context import MediaInfo, TorrentInfo
+from app.domain.metainfo import MetaInfo
+from app.schemas.types import MediaSource, MediaType
+
+
+def _media() -> MediaInfo:
+    """构造具备稳定来源身份和别名的测试媒体。"""
+    return MediaInfo(
+        media_source=MediaSource.TMDB,
+        media_id="100",
+        tmdb_id=100,
+        title="测试电影",
+        names=["Test Movie"],
+        type=MediaType.MOVIE,
+        year="2026",
+    )
+
+
+def _torrent(title: str) -> TorrentInfo:
+    """构造一条无需外部 provider 的测试资源。"""
+    return TorrentInfo(title=title, description="description")
+
+
+def _chain() -> SearchChain:
+    """绕过运行时组合根构造可局部注入的搜索链。"""
+    return object.__new__(SearchChain)
+
+
+def test_id_search_sync_async_share_request_and_missing_plan(monkeypatch):
+    """ID 搜索双入口应产生完全相同的识别、缓存和缺集参数。"""
+    chain = _chain()
+    target = _media()
+    sync_calls: list[dict[str, Any]] = []
+    async_calls: list[dict[str, Any]] = []
+    saved_sync: list[dict[str, Any]] = []
+    saved_async: list[dict[str, Any]] = []
+    result = [SimpleNamespace(name="context")]
+
+    class FakeMediaChain:
+        """记录同步与异步识别参数并返回同一媒体快照。"""
+
+        def recognize_media(self, **kwargs):
+            """返回同步识别结果。"""
+            sync_calls.append(kwargs)
+            return target
+
+        async def async_recognize_media(self, **kwargs):
+            """返回异步识别结果。"""
+            async_calls.append(kwargs)
+            return target
+
+    def process(**kwargs):
+        """记录同步处理计划。"""
+        sync_calls.append(kwargs)
+        return result
+
+    async def async_process(**kwargs):
+        """记录异步处理计划。"""
+        async_calls.append(kwargs)
+        return result
+
+    async def async_save_params(**kwargs):
+        """记录异步缓存参数。"""
+        saved_async.append(kwargs)
+
+    async def async_save_results(_contexts):
+        """模拟异步结果缓存。"""
+
+    monkeypatch.setattr(media_module, "MediaChain", FakeMediaChain)
+    chain.cancel_ai_recommend = lambda: None
+    chain.save_last_search_params = lambda **kwargs: saved_sync.append(kwargs)
+    chain.async_save_last_search_params = async_save_params
+    chain.process = process
+    chain.async_process = async_process
+    chain._save_results = lambda _contexts: None
+    chain._async_save_results = async_save_results
+
+    kwargs = {
+        "media_source": MediaSource.TMDB,
+        "media_id": "100",
+        "mtype": MediaType.TV,
+        "area": "imdbid",
+        "season": 2,
+        "sites": [1, 3],
+        "cache_local": True,
+        "music_type": "recording",
+    }
+    sync_result = chain.search_by_id(**kwargs)
+    async_result = asyncio.run(chain.async_search_by_id(**kwargs))
+
+    assert sync_result is result
+    assert async_result is result
+    assert sync_calls[0] == async_calls[0] == {
+        "media_source": MediaSource.TMDB,
+        "media_id": "100",
+        "mtype": MediaType.TV,
+        "music_type": "recording",
+    }
+    assert sync_calls[1] == async_calls[1]
+    assert sync_calls[1]["sites"] == [1, 3]
+    assert sync_calls[1]["area"] == "imdbid"
+    missing = sync_calls[1]["no_exists"]
+    assert list(missing) == ["tmdb:100"]
+    assert list(missing["tmdb:100"]) == [2]
+    assert saved_sync == saved_async
+
+
+def test_media_process_sync_async_share_keyword_stop_decision(monkeypatch):
+    """双入口应按相同关键字顺序搜索，并在首个有效结果后同时停止。"""
+    chain = _chain()
+    target = _media()
+    found = _torrent("测试电影 2026 1080p")
+    sync_keywords: list[str] = []
+    async_keywords: list[str] = []
+    sync_sleeps: list[int] = []
+    async_sleeps: list[int] = []
+    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 search_all_sites(**kwargs):
+        """第二个关键字开始返回资源。"""
+        keyword = kwargs["keyword"]
+        sync_keywords.append(keyword)
+        return [found] if keyword == "second" else []
+
+    async def async_search_all_sites(**kwargs):
+        """第二个关键字开始异步返回资源。"""
+        keyword = kwargs["keyword"]
+        async_keywords.append(keyword)
+        return [found] if keyword == "second" else []
+
+    def parse_result(**kwargs):
+        """记录进入统一结果处理边界的资源顺序。"""
+        parsed.append(list(kwargs["torrents"]))
+        return kwargs["torrents"]
+
+    async def fake_async_sleep(delay):
+        """记录异步关键字退避但不实际等待。"""
+        async_sleeps.append(delay)
+
+    monkeypatch.setattr(media_module, "MediaChain", FakeMediaChain)
+    monkeypatch.setattr(media_module.random, "randint", lambda _start, _end: 1)
+    monkeypatch.setattr(media_module.time, "sleep", sync_sleeps.append)
+    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__search_all_sites = search_all_sites
+    chain._SearchChain__async_search_all_sites = async_search_all_sites
+    chain._parse_result = parse_result
+
+    sync_result = chain.process(target)
+    async_result = asyncio.run(chain.async_process(target))
+
+    assert sync_keywords == async_keywords == ["first", "second"]
+    assert sync_sleeps == async_sleeps == [1]
+    assert parsed == [[found], [found]]
+    assert sync_result == async_result == [found]
+
+
+def test_id_search_stream_failure_preserves_order_and_skips_result_cache(
+    monkeypatch,
+):
+    """识别失败时流入口应先保存请求状态，只输出错误且不保存空结果。"""
+    chain = _chain()
+    order: list[str] = []
+
+    class FakeMediaChain:
+        """模拟异步识别失败。"""
+
+        async def async_recognize_media(self, **_kwargs):
+            """记录识别顺序并返回失败。"""
+            order.append("recognize")
+            return None
+
+    async def save_params(**_kwargs):
+        """记录请求状态保存顺序。"""
+        order.append("params")
+
+    async def save_results(_contexts):
+        """若失败路径错误保存结果则留下可断言记录。"""
+        order.append("results")
+
+    async def collect_events():
+        """完整消费识别失败的搜索流。"""
+        return [
+            event
+            async for event in chain.async_search_by_id_stream(
+                media_source=MediaSource.TMDB,
+                media_id="404",
+                cache_local=True,
+            )
+        ]
+
+    monkeypatch.setattr(media_module, "MediaChain", FakeMediaChain)
+    chain.cancel_ai_recommend = lambda: order.append("cancel")
+    chain.async_save_last_search_params = save_params
+    chain._async_save_results = save_results
+
+    events = asyncio.run(collect_events())
+
+    assert order == ["cancel", "params", "recognize"]
+    assert events == [
+        {"type": "error", "success": False, "message": "媒体信息识别失败"}
+    ]
+
+
+def test_title_search_sync_async_share_plan_filter_and_projection():
+    """标题搜索双入口应共享 provider 参数、过滤决策和可替换元数据投影。"""
+    chain = _chain()
+    keep = _torrent("Movie 2026 1080p")
+    drop = _torrent("Movie 2026 2160p")
+    provider_calls: list[dict[str, Any]] = []
+    filter_calls: list[list[TorrentInfo]] = []
+    meta_calls: list[str] = []
+
+    def search_all_sites(**kwargs):
+        """记录同步 provider 参数。"""
+        provider_calls.append(kwargs)
+        return [keep, drop]
+
+    async def async_search_all_sites(**kwargs):
+        """记录异步 provider 参数。"""
+        provider_calls.append(kwargs)
+        return [keep, drop]
+
+    def filter_torrents(torrents, rule_groups=None):
+        """记录共享过滤输入并保留首条资源。"""
+        assert rule_groups == ["quality"]
+        filter_calls.append(list(torrents))
+        return [keep]
+
+    def build_meta(torrent, _mtype):
+        """模拟插件或测试替换的稳定元数据构造点。"""
+        meta_calls.append(torrent.title)
+        return MetaInfo(title=f"mapped:{torrent.title}")
+
+    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 = build_meta
+
+    sync_result = chain.search_by_title(
+        title="Movie", page=2, sites=[1], mtype=MediaType.MOVIE, rule_groups=["quality"]
+    )
+    async_result = asyncio.run(
+        chain.async_search_by_title(
+            title="Movie",
+            page=2,
+            sites=[1],
+            mtype=MediaType.MOVIE,
+            rule_groups=["quality"],
+        )
+    )
+
+    assert provider_calls == [
+        {"keyword": "Movie", "sites": [1], "page": 2, "mtype": MediaType.MOVIE},
+        {"keyword": "Movie", "sites": [1], "page": 2, "mtype": MediaType.MOVIE},
+    ]
+    assert filter_calls == [[keep, drop], [keep, drop]]
+    assert meta_calls == [keep.title, keep.title]
+    assert [context.meta_info.title for context in sync_result] == [
+        context.meta_info.title for context in async_result
+    ] == [f"mapped:{keep.title}"]
+
+
+def test_title_stream_preserves_provider_and_result_order():
+    """标题流应按 provider 批次顺序追加，并以相同顺序发布累计完成结果。"""
+    chain = _chain()
+    first = _torrent("Movie 2026 1080p")
+    second = _torrent("Movie 2026 2160p")
+    provider_params: list[dict[str, Any]] = []
+
+    async def search_stream(**kwargs):
+        """按固定顺序输出两个 provider 批次。"""
+        provider_params.append(kwargs)
+        yield {"type": "append", "site_id": 1, "items": [first]}
+        yield {"type": "append", "site_id": 2, "items": [second]}
+
+    async def collect_events():
+        """完整消费标题搜索流。"""
+        return [
+            event
+            async for event in chain.async_search_by_title_stream(
+                title="Movie", sites=[1, 2], rule_groups=[]
+            )
+        ]
+
+    chain._SearchChain__async_search_all_sites_stream = search_stream
+    chain._filter_title_search_torrents = lambda torrents, rule_groups=None: torrents
+    chain._build_title_search_meta = (
+        lambda torrent, _mtype: MetaInfo(title=torrent.title)
+    )
+
+    events = asyncio.run(collect_events())
+
+    assert provider_params == [
+        {"keyword": "Movie", "sites": [1, 2], "page": 0, "mtype": None}
+    ]
+    assert [event["type"] for event in events] == ["append", "append", "done"]
+    assert [event["total_items"] for event in events] == [1, 2, 2]
+    assert [
+        item["torrent_info"]["title"] for item in events[-1]["items"]
+    ] == [first.title, second.title]
+    assert events[-1]["candidate_items"] == 2