refactor(subscribe): satisfy global governance gates

app/api/dependencies/subscription.py

@@ -43,7 +43,10 @@ from app.application.subscription.mutation import (
 )
 from app.application.subscription.query import SubscriptionQueryService
 from app.application.subscription.search import SearchSubscriptionsCommand
-from app.application.subscription.status import SubscriptionExecutionStatusService
+from app.application.subscription.status import (
+    SubscriptionExecutionReadRepository,
+    SubscriptionExecutionStatusService,
+)
 from app.application.subscription.write import (
     SubscriptionBatchWritePort,
 )
@@ -187,7 +190,9 @@ def get_subscription_execution_status_service(
     if factory is None:
         raise RuntimeError("订阅执行状态仓储未注册")
     repository = factory(db)
-    return SubscriptionExecutionStatusService(repository=repository)  # type: ignore[arg-type]
+    return SubscriptionExecutionStatusService(
+        repository=cast(SubscriptionExecutionReadRepository, repository)
+    )
 
 
 def get_subscription_search_repository(

app/api/endpoints/subexecution.py

@@ -0,0 +1,110 @@
+"""订阅执行批次状态与取消端点。"""
+
+from typing import Any, List, Optional
+
+from fastapi import Depends, HTTPException
+
+from app.api.dependencies.auth import get_current_active_user_async
+from app.api.dependencies.subscription import (
+    get_subscription_execution_status_service,
+    get_subscription_query_service,
+    get_subscription_search_repository,
+)
+from app.api.principal import ApiPrincipal
+from app.api.response import ResponseAPIRouter
+from app.application.subscription.execution import SubscriptionSearchRepository
+from app.application.subscription.query import SubscriptionQueryService
+from app.application.subscription.status import SubscriptionExecutionStatusService
+from app.runtime.execution import run_in_threadpool
+from app.schemas.response import Response as _SchemaResponse
+from app.schemas.subscribe import SubscriptionBatchStatus as _SchemaSubscriptionBatchStatus
+
+router = ResponseAPIRouter()
+
+
+async def _accessible_subscription_ids(
+    query: SubscriptionQueryService,
+    current_user: ApiPrincipal,
+) -> Optional[set[int]]:
+    """返回普通用户可访问订阅 ID；超级用户以 None 表示不限制。"""
+    if current_user.is_superuser:
+        return None
+    subscribes = await query.list_public(current_user.name)
+    return {item.id for item in subscribes if item.id is not None}
+
+
+@router.get(  # type: ignore[misc]
+    "/execution/batches",
+    summary="查询订阅搜索批次状态",
+    response_model=List[_SchemaSubscriptionBatchStatus],
+)
+async def list_subscription_execution_batches(
+    limit: int = 10,
+    status_service: SubscriptionExecutionStatusService = Depends(
+        get_subscription_execution_status_service
+    ),
+    query: SubscriptionQueryService = Depends(get_subscription_query_service),
+    current_user: ApiPrincipal = Depends(get_current_active_user_async),
+) -> Any:
+    """返回当前用户完整可见的最近搜索批次。"""
+    accessible_ids = await _accessible_subscription_ids(query, current_user)
+    batches = await status_service.list_batches(
+        accessible_subscription_ids=accessible_ids,
+        limit=limit,
+    )
+    return [_SchemaSubscriptionBatchStatus.model_validate(batch) for batch in batches]
+
+
+@router.get(  # type: ignore[misc]
+    "/execution/batches/{batch_id}",
+    summary="查询订阅搜索批次",
+    response_model=_SchemaSubscriptionBatchStatus,
+)
+async def get_subscription_execution_batch(
+    batch_id: str,
+    status_service: SubscriptionExecutionStatusService = Depends(
+        get_subscription_execution_status_service
+    ),
+    query: SubscriptionQueryService = Depends(get_subscription_query_service),
+    current_user: ApiPrincipal = Depends(get_current_active_user_async),
+) -> Any:
+    """按稳定 ID 返回当前用户可访问的搜索批次。"""
+    accessible_ids = await _accessible_subscription_ids(query, current_user)
+    batch = await status_service.get_batch(
+        batch_id,
+        accessible_subscription_ids=accessible_ids,
+    )
+    if batch is None:
+        raise HTTPException(status_code=404, detail="订阅搜索批次不存在")
+    return _SchemaSubscriptionBatchStatus.model_validate(batch)
+
+
+@router.put(  # type: ignore[misc]
+    "/execution/batches/{batch_id}/cancel",
+    summary="取消订阅搜索批次",
+    response_model=_SchemaResponse[None],
+)
+async def cancel_subscription_execution_batch(
+    batch_id: str,
+    status_service: SubscriptionExecutionStatusService = Depends(
+        get_subscription_execution_status_service
+    ),
+    query: SubscriptionQueryService = Depends(get_subscription_query_service),
+    search_repository: SubscriptionSearchRepository = Depends(
+        get_subscription_search_repository
+    ),
+    current_user: ApiPrincipal = Depends(get_current_active_user_async),
+) -> Any:
+    """在权限校验后请求取消尚未越过下载副作用边界的任务。"""
+    accessible_ids = await _accessible_subscription_ids(query, current_user)
+    batch = await status_service.get_batch(
+        batch_id,
+        accessible_subscription_ids=accessible_ids,
+    )
+    if batch is None:
+        return _SchemaResponse(success=False, message="订阅搜索批次不存在")
+    cancelled = await run_in_threadpool(search_repository.request_cancel, batch_id)
+    return _SchemaResponse(
+        success=bool(cancelled),
+        message="" if cancelled else "订阅搜索批次已结束或无法取消",
+    )

app/api/endpoints/subscribe.py

@@ -25,7 +25,6 @@ from app.api.dependencies.subscription import (
     get_subscription_execution_status_service,
     get_subscription_mutation_service,
     get_subscription_query_service,
-    get_subscription_search_repository,
 )
 from app.api.principal import ApiPrincipal
 from app.api.response import (
@@ -46,7 +45,6 @@ from app.application.subscription.delete import (
     DeleteSubscribeCommand,
     SubscribeDeletionActor,
 )
-from app.application.subscription.execution import SubscriptionSearchRepository
 from app.application.subscription.identity import (
     DeleteSubscriptionsByIdentityCommand,
 )
@@ -72,7 +70,6 @@ from app.schemas.subscribe import SubscrbieInfo as _SchemaSubscrbieInfo
 from app.schemas.subscribe import SubscribeDeletionResult as _SchemaSubscribeDeletionResult
 from app.schemas.subscribe import SubscribeShare as _SchemaSubscribeShare
 from app.schemas.subscribe import SubscribeShareStatistics as _SchemaSubscribeShareStatistics
-from app.schemas.subscribe import SubscriptionBatchStatus as _SchemaSubscriptionBatchStatus
 from app.schemas.subscribe import SubscriptionExecutionStatus as _SchemaSubscriptionExecutionStatus
 from app.schemas.token import TokenPayload as _SchemaTokenPayload
 from app.schemas.types import (
@@ -105,17 +102,6 @@ async def _attach_execution_status(
     return subscribes
 
 
-async def _accessible_subscription_ids(
-    query: SubscriptionQueryService,
-    current_user: ApiPrincipal,
-) -> Optional[set[int]]:
-    """返回普通用户可访问订阅 ID；超级用户以 None 表示不限制。"""
-    if current_user.is_superuser:
-        return None
-    subscribes = await query.list_public(current_user.name)
-    return {item.id for item in subscribes if item.id is not None}
-
-
 def start_subscribe_add(
     title: str,
     year: str,
@@ -502,83 +488,6 @@ async def search_subscribe(
     return _SchemaResponse(success=True)
 
 
-@router.get(
-    "/execution/batches",
-    summary="查询订阅搜索批次状态",
-    response_model=List[_SchemaSubscriptionBatchStatus],
-)
-async def list_subscription_execution_batches(
-    limit: int = 10,
-    status_service: SubscriptionExecutionStatusService = Depends(
-        get_subscription_execution_status_service
-    ),
-    query: SubscriptionQueryService = Depends(get_subscription_query_service),
-    current_user: ApiPrincipal = Depends(get_current_active_user_async),
-) -> Any:
-    """返回当前用户完整可见的最近搜索批次。"""
-    accessible_ids = await _accessible_subscription_ids(query, current_user)
-    batches = await status_service.list_batches(
-        accessible_subscription_ids=accessible_ids,
-        limit=limit,
-    )
-    return [_SchemaSubscriptionBatchStatus.model_validate(batch) for batch in batches]
-
-
-@router.get(
-    "/execution/batches/{batch_id}",
-    summary="查询订阅搜索批次",
-    response_model=_SchemaSubscriptionBatchStatus,
-)
-async def get_subscription_execution_batch(
-    batch_id: str,
-    status_service: SubscriptionExecutionStatusService = Depends(
-        get_subscription_execution_status_service
-    ),
-    query: SubscriptionQueryService = Depends(get_subscription_query_service),
-    current_user: ApiPrincipal = Depends(get_current_active_user_async),
-) -> Any:
-    """按稳定 ID 返回当前用户可访问的搜索批次。"""
-    accessible_ids = await _accessible_subscription_ids(query, current_user)
-    batch = await status_service.get_batch(
-        batch_id,
-        accessible_subscription_ids=accessible_ids,
-    )
-    if batch is None:
-        raise HTTPException(status_code=404, detail="订阅搜索批次不存在")
-    return _SchemaSubscriptionBatchStatus.model_validate(batch)
-
-
-@router.put(
-    "/execution/batches/{batch_id}/cancel",
-    summary="取消订阅搜索批次",
-    response_model=_SchemaResponse[None],
-)
-async def cancel_subscription_execution_batch(
-    batch_id: str,
-    status_service: SubscriptionExecutionStatusService = Depends(
-        get_subscription_execution_status_service
-    ),
-    query: SubscriptionQueryService = Depends(get_subscription_query_service),
-    search_repository: SubscriptionSearchRepository = Depends(
-        get_subscription_search_repository
-    ),
-    current_user: ApiPrincipal = Depends(get_current_active_user_async),
-) -> Any:
-    """在权限校验后请求取消尚未越过下载副作用边界的任务。"""
-    accessible_ids = await _accessible_subscription_ids(query, current_user)
-    batch = await status_service.get_batch(
-        batch_id,
-        accessible_subscription_ids=accessible_ids,
-    )
-    if batch is None:
-        return _SchemaResponse(success=False, message="订阅搜索批次不存在")
-    cancelled = await run_in_threadpool(search_repository.request_cancel, batch_id)
-    return _SchemaResponse(
-        success=bool(cancelled),
-        message="" if cancelled else "订阅搜索批次已结束或无法取消",
-    )
-
-
 @router.delete("/media/{media_id}", summary="删除订阅", response_model=_SchemaResponse[None])
 async def delete_subscribe_by_media_identity(
     media_id: str,

app/api/routers.py

@@ -29,6 +29,7 @@ from app.api.endpoints import (
     search,
     site,
     storage,
+    subexecution,
     subscribe,
     system,
     tmdb,
@@ -58,6 +59,7 @@ API_V1_ROUTER_SPECS: tuple[RouterSpec, ...] = (
     RouterSpec(agent.router, "/message/agent", ("agent",)),
     RouterSpec(webhook.router, "/webhook", ("webhook",)),
     RouterSpec(subscribe.router, "/subscribe", ("subscribe",)),
+    RouterSpec(subexecution.router, "/subscribe", ("subscribe",)),
     RouterSpec(music.router, "/music", ("music",)),
     RouterSpec(media.router, "/media", ("media",)),
     RouterSpec(search.router, "/search", ("search",)),

app/application/subscription/sitebudget.py

@@ -8,7 +8,7 @@ from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 from typing import Callable, Optional, Protocol
 
-from app.application.site.search_observation import SiteSearchObservation
+from app.application.site.observation import SiteSearchObservation
 from app.runtime.stop import StopState
 
 

app/chain/base.py

@@ -43,7 +43,6 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin,
     """
     处理链基类
     """
-
     def __init__(self, runtime_context: Optional[ChainRuntimeContext] = None):
         """
         公共初始化；未显式传入上下文时继续使用兼容运行时 provider。
@@ -86,7 +85,6 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin,
         )
         self._legacy_transfer_command = context.legacy_transfer_command
         self.messagequeue = context.message_queue.bind(self.run_module)
-
     @property
     def runtime_config(self) -> ChainRuntimeConfig:
         """返回实例快照；兼容绕过构造器的旧调用并按需取得当前快照。"""

app/chain/download/batch.py

@@ -40,8 +40,7 @@ class DownloadBatchOwner(_DownloadOwnerBase):
                        userid: Optional[str] = None,
                        username: Optional[str] = None,
                        downloader: Optional[str] = None,
-                       custom_words: Optional[str] = None,
-                       governance: Optional[SubscriptionDownloadGovernance] = None,
+                       custom_words: Optional[str] = None, governance: Optional[SubscriptionDownloadGovernance] = None,
                        ) -> Tuple[
                            List[Context],
                            Optional[Dict[str, Dict[int, NotExistMediaInfo]]],
@@ -61,8 +60,7 @@ class DownloadBatchOwner(_DownloadOwnerBase):
             userid=userid,
             username=username,
             downloader=downloader,
-            custom_words=custom_words,
-            governance=governance,
+            custom_words=custom_words, governance=governance,
         )
 
     def _execute_batch_download(self,
@@ -74,8 +72,7 @@ class DownloadBatchOwner(_DownloadOwnerBase):
                        userid: Optional[str] = None,
                        username: Optional[str] = None,
                        downloader: Optional[str] = None,
-                       custom_words: Optional[str] = None,
-                       governance: Optional[SubscriptionDownloadGovernance] = None,
+                       custom_words: Optional[str] = None, governance: Optional[SubscriptionDownloadGovernance] = None,
                        ) -> Tuple[
                            List[Context],
                            Optional[Dict[str, Dict[int, NotExistMediaInfo]]],
@@ -90,8 +87,7 @@ class DownloadBatchOwner(_DownloadOwnerBase):
         :param userid:  用户ID
         :param username: 调用下载的用户名/插件名
         :param downloader: 下载器
-        :param custom_words: 下载来源（如订阅）的完整自定义识别词文本，随下载记录存档，供整理时原样复现识别
-        :param governance: 订阅下载幂等、入口任务和取消边界；非订阅调用保持为空
+        :param custom_words: 下载来源自定义词；governance: 订阅幂等、入口任务和取消边界
         :return: 已下载资源列表及剩余缺集，键格式为 no_exists[source:id]
         """
         no_exists_was_none = no_exists is None
@@ -190,8 +186,7 @@ class DownloadBatchOwner(_DownloadOwnerBase):
             userid=userid,
             username=username,
             downloader=downloader,
-            custom_words=custom_words,
-            governance=governance,
+            custom_words=custom_words, governance=governance,
         )
 
         # 电视剧整季匹配
@@ -309,8 +304,7 @@ class DownloadBatchOwner(_DownloadOwnerBase):
                                         userid=userid,
                                         username=username,
                                         downloader=downloader,
-                                        custom_words=custom_words,
-                                        governance=governance,
+                                        custom_words=custom_words, governance=governance,
                                     )
                             else:
                                 # 下载
@@ -319,8 +313,7 @@ class DownloadBatchOwner(_DownloadOwnerBase):
                                                                    channel=channel, source=source,
                                                                    userid=userid, username=username,
                                                                    downloader=downloader,
-                                                                   custom_words=custom_words,
-                                                                   governance=governance)
+                                                                   custom_words=custom_words, governance=governance)
 
                             if download_id:
                                 # 下载成功
@@ -409,8 +402,7 @@ class DownloadBatchOwner(_DownloadOwnerBase):
                                                                    channel=channel, source=source,
                                                                    userid=userid, username=username,
                                                                    downloader=downloader,
-                                                                   custom_words=custom_words,
-                                                                   governance=governance)
+                                                                   custom_words=custom_words, governance=governance)
                                 if download_id:
                                     # 下载成功
                                     if __requires_complete_coverage(missing_info):
@@ -527,8 +519,7 @@ class DownloadBatchOwner(_DownloadOwnerBase):
                                 userid=userid,
                                 username=username,
                                 downloader=downloader,
-                                custom_words=custom_words,
-                                governance=governance,
+                                custom_words=custom_words, governance=governance,
                             )
                             if not download_id:
                                 __remember_context_failure(context)

app/chain/download/submission.py

@@ -3,6 +3,7 @@
 import base64
 import json
 import re
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Dict, Optional, Set, Tuple, Union, cast
 from urllib.parse import urlencode, urljoin, urlparse
@@ -27,6 +28,7 @@ from app.domain.context import (
     MusicInfo,
     TorrentInfo,
 )
+from app.domain.meta.metabase import MetaBase
 from app.runtime.cache import FileCache
 from app.runtime.events import eventmanager
 from app.runtime.log import logger
@@ -40,8 +42,24 @@ from app.schemas.types import (
 )
 
 
-class DownloadSubmissionOwner(_DownloadOwnerBase):
-    """种子获取与单任务提交 owner。"""
+@dataclass(frozen=True, slots=True)
+class _PreparedDownload:
+    """下载器调用前已经验证并规范化的本地提交事实。"""
+
+    torrent: TorrentInfo
+    media: Union[MediaInfo, MusicInfo]
+    meta: MetaBase
+    torrent_content: Union[str, bytes]
+    folder_name: str
+    file_list: list[str]
+    download_dir: Path
+    download_uri: str
+    download_episodes: Optional[str]
+    site_downloader: Optional[str]
+
+
+class _DownloadResourceOwner(_DownloadOwnerBase):
+    """种子获取、间接地址解析与资源下载事件 owner。"""
 
 
     @staticmethod
@@ -262,6 +280,10 @@ class DownloadSubmissionOwner(_DownloadOwnerBase):
             logger.warn(str(err))
             return save_path, str(err)
 
+
+class DownloadSubmissionOwner(_DownloadResourceOwner):
+    """单任务下载准备、提交与结算 owner。"""
+
     def download_single(self, context: Context,
                         torrent_file: Optional[Path] = None,
                         torrent_content: Optional[Union[str, bytes]] = None,
@@ -299,146 +321,183 @@ class DownloadSubmissionOwner(_DownloadOwnerBase):
             governance=governance,
         )
 
-    def _execute_download_single(self, context: Context,
-                        torrent_file: Optional[Path] = None,
-                        torrent_content: Optional[Union[str, bytes]] = None,
-                        episodes: Optional[Set[int]] = None,
-                        channel: Optional[NotificationChannel] = None,
-                        source: Optional[str] = None,
-                        downloader: Optional[str] = None,
-                        save_path: Optional[str] = None,
-                        userid: Union[str, int, None] = None,
-                        username: Optional[str] = None,
-                        label: Optional[str] = None,
-                        return_detail: bool = False,
-                        custom_words: Optional[str] = None,
-                        governance: Optional[SubscriptionDownloadGovernance] = None,
-                        ) -> Union[Optional[str], Tuple[Optional[str], Optional[str]]]:
-        """
-        下载及发送通知
-        :param context: 资源上下文
-        :param torrent_file: 种子文件路径
-        :param torrent_content: 种子内容（磁力链或种子文件内容）
-        :param episodes: 需要下载的集数
-        :param channel: 通知渠道
-        :param source: 来源（消息通知、Subscribe、Manual等）
-        :param downloader: 下载器
-        :param save_path: 保存路径, 支持<storage>:<path>, 如rclone:/MP, smb:/server/share/Movies等
-        :param userid: 用户ID
-        :param username: 调用下载的用户名/插件名
-        :param label: 自定义标签
-        :param return_detail: 是否返回详细结果；False 时返回下载任务 hash 或 None，True 时返回 (hash, error_msg)
-        :param custom_words: 下载来源（如订阅）的完整自定义识别词文本，随下载记录存档，供整理时原样复现识别
-        :param governance: 订阅级提交幂等、入口任务与取消检查；非订阅调用保持为空
-        :return: return_detail=False 时返回下载任务 hash 或 None；return_detail=True 时返回 (hash, error_msg)
-        """
-        _torrent = context.torrent_info
-        _media = context.media_info
-        _meta = context.meta_info
-        if _torrent is None or _media is None or _meta is None:
-            error_message = "下载上下文缺少媒体、元数据或种子信息"
-            return (None, error_message) if return_detail else None
-        _site_downloader = _torrent.site_downloader
+    def _execute_download_single(
+        self,
+        context: Context,
+        torrent_file: Optional[Path] = None,
+        torrent_content: Optional[Union[str, bytes]] = None,
+        episodes: Optional[Set[int]] = None,
+        channel: Optional[NotificationChannel] = None,
+        source: Optional[str] = None,
+        downloader: Optional[str] = None,
+        save_path: Optional[str] = None,
+        userid: Union[str, int, None] = None,
+        username: Optional[str] = None,
+        label: Optional[str] = None,
+        return_detail: bool = False,
+        custom_words: Optional[str] = None,
+        governance: Optional[SubscriptionDownloadGovernance] = None,
+    ) -> Union[Optional[str], Tuple[Optional[str], Optional[str]]]:
+        """准备下载事实，提交下载器并按订阅治理合同结算结果。"""
+        prepared, error_msg = self._prepare_download_single(
+            context=context,
+            torrent_file=torrent_file,
+            torrent_content=torrent_content,
+            episodes=episodes,
+            channel=channel,
+            source=source,
+            downloader=downloader,
+            save_path=save_path,
+            userid=userid,
+            username=username,
+        )
+        if prepared is None:
+            return (None, error_msg) if return_detail else None
+        download_hash, error_msg = self._submit_prepared_download(
+            prepared=prepared,
+            context=context,
+            episodes=episodes,
+            channel=channel,
+            source=source,
+            downloader=downloader,
+            userid=userid,
+            username=username,
+            label=label,
+            custom_words=custom_words,
+            governance=governance,
+        )
+        return (download_hash, error_msg) if return_detail else download_hash
 
-        # 下载目录和下载器分类依赖 TMDB 辅助分类，但媒体主身份保持不变。
-        supplemented_media = MediaChain().supplement_tmdb_info(_media, _meta)
+    def _prepare_download_single(
+        self,
+        *,
+        context: Context,
+        torrent_file: Optional[Path],
+        torrent_content: Optional[Union[str, bytes]],
+        episodes: Optional[Set[int]],
+        channel: Optional[NotificationChannel],
+        source: Optional[str],
+        downloader: Optional[str],
+        save_path: Optional[str],
+        userid: Union[str, int, None],
+        username: Optional[str],
+    ) -> tuple[Optional[_PreparedDownload], Optional[str]]:
+        """补全媒体、读取种子并解析出可信下载目录。"""
+        torrent, media, meta = context.torrent_info, context.media_info, context.meta_info
+        if torrent is None or media is None or meta is None:
+            return None, "下载上下文缺少媒体、元数据或种子信息"
+        site_downloader = torrent.site_downloader
+        supplemented_media = MediaChain().supplement_tmdb_info(media, meta)
         if not isinstance(supplemented_media, (MediaInfo, MusicInfo)):
-            error_message = "媒体信息补全失败"
-            return (None, error_message) if return_detail else None
-        _media = supplemented_media
-        context.media_info = _media
-
+            return None, "媒体信息补全失败"
+        media = supplemented_media
+        context.media_info = media
         save_path, event_error = self._apply_resource_download_event(
-            context, episodes, channel, source, downloader, save_path,
-            userid, username,
+            context, episodes, channel, source, downloader, save_path, userid, username
         )
         if event_error:
-            return (None, event_error) if return_detail else None
-
-        # 实际下载的集数
+            return None, event_error
         download_episodes = episode_rules.format_ranges(list(episodes)) if episodes else None
         if episodes is not None:
             context.selected_episodes = sorted(set(episodes))
-        elif _meta and _meta.episode_list:
-            context.selected_episodes = sorted(set(_meta.episode_list))
+        elif meta.episode_list:
+            context.selected_episodes = sorted(set(meta.episode_list))
         else:
             context.selected_episodes = []
-        _folder_name = ""
         if not torrent_file and not torrent_content:
-            # 下载种子文件，得到的可能是文件也可能是磁力链
-            torrent_content, _folder_name, _file_list = self.download_torrent(_torrent,
-                                                                              channel=channel,
-                                                                              source=source,
-                                                                              userid=userid)
+            torrent_content, _, _ = self.download_torrent(
+                torrent, channel=channel, source=source, userid=userid
+            )
         elif torrent_file:
-            if torrent_file.exists():
-                torrent_content = torrent_file.read_bytes()
-            else:
-                # 缓存处理器
-                cache_backend = FileCache()
-                # 读取缓存的种子文件
-                torrent_content = cache_backend.get(torrent_file.as_posix(), region="torrents")
-
+            torrent_content = (
+                torrent_file.read_bytes()
+                if torrent_file.exists()
+                else FileCache().get(torrent_file.as_posix(), region="torrents")
+            )
         if not torrent_content:
             self._record_download_failure(
                 context=context,
                 error_msg="下载种子内容为空",
-                downloader=downloader or _site_downloader,
+                downloader=downloader or site_downloader,
                 source=source,
                 episodes=episodes,
             )
-            return (None, "下载种子内容为空") if return_detail else None
-
-        # 获取种子文件的文件夹名和文件清单
-        _folder_name, _file_list = cast(
-            Any, TorrentHelper
-        )().get_fileinfo_from_torrent_content(torrent_content)
-
-        album_validation_error = self._validate_music_album_resource(context, _file_list)
-        if album_validation_error:
-            logger.info(f"{_torrent.title} {album_validation_error}，跳过该资源")
+            return None, "下载种子内容为空"
+        folder_name, file_list = cast(Any, TorrentHelper)().get_fileinfo_from_torrent_content(
+            torrent_content
+        )
+        album_error = self._validate_music_album_resource(context, file_list)
+        if album_error:
+            logger.info(f"{torrent.title} {album_error}，跳过该资源")
             self._record_download_failure(
                 context=context,
-                error_msg=album_validation_error,
-                downloader=downloader or _site_downloader,
+                error_msg=album_error,
+                downloader=downloader or site_downloader,
                 source=source,
                 episodes=episodes,
             )
-            return (None, album_validation_error) if return_detail else None
-
+            return None, album_error
         storage, download_dir, error_msg = self._resolve_media_download_dir(
-            media_info=_media,
+            media_info=media,
             save_path=save_path,
         )
         if not download_dir or not storage:
             if error_msg == "未找到下载目录":
-                self.messagehelper.put(f"{_media.type.value} {_media.title_year} 未找到下载目录！",
-                                       title="下载失败", role="system")
-            return (None, error_msg or "未找到下载目录") if return_detail else None
-        file_uri = FileURI(storage=storage, path=download_dir.as_posix())
-        download_dir = Path(file_uri.uri)
+                self.messagehelper.put(
+                    f"{media.type.value} {media.title_year} 未找到下载目录！",
+                    title="下载失败",
+                    role="system",
+                )
+            return None, error_msg or "未找到下载目录"
+        download_uri = FileURI(storage=storage, path=download_dir.as_posix()).uri
+        return _PreparedDownload(
+            torrent=torrent,
+            media=media,
+            meta=meta,
+            torrent_content=torrent_content,
+            folder_name=folder_name,
+            file_list=file_list,
+            download_dir=Path(download_uri),
+            download_uri=download_uri,
+            download_episodes=download_episodes,
+            site_downloader=site_downloader,
+        ), None
 
+    def _submit_prepared_download(
+        self,
+        *,
+        prepared: _PreparedDownload,
+        context: Context,
+        episodes: Optional[Set[int]],
+        channel: Optional[NotificationChannel],
+        source: Optional[str],
+        downloader: Optional[str],
+        userid: Union[str, int, None],
+        username: Optional[str],
+        label: Optional[str],
+        custom_words: Optional[str],
+        governance: Optional[SubscriptionDownloadGovernance],
+    ) -> tuple[Optional[str], Optional[str]]:
+        """认领幂等提交权，调用下载器并分派成功或拒绝结算。"""
         if self._subscription_download_cancelled(governance):
-            cancel_error = "订阅下载在提交前已取消"
-            return (None, cancel_error) if return_detail else None
+            return None, "订阅下载在提交前已取消"
         admission, duplicate_hash = self._claim_subscription_download(
             context=context,
             episodes=episodes,
             governance=governance,
-            downloader=downloader or _site_downloader,
-            download_uri=file_uri.uri,
+            downloader=downloader or prepared.site_downloader,
+            download_uri=prepared.download_uri,
         )
         if duplicate_hash:
             if governance and governance.mark_started:
                 governance.mark_started()
-            logger.info(f"{_torrent.title} 已由重叠订阅入口提交，复用任务 {duplicate_hash}")
-            return (duplicate_hash, "下载任务已由重叠入口提交") if return_detail else duplicate_hash
+            logger.info(f"{prepared.torrent.title} 已由重叠订阅入口提交，复用任务 {duplicate_hash}")
+            return duplicate_hash, "下载任务已由重叠入口提交"
         if admission is not None and not admission.acquired:
-            wait_error = (
+            return None, (
                 f"订阅下载提交当前为 {admission.snapshot.state}，"
                 f"最早可重试：{admission.snapshot.available_at or '待下一轮'}"
             )
-            return (None, wait_error) if return_detail else None
         attempt_token = admission.snapshot.attempt_token if admission is not None else None
         admission_key = admission.snapshot.idempotency_key if admission is not None else None
         if admission is not None and self._subscription_download_cancelled(governance):
@@ -446,21 +505,18 @@ class DownloadSubmissionOwner(_DownloadOwnerBase):
                 idempotency_key=admission.snapshot.idempotency_key,
                 attempt_token=admission.snapshot.attempt_token or "",
             )
-            cancel_error = "订阅下载在下载器调用前已取消"
-            return (None, cancel_error) if return_detail else None
-
-        # 添加下载
+            return None, "订阅下载在下载器调用前已取消"
         if governance and governance.mark_started:
             governance.mark_started()
         try:
-            result: Optional[Tuple[Optional[str], Optional[str], Optional[str], str]] = self.download(
-                content=torrent_content,
-                cookie=_torrent.site_cookie,
+            result = self.download(
+                content=prepared.torrent_content,
+                cookie=prepared.torrent.site_cookie,
                 episodes=cast(Set[int], episodes),
-                download_dir=download_dir,
-                category=_media.category,
+                download_dir=prepared.download_dir,
+                category=prepared.media.category,
                 label=label,
-                downloader=downloader or _site_downloader,
+                downloader=downloader or prepared.site_downloader,
             )
         except Exception as err:
             if admission_key and attempt_token:
@@ -470,110 +526,173 @@ class DownloadSubmissionOwner(_DownloadOwnerBase):
                     error=f"下载器调用异常：{str(err)}",
                 )
                 raise DownloadReconciliationRequired(
-                    f"{_torrent.title} 下载器结果不确定，已冻结自动重试"
+                    f"{prepared.torrent.title} 下载器结果不确定，已冻结自动重试"
                 ) from err
             raise
-        if result:
-            _downloader, _hash, _layout, error_msg = result
+        actual_downloader, download_hash, layout, error_msg = (
+            result if result else (None, None, None, "未找到下载器")
+        )
+        if download_hash:
+            self._settle_accepted_download(
+                prepared=prepared,
+                context=context,
+                episodes=episodes,
+                channel=channel,
+                source=source,
+                userid=userid,
+                username=username,
+                custom_words=custom_words,
+                actual_downloader=actual_downloader,
+                download_hash=download_hash,
+                layout=layout,
+                admission_key=admission_key,
+                attempt_token=attempt_token,
+            )
         else:
-            _downloader, _hash, _layout, error_msg = None, None, None, "未找到下载器"
+            self._record_rejected_download(
+                prepared=prepared,
+                context=context,
+                episodes=episodes,
+                channel=channel,
+                source=source,
+                downloader=downloader,
+                userid=userid,
+                actual_downloader=actual_downloader,
+                error_msg=error_msg,
+                admission_key=admission_key,
+                attempt_token=attempt_token,
+            )
+        return download_hash, error_msg
 
-        if _hash:
+    def _settle_accepted_download(
+        self,
+        *,
+        prepared: _PreparedDownload,
+        context: Context,
+        episodes: Optional[Set[int]],
+        channel: Optional[NotificationChannel],
+        source: Optional[str],
+        userid: Union[str, int, None],
+        username: Optional[str],
+        custom_words: Optional[str],
+        actual_downloader: Optional[str],
+        download_hash: str,
+        layout: Optional[str],
+        admission_key: Optional[str],
+        attempt_token: Optional[str],
+    ) -> None:
+        """持久化下载器接受事实，执行本地结算并确认成功终态。"""
+        if admission_key and attempt_token:
+            accepted = self._subscription_download_repository().mark_accepted(
+                idempotency_key=admission_key,
+                attempt_token=attempt_token,
+                downloader=actual_downloader,
+                download_hash=download_hash,
+            )
+            if not accepted:
+                raise DownloadReconciliationRequired(
+                    f"{prepared.torrent.title} 已被下载器接受，但本地接受状态写入失败"
+                )
+        try:
+            self._settle_download_success(
+                context=context,
+                media=prepared.media,
+                meta=prepared.meta,
+                torrent=prepared.torrent,
+                folder_name=prepared.folder_name,
+                file_list=prepared.file_list,
+                download_dir=prepared.download_dir,
+                layout=layout,
+                downloader=actual_downloader,
+                download_hash=download_hash,
+                download_episodes=prepared.download_episodes,
+                episodes=episodes,
+                channel=channel,
+                source=source,
+                userid=userid,
+                username=username,
+                torrent_content=prepared.torrent_content,
+                custom_words=custom_words,
+            )
+        except Exception as err:
             if admission_key and attempt_token:
-                accepted = self._subscription_download_repository().mark_accepted(
+                self._subscription_download_repository().mark_reconcile_required(
                     idempotency_key=admission_key,
                     attempt_token=attempt_token,
-                    downloader=_downloader,
-                    download_hash=_hash,
+                    error=f"下载器已接受但本地结算失败：{str(err)}",
+                    downloader=actual_downloader,
+                    download_hash=download_hash,
                 )
-                if not accepted:
-                    raise DownloadReconciliationRequired(
-                        f"{_torrent.title} 已被下载器接受，但本地接受状态写入失败"
-                    )
-            try:
-                self._settle_download_success(
-                    context=context,
-                    media=_media,
-                    meta=_meta,
-                    torrent=_torrent,
-                    folder_name=_folder_name,
-                    file_list=_file_list,
-                    download_dir=download_dir,
-                    layout=_layout,
-                    downloader=_downloader,
-                    download_hash=_hash,
-                    download_episodes=download_episodes,
-                    episodes=episodes,
-                    channel=channel,
-                    source=source,
-                    userid=userid,
-                    username=username,
-                    torrent_content=torrent_content,
-                    custom_words=custom_words,
-                )
-            except Exception as err:
-                if admission_key and attempt_token:
-                    self._subscription_download_repository().mark_reconcile_required(
-                        idempotency_key=admission_key,
-                        attempt_token=attempt_token,
-                        error=f"下载器已接受但本地结算失败：{str(err)}",
-                        downloader=_downloader,
-                        download_hash=_hash,
-                    )
-                    raise DownloadReconciliationRequired(
-                        f"{_torrent.title} 已被下载器接受但本地结算失败，已转待对账"
-                    ) from err
-                raise
-            if admission_key and attempt_token:
-                succeeded = self._subscription_download_repository().mark_succeeded(
+                raise DownloadReconciliationRequired(
+                    f"{prepared.torrent.title} 已被下载器接受但本地结算失败，已转待对账"
+                ) from err
+            raise
+        if admission_key and attempt_token:
+            succeeded = self._subscription_download_repository().mark_succeeded(
+                idempotency_key=admission_key,
+                attempt_token=attempt_token,
+            )
+            if not succeeded:
+                self._subscription_download_repository().mark_reconcile_required(
                     idempotency_key=admission_key,
                     attempt_token=attempt_token,
+                    error="本地结算完成但幂等成功终态写入失败",
+                    downloader=actual_downloader,
+                    download_hash=download_hash,
                 )
-                if not succeeded:
-                    self._subscription_download_repository().mark_reconcile_required(
-                        idempotency_key=admission_key,
-                        attempt_token=attempt_token,
-                        error="本地结算完成但幂等成功终态写入失败",
-                        downloader=_downloader,
-                        download_hash=_hash,
-                    )
-                    raise DownloadReconciliationRequired(
-                        f"{_torrent.title} 本地结算完成但提交终态未确认，已转待对账"
-                    )
-        else:
-            if admission_key and attempt_token:
-                retry_at = self._subscription_download_retry_at(
+                raise DownloadReconciliationRequired(
+                    f"{prepared.torrent.title} 本地结算完成但提交终态未确认，已转待对账"
+                )
+
+    def _record_rejected_download(
+        self,
+        *,
+        prepared: _PreparedDownload,
+        context: Context,
+        episodes: Optional[Set[int]],
+        channel: Optional[NotificationChannel],
+        source: Optional[str],
+        downloader: Optional[str],
+        userid: Union[str, int, None],
+        actual_downloader: Optional[str],
+        error_msg: str,
+        admission_key: Optional[str],
+        attempt_token: Optional[str],
+    ) -> None:
+        """记录无外部副作用的明确拒绝，并通知原调用渠道。"""
+        if admission_key and attempt_token:
+            self._subscription_download_repository().mark_retryable(
+                idempotency_key=admission_key,
+                attempt_token=attempt_token,
+                available_at=self._subscription_download_retry_at(
                     error_msg,
                     self._download_failure_ttl(error_msg),
-                )
-                self._subscription_download_repository().mark_retryable(
-                    idempotency_key=admission_key,
-                    attempt_token=attempt_token,
-                    available_at=retry_at,
-                    error=error_msg,
-                )
-            # 下载失败
-            logger.error(f"{_media.title_year} 添加下载任务失败："
-                         f"{_torrent.title} - {_torrent.enclosure}，{error_msg}")
-            self._record_download_failure(
-                context=context,
-                error_msg=error_msg,
-                downloader=_downloader or downloader or _site_downloader,
-                source=source,
-                episodes=episodes,
+                ),
+                error=error_msg,
             )
-            # 只发送给对应渠道和用户
-            self.post_message(Message(
+        logger.error(
+            f"{prepared.media.title_year} 添加下载任务失败："
+            f"{prepared.torrent.title} - {prepared.torrent.enclosure}，{error_msg}"
+        )
+        self._record_download_failure(
+            context=context,
+            error_msg=error_msg,
+            downloader=actual_downloader or downloader or prepared.site_downloader,
+            source=source,
+            episodes=episodes,
+        )
+        self.post_message(
+            Message(
                 channel=channel,
                 source=source if channel else None,
                 mtype=MessageType.Manual,
-                title="添加下载任务失败：%s %s"
-                      % (_media.title_year, _meta.season_episode),
-                text=f"站点：{_torrent.site_name}\n"
-                     f"种子名称：{_meta.org_string}\n"
-                     f"错误信息：{error_msg}",
-                image=_media.get_message_image(),
-                userid=userid))
-        if return_detail:
-            return _hash, error_msg
-        return _hash
+                title=f"添加下载任务失败：{prepared.media.title_year} {prepared.meta.season_episode}",
+                text=(
+                    f"站点：{prepared.torrent.site_name}\n"
+                    f"种子名称：{prepared.meta.org_string}\n"
+                    f"错误信息：{error_msg}"
+                ),
+                image=prepared.media.get_message_image(),
+                userid=userid,
+            )
+        )

app/chain/search/provider.py

@@ -10,7 +10,7 @@ from datetime import datetime
 from typing import Any, AsyncGenerator, Awaitable, Callable, Dict, Iterator, List, Optional
 
 from app.application.configuration import get_configured_system_config
-from app.application.site.search_observation import (
+from app.application.site.observation import (
     capture_site_search_observation,
     report_site_search_outcome,
 )
@@ -107,8 +107,8 @@ class ProviderBatch:
     continued: bool
 
 
-class SearchProviderOwner(_SearchOwnerBase):
-    """站点与插件资源 provider fan-out owner。"""
+class _SearchProviderSyncOwner(_SearchOwnerBase):
+    """同步站点 provider 选择、预算与 fan-out owner。"""
 
     @staticmethod
     def _selected_site_ids(sites: Optional[List[int]]) -> List[int]:
@@ -404,6 +404,10 @@ class SearchProviderOwner(_SearchOwnerBase):
         finally:
             progress.end()
 
+
+class SearchProviderOwner(_SearchProviderSyncOwner):
+    """异步站点与插件资源 provider fan-out owner。"""
+
     async def _iter_provider_batches(
         self,
         *,

app/chain/subscribe/contract.py

@@ -49,20 +49,28 @@ if TYPE_CHECKING:
         _SubscribeChain__notify_subscribe_create_failure: Callable[..., Any]
         _SubscribeChain__post_subscribe_added: Callable[..., Any]
         _SubscribeChain__prepare_best_version_tv_candidate: Callable[..., Any]
+        _SubscribeChain__prepare_subscribe_progress_fields: Callable[..., Any]
         _SubscribeChain__prepare_total_episode_change_fields: Callable[..., Any]
         _SubscribeChain__refresh_subscribe_progress_with_no_exists: Callable[..., Any]
         _SubscribeChain__refresh_total_episode_before_completion: Callable[..., Any]
         _SubscribeChain__report_completed: Callable[..., Any]
+        _SubscribeChain__resolve_total_episode_decrease: Callable[..., Any]
         _acquire_run_lock: Callable[..., Any]
         _async_recognize_music_subscribe: Callable[..., Awaitable[Any]]
         _get_pending_best_version_episodes: Callable[..., Any]
         _is_episode_range_covered: Callable[..., Any]
         _is_music_download_complete: Callable[..., Any]
+        _load_search_subscriptions: Callable[..., Any]
         _match_music_subscribe: Callable[..., Any]
+        _notify_manual_search: Callable[..., Any]
+        _process_search_subscription: Callable[..., Any]
         _recognize_music_subscribe: Callable[..., Any]
+        _report_search_progress: Callable[..., Any]
         _search_music_subscribe: Callable[..., Any]
         _subscription_query: Callable[..., Any]
+        _defer_recent_subscription: Callable[..., Any]
         _validate_music_subscribe_target: Callable[..., Any]
+        _wait_before_scheduled_search: Callable[..., Any]
         add: Callable[..., Any]
         async def async_obtain_images(self, mediainfo: MediaInfo) -> Optional[MediaInfo]:
             """异步补全媒体图片。"""

app/chain/subscribe/match.py

@@ -2,7 +2,7 @@
 
 import copy
 from datetime import datetime
-from typing import Callable, Dict, List, Optional
+from typing import Any, Callable, Dict, List, Optional
 
 from app.application.configuration import get_configured_system_config
 from app.application.subscription.candidates import CandidateBatch, CandidateIndex
@@ -29,8 +29,51 @@ from app.schemas.types import (
 )
 
 
+def _prepare_subscription_match(
+    owner: Any,
+    subscribe: Any,
+    candidate_index: CandidateIndex,
+    fresh_fact_lease: FreshFactLease,
+) -> Optional[
+    tuple[MetaBase, MediaInfo, Dict[str, List[Context]], List[str], List[int]]
+]:
+    """为单个订阅路由候选，并在本轮新鲜事实租约中加载媒体信息。"""
+    try:
+        meta = build_subscribe_meta(subscribe)
+    except ValueError:
+        logger.error(f"订阅 {subscribe.name} 类型错误：{subscribe.type}")
+        return None
+    domains = owner.site_repository.get_domains_by_ids(subscribe.sites) if subscribe.sites else []
+    sub_sites = owner.get_sub_sites(subscribe)
+    routed_torrents = candidate_index.route_for_match(
+        subscribe,
+        domains=set(domains) if domains else None,
+        site_ids=set(sub_sites) if sub_sites else None,
+    )
+    if not routed_torrents:
+        logger.info(f"订阅 {subscribe.name} 本轮没有可能相关的资源，跳过资源匹配准备")
+        return None
+    mediainfo = fresh_fact_lease.get_or_load(
+        subscribe,
+        lambda: MediaChain().recognize_media(
+            meta=meta,
+            mtype=meta.type,
+            **subscribe_recognize_kwargs(subscribe),
+            episode_group=subscribe.episode_group,
+            cache=False,
+        ),
+    )
+    if not mediainfo:
+        logger.warning(
+            f"未识别到媒体信息，标题：{subscribe.name}，"
+            f"媒体来源：{subscribe.media_source}，媒体 ID：{subscribe.media_id}"
+        )
+        return None
+    return meta, mediainfo, routed_torrents, domains, sub_sites
+
+
 class SubscribeMatchOwner(_SubscribeOwnerBase):
-    """订阅候选准备、身份复核与资源匹配，作为 SubscribeChain 的单一职责实现 owner。"""
+    """订阅候选准备、身份复核与资源匹配 owner。"""
 
     def _prepare_match_torrents(
         self,
@@ -142,8 +185,7 @@ class SubscribeMatchOwner(_SubscribeOwnerBase):
                 return
 
             processed_torrents = self._prepare_match_torrents(torrents)
-            candidate_index = CandidateIndex(processed_torrents)
-            fresh_fact_lease = FreshFactLease()
+            candidate_index, fresh_fact_lease = CandidateIndex(processed_torrents), FreshFactLease()
 
             # 所有订阅
             subscribes = self.subscription_repository.list(self.get_states_for_search("R"))
@@ -174,41 +216,12 @@ class SubscribeMatchOwner(_SubscribeOwnerBase):
                         self._match_music_subscribe(subscribe, music_contexts)
                         continue
                     mediakey = subscribe_media_key(subscribe)
-                    try:
-                        meta = build_subscribe_meta(subscribe)
-                    except ValueError:
-                        logger.error(f"订阅 {subscribe.name} 类型错误：{subscribe.type}")
-                        continue
-                    # 订阅的站点域名列表
-                    domains = []
-                    if subscribe.sites:
-                        domains = self.site_repository.get_domains_by_ids(subscribe.sites)
-                    sub_sites = self.get_sub_sites(subscribe)
-                    routed_torrents = candidate_index.route_for_match(
-                        subscribe,
-                        domains=set(domains) if domains else None,
-                        site_ids=set(sub_sites) if sub_sites else None,
+                    prepared = _prepare_subscription_match(
+                        self, subscribe, candidate_index, fresh_fact_lease
                     )
-                    if not routed_torrents:
-                        logger.info(f"订阅 {subscribe.name} 本轮没有可能相关的资源，跳过资源匹配准备")
-                        continue
-                    # 识别媒体信息
-                    mediainfo = fresh_fact_lease.get_or_load(
-                        subscribe,
-                        lambda: MediaChain().recognize_media(
-                            meta=meta,
-                            mtype=meta.type,
-                            **subscribe_recognize_kwargs(subscribe),
-                            episode_group=subscribe.episode_group,
-                            cache=False,
-                        ),
-                    )
-                    if not mediainfo:
-                        logger.warn(
-                            f"未识别到媒体信息，标题：{subscribe.name}，"
-                            f"媒体来源：{subscribe.media_source}，媒体 ID：{subscribe.media_id}"
-                        )
+                    if prepared is None:
                         continue
+                    meta, mediainfo, routed_torrents, domains, sub_sites = prepared
 
                     # 如果媒体已存在或已下载完毕，跳过当前订阅处理
                     exist_flag, no_exists = self.check_and_handle_existing_media(
@@ -505,7 +518,6 @@ class SubscribeMatchOwner(_SubscribeOwnerBase):
             if lock_acquired:
                 self._rlock.release()
                 logger.debug(f"match Lock released at {datetime.now()}")
-
     @staticmethod
     def _SubscribeChain__get_media_id_match_source(mediainfo: Optional[MediaInfo]) -> str:
         """

app/chain/subscribe/metadata.py

@@ -0,0 +1,144 @@
+"""订阅单条元数据刷新与完成对账协作者。"""
+
+from collections.abc import Callable
+from typing import Any, Optional, cast
+
+from app.application.subscription.contract import (
+    SubscriptionSnapshot,
+    build_subscribe_meta,
+    subscribe_media_key,
+)
+from app.application.subscription.facts import FreshFactLease
+from app.chain.subscribe.contract import _SubscribeOwnerBase
+from app.chain.subscribe.identity import subscribe_recognize_kwargs
+from app.runtime.log import logger
+from app.schemas.media import resolve_media_identity
+from app.schemas.types import MUSIC_ENTITY_ALBUM, MediaType
+
+
+class SubscribeMetadataOwner(_SubscribeOwnerBase):
+    """刷新一条订阅的新鲜媒体事实、持久字段与可选完成状态。"""
+
+    def _check_subscription(
+        self,
+        subscribe: SubscriptionSnapshot,
+        fresh_fact_lease: FreshFactLease,
+        *,
+        reconcile_completion: bool,
+        media_chain_factory: Callable[[], Any],
+    ) -> Optional[SubscriptionSnapshot]:
+        """刷新单条订阅元数据，并按需复用同一事实执行完成对账。"""
+        try:
+            meta = build_subscribe_meta(subscribe)
+        except ValueError:
+            logger.error(f"订阅 {subscribe.name} 类型错误：{subscribe.type}")
+            return None
+        if meta.type == MediaType.MUSIC:
+            mediainfo = self._recognize_music_subscribe(subscribe)
+        else:
+            mediainfo = fresh_fact_lease.get_or_load(
+                subscribe,
+                lambda: media_chain_factory().recognize_media(
+                    meta=meta,
+                    mtype=meta.type,
+                    **subscribe_recognize_kwargs(subscribe),
+                    episode_group=subscribe.episode_group,
+                    cache=False,
+                ),
+            )
+        if not mediainfo:
+            logger.warning(
+                f"未识别到媒体信息，标题：{subscribe.name}，"
+                f"媒体来源：{subscribe.media_source}，媒体 ID：{subscribe.media_id}"
+            )
+            return None
+        episodes = (
+            mediainfo.seasons.get(subscribe.season) or []
+            if meta.type == MediaType.TV
+            else []
+        )
+        progress_update: dict[str, Any]
+        if (
+            subscribe.type == MediaType.TV.value
+            and not subscribe.manual_total_episode
+            and episodes
+        ):
+            current_total_episode = len(episodes)
+            total_episode = self._SubscribeChain__apply_episodes_refresh(
+                current_total_episode,
+                season=subscribe.season,
+                mediainfo=mediainfo,
+                media_source=subscribe.media_source,
+                media_id=subscribe.media_id,
+                subscribe_id=subscribe.id,
+                scene="refresh",
+            )
+            old_total_episode = subscribe.total_episode or 0
+            if total_episode and total_episode < old_total_episode:
+                total_episode = self._SubscribeChain__resolve_total_episode_decrease(
+                    subscribe=subscribe,
+                    candidate_total=total_episode,
+                    meta=meta,
+                    mediainfo=mediainfo,
+                    mediakey=subscribe_media_key(subscribe),
+                )
+            if total_episode and total_episode != old_total_episode:
+                progress_update = self._SubscribeChain__prepare_total_episode_change_fields(
+                    subscribe=subscribe,
+                    total_episode=total_episode,
+                    old_total_episode=old_total_episode,
+                )
+            else:
+                total_episode = subscribe.total_episode
+                progress_update = {"lack_episode": subscribe.lack_episode}
+                if subscribe.best_version:
+                    progress_update = self._SubscribeChain__prepare_subscribe_progress_fields(
+                        subscribe=subscribe,
+                        no_exists={},
+                    )
+            logger.info(
+                f"订阅 {subscribe.name} 总集数变化，更新总集数为{total_episode}，"
+                f"缺失集数为{progress_update.get('lack_episode', subscribe.lack_episode)} ..."
+            )
+        else:
+            total_episode = subscribe.total_episode
+            progress_update = {"lack_episode": subscribe.lack_episode}
+            if subscribe.best_version and subscribe.type == MediaType.TV.value:
+                progress_update = self._SubscribeChain__prepare_subscribe_progress_fields(
+                    subscribe=subscribe,
+                    no_exists={},
+                )
+        update_data = {
+            "name": mediainfo.title,
+            "year": str(mediainfo.year) if mediainfo.year is not None else None,
+            "vote": mediainfo.vote_average,
+            "poster": mediainfo.get_poster_image(),
+            "backdrop": mediainfo.get_backdrop_image(),
+            "description": mediainfo.overview,
+            "media_source": resolve_media_identity(media=mediainfo)[0],
+            "media_id": resolve_media_identity(media=mediainfo)[1],
+            "total_episode": total_episode,
+        }
+        if meta.type == MediaType.MUSIC:
+            music_type = getattr(mediainfo, "music_type", None)
+            update_data.update(
+                {
+                    "music_type": music_type,
+                    "total_tracks": getattr(mediainfo, "total_tracks", None)
+                    if music_type == MUSIC_ENTITY_ALBUM
+                    else None,
+                }
+            )
+        update_data.update(progress_update)
+        updated = self._SubscribeChain__apply_subscribe_update(
+            subscribe,
+            update_data,
+            scene="metadata_refresh",
+        )
+        if reconcile_completion and updated.state in self.get_states_for_search("R"):
+            self.reconcile_subscription_completion(
+                subscribe=updated,
+                meta=meta,
+                mediainfo=mediainfo,
+            )
+        return cast(SubscriptionSnapshot, updated)

app/chain/subscribe/policy.py

@@ -21,8 +21,8 @@ from app.schemas.types import (
 )
 
 
-class SubscribePolicyOwner(_SubscribeOwnerBase):
-    """订阅优先级、剧集范围与来源编码策略，作为 SubscribeChain 的单一职责实现 owner。"""
+class _SubscribePriorityPolicyOwner(_SubscribeOwnerBase):
+    """订阅优先级、剧集范围与来源编码策略 owner。"""
 
     @staticmethod
     def _SubscribeChain__normalize_episode_priority(
@@ -179,6 +179,10 @@ class SubscribePolicyOwner(_SubscribeOwnerBase):
         )
         return result
 
+
+class SubscribePolicyOwner(_SubscribePriorityPolicyOwner):
+    """订阅提交前复核与下载治理策略 owner。"""
+
     @staticmethod
     def _SubscribeChain__get_downloaded_episodes(downloads: Optional[List[Context]]) -> List[int]:
         """获取本次下载实际涉及的剧集。"""

app/chain/subscribe/refresh.py

@@ -5,6 +5,7 @@ from datetime import datetime
 from typing import Any, Callable, Dict, List, Optional, Tuple, Union, cast
 
 from app.application.subscription import priority as _priority
+from app.application.subscription.candidates import CandidateBatch
 from app.application.subscription.contract import (
     SubscriptionSnapshot,
     build_subscribe_meta,
@@ -14,8 +15,8 @@ from app.application.subscription.contract import (
 from app.application.subscription.facts import FreshFactLease
 from app.chain.download import DownloadChain
 from app.chain.media import MediaChain
-from app.chain.subscribe.contract import _SubscribeOwnerBase
 from app.chain.subscribe.identity import subscribe_recognize_kwargs
+from app.chain.subscribe.metadata import SubscribeMetadataOwner
 from app.chain.tmdb import TmdbChain
 from app.chain.torrents import TorrentsChain
 from app.domain.context import (
@@ -26,17 +27,15 @@ from app.runtime.events import eventmanager
 from app.runtime.log import logger
 from app.runtime.stop import runtime_stop_state
 from app.schemas.event import SubscribeEpisodesRefreshEventData
-from app.schemas.media import resolve_media_identity
 from app.schemas.mediaserver import NotExistMediaInfo as _SchemaNotExistMediaInfo
 from app.schemas.types import (
-    MUSIC_ENTITY_ALBUM,
     ChainEventType,
     MediaSource,
     MediaType,
 )
 
 
-class SubscribeRefreshOwner(_SubscribeOwnerBase):
+class SubscribeRefreshOwner(SubscribeMetadataOwner):
     """订阅元数据、进度与剧集范围刷新，作为 SubscribeChain 的单一职责实现 owner。"""
 
     def refresh(self, progress_callback: Optional[Callable[..., None]] = None) -> None:
@@ -78,12 +77,20 @@ class SubscribeRefreshOwner(_SubscribeOwnerBase):
                     data=data,
                 )
 
-        candidate_batch = TorrentsChain().refresh_batch(
+        torrents_chain = TorrentsChain()
+        candidate_batch = torrents_chain.refresh_batch(
             sites=sites,
             progress_callback=_update_refresh_progress if progress_callback else None,
             # 存在音乐订阅时额外抓取站点音乐专用入口，音乐不一定在默认种子首页
             include_music=self.has_music_subscribe(),
         )
+        if not isinstance(candidate_batch, CandidateBatch):
+            legacy_candidates = torrents_chain.refresh(
+                sites=sites,
+                progress_callback=_update_refresh_progress if progress_callback else None,
+                include_music=self.has_music_subscribe(),
+            )
+            candidate_batch = CandidateBatch.from_legacy(legacy_candidates or {}, source="refresh")
         self.match_batch(
             candidate_batch,
             progress_callback=_update_match_progress if progress_callback else None,
@@ -128,114 +135,15 @@ class SubscribeRefreshOwner(_SubscribeOwnerBase):
                         "current": subscribe.id,
                     },
                 )
-            try:
-                meta = build_subscribe_meta(subscribe)
-            except ValueError:
-                logger.error(f"订阅 {subscribe.name} 类型错误：{subscribe.type}")
-                continue
-            # 识别媒体信息
-            if meta.type == MediaType.MUSIC:
-                mediainfo = self._recognize_music_subscribe(subscribe)
-            else:
-                mediainfo = fresh_fact_lease.get_or_load(
-                    subscribe,
-                    lambda: MediaChain().recognize_media(
-                        meta=meta,
-                        mtype=meta.type,
-                        **subscribe_recognize_kwargs(subscribe),
-                        episode_group=subscribe.episode_group,
-                        cache=False,
-                    ),
-                )
-            if not mediainfo:
-                logger.warn(
-                    f"未识别到媒体信息，标题：{subscribe.name}，"
-                    f"媒体来源：{subscribe.media_source}，媒体 ID：{subscribe.media_id}"
-                )
-                continue
-            # 对于电视剧，获取当前季的总集数
-            episodes = (mediainfo.seasons.get(subscribe.season) or []) if meta.type == MediaType.TV else []
-            progress_update = {}
-            if subscribe.type == MediaType.TV.value and not subscribe.manual_total_episode and len(episodes):
-                current_total_episode = len(episodes)
-                # 外部事件只能向上覆盖主程序本次识别到的 TMDB 当前季总集数，已有订阅按最终 total 跟随持久化。
-                total_episode = self._SubscribeChain__apply_episodes_refresh(
-                    current_total_episode,
-                    season=subscribe.season,
-                    mediainfo=mediainfo,
-                    media_source=subscribe.media_source,
-                    media_id=subscribe.media_id,
-                    subscribe_id=subscribe.id,
-                    scene="refresh",
-                )
-                old_total_episode = subscribe.total_episode or 0
-                if total_episode and total_episode < old_total_episode:
-                    total_episode = self._SubscribeChain__resolve_total_episode_decrease(
-                        subscribe=subscribe,
-                        candidate_total=total_episode,
-                        meta=meta,
-                        mediainfo=mediainfo,
-                        mediakey=subscribe_media_key(subscribe),
-                    )
-                if total_episode and total_episode != old_total_episode:
-                    progress_update = self._SubscribeChain__prepare_total_episode_change_fields(
-                        subscribe=subscribe,
-                        total_episode=total_episode,
-                        old_total_episode=old_total_episode,
-                    )
-                else:
-                    total_episode = subscribe.total_episode
-                    progress_update = {"lack_episode": subscribe.lack_episode}
-                    if subscribe.best_version and subscribe.type == MediaType.TV.value:
-                        progress_update = self._SubscribeChain__prepare_subscribe_progress_fields(
-                            subscribe=subscribe, no_exists={}
-                        )
-                logger.info(
-                    f"订阅 {subscribe.name} 总集数变化，更新总集数为{total_episode}，"
-                    f"缺失集数为{progress_update.get('lack_episode', subscribe.lack_episode)} ..."
-                )
-            else:
-                total_episode = subscribe.total_episode
-                progress_update = {"lack_episode": subscribe.lack_episode}
-                if subscribe.best_version and subscribe.type == MediaType.TV.value:
-                    progress_update = self._SubscribeChain__prepare_subscribe_progress_fields(
-                        subscribe=subscribe, no_exists={}
-                    )
-            # 更新TMDB信息
-            update_data = {
-                "name": mediainfo.title,
-                "year": str(mediainfo.year) if mediainfo.year is not None else None,
-                "vote": mediainfo.vote_average,
-                "poster": mediainfo.get_poster_image(),
-                "backdrop": mediainfo.get_backdrop_image(),
-                "description": mediainfo.overview,
-                "media_source": resolve_media_identity(media=mediainfo)[0],
-                "media_id": resolve_media_identity(media=mediainfo)[1],
-                "total_episode": total_episode,
-            }
-            if meta.type == MediaType.MUSIC:
-                music_type = getattr(mediainfo, "music_type", None)
-                update_data.update(
-                    {
-                        "music_type": music_type,
-                        "total_tracks": getattr(mediainfo, "total_tracks", None)
-                        if music_type == MUSIC_ENTITY_ALBUM
-                        else None,
-                    }
-                )
-            update_data.update(progress_update)
-            subscribe = self._SubscribeChain__apply_subscribe_update(
+            updated_subscribe = self._check_subscription(
                 subscribe,
-                update_data,
-                scene="metadata_refresh",
+                fresh_fact_lease,
+                reconcile_completion=reconcile_completion,
+                media_chain_factory=MediaChain,
             )
-            if reconcile_completion and subscribe.state in self.get_states_for_search("R"):
-                self.reconcile_subscription_completion(
-                    subscribe=subscribe,
-                    meta=meta,
-                    mediainfo=mediainfo,
-                )
-            logger.info(f"{subscribe.name} 订阅元数据更新完成")
+            if updated_subscribe is None:
+                continue
+            logger.info(f"{updated_subscribe.name} 订阅元数据更新完成")
             if progress_callback:
                 progress_callback(
                     value=index / total_num * 100 if total_num else 100,

app/chain/subscribe/search.py

@@ -36,8 +36,8 @@ from app.schemas.types import (
 )
 
 
-class SubscribeSearchOwner(_SubscribeOwnerBase):
-    """订阅主动搜索编排，作为 SubscribeChain 的单一职责实现 owner。"""
+class _SubscribeSearchQueueOwner(_SubscribeOwnerBase):
+    """订阅主动搜索入口与持久队列消费 owner。"""
 
     def _subscription_query(self) -> SubscriptionQueryService:
         """构造绑定订阅 Oper 的查询应用服务。"""
@@ -481,6 +481,10 @@ class SubscribeSearchOwner(_SubscribeOwnerBase):
         )
         return queue.get_batch(batch_id) if queue else None
 
+
+class SubscribeSearchOwner(_SubscribeSearchQueueOwner):
+    """订阅搜索加载、单项处理与结果通知 owner。"""
+
     def _load_search_subscriptions(
         self,
         sid: Optional[int],

app/db/oper/subscriptiondownload.py

@@ -1,7 +1,7 @@
 """订阅下载提交账本的事务内状态转换。"""
 
 from datetime import datetime, timezone
-from typing import Optional
+from typing import Optional, cast
 from uuid import uuid4
 
 from sqlalchemy import and_, or_, select, update
@@ -106,11 +106,11 @@ class SubscriptionDownloadOper(DbOper):
         """按稳定幂等键读取提交记录。"""
         if not isinstance(self._db, Session):
             raise RuntimeError("订阅下载提交查询需要调用方提供同步 Session")
-        return self._db.execute(
+        return cast(Optional[SubscriptionDownloadSubmission], self._db.execute(
             select(SubscriptionDownloadSubmission).where(
                 SubscriptionDownloadSubmission.idempotency_key == idempotency_key
             )
-        ).scalars().first()
+        ).scalars().first())
 
     def mark_accepted(
         self,

app/locales/en-US.json

@@ -346,6 +346,8 @@
     "微信 ClawBot 通知未启用或配置尚未保存，请先保存并启用当前渠道": "WeChat ClawBot notification is not enabled or the configuration has not been saved. Please save and enable this channel first",
     "请输入至少一个有效的站点 ID": "Enter at least one valid site ID",
     "所有订阅搜索完成": "All subscription searches are complete",
+    "订阅搜索批次不存在": "The subscription search batch does not exist",
+    "订阅搜索批次已结束或无法取消": "The subscription search batch has ended or cannot be cancelled",
     "订阅搜索锁等待超时，已跳过本轮": "Subscription search lock timed out, this round was skipped",
     "订阅匹配锁等待超时，已跳过本轮": "Subscription matching lock timed out, this round was skipped",
     "请输入订阅 ID，多个 ID 用空格分隔，或输入 all": "Enter subscription IDs separated by spaces, or enter all",

app/locales/zh-TW.json

@@ -342,6 +342,8 @@
     "微信 ClawBot 通知未启用或配置尚未保存，请先保存并启用当前渠道": "微信 ClawBot 通知未啟用或設定尚未儲存，請先儲存並啟用目前渠道",
     "请输入至少一个有效的站点 ID": "請輸入至少一個有效的站點 ID",
     "所有订阅搜索完成": "所有訂閱搜尋完成",
+    "订阅搜索批次不存在": "訂閱搜尋批次不存在",
+    "订阅搜索批次已结束或无法取消": "訂閱搜尋批次已結束或無法取消",
     "请输入订阅 ID，多个 ID 用空格分隔，或输入 all": "請輸入訂閱 ID，多個 ID 以空格分隔，或輸入 all",
     "请输入至少一个有效的订阅 ID": "請輸入至少一個有效的訂閱 ID",
     "格式错误，请输入：cookie <id> <username> <password> [2fa_code/secret]": "格式錯誤，請輸入：cookie <id> <username> <password> [2fa_code/secret]",

app/modules/indexer/__init__.py

@@ -4,8 +4,8 @@ from types import MappingProxyType
 from typing import Any, Callable, List, Mapping, Optional, Tuple, Union, cast
 
 from app.application.site.health import get_configured_site_health_service
+from app.application.site.observation import report_site_search_outcome
 from app.application.site.query import get_configured_site_query_service
-from app.application.site.search_observation import report_site_search_outcome
 from app.application.site.sites import SitesHelper  # pylint: disable=import-error,no-name-in-module
 from app.domain import site as site_rules
 from app.domain.context import Context, SubtitleInfo, TorrentInfo

app/scheduler/catalog.py

@@ -23,6 +23,7 @@ from app.application.scheduling import (  # noqa: E402
 )
 from app.runtime.scheduling import TimerUtils
 from app.scheduler.contract import _SchedulerOwnerBase
+from app.scheduler.services import SchedulerServices
 from app.schemas.system import MediaServerConf as _SchemaMediaServerConf
 
 
@@ -35,6 +36,29 @@ class _MediaServerSchedule(TypedDict):
     interval: int
 
 
+def _subscription_search_job_specs(services: SchedulerServices) -> tuple[JobSpec, ...]:
+    """构造订阅搜索、新增搜索与持久队列恢复任务目录。"""
+    return (
+        JobSpec(
+            "subscribe_search", "订阅搜索补全", services.search_subscribe, "subscription", kwargs={"state": "R"}
+        ),
+        JobSpec(
+            "new_subscribe_search",
+            "新增订阅搜索",
+            services.search_subscribe,
+            "subscription",
+            kwargs={"state": "N"},
+        ),
+        JobSpec(
+            "subscribe_search_queue",
+            "恢复订阅搜索队列",
+            services.resume_subscribe_search,
+            "subscription",
+            recovery=JobRecoveryPolicy.DURABLE_QUEUE,
+        ),
+    )
+
+
 class SchedulerCatalogOwner(_SchedulerOwnerBase):
     """调度器静态任务目录与 APScheduler 投影。"""
 
@@ -123,6 +147,18 @@ class SchedulerCatalogOwner(_SchedulerOwnerBase):
             replace_existing=True,
         )
 
+    def _register_subscription_search_queue_job(self, config: SchedulerRuntimeConfig) -> None:
+        """注册短周期持久搜索队列恢复任务。"""
+        self._scheduler.add_job(
+            self.start,
+            "interval",
+            id="subscribe_search_queue",
+            name="恢复订阅搜索队列",
+            minutes=1,
+            next_run_time=datetime.now(pytz.timezone(config.timezone)) + timedelta(seconds=10),
+            kwargs={"job_id": "subscribe_search_queue"},
+        )
+
     def _initialize_catalog(self, config: SchedulerRuntimeConfig) -> None:
         """构建完整任务目录并投影到尚未启动的 APScheduler。"""
         services = self._scheduler_services()
@@ -132,23 +168,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"),
-                JobSpec(
-                    "subscribe_search", "订阅搜索补全", services.search_subscribe, "subscription", kwargs={"state": "R"}
-                ),
-                JobSpec(
-                    "new_subscribe_search",
-                    "新增订阅搜索",
-                    services.search_subscribe,
-                    "subscription",
-                    kwargs={"state": "N"},
-                ),
-                JobSpec(
-                    "subscribe_search_queue",
-                    "恢复订阅搜索队列",
-                    services.resume_subscribe_search,
-                    "subscription",
-                    recovery=JobRecoveryPolicy.DURABLE_QUEUE,
-                ),
+                *_subscription_search_job_specs(services),
                 JobSpec("subscribe_refresh", "订阅刷新", services.refresh_subscribe, "subscription"),
                 JobSpec("subscribe_follow", "关注的订阅分享", services.follow_subscribe, "subscription"),
                 JobSpec(
@@ -253,15 +273,7 @@ class SchedulerCatalogOwner(_SchedulerOwnerBase):
             )
 
         # 新增订阅时搜索（5分钟检查一次）
-        self._scheduler.add_job(
-            self.start,
-            "interval",
-            id="subscribe_search_queue",
-            name="恢复订阅搜索队列",
-            minutes=1,
-            next_run_time=datetime.now(pytz.timezone(config.timezone)) + timedelta(seconds=10),
-            kwargs={"job_id": "subscribe_search_queue"},
-        )
+        self._register_subscription_search_queue_job(config)
 
         self._scheduler.add_job(
             self.start,

app/schemas/exports.py

@@ -396,6 +396,8 @@ SCHEMA_EXPORTS = {
     'SubscribeShare': ('app.schemas.subscribe', 'SubscribeShare'),
     'SubscribeShareStatistics': ('app.schemas.subscribe', 'SubscribeShareStatistics'),
     'Subscription': ('app.schemas.message', 'Subscription'),
+    'SubscriptionBatchStatus': ('app.schemas.subscribe', 'SubscriptionBatchStatus'),
+    'SubscriptionExecutionStatus': ('app.schemas.subscribe', 'SubscriptionExecutionStatus'),
     'SubscriptionMessage': ('app.schemas.message', 'SubscriptionMessage'),
     'SubtitleDownloadData': ('app.schemas.download', 'SubtitleDownloadData'),
     'SubtitleInfo': ('app.schemas.search', 'SubtitleInfo'),

app/schemas/subscribe.py

@@ -45,7 +45,7 @@ def compute_subscribe_completed_episode(subscribe: "Subscribe") -> Optional[int]
     return min(max(start_episode - 1, 0), total_episode) + priority_completed
 
 
-class SubscriptionExecutionStatus(BaseModel):
+class SubscriptionExecutionStatus(BaseModel):  # type: ignore[misc]
     """订阅列表可见的当前业务执行状态。"""
 
     state: str
@@ -63,7 +63,7 @@ class SubscriptionExecutionStatus(BaseModel):
     model_config = ConfigDict(from_attributes=True)
 
 
-class SubscriptionBatchStatus(BaseModel):
+class SubscriptionBatchStatus(BaseModel):  # type: ignore[misc]
     """订阅搜索批次的用户可见进度和操作能力。"""
 
     batch_id: str

docs/architecture-overview.md

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

docs/architecture/agent-api-surface-audit.json

@@ -6,7 +6,7 @@
     "provider-skill": 11,
     "stream_or_binary": 10,
     "transport_or_identity": 66,
-    "ui_presentation": 5
+    "ui_presentation": 8
   },
   "dynamic_gateway_routes": [
     {
@@ -21,7 +21,7 @@
   "gateway_http_route_count": 201,
   "gateway_operation_count": 203,
   "matched_gateway_http_route_count": 200,
-  "openapi_operation_count": 375,
+  "openapi_operation_count": 378,
   "operations": [
     {
       "disposition": "consolidated",
@@ -3367,6 +3367,42 @@
         "subscribe"
       ]
     },
+    {
+      "disposition": "ui_presentation",
+      "method": "GET",
+      "operation_ids": [],
+      "owner": "host-ui",
+      "path": "/api/v1/subscribe/execution/batches",
+      "reason": "Background subscription execution status and cancellation are owned by the authenticated frontend workflow; they are not yet a stable Agent gateway contract.",
+      "summary": "查询订阅搜索批次状态",
+      "tags": [
+        "subscribe"
+      ]
+    },
+    {
+      "disposition": "ui_presentation",
+      "method": "GET",
+      "operation_ids": [],
+      "owner": "host-ui",
+      "path": "/api/v1/subscribe/execution/batches/{batch_id}",
+      "reason": "Background subscription execution status and cancellation are owned by the authenticated frontend workflow; they are not yet a stable Agent gateway contract.",
+      "summary": "查询订阅搜索批次",
+      "tags": [
+        "subscribe"
+      ]
+    },
+    {
+      "disposition": "ui_presentation",
+      "method": "PUT",
+      "operation_ids": [],
+      "owner": "host-ui",
+      "path": "/api/v1/subscribe/execution/batches/{batch_id}/cancel",
+      "reason": "Background subscription execution status and cancellation are owned by the authenticated frontend workflow; they are not yet a stable Agent gateway contract.",
+      "summary": "取消订阅搜索批次",
+      "tags": [
+        "subscribe"
+      ]
+    },
     {
       "disposition": "gateway",
       "method": "GET",

docs/architecture/agent-api-surface-audit.md

@@ -5,7 +5,7 @@
 
 ## Result
 
-- OpenAPI HTTP operations: **375**
+- OpenAPI HTTP operations: **378**
 - Stable `moviepilot_api` operations: **203**
 - Exact HTTP routes used by the gateway: **201**
 - OpenAPI routes matched directly by the gateway: **200**
@@ -23,7 +23,7 @@
 | `provider-skill` | 11 | Low-level downloader or media-server capability owned by a provider Skill. |
 | `stream_or_binary` | 10 | Streaming or binary response owned by a direct client transport. |
 | `transport_or_identity` | 66 | Authentication, protocol, callback, account, or conversation transport boundary. |
-| `ui_presentation` | 5 | Frontend or plugin-rendered presentation contract. |
+| `ui_presentation` | 8 | Frontend or plugin-rendered presentation contract. |
 
 ## Bounded Dynamic Routes
 
@@ -291,6 +291,9 @@
 | `POST` | `/api/v1/subscribe/` | subscribe | `gateway` | subscription.add | 新增订阅 |
 | `PUT` | `/api/v1/subscribe/` | subscribe | `gateway` | subscription.update | 更新订阅 |
 | `GET` | `/api/v1/subscribe/check` | subscribe | `gateway` | subscription.metadata.refresh | 刷新订阅 TMDB 信息 |
+| `GET` | `/api/v1/subscribe/execution/batches` | subscribe | `ui_presentation` | host-ui | 查询订阅搜索批次状态 |
+| `GET` | `/api/v1/subscribe/execution/batches/{batch_id}` | subscribe | `ui_presentation` | host-ui | 查询订阅搜索批次 |
+| `PUT` | `/api/v1/subscribe/execution/batches/{batch_id}/cancel` | subscribe | `ui_presentation` | host-ui | 取消订阅搜索批次 |
 | `GET` | `/api/v1/subscribe/files/{subscribe_id}` | subscribe | `gateway` | subscription.files | 订阅相关文件信息 |
 | `DELETE` | `/api/v1/subscribe/follow` | subscribe | `gateway` | subscription.follow.delete | 取消Follow订阅分享人 |
 | `GET` | `/api/v1/subscribe/follow` | subscribe | `gateway` | subscription.follow.list | 查询已Follow的订阅分享人 |

docs/architecture/optimization-checklist.md

@@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
 
 | 指标 | 当前值 | 解释 |
 |---|---:|---|
-| 宿主 Python 模块 / 内部依赖边 | 932 / 7,784 | `dependency-baseline.json` 当前快照 |
+| 宿主 Python 模块 / 内部依赖边 | 936 / 7,835 | `dependency-baseline.json` 当前快照 |
 | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 |
 | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 |
 | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |
@@ -102,8 +102,8 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
 | Event Contract | 53 | 均已有 payload model，但当前全部是 diagnostic enforcement |
 | Python 源码量 | 305,884 行 | 排除 `app/plugins/**`；61 个文件超过 1,000 行，11 个超过 2,000 行 |
 | 长方法 | 290 个超过 80 行 | AST 统计排除 `app/plugins/**`；65 个超过 150 行，21 个超过 250 行 |
-| 全量 mypy 历史债务 | 9,589 / 517 文件 | Agent API 重构后的现状基线；canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降 |
-| Ruff 历史诊断 | 568 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
+| 全量 mypy 历史债务 | 9,588 / 517 文件 | Agent API 重构后的现状基线；canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降 |
+| Ruff 历史诊断 | 567 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
 | 覆盖率固定基线 | Application 80.00%，Domain 80.00% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 |
 
 ### 3.3 热点文件

scripts/generate_agent_api_surface_audit.py

@@ -106,6 +106,7 @@ EXPLICIT_TRANSPORT_PATHS = frozenset(
         "/api/v1/system/ping",
     }
 )
+SUBSCRIPTION_EXECUTION_UI_PREFIX = "/api/v1/subscribe/execution/"
 
 
 def _gateway_routes() -> dict[tuple[str, str], list[str]]:
@@ -162,6 +163,13 @@ def _classify(
             "Plugin-rendered page, dashboard, or navigation metadata owned by the frontend presentation contract rather than an Agent business action.",
             [],
         )
+    if path.startswith(SUBSCRIPTION_EXECUTION_UI_PREFIX):
+        return (
+            "ui_presentation",
+            "host-ui",
+            "Background subscription execution status and cancellation are owned by the authenticated frontend workflow; they are not yet a stable Agent gateway contract.",
+            [],
+        )
     if path in EXPLICIT_TRANSPORT_PATHS:
         return (
             "transport_or_identity",

scripts/generate_agent_skill_docs.py

@@ -116,6 +116,26 @@ DATABASE_TABLE_GUIDES: dict[str, tuple[str, str, str]] = {
         "Auditing historical subscriptions, media identity, completion criteria, and filter configuration.",
         "Generated by subscription completion and archival; restore or delete through its business API.",
     ),
+    "subscriptiondownloadsubmission": (
+        "Stores subscription download idempotency claims, downloader acceptance facts, retries, and reconciliation freezes.",
+        "Diagnosing duplicate suppression, uncertain downloader outcomes, retry timing, and task ownership.",
+        "Owned by the subscription download submission state machine; never force accepted, succeeded, or retryable states.",
+    ),
+    "subscriptionsearchbatch": (
+        "Stores durable subscription search batches, source, aggregate state, counts, and cancellation requests.",
+        "Inspecting user-visible search progress, recovery state, cancellation, and terminal outcomes.",
+        "Owned by subscription search orchestration; create and cancel batches through the subscription API.",
+    ),
+    "subscriptionsearchtask": (
+        "Stores one durable subscription search task per batch and subscription with leases and execution phases.",
+        "Diagnosing queued, running, failed, cancelled, or recovered work and its current site.",
+        "Advanced only by the search queue lease state machine; never rewrite leases or terminal states manually.",
+    ),
+    "subscriptionsitebudget": (
+        "Stores per-site subscription search concurrency, cooldown, health, and fairness state.",
+        "Diagnosing site pressure, cooldown deferrals, recent failures, and active search ownership.",
+        "Owned by the subscription site-budget coordinator; do not clear cooldowns or counters by direct SQL.",
+    ),
     "systemconfig": (
         "Stores JSON business configuration values keyed by SystemConfigKey.",
         "Verifying the physical value only when the managed settings API behaves unexpectedly.",

skills/database-operation/SKILL.md

@@ -236,6 +236,30 @@ python scripts/mp-db.py write "UPDATE subscribe SET state = 'S' WHERE id = 123"
 - 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`, `filter_groups`, `episode_group`
 
+### `subscriptiondownloadsubmission`
+- Purpose: Stores subscription download idempotency claims, downloader acceptance facts, retries, and reconciliation freezes.
+- Useful queries: Diagnosing duplicate suppression, uncertain downloader outcomes, retry timing, and task ownership.
+- Write boundary: Owned by the subscription download submission state machine; never force accepted, succeeded, or retryable states.
+- Columns: `id`, `idempotency_key`, `subscription_id`, `task_id`, `logical_identity`, `resource_key`, `coverage`, `mode`, `delivery_scope`, `state`, `attempt_count`, `attempt_token`, `downloader`, `download_hash`, `available_at`, `last_error`, `created_at`, `updated_at`, `started_at`, `finished_at`
+
+### `subscriptionsearchbatch`
+- Purpose: Stores durable subscription search batches, source, aggregate state, counts, and cancellation requests.
+- Useful queries: Inspecting user-visible search progress, recovery state, cancellation, and terminal outcomes.
+- Write boundary: Owned by subscription search orchestration; create and cancel batches through the subscription API.
+- Columns: `id`, `batch_id`, `source`, `state`, `priority`, `total_count`, `finished_count`, `failed_count`, `cancelled_count`, `cancel_requested`, `created_at`, `updated_at`, `started_at`, `finished_at`, `last_error`
+
+### `subscriptionsearchtask`
+- Purpose: Stores one durable subscription search task per batch and subscription with leases and execution phases.
+- Useful queries: Diagnosing queued, running, failed, cancelled, or recovered work and its current site.
+- Write boundary: Advanced only by the search queue lease state machine; never rewrite leases or terminal states manually.
+- Columns: `id`, `task_id`, `batch_id`, `subscription_id`, `active_key`, `source`, `priority`, `position`, `state`, `phase`, `current_site_id`, `attempt_count`, `cancel_requested`, `lease_owner`, `lease_token`, `lease_expires_at`, `available_at`, `created_at`, `updated_at`, `started_at`, `finished_at`, `last_error`
+
+### `subscriptionsitebudget`
+- Purpose: Stores per-site subscription search concurrency, cooldown, health, and fairness state.
+- Useful queries: Diagnosing site pressure, cooldown deferrals, recent failures, and active search ownership.
+- Write boundary: Owned by the subscription site-budget coordinator; do not clear cooldowns or counters by direct SQL.
+- Columns: `id`, `site_id`, `lease_owner`, `lease_token`, `lease_expires_at`, `next_allowed_at`, `consecutive_failures`, `success_streak`, `last_outcome`, `last_error`, `updated_at`
+
 ### `systemconfig`
 - Purpose: Stores JSON business configuration values keyed by SystemConfigKey.
 - Useful queries: Verifying the physical value only when the managed settings API behaves unexpectedly.

tests/fixtures/architecture/dependency-baseline.json

@@ -1089,8 +1089,8 @@
       "runtime_only": true
     }
   },
-  "edge_count": 7784,
-  "edge_sha256": "602f1b401b10d5a26a0782fee6ca31b1276dd9861ec9006be35d1372c4974e59",
+  "edge_count": 7835,
+  "edge_sha256": "491127de88d2727e18db050dba310fd4df5a0fe32c67f4976c9f6d1145e2e53f",
   "edges": [
     "app -> app.foundation",
     "app -> app.foundation.environment",
@@ -1997,10 +1997,12 @@
     "app.api.dependencies.subscription -> app.application.subscription",
     "app.api.dependencies.subscription -> app.application.subscription.contract",
     "app.api.dependencies.subscription -> app.application.subscription.delete",
+    "app.api.dependencies.subscription -> app.application.subscription.execution",
     "app.api.dependencies.subscription -> app.application.subscription.identity",
     "app.api.dependencies.subscription -> app.application.subscription.mutation",
     "app.api.dependencies.subscription -> app.application.subscription.query",
     "app.api.dependencies.subscription -> app.application.subscription.search",
+    "app.api.dependencies.subscription -> app.application.subscription.status",
     "app.api.dependencies.subscription -> app.application.subscription.write",
     "app.api.dependencies.subscription -> app.runtime",
     "app.api.dependencies.subscription -> app.runtime.events",
@@ -2631,6 +2633,22 @@
     "app.api.endpoints.storage -> app.schemas.system",
     "app.api.endpoints.storage -> app.schemas.types",
     "app.api.endpoints.storage -> app.schemas.workflow",
+    "app.api.endpoints.subexecution -> app.api",
+    "app.api.endpoints.subexecution -> app.api.dependencies",
+    "app.api.endpoints.subexecution -> app.api.dependencies.auth",
+    "app.api.endpoints.subexecution -> app.api.dependencies.subscription",
+    "app.api.endpoints.subexecution -> app.api.principal",
+    "app.api.endpoints.subexecution -> app.api.response",
+    "app.api.endpoints.subexecution -> app.application",
+    "app.api.endpoints.subexecution -> app.application.subscription",
+    "app.api.endpoints.subexecution -> app.application.subscription.execution",
+    "app.api.endpoints.subexecution -> app.application.subscription.query",
+    "app.api.endpoints.subexecution -> app.application.subscription.status",
+    "app.api.endpoints.subexecution -> app.runtime",
+    "app.api.endpoints.subexecution -> app.runtime.execution",
+    "app.api.endpoints.subexecution -> app.schemas",
+    "app.api.endpoints.subexecution -> app.schemas.response",
+    "app.api.endpoints.subexecution -> app.schemas.subscribe",
     "app.api.endpoints.subscribe -> app.adapters",
     "app.api.endpoints.subscribe -> app.adapters.external",
     "app.api.endpoints.subscribe -> app.adapters.external.server",
@@ -2654,6 +2672,7 @@
     "app.api.endpoints.subscribe -> app.application.subscription.mutation",
     "app.api.endpoints.subscribe -> app.application.subscription.query",
     "app.api.endpoints.subscribe -> app.application.subscription.search",
+    "app.api.endpoints.subscribe -> app.application.subscription.status",
     "app.api.endpoints.subscribe -> app.chain",
     "app.api.endpoints.subscribe -> app.chain.subscribe",
     "app.api.endpoints.subscribe -> app.chain.subscribe.facade",
@@ -2867,6 +2886,7 @@
     "app.api.routers -> app.api.endpoints.search",
     "app.api.routers -> app.api.endpoints.site",
     "app.api.routers -> app.api.endpoints.storage",
+    "app.api.routers -> app.api.endpoints.subexecution",
     "app.api.routers -> app.api.endpoints.subscribe",
     "app.api.routers -> app.api.endpoints.system",
     "app.api.routers -> app.api.endpoints.tmdb",
@@ -3497,9 +3517,14 @@
     "app.application.subscription.search -> app.application.subscription.contract",
     "app.application.subscription.sitebudget -> app.application",
     "app.application.subscription.sitebudget -> app.application.site",
-    "app.application.subscription.sitebudget -> app.application.site.search_observation",
+    "app.application.subscription.sitebudget -> app.application.site.observation",
     "app.application.subscription.sitebudget -> app.runtime",
     "app.application.subscription.sitebudget -> app.runtime.stop",
+    "app.application.subscription.status -> app.application",
+    "app.application.subscription.status -> app.application.download",
+    "app.application.subscription.status -> app.application.download.admission",
+    "app.application.subscription.status -> app.application.subscription",
+    "app.application.subscription.status -> app.application.subscription.execution",
     "app.application.subscription.write -> app.application",
     "app.application.subscription.write -> app.application.outbox",
     "app.application.subscription.write -> app.application.subscription",
@@ -3842,6 +3867,8 @@
     "app.chain.download.submission -> app.domain",
     "app.chain.download.submission -> app.domain.context",
     "app.chain.download.submission -> app.domain.episode",
+    "app.chain.download.submission -> app.domain.meta",
+    "app.chain.download.submission -> app.domain.meta.metabase",
     "app.chain.download.submission -> app.runtime",
     "app.chain.download.submission -> app.runtime.cache",
     "app.chain.download.submission -> app.runtime.events",
@@ -4274,7 +4301,7 @@
     "app.chain.search.provider -> app.application",
     "app.chain.search.provider -> app.application.configuration",
     "app.chain.search.provider -> app.application.site",
-    "app.chain.search.provider -> app.application.site.search_observation",
+    "app.chain.search.provider -> app.application.site.observation",
     "app.chain.search.provider -> app.application.subscription",
     "app.chain.search.provider -> app.application.subscription.sitebudget",
     "app.chain.search.provider -> app.chain",
@@ -4517,6 +4544,19 @@
     "app.chain.subscribe.match -> app.schemas",
     "app.chain.subscribe.match -> app.schemas.media",
     "app.chain.subscribe.match -> app.schemas.types",
+    "app.chain.subscribe.metadata -> app.application",
+    "app.chain.subscribe.metadata -> app.application.subscription",
+    "app.chain.subscribe.metadata -> app.application.subscription.contract",
+    "app.chain.subscribe.metadata -> app.application.subscription.facts",
+    "app.chain.subscribe.metadata -> app.chain",
+    "app.chain.subscribe.metadata -> app.chain.subscribe",
+    "app.chain.subscribe.metadata -> app.chain.subscribe.contract",
+    "app.chain.subscribe.metadata -> app.chain.subscribe.identity",
+    "app.chain.subscribe.metadata -> app.runtime",
+    "app.chain.subscribe.metadata -> app.runtime.log",
+    "app.chain.subscribe.metadata -> app.schemas",
+    "app.chain.subscribe.metadata -> app.schemas.media",
+    "app.chain.subscribe.metadata -> app.schemas.types",
     "app.chain.subscribe.notify -> app.application",
     "app.chain.subscribe.notify -> app.application.configuration",
     "app.chain.subscribe.notify -> app.application.messaging",
@@ -4591,6 +4631,7 @@
     "app.chain.subscribe.reconcile -> app.schemas.types",
     "app.chain.subscribe.refresh -> app.application",
     "app.chain.subscribe.refresh -> app.application.subscription",
+    "app.chain.subscribe.refresh -> app.application.subscription.candidates",
     "app.chain.subscribe.refresh -> app.application.subscription.contract",
     "app.chain.subscribe.refresh -> app.application.subscription.facts",
     "app.chain.subscribe.refresh -> app.application.subscription.priority",
@@ -4598,8 +4639,8 @@
     "app.chain.subscribe.refresh -> app.chain.download",
     "app.chain.subscribe.refresh -> app.chain.media",
     "app.chain.subscribe.refresh -> app.chain.subscribe",
-    "app.chain.subscribe.refresh -> app.chain.subscribe.contract",
     "app.chain.subscribe.refresh -> app.chain.subscribe.identity",
+    "app.chain.subscribe.refresh -> app.chain.subscribe.metadata",
     "app.chain.subscribe.refresh -> app.chain.tmdb",
     "app.chain.subscribe.refresh -> app.chain.torrents",
     "app.chain.subscribe.refresh -> app.domain",
@@ -4612,7 +4653,6 @@
     "app.chain.subscribe.refresh -> app.runtime.stop",
     "app.chain.subscribe.refresh -> app.schemas",
     "app.chain.subscribe.refresh -> app.schemas.event",
-    "app.chain.subscribe.refresh -> app.schemas.media",
     "app.chain.subscribe.refresh -> app.schemas.mediaserver",
     "app.chain.subscribe.refresh -> app.schemas.types",
     "app.chain.subscribe.search -> app.application",
@@ -5185,6 +5225,15 @@
     "app.db.adapters.subscriptionsearch -> app.db.oper",
     "app.db.adapters.subscriptionsearch -> app.db.oper.subscriptionsearch",
     "app.db.adapters.subscriptionsearch -> app.db.uow",
+    "app.db.adapters.subscriptionstatus -> app.application",
+    "app.db.adapters.subscriptionstatus -> app.application.download",
+    "app.db.adapters.subscriptionstatus -> app.application.download.admission",
+    "app.db.adapters.subscriptionstatus -> app.application.subscription",
+    "app.db.adapters.subscriptionstatus -> app.application.subscription.execution",
+    "app.db.adapters.subscriptionstatus -> app.db",
+    "app.db.adapters.subscriptionstatus -> app.db.models",
+    "app.db.adapters.subscriptionstatus -> app.db.models.subscriptiondownload",
+    "app.db.adapters.subscriptionstatus -> app.db.models.subscriptionsearch",
     "app.db.adapters.transaction -> app.db",
     "app.db.adapters.transaction -> app.db.uow",
     "app.db.adapters.transfer.admission -> app.application",
@@ -6186,8 +6235,8 @@
     "app.modules.indexer -> app.application",
     "app.modules.indexer -> app.application.site",
     "app.modules.indexer -> app.application.site.health",
+    "app.modules.indexer -> app.application.site.observation",
     "app.modules.indexer -> app.application.site.query",
-    "app.modules.indexer -> app.application.site.search_observation",
     "app.modules.indexer -> app.domain",
     "app.modules.indexer -> app.domain.context",
     "app.modules.indexer -> app.domain.site",
@@ -7694,6 +7743,7 @@
     "app.scheduler.catalog -> app.runtime.scheduling",
     "app.scheduler.catalog -> app.scheduler",
     "app.scheduler.catalog -> app.scheduler.contract",
+    "app.scheduler.catalog -> app.scheduler.services",
     "app.scheduler.catalog -> app.schemas",
     "app.scheduler.catalog -> app.schemas.system",
     "app.scheduler.chain -> app.application",
@@ -8289,6 +8339,7 @@
     "app.startup.composition.runtime -> app.db.adapters.site",
     "app.startup.composition.runtime -> app.db.adapters.subscription",
     "app.startup.composition.runtime -> app.db.adapters.subscriptionsearch",
+    "app.startup.composition.runtime -> app.db.adapters.subscriptionstatus",
     "app.startup.composition.runtime -> app.db.adapters.transfer",
     "app.startup.composition.runtime -> app.db.adapters.transfer.execution",
     "app.startup.composition.runtime -> app.db.oper",
@@ -8877,7 +8928,7 @@
     "app.workflow.actions.transfer_file -> app.workflow",
     "app.workflow.actions.transfer_file -> app.workflow.actions"
   ],
-  "module_count": 932,
+  "module_count": 936,
   "modules": [
     "app",
     "app.adapters",
@@ -9051,6 +9102,7 @@
     "app.api.endpoints.search",
     "app.api.endpoints.site",
     "app.api.endpoints.storage",
+    "app.api.endpoints.subexecution",
     "app.api.endpoints.subscribe",
     "app.api.endpoints.system",
     "app.api.endpoints.tmdb",
@@ -9162,8 +9214,8 @@
     "app.application.site.contract",
     "app.application.site.health",
     "app.application.site.mutation",
+    "app.application.site.observation",
     "app.application.site.query",
-    "app.application.site.search_observation",
     "app.application.storage",
     "app.application.subscription",
     "app.application.subscription.candidates",
@@ -9178,6 +9230,7 @@
     "app.application.subscription.query",
     "app.application.subscription.search",
     "app.application.subscription.sitebudget",
+    "app.application.subscription.status",
     "app.application.subscription.write",
     "app.application.system",
     "app.application.torrent",
@@ -9260,6 +9313,7 @@
     "app.chain.subscribe.identity",
     "app.chain.subscribe.interaction",
     "app.chain.subscribe.match",
+    "app.chain.subscribe.metadata",
     "app.chain.subscribe.notify",
     "app.chain.subscribe.policy",
     "app.chain.subscribe.query",
@@ -9309,6 +9363,7 @@
     "app.db.adapters.subscription",
     "app.db.adapters.subscriptiondownload",
     "app.db.adapters.subscriptionsearch",
+    "app.db.adapters.subscriptionstatus",
     "app.db.adapters.transaction",
     "app.db.adapters.transfer",
     "app.db.adapters.transfer.admission",

tests/fixtures/architecture/mypy-baseline.json

@@ -994,7 +994,7 @@
     "var-annotated": 2
   },
   "app/chain/subscribe/search.py": {
-    "assignment": 3
+    "assignment": 2
   },
   "app/chain/system.py": {
     "attr-defined": 1,

tests/fixtures/architecture/ruff-baseline.json

@@ -570,9 +570,6 @@
   "app/schemas/response.py": {
     "I001": 1
   },
-  "app/schemas/subscribe.py": {
-    "I001": 1
-  },
   "app/schemas/types.py": {
     "I001": 1
   },

tests/fixtures/architecture/runtime-contract-baseline.json

@@ -1669,7 +1669,7 @@
       "ChainEventType.ResourceDownload": {
         "consumer_fingerprints": [],
         "producer_fingerprints": [
-          "aac2e9ac03b43fd89c42f3efba186365152568edc05881f24204a88fbd74bdbf"
+          "26b919353eed404b502c7fdf899981f372b9a25ca75a90ba205f61c5ff90c41a"
         ]
       },
       "ChainEventType.ResourceSelection": {
@@ -2237,10 +2237,10 @@
         "events": [
           "ChainEventType.ResourceDownload"
         ],
-        "fingerprint": "aac2e9ac03b43fd89c42f3efba186365152568edc05881f24204a88fbd74bdbf",
+        "fingerprint": "26b919353eed404b502c7fdf899981f372b9a25ca75a90ba205f61c5ff90c41a",
         "invalid": false,
         "method": "send_event",
-        "qualname": "DownloadSubmissionOwner._apply_resource_download_event",
+        "qualname": "_DownloadResourceOwner._apply_resource_download_event",
         "receiver_kind": "canonical_singleton"
       },
       {

tests/test_subscription_architecture.py

@@ -234,6 +234,7 @@ def test_subscribe_chain_package_has_single_responsibility_owners() -> None:
         "identity.py",
         "interaction.py",
         "match.py",
+        "metadata.py",
         "notify.py",
         "policy.py",
         "query.py",

tests/test_subscription_governance_joint.py

@@ -3,7 +3,7 @@
 from datetime import datetime, timedelta, timezone
 from types import SimpleNamespace
 
-from app.application.site.search_observation import report_site_search_outcome
+from app.application.site.observation import report_site_search_outcome
 from app.application.subscription.sitebudget import SiteBudgetClaim, SubscriptionSiteBudget
 from app.chain.search.facade import SearchChain
 from app.chain.search.provider import SearchProviderOwner

tests/test_subscription_site_budget.py

@@ -6,7 +6,7 @@ import pytest
 from sqlalchemy import create_engine, select
 from sqlalchemy.orm import Session, sessionmaker
 
-from app.application.site.search_observation import (
+from app.application.site.observation import (
     SiteSearchObservation,
     report_site_search_outcome,
 )