diff --git a/app/application/data_query.py b/app/application/query.py
similarity index 100%
rename from app/application/data_query.py
rename to app/application/query.py
diff --git a/app/db/adapters/data_query.py b/app/db/adapters/data_query.py
deleted file mode 100644
index 7d2d43cc3..000000000
--- a/app/db/adapters/data_query.py
+++ /dev/null
@@ -1,374 +0,0 @@
-"""插件只读数据查询的 SQLAlchemy 持久化适配器。"""
-
-from __future__ import annotations
-
-from collections.abc import Callable, Iterable
-from enum import Enum
-from typing import Any, TypeVar
-
-from pydantic import BaseModel
-from sqlalchemy import func, or_, select
-from sqlalchemy.orm import Session
-
-from app.application.data_query import QueryRows
-from app.db.base import Base
-from app.db.models.downloadhistory import DownloadHistory
-from app.db.models.subscribe import Subscribe
-from app.db.models.subscribehistory import SubscribeHistory as SubscribeHistoryModel
-from app.db.models.transferhistory import TransferHistory
-from app.schemas.query import (
-    DownloadHistoryFilter,
-    DownloadHistorySnapshot,
-    QueryPageRequest,
-    QuerySortDirection,
-    QuerySortField,
-    SubscriptionFilter,
-    SubscriptionHistoryFilter,
-    SubscriptionHistorySnapshot,
-    SubscriptionSnapshot,
-    TransferHistoryFilter,
-    TransferHistorySnapshot,
-)
-from app.schemas.types import MUSIC_ENTITY_RECORDING
-
-_ModelT = TypeVar("_ModelT", bound=Base)
-_ViewT = TypeVar("_ViewT", bound=BaseModel)
-
-
-def _enum_value(value: Any) -> Any:
-    """返回枚举筛选值的稳定数据库表示。"""
-    return value.value if isinstance(value, Enum) else value
-
-
-def _values(values: Iterable[Any]) -> tuple[Any, ...]:
-    """去除空筛选值并保留调用方声明的顺序。"""
-    normalized: list[Any] = []
-    for value in values:
-        value = _enum_value(value)
-        if value in (None, ""):
-            continue
-        normalized.append(value)
-    return tuple(dict.fromkeys(normalized))
-
-
-def _contains(column: Any, value: str) -> Any:
-    """构造不区分大小写且不解释通配符的字面包含筛选。"""
-    escaped = value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
-    return column.ilike(f"%{escaped}%", escape="\\")
-
-
-def _music_type_condition(column: Any, music_type: str | None) -> Any | None:
-    """兼容未标注音乐类型的历史单曲记录。"""
-    if not music_type:
-        return None
-    if music_type == MUSIC_ENTITY_RECORDING:
-        return or_(column == music_type, column.is_(None))
-    return column == music_type
-
-
-class SqlAlchemyDataQueryAdapter:
-    """以短生命周期同步 Session 执行统一只读分页查询。
-
-    这个适配器是查询层唯一接触 SQLAlchemy Model 的边界。每个公开方法在同一
-    Session 中先统计再读取当前页，并在 Session 仍有效时转换成 Pydantic DTO，
-    因而调用方不会持有 ORM 实例或延迟加载状态。
-    """
-
-    def __init__(self, session_factory: Callable[[], Session]) -> None:
-        """保存由启动组合根提供的同步 Session 工厂。"""
-        self._session_factory = session_factory
-
-    @staticmethod
-    def _identity_conditions(model: Any, query: Any) -> list[Any]:
-        """按媒体来源与原生 ID 的成对合同构造筛选条件。"""
-        media_source = query.media_source
-        media_id = query.media_id
-        if (media_source is None) != (media_id is None):
-            # Pydantic 合同会先拒绝这种输入；这里仍保留拒绝，避免未校验对象
-            # 在持久化边界退化成只按 NULL 查询而扩大结果集。
-            raise ValueError("media_source 和 media_id 必须同时提供")
-        if media_source is None:
-            return []
-        normalized_id = str(media_id).strip()
-        if not normalized_id or normalized_id == "0":
-            raise ValueError("media_id 必须是非零的来源原生 ID")
-        return [
-            model.media_source == _enum_value(media_source),
-            model.media_id == normalized_id,
-        ]
-
-    @staticmethod
-    def _require_media_identity(model: Any) -> list[Any]:
-        """只保留来源和原生 ID 均存在且非空的记录。"""
-        return [
-            model.media_source.is_not(None),
-            func.trim(model.media_source) != "",
-            model.media_id.is_not(None),
-            func.trim(model.media_id) != "",
-            func.trim(model.media_id) != "0",
-        ]
-
-    @staticmethod
-    def _order_by(model: Any, request: QueryPageRequest) -> tuple[Any, ...]:
-        """构造可跨页复现的排序，日期相同时始终以主键打破平局。"""
-        descending = request.sort.direction == QuerySortDirection.DESC
-        if request.sort.field == QuerySortField.ID:
-            primary = model.id.desc() if descending else model.id.asc()
-            secondary = model.date.desc() if descending else model.date.asc()
-            return primary, secondary
-        primary = model.date.desc().nullslast() if descending else model.date.asc().nullsfirst()
-        secondary = model.id.desc() if descending else model.id.asc()
-        return primary, secondary
-
-    def _page(
-        self,
-        *,
-        model: type[_ModelT],
-        view_model: type[_ViewT],
-        conditions: Iterable[Any],
-        page: QueryPageRequest,
-    ) -> QueryRows[_ViewT]:
-        """在单个 Session 内完成 count、分页读取和 DTO 投影。"""
-        conditions = tuple(conditions)
-        count_statement = select(func.count(model.id))
-        page_statement = select(model)
-        if conditions:
-            count_statement = count_statement.where(*conditions)
-            page_statement = page_statement.where(*conditions)
-        page_statement = (
-            page_statement.order_by(*self._order_by(model, page)).offset((page.page - 1) * page.count).limit(page.count)
-        )
-
-        with self._session_factory() as session:
-            total = int(session.execute(count_statement).scalar_one() or 0)
-            records = session.execute(page_statement).scalars().all()
-            # model_validate 必须在会话内完成；返回值只包含 Pydantic 数据。
-            items = [view_model.model_validate(record) for record in records]
-            return QueryRows(items=items, total=total)
-
-    def _get(
-        self,
-        *,
-        model: type[_ModelT],
-        view_model: type[_ViewT],
-        record_id: int,
-    ) -> _ViewT | None:
-        """在短 Session 内按主键读取并冻结单条 Pydantic 投影。"""
-        statement = select(model).where(model.id == record_id)
-        with self._session_factory() as session:
-            record = session.execute(statement).scalars().first()
-            return view_model.model_validate(record) if record is not None else None
-
-    def list_subscriptions(
-        self,
-        *,
-        filters: SubscriptionFilter,
-        page: QueryPageRequest,
-    ) -> QueryRows[SubscriptionSnapshot]:
-        """按受控组合条件分页查询当前订阅。"""
-        query = filters
-        conditions = self._identity_conditions(Subscribe, query)
-        ids = _values(query.ids)
-        names = _values(query.names)
-        states = _values(query.states)
-        usernames = _values(query.usernames)
-        media_types = _values(query.media_types)
-        if ids:
-            conditions.append(Subscribe.id.in_(ids))
-        if names:
-            conditions.append(Subscribe.name.in_(names))
-        if states:
-            conditions.append(Subscribe.state.in_(states))
-        if usernames:
-            conditions.append(Subscribe.username.in_(usernames))
-        if media_types:
-            conditions.append(Subscribe.type.in_(media_types))
-        if query.season is not None:
-            conditions.append(Subscribe.season == query.season)
-        if query.episode_group is not None:
-            conditions.append(Subscribe.episode_group == query.episode_group)
-        music_condition = _music_type_condition(Subscribe.music_type, query.music_type)
-        if music_condition is not None:
-            conditions.append(music_condition)
-        return self._page(
-            model=Subscribe,
-            view_model=SubscriptionSnapshot,
-            conditions=conditions,
-            page=page,
-        )
-
-    def get_subscription(self, subscription_id: int) -> SubscriptionSnapshot | None:
-        """按主键查询订阅并返回脱离 Session 的 DTO。"""
-        return self._get(
-            model=Subscribe,
-            view_model=SubscriptionSnapshot,
-            record_id=subscription_id,
-        )
-
-    def list_subscription_history(
-        self,
-        *,
-        filters: SubscriptionHistoryFilter,
-        page: QueryPageRequest,
-    ) -> QueryRows[SubscriptionHistorySnapshot]:
-        """按受控组合条件分页查询订阅完成历史。"""
-        query = filters
-        conditions = self._identity_conditions(SubscribeHistoryModel, query)
-        ids = _values(query.ids)
-        names = _values(query.names)
-        usernames = _values(query.usernames)
-        media_types = _values(query.media_types)
-        if ids:
-            conditions.append(SubscribeHistoryModel.id.in_(ids))
-        if names:
-            conditions.append(SubscribeHistoryModel.name.in_(names))
-        if usernames:
-            conditions.append(SubscribeHistoryModel.username.in_(usernames))
-        if media_types:
-            conditions.append(SubscribeHistoryModel.type.in_(media_types))
-        if query.season is not None:
-            conditions.append(SubscribeHistoryModel.season == query.season)
-        if query.episode_group is not None:
-            conditions.append(SubscribeHistoryModel.episode_group == query.episode_group)
-        music_condition = _music_type_condition(
-            SubscribeHistoryModel.music_type,
-            query.music_type,
-        )
-        if music_condition is not None:
-            conditions.append(music_condition)
-        return self._page(
-            model=SubscribeHistoryModel,
-            view_model=SubscriptionHistorySnapshot,
-            conditions=conditions,
-            page=page,
-        )
-
-    def get_subscription_history(
-        self,
-        history_id: int,
-    ) -> SubscriptionHistorySnapshot | None:
-        """按主键查询订阅完成历史并返回稳定 DTO。"""
-        return self._get(
-            model=SubscribeHistoryModel,
-            view_model=SubscriptionHistorySnapshot,
-            record_id=history_id,
-        )
-
-    def list_download_history(
-        self,
-        *,
-        filters: DownloadHistoryFilter,
-        page: QueryPageRequest,
-    ) -> QueryRows[DownloadHistorySnapshot]:
-        """按受控组合条件分页查询下载历史。"""
-        query = filters
-        conditions = self._identity_conditions(DownloadHistory, query)
-        ids = _values(query.ids)
-        media_types = _values(query.media_types)
-        usernames = _values(query.usernames)
-        if ids:
-            conditions.append(DownloadHistory.id.in_(ids))
-        if media_types:
-            conditions.append(DownloadHistory.type.in_(media_types))
-        for column, value in (
-            (DownloadHistory.title, query.title),
-            (DownloadHistory.year, query.year),
-            (DownloadHistory.seasons, query.seasons),
-            (DownloadHistory.episodes, query.episodes),
-            (DownloadHistory.path, query.path),
-            (DownloadHistory.download_hash, query.download_hash),
-            (DownloadHistory.username, query.username),
-            (DownloadHistory.episode_group, query.episode_group),
-        ):
-            if value is not None and value != "":
-                conditions.append(column == value)
-        if query.text:
-            conditions.append(
-                or_(
-                    _contains(DownloadHistory.title, query.text),
-                    _contains(DownloadHistory.path, query.text),
-                )
-            )
-        if usernames:
-            conditions.append(DownloadHistory.username.in_(usernames))
-        music_condition = _music_type_condition(DownloadHistory.music_type, query.music_type)
-        if music_condition is not None:
-            conditions.append(music_condition)
-        return self._page(
-            model=DownloadHistory,
-            view_model=DownloadHistorySnapshot,
-            conditions=conditions,
-            page=page,
-        )
-
-    def get_download_history(self, history_id: int) -> DownloadHistorySnapshot | None:
-        """按主键查询下载历史并返回稳定 DTO。"""
-        return self._get(
-            model=DownloadHistory,
-            view_model=DownloadHistorySnapshot,
-            record_id=history_id,
-        )
-
-    def list_transfer_history(
-        self,
-        *,
-        filters: TransferHistoryFilter,
-        page: QueryPageRequest,
-    ) -> QueryRows[TransferHistorySnapshot]:
-        """按受控组合条件分页查询整理历史。"""
-        query = filters
-        conditions = self._identity_conditions(TransferHistory, query)
-        ids = _values(query.ids)
-        media_types = _values(query.media_types)
-        media_sources = _values(query.media_sources)
-        if ids:
-            conditions.append(TransferHistory.id.in_(ids))
-        if media_types:
-            conditions.append(TransferHistory.type.in_(media_types))
-        if media_sources:
-            conditions.append(TransferHistory.media_source.in_(media_sources))
-        if query.require_media_identity:
-            conditions.extend(self._require_media_identity(TransferHistory))
-        if query.title:
-            conditions.append(TransferHistory.title == query.title)
-        if query.text:
-            conditions.append(
-                or_(
-                    _contains(TransferHistory.title, query.text),
-                    _contains(TransferHistory.src, query.text),
-                    _contains(TransferHistory.dest, query.text),
-                )
-            )
-        for column, value in (
-            (TransferHistory.year, query.year),
-            (TransferHistory.seasons, query.seasons),
-            (TransferHistory.episodes, query.episodes),
-            (TransferHistory.src, query.src),
-            (TransferHistory.dest, query.dest),
-            (TransferHistory.download_hash, query.download_hash),
-            (TransferHistory.episode_group, query.episode_group),
-        ):
-            if value is not None and value != "":
-                conditions.append(column == value)
-        if query.status is not None:
-            conditions.append(TransferHistory.status == query.status)
-        music_condition = _music_type_condition(TransferHistory.music_type, query.music_type)
-        if music_condition is not None:
-            conditions.append(music_condition)
-        return self._page(
-            model=TransferHistory,
-            view_model=TransferHistorySnapshot,
-            conditions=conditions,
-            page=page,
-        )
-
-    def get_transfer_history(self, history_id: int) -> TransferHistorySnapshot | None:
-        """按主键查询整理历史并返回稳定 DTO。"""
-        return self._get(
-            model=TransferHistory,
-            view_model=TransferHistorySnapshot,
-            record_id=history_id,
-        )
-
-
-__all__ = ["SqlAlchemyDataQueryAdapter"]
diff --git a/app/db/adapters/query.py b/app/db/adapters/query.py
new file mode 100644
index 000000000..e999e4cd0
--- /dev/null
+++ b/app/db/adapters/query.py
@@ -0,0 +1,140 @@
+"""插件只读数据查询的持久化端口适配器。"""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Sequence
+from typing import TypeVar
+
+from pydantic import BaseModel
+from sqlalchemy.orm import Session
+
+from app.application.query import QueryRows
+from app.db.oper.downloadhistory import DownloadHistoryOper
+from app.db.oper.subscribe import SubscribeOper
+from app.db.oper.subscribehistory import SubscribeHistoryOper
+from app.db.oper.transferhistory import TransferHistoryOper
+from app.schemas.query import (
+    DownloadHistoryFilter,
+    DownloadHistorySnapshot,
+    QueryPageRequest,
+    SubscriptionFilter,
+    SubscriptionHistoryFilter,
+    SubscriptionHistorySnapshot,
+    SubscriptionSnapshot,
+    TransferHistoryFilter,
+    TransferHistorySnapshot,
+)
+
+_SnapshotT = TypeVar("_SnapshotT", bound=BaseModel)
+
+
+class SqlAlchemyDataQueryAdapter:
+    """以短生命周期 Session 和显式 Oper 实现统一查询端口。
+
+    Oper 拥有表级筛选、排序和分页语义；适配器只管理一次操作使用的 Session，
+    并在 Session 关闭前把持久化记录冻结为稳定 Pydantic 快照。
+    """
+
+    def __init__(self, session_factory: Callable[[], Session]) -> None:
+        """保存由启动组合根提供的同步 Session 工厂。"""
+        self._session_factory = session_factory
+
+    @staticmethod
+    def _rows(
+        snapshot_type: type[_SnapshotT],
+        records: Sequence[object],
+        total: int,
+    ) -> QueryRows[_SnapshotT]:
+        """在会话有效期内投影一页记录，并保护非负总数合同。"""
+        return QueryRows(
+            items=[snapshot_type.model_validate(record) for record in records],
+            total=max(int(total), 0),
+        )
+
+    @staticmethod
+    def _item(
+        snapshot_type: type[_SnapshotT],
+        record: object | None,
+    ) -> _SnapshotT | None:
+        """把单条持久化记录投影为稳定快照。"""
+        return snapshot_type.model_validate(record) if record is not None else None
+
+    def list_subscriptions(
+        self,
+        *,
+        filters: SubscriptionFilter,
+        page: QueryPageRequest,
+    ) -> QueryRows[SubscriptionSnapshot]:
+        """通过订阅 Oper 读取并冻结一页当前订阅。"""
+        with self._session_factory() as session:
+            records, total = SubscribeOper(session).query(filters, page)
+            return self._rows(SubscriptionSnapshot, records, total)
+
+    def get_subscription(self, subscription_id: int) -> SubscriptionSnapshot | None:
+        """通过订阅 Oper 按 ID 读取并冻结当前订阅。"""
+        with self._session_factory() as session:
+            record = SubscribeOper(session).get_by_id(subscription_id)
+            return self._item(SubscriptionSnapshot, record)
+
+    def list_subscription_history(
+        self,
+        *,
+        filters: SubscriptionHistoryFilter,
+        page: QueryPageRequest,
+    ) -> QueryRows[SubscriptionHistorySnapshot]:
+        """通过订阅历史 Oper 读取并冻结一页完成历史。"""
+        with self._session_factory() as session:
+            records, total = SubscribeHistoryOper(session).query(filters, page)
+            return self._rows(SubscriptionHistorySnapshot, records, total)
+
+    def get_subscription_history(
+        self,
+        history_id: int,
+    ) -> SubscriptionHistorySnapshot | None:
+        """通过订阅历史 Oper 按 ID 读取并冻结完成历史。"""
+        with self._session_factory() as session:
+            record = SubscribeHistoryOper(session).get_by_id(history_id)
+            return self._item(SubscriptionHistorySnapshot, record)
+
+    def list_download_history(
+        self,
+        *,
+        filters: DownloadHistoryFilter,
+        page: QueryPageRequest,
+    ) -> QueryRows[DownloadHistorySnapshot]:
+        """通过下载历史 Oper 读取并冻结一页下载记录。"""
+        with self._session_factory() as session:
+            records, total = DownloadHistoryOper(session).query(filters, page)
+            return self._rows(DownloadHistorySnapshot, records, total)
+
+    def get_download_history(
+        self,
+        history_id: int,
+    ) -> DownloadHistorySnapshot | None:
+        """通过下载历史 Oper 按 ID 读取并冻结下载记录。"""
+        with self._session_factory() as session:
+            record = DownloadHistoryOper(session).get_by_id(history_id)
+            return self._item(DownloadHistorySnapshot, record)
+
+    def list_transfer_history(
+        self,
+        *,
+        filters: TransferHistoryFilter,
+        page: QueryPageRequest,
+    ) -> QueryRows[TransferHistorySnapshot]:
+        """通过整理历史 Oper 读取并冻结一页整理记录。"""
+        with self._session_factory() as session:
+            records, total = TransferHistoryOper(session).query(filters, page)
+            return self._rows(TransferHistorySnapshot, records, total)
+
+    def get_transfer_history(
+        self,
+        history_id: int,
+    ) -> TransferHistorySnapshot | None:
+        """通过整理历史 Oper 按 ID 读取并冻结整理记录。"""
+        with self._session_factory() as session:
+            record = TransferHistoryOper(session).get_by_id(history_id)
+            return self._item(TransferHistorySnapshot, record)
+
+
+__all__ = ["SqlAlchemyDataQueryAdapter"]
diff --git a/app/db/oper/downloadhistory.py b/app/db/oper/downloadhistory.py
index 9c173fa6d..5edaba329 100644
--- a/app/db/oper/downloadhistory.py
+++ b/app/db/oper/downloadhistory.py
@@ -1,10 +1,25 @@
 from typing import Dict, List, Optional, cast
 
-from sqlalchemy import delete as sqlalchemy_delete, update as sqlalchemy_update
+from sqlalchemy import delete as sqlalchemy_delete
+from sqlalchemy import func, select
+from sqlalchemy import update as sqlalchemy_update
 from sqlalchemy.orm import Session
 
 from app.db.base import DbOper
-from app.db.models.downloadhistory import DownloadHistory, DownloadFiles
+from app.db.models.downloadhistory import DownloadFiles, DownloadHistory
+from app.db.oper.query import (
+    descending,
+    enum_values,
+    execute_page,
+    literal_contains,
+    media_identity_conditions,
+    music_type_condition,
+)
+from app.schemas.query import (
+    DownloadHistoryFilter,
+    QueryPageRequest,
+    QuerySortField,
+)
 from app.schemas.types import MediaSource
 
 
@@ -13,6 +28,95 @@ class DownloadHistoryOper(DbOper):
     下载历史管理
     """
 
+    def get_by_id(self, record_id: int) -> Optional[DownloadHistory]:
+        """按稳定记录 ID 读取单条下载历史。"""
+        return cast(
+            Optional[DownloadHistory],
+            self._execute_sync_query(
+                lambda session: session.execute(
+                    select(DownloadHistory).where(DownloadHistory.id == record_id)
+                ).scalars().first()
+            ),
+        )
+
+    def query(
+        self,
+        filters: DownloadHistoryFilter,
+        page: QueryPageRequest,
+    ) -> tuple[list[DownloadHistory], int]:
+        """按稳定筛选和分页合同读取下载历史记录及总数。"""
+        def execute(session: Session) -> tuple[list[DownloadHistory], int]:
+            """在同一会话中构造并执行下载历史 count/page 查询。"""
+            conditions = media_identity_conditions(DownloadHistory, filters)
+            ids = enum_values(filters.ids)
+            media_types = enum_values(filters.media_types)
+            usernames = enum_values(filters.usernames)
+            if ids:
+                conditions.append(DownloadHistory.id.in_(ids))
+            if media_types:
+                conditions.append(DownloadHistory.type.in_(media_types))
+            for column, value in (
+                (DownloadHistory.title, filters.title),
+                (DownloadHistory.year, filters.year),
+                (DownloadHistory.seasons, filters.seasons),
+                (DownloadHistory.episodes, filters.episodes),
+                (DownloadHistory.path, filters.path),
+                (DownloadHistory.download_hash, filters.download_hash),
+                (DownloadHistory.username, filters.username),
+                (DownloadHistory.episode_group, filters.episode_group),
+            ):
+                if value is not None and value != "":
+                    conditions.append(column == value)
+            if filters.text:
+                conditions.append(
+                    literal_contains(DownloadHistory.title, filters.text)
+                    | literal_contains(DownloadHistory.path, filters.text)
+                )
+            if usernames:
+                conditions.append(DownloadHistory.username.in_(usernames))
+            music_condition = music_type_condition(
+                DownloadHistory.music_type,
+                filters.music_type,
+            )
+            if music_condition is not None:
+                conditions.append(music_condition)
+
+            count_statement = select(func.count(DownloadHistory.id))
+            page_statement = select(DownloadHistory)
+            if conditions:
+                count_statement = count_statement.where(*conditions)
+                page_statement = page_statement.where(*conditions)
+            descending_order = descending(page)
+            if page.sort.field == QuerySortField.ID:
+                primary = (
+                    DownloadHistory.id.desc()
+                    if descending_order
+                    else DownloadHistory.id.asc()
+                )
+                secondary = (
+                    DownloadHistory.date.desc()
+                    if descending_order
+                    else DownloadHistory.date.asc()
+                )
+            else:
+                primary = (
+                    DownloadHistory.date.desc().nullslast()
+                    if descending_order
+                    else DownloadHistory.date.asc().nullsfirst()
+                )
+                secondary = (
+                    DownloadHistory.id.desc()
+                    if descending_order
+                    else DownloadHistory.id.asc()
+                )
+            page_statement = page_statement.order_by(primary, secondary)
+            return cast(
+                tuple[list[DownloadHistory], int],
+                execute_page(session, count_statement, page_statement, page),
+            )
+
+        return self._execute_sync_query(execute)
+
     def get_by_path(self, path: str) -> Optional[DownloadHistory]:
         """
         按路径查询下载记录
diff --git a/app/db/oper/query.py b/app/db/oper/query.py
new file mode 100644
index 000000000..49f4fc112
--- /dev/null
+++ b/app/db/oper/query.py
@@ -0,0 +1,113 @@
+"""只读查询 Oper 共享的持久化原语。
+
+本模块只承载跨表完全相同的查询表示规则和分页执行步骤。每个具体 Oper 仍负责
+声明本表模型、筛选字段、SQLAlchemy 条件、count/page 语句以及排序，避免这里退化
+成任意表的 Repository。
+"""
+
+from collections.abc import Iterable
+from enum import Enum
+from typing import Any, TypeVar
+
+from sqlalchemy import func, or_
+from sqlalchemy.orm import Session
+
+from app.db.base import Base
+from app.schemas.query import QueryPageRequest, QuerySortDirection
+from app.schemas.types import MEDIA_SOURCE_IDENTIFIER_PATTERN, MUSIC_ENTITY_RECORDING
+
+ModelT = TypeVar("ModelT", bound=Base)
+
+
+def enum_value(value: Any) -> Any:
+    """返回筛选枚举对应的数据库值。"""
+    return value.value if isinstance(value, Enum) else value
+
+
+def enum_values(values: Iterable[Any]) -> tuple[Any, ...]:
+    """去除空筛选值、归一枚举，并保留调用方顺序。"""
+    normalized: list[Any] = []
+    for value in values:
+        value = enum_value(value)
+        if value in (None, ""):
+            continue
+        normalized.append(value)
+    return tuple(dict.fromkeys(normalized))
+
+
+def literal_contains(column: Any, value: str) -> Any:
+    """构造不区分大小写且不解释 ``%``、``_`` 通配符的字面包含条件。"""
+    escaped = value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+    return column.ilike(f"%{escaped}%", escape="\\")
+
+
+def media_identity_conditions(model: Any, query: Any) -> list[Any]:
+    """按媒体来源与原生 ID 的成对合同构造条件，并对非法身份 fail-closed。"""
+    media_source = query.media_source
+    media_id = query.media_id
+    if (media_source is None) != (media_id is None):
+        raise ValueError("media_source 和 media_id 必须同时提供")
+    if media_source is None:
+        return []
+    normalized_id = str(media_id).strip()
+    if not normalized_id or normalized_id == "0":
+        raise ValueError("media_id 必须是非零的来源原生 ID")
+    return [
+        model.media_source == enum_value(media_source),
+        model.media_id == normalized_id,
+    ]
+
+
+def required_media_identity_conditions(model: Any) -> list[Any]:
+    """构造只保留可解析来源与非空、非零原生 ID 的条件。"""
+    return [
+        model.media_source.is_not(None),
+        func.trim(model.media_source) != "",
+        func.lower(func.trim(model.media_source)).regexp_match(
+            MEDIA_SOURCE_IDENTIFIER_PATTERN
+        ),
+        model.media_id.is_not(None),
+        func.trim(model.media_id) != "",
+        func.trim(model.media_id) != "0",
+    ]
+
+
+def music_type_condition(column: Any, music_type: str | None) -> Any | None:
+    """兼容未标注音乐类型的历史单曲记录。"""
+    music_type = enum_value(music_type)
+    if not music_type:
+        return None
+    if music_type == MUSIC_ENTITY_RECORDING:
+        return or_(column == music_type, column.is_(None))
+    return column == music_type
+
+
+def execute_page(
+    session: Session,
+    count_statement: Any,
+    page_statement: Any,
+    page: QueryPageRequest,
+) -> tuple[list[ModelT], int]:
+    """在同一同步 Session 内先统计再读取一页已构造的查询语句。"""
+    total = int(session.execute(count_statement).scalar_one() or 0)
+    records = list(
+        session.execute(page_statement.offset((page.page - 1) * page.count).limit(page.count)).scalars().all()
+    )
+    return records, total
+
+
+def descending(page: QueryPageRequest) -> bool:
+    """返回公开分页排序是否为降序，集中处理枚举表示。"""
+    return page.sort.direction == QuerySortDirection.DESC
+
+
+__all__ = [
+    "descending",
+    "enum_value",
+    "enum_values",
+    "execute_page",
+    "literal_contains",
+    "media_identity_conditions",
+    "music_type_condition",
+    "required_media_identity_conditions",
+]
diff --git a/app/db/oper/subscribe.py b/app/db/oper/subscribe.py
index b1e18717a..8d5468723 100644
--- a/app/db/oper/subscribe.py
+++ b/app/db/oper/subscribe.py
@@ -9,9 +9,10 @@
 """
 import time
 from collections.abc import Awaitable, Callable
-from typing import Any, Tuple, List, Optional
+from typing import Any, List, Optional, Tuple, cast
 
 from sqlalchemy import delete as sqlalchemy_delete
+from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Session
 
@@ -19,6 +20,14 @@ from app.application.subscription.delete import SubscribeDeletionCandidate
 from app.db.base import DbOper
 from app.db.models.subscribe import Subscribe
 from app.db.models.subscribehistory import SubscribeHistory
+from app.db.oper.query import (
+    descending,
+    enum_values,
+    execute_page,
+    media_identity_conditions,
+    music_type_condition,
+)
+from app.schemas.query import QueryPageRequest, QuerySortField, SubscriptionFilter
 from app.schemas.types import MediaSource
 
 INTEGER_FLAG_FIELDS = ("best_version", "best_version_full", "search_imdbid", "manual_total_episode")
@@ -276,7 +285,77 @@ class SubscribeOper(DbOper):
         """
         获取订阅
         """
-        return self._execute_sync_query(lambda session: Subscribe.get(session, sid))
+        return self.get_by_id(sid)
+
+    def get_by_id(self, record_id: int) -> Optional[Subscribe]:
+        """按稳定记录 ID 读取单条订阅。"""
+        return cast(
+            Optional[Subscribe],
+            self._execute_sync_query(
+                lambda session: session.execute(
+                    select(Subscribe).where(Subscribe.id == record_id)
+                ).scalars().first()
+            ),
+        )
+
+    def query(
+        self,
+        filters: SubscriptionFilter,
+        page: QueryPageRequest,
+    ) -> tuple[list[Subscribe], int]:
+        """按稳定筛选和分页合同读取订阅记录及总数。"""
+        def execute(session: Session) -> tuple[list[Subscribe], int]:
+            """在同一会话中构造并执行订阅 count/page 查询。"""
+            conditions = media_identity_conditions(Subscribe, filters)
+            ids = enum_values(filters.ids)
+            names = enum_values(filters.names)
+            states = enum_values(filters.states)
+            usernames = enum_values(filters.usernames)
+            media_types = enum_values(filters.media_types)
+            if ids:
+                conditions.append(Subscribe.id.in_(ids))
+            if names:
+                conditions.append(Subscribe.name.in_(names))
+            if states:
+                conditions.append(Subscribe.state.in_(states))
+            if usernames:
+                conditions.append(Subscribe.username.in_(usernames))
+            if media_types:
+                conditions.append(Subscribe.type.in_(media_types))
+            if filters.season is not None:
+                conditions.append(Subscribe.season == filters.season)
+            if filters.episode_group is not None:
+                conditions.append(Subscribe.episode_group == filters.episode_group)
+            music_condition = music_type_condition(
+                Subscribe.music_type,
+                filters.music_type,
+            )
+            if music_condition is not None:
+                conditions.append(music_condition)
+
+            count_statement = select(func.count(Subscribe.id))
+            page_statement = select(Subscribe)
+            if conditions:
+                count_statement = count_statement.where(*conditions)
+                page_statement = page_statement.where(*conditions)
+            descending_order = descending(page)
+            if page.sort.field == QuerySortField.ID:
+                primary = Subscribe.id.desc() if descending_order else Subscribe.id.asc()
+                secondary = Subscribe.date.desc() if descending_order else Subscribe.date.asc()
+            else:
+                primary = (
+                    Subscribe.date.desc().nullslast()
+                    if descending_order
+                    else Subscribe.date.asc().nullsfirst()
+                )
+                secondary = Subscribe.id.desc() if descending_order else Subscribe.id.asc()
+            page_statement = page_statement.order_by(primary, secondary)
+            return cast(
+                tuple[list[Subscribe], int],
+                execute_page(session, count_statement, page_statement, page),
+            )
+
+        return self._execute_sync_query(execute)
 
     async def async_get(self, sid: int) -> Optional[Subscribe]:
         """
diff --git a/app/db/oper/subscribehistory.py b/app/db/oper/subscribehistory.py
index 35d8d0421..869af34f9 100644
--- a/app/db/oper/subscribehistory.py
+++ b/app/db/oper/subscribehistory.py
@@ -1,10 +1,23 @@
-from typing import List, Optional
+from typing import List, Optional, cast
 
-from sqlalchemy import select
+from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import Session
 
 from app.db.base import DbOper
 from app.db.models.subscribehistory import SubscribeHistory
+from app.db.oper.query import (
+    descending,
+    enum_values,
+    execute_page,
+    media_identity_conditions,
+    music_type_condition,
+)
+from app.schemas.query import (
+    QueryPageRequest,
+    QuerySortField,
+    SubscriptionHistoryFilter,
+)
 
 
 class SubscribeHistoryOper(DbOper):
@@ -12,6 +25,87 @@ class SubscribeHistoryOper(DbOper):
     订阅历史管理。
     """
 
+    def get_by_id(self, record_id: int) -> Optional[SubscribeHistory]:
+        """按稳定记录 ID 读取单条订阅历史。"""
+        return cast(
+            Optional[SubscribeHistory],
+            self._execute_sync_query(
+                lambda session: session.execute(
+                    select(SubscribeHistory).where(SubscribeHistory.id == record_id)
+                ).scalars().first()
+            ),
+        )
+
+    def query(
+        self,
+        filters: SubscriptionHistoryFilter,
+        page: QueryPageRequest,
+    ) -> tuple[list[SubscribeHistory], int]:
+        """按稳定筛选和分页合同读取订阅历史记录及总数。"""
+        def execute(session: Session) -> tuple[list[SubscribeHistory], int]:
+            """在同一会话中构造并执行订阅历史 count/page 查询。"""
+            conditions = media_identity_conditions(SubscribeHistory, filters)
+            ids = enum_values(filters.ids)
+            names = enum_values(filters.names)
+            usernames = enum_values(filters.usernames)
+            media_types = enum_values(filters.media_types)
+            if ids:
+                conditions.append(SubscribeHistory.id.in_(ids))
+            if names:
+                conditions.append(SubscribeHistory.name.in_(names))
+            if usernames:
+                conditions.append(SubscribeHistory.username.in_(usernames))
+            if media_types:
+                conditions.append(SubscribeHistory.type.in_(media_types))
+            if filters.season is not None:
+                conditions.append(SubscribeHistory.season == filters.season)
+            if filters.episode_group is not None:
+                conditions.append(
+                    SubscribeHistory.episode_group == filters.episode_group
+                )
+            music_condition = music_type_condition(
+                SubscribeHistory.music_type,
+                filters.music_type,
+            )
+            if music_condition is not None:
+                conditions.append(music_condition)
+
+            count_statement = select(func.count(SubscribeHistory.id))
+            page_statement = select(SubscribeHistory)
+            if conditions:
+                count_statement = count_statement.where(*conditions)
+                page_statement = page_statement.where(*conditions)
+            descending_order = descending(page)
+            if page.sort.field == QuerySortField.ID:
+                primary = (
+                    SubscribeHistory.id.desc()
+                    if descending_order
+                    else SubscribeHistory.id.asc()
+                )
+                secondary = (
+                    SubscribeHistory.date.desc()
+                    if descending_order
+                    else SubscribeHistory.date.asc()
+                )
+            else:
+                primary = (
+                    SubscribeHistory.date.desc().nullslast()
+                    if descending_order
+                    else SubscribeHistory.date.asc().nullsfirst()
+                )
+                secondary = (
+                    SubscribeHistory.id.desc()
+                    if descending_order
+                    else SubscribeHistory.id.asc()
+                )
+            page_statement = page_statement.order_by(primary, secondary)
+            return cast(
+                tuple[list[SubscribeHistory], int],
+                execute_page(session, count_statement, page_statement, page),
+            )
+
+        return self._execute_sync_query(execute)
+
     async def async_list_by_type(
         self,
         mtype: str,
diff --git a/app/db/oper/transferhistory.py b/app/db/oper/transferhistory.py
index 8b3b44fc2..4c1049c2d 100644
--- a/app/db/oper/transferhistory.py
+++ b/app/db/oper/transferhistory.py
@@ -1,12 +1,27 @@
 import time
-from typing import Any, List, Optional
+from typing import Any, List, Optional, cast
 
 from sqlalchemy import delete as sqlalchemy_delete
+from sqlalchemy import func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Session
 
 from app.db.base import DbOper
 from app.db.models.transferhistory import TransferHistory
+from app.db.oper.query import (
+    descending,
+    enum_values,
+    execute_page,
+    literal_contains,
+    media_identity_conditions,
+    music_type_condition,
+    required_media_identity_conditions,
+)
+from app.schemas.query import (
+    QueryPageRequest,
+    QuerySortField,
+    TransferHistoryFilter,
+)
 from app.schemas.types import MediaSource
 
 
@@ -20,10 +35,111 @@ class TransferHistoryOper(DbOper):
         获取转移历史
         :param historyid: 转移历史id
         """
-        return self._execute_sync_query(
-            lambda session: TransferHistory.get(session, historyid)
+        return self.get_by_id(historyid)
+
+    def get_by_id(self, record_id: int) -> Optional[TransferHistory]:
+        """按稳定记录 ID 读取单条整理历史。"""
+        return cast(
+            Optional[TransferHistory],
+            self._execute_sync_query(
+                lambda session: session.execute(
+                    select(TransferHistory).where(TransferHistory.id == record_id)
+                ).scalars().first()
+            ),
         )
 
+    def query(
+        self,
+        filters: TransferHistoryFilter,
+        page: QueryPageRequest,
+    ) -> tuple[list[TransferHistory], int]:
+        """按稳定筛选和分页合同读取整理历史记录及总数。"""
+        def execute(session: Session) -> tuple[list[TransferHistory], int]:
+            """在同一会话中构造并执行整理历史 count/page 查询。"""
+            conditions = media_identity_conditions(TransferHistory, filters)
+            ids = enum_values(filters.ids)
+            media_types = enum_values(filters.media_types)
+            media_sources = enum_values(filters.media_sources)
+            if ids:
+                conditions.append(TransferHistory.id.in_(ids))
+            if media_types:
+                conditions.append(TransferHistory.type.in_(media_types))
+            if media_sources:
+                conditions.append(TransferHistory.media_source.in_(media_sources))
+            if filters.require_media_identity:
+                conditions.extend(required_media_identity_conditions(TransferHistory))
+            if filters.title:
+                conditions.append(TransferHistory.title == filters.title)
+            if filters.text:
+                conditions.append(
+                    literal_contains(TransferHistory.title, filters.text)
+                    | literal_contains(TransferHistory.src, filters.text)
+                    | literal_contains(TransferHistory.dest, filters.text)
+                )
+            for column, value in (
+                (TransferHistory.year, filters.year),
+                (TransferHistory.seasons, filters.seasons),
+                (TransferHistory.episodes, filters.episodes),
+                (TransferHistory.src, filters.src),
+                (TransferHistory.dest, filters.dest),
+                (TransferHistory.download_hash, filters.download_hash),
+                (TransferHistory.episode_group, filters.episode_group),
+            ):
+                if value is not None and value != "":
+                    conditions.append(column == value)
+            if filters.status is not None:
+                if filters.status:
+                    conditions.append(TransferHistory.status.is_(True))
+                else:
+                    conditions.append(
+                        or_(
+                            TransferHistory.status.is_(False),
+                            TransferHistory.status.is_(None),
+                        )
+                    )
+            music_condition = music_type_condition(
+                TransferHistory.music_type,
+                filters.music_type,
+            )
+            if music_condition is not None:
+                conditions.append(music_condition)
+
+            count_statement = select(func.count(TransferHistory.id))
+            page_statement = select(TransferHistory)
+            if conditions:
+                count_statement = count_statement.where(*conditions)
+                page_statement = page_statement.where(*conditions)
+            descending_order = descending(page)
+            if page.sort.field == QuerySortField.ID:
+                primary = (
+                    TransferHistory.id.desc()
+                    if descending_order
+                    else TransferHistory.id.asc()
+                )
+                secondary = (
+                    TransferHistory.date.desc()
+                    if descending_order
+                    else TransferHistory.date.asc()
+                )
+            else:
+                primary = (
+                    TransferHistory.date.desc().nullslast()
+                    if descending_order
+                    else TransferHistory.date.asc().nullsfirst()
+                )
+                secondary = (
+                    TransferHistory.id.desc()
+                    if descending_order
+                    else TransferHistory.id.asc()
+                )
+            page_statement = page_statement.order_by(primary, secondary)
+            return cast(
+                tuple[list[TransferHistory], int],
+                execute_page(session, count_statement, page_statement, page),
+            )
+
+        return self._execute_sync_query(execute)
+
     async def async_get(self, historyid: int) -> Optional[TransferHistory]:
         """
         异步获取转移历史。
diff --git a/app/schemas/query.py b/app/schemas/query.py
index 8cc31c0c7..1fc3f42e6 100644
--- a/app/schemas/query.py
+++ b/app/schemas/query.py
@@ -6,7 +6,7 @@ from typing import Any, Generic, Optional, TypeVar
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
 from app.schemas.common import JsonData
-from app.schemas.media import OptionalMediaIdentityMixin, normalize_media_source
+from app.schemas.media import normalize_media_source
 from app.schemas.types import MediaSource, MediaType
 
 T = TypeVar("T")
@@ -36,14 +36,18 @@ class QuerySortDirection(str, Enum):
 
 
 class QuerySort(_QueryInput):
-    """声明公开查询的稳定排序字段与方向。"""
+    """声明公开查询的稳定排序字段与方向。
+
+    日期排序将空日期置于升序开头、降序末尾，并以同方向 ID 打破日期并列；ID
+    本身唯一，因此按 ID 排序不依赖数据库的隐式行顺序。
+    """
 
     field: QuerySortField = QuerySortField.DATE
     direction: QuerySortDirection = QuerySortDirection.DESC
 
 
 class QueryPageRequest(_QueryInput):
-    """限制插件单次读取规模，并为跨页扫描提供稳定顺序。"""
+    """限制单次读取规模，并通过显式排序为跨页扫描提供稳定顺序。"""
 
     page: int = Field(default=1, ge=1)
     count: int = Field(
@@ -72,15 +76,25 @@ class QueryPage(BaseModel, Generic[T]):  # type: ignore[misc]  # Pydantic import
         return self.page * self.count < self.total
 
 
-class MediaIdentityQuery(
-    OptionalMediaIdentityMixin,
-    _QueryInput,
-):
+class MediaIdentityQuery(_QueryInput):
     """允许省略身份，但显式筛选时要求来源与原生 ID 成对有效。"""
 
     media_source: Optional[MediaSource] = None
     media_id: Optional[str] = None
 
+    @model_validator(mode="after")  # type: ignore[misc]
+    def _validate_media_identity(self) -> "MediaIdentityQuery":
+        """规范化 ID，并拒绝显式半对、空白或零值身份。"""
+        source_provided = "media_source" in self.model_fields_set
+        id_provided = "media_id" in self.model_fields_set
+        normalized_id = str(self.media_id).strip() if self.media_id is not None else None
+        if source_provided != id_provided or bool(self.media_source) != bool(normalized_id):
+            raise ValueError("media_source 和 media_id 必须同时提供")
+        if normalized_id == "0":
+            raise ValueError("media_id 不能为 0")
+        object.__setattr__(self, "media_id", normalized_id)
+        return self
+
 
 class _QuerySnapshot(BaseModel):  # type: ignore[misc]  # Pydantic imports are skipped by strict mypy
     """只读查询 DTO 的媒体身份与脏数据归一化边界。"""
@@ -279,7 +293,11 @@ class TransferHistorySnapshot(_QuerySnapshot):
 
 
 class SubscriptionFilter(MediaIdentityQuery):
-    """订阅查询允许组合的稳定业务字段。"""
+    """当前订阅的组合筛选合同。
+
+    所有非空字段按 AND 组合，tuple 字段内部按 IN 匹配，其余字段均精确匹配；
+    ``music_type=recording`` 同时匹配未标注音乐类型的旧单曲记录。
+    """
 
     ids: tuple[int, ...] = ()
     names: tuple[str, ...] = ()
@@ -292,7 +310,11 @@ class SubscriptionFilter(MediaIdentityQuery):
 
 
 class SubscriptionHistoryFilter(MediaIdentityQuery):
-    """订阅完成历史查询允许组合的稳定业务字段。"""
+    """订阅完成历史的组合筛选合同。
+
+    所有非空字段按 AND 组合，tuple 字段内部按 IN 匹配，其余字段均精确匹配；
+    ``music_type=recording`` 同时匹配未标注音乐类型的旧单曲记录。
+    """
 
     ids: tuple[int, ...] = ()
     names: tuple[str, ...] = ()
@@ -304,7 +326,12 @@ class SubscriptionHistoryFilter(MediaIdentityQuery):
 
 
 class DownloadHistoryFilter(MediaIdentityQuery):
-    """下载历史查询允许组合的稳定业务字段。"""
+    """下载历史的组合筛选合同。
+
+    所有非空字段按 AND 组合，tuple 字段内部按 IN 匹配；除 ``text`` 外的字符串
+    字段均精确匹配。``text`` 对标题和路径执行转义后的字面包含查询，不把 ``%``
+    或 ``_`` 解释为通配符。``music_type=recording`` 同时匹配旧 NULL 记录。
+    """
 
     ids: tuple[int, ...] = ()
     media_types: tuple[MediaType, ...] = ()
@@ -322,7 +349,13 @@ class DownloadHistoryFilter(MediaIdentityQuery):
 
 
 class TransferHistoryFilter(MediaIdentityQuery):
-    """整理历史查询允许组合的稳定业务字段。"""
+    """整理历史的组合筛选合同。
+
+    所有非空字段按 AND 组合，tuple 字段内部按 IN 匹配；除 ``text`` 外的字符串
+    字段均精确匹配。``text`` 对标题、源路径和目标路径执行转义后的字面包含查询。
+    ``require_media_identity`` 只保留来源可解析且原生 ID 非空、非零的记录；
+    ``status=False`` 包含旧 NULL 状态，``music_type=recording`` 包含旧 NULL 类型。
+    """
 
     ids: tuple[int, ...] = ()
     media_types: tuple[MediaType, ...] = ()
diff --git a/app/sdk/queries.py b/app/sdk/queries.py
index d9150dae5..8346397fa 100644
--- a/app/sdk/queries.py
+++ b/app/sdk/queries.py
@@ -118,7 +118,7 @@ class _DataQueryBackend(Protocol):
 
 def _service() -> _DataQueryBackend:
     """获取启动阶段登记的查询服务，避免把应用服务暴露为 SDK 合同。"""
-    from app.application.data_query import (
+    from app.application.query import (
         get_configured_data_query_service,
     )
 
diff --git a/app/startup/initializers/modules.py b/app/startup/initializers/modules.py
index 65405992c..3b35a7071 100644
--- a/app/startup/initializers/modules.py
+++ b/app/startup/initializers/modules.py
@@ -60,10 +60,6 @@ from app.application.configuration import (
     configure_transfer_retry_config,
     get_configured_system_config,
 )
-from app.application.data_query import (
-    DataQueryService,
-    configure_data_query_service,
-)
 from app.application.database import configure_database_governance
 from app.application.history import configure_transfer_history_provider
 from app.application.image import configure_wallpaper_providers
@@ -92,6 +88,10 @@ from app.application.outbox import (
     validate_durable_event_handlers,
 )
 from app.application.plugin.runtime import configure_plugin_runtime
+from app.application.query import (
+    DataQueryService,
+    configure_data_query_service,
+)
 from app.application.security.auth import AuthService, build_superuser_token_payload, configure_auth_service
 from app.application.security.passkey import PasskeyService, configure_passkey_service
 from app.application.security.url import close_image_proxy_block_log_coalescer
@@ -109,9 +109,9 @@ from app.application.subscription.write import configure_subscribe_writer
 from app.application.workflow import WorkflowQueryService, configure_workflow_query
 from app.command import CommandChain
 from app.db.adapters.chain import TransactionalChainDurableEventWriter
-from app.db.adapters.data_query import SqlAlchemyDataQueryAdapter
 from app.db.adapters.download import TransactionalDownloadFailureRepository
 from app.db.adapters.outbox import SqlAlchemyAsyncOutboxStager, SqlAlchemyOutboxRepository
+from app.db.adapters.query import SqlAlchemyDataQueryAdapter
 from app.db.adapters.site import TransactionalSiteRepository
 from app.db.adapters.subscription import TransactionalSubscribeWriter
 from app.db.adapters.transaction import TransactionalWriteRunner
diff --git a/docs/architecture-optimization-checklist.md b/docs/architecture-optimization-checklist.md
index dc87fd4e0..ad3756f5f 100644
--- a/docs/architecture-optimization-checklist.md
+++ b/docs/architecture-optimization-checklist.md
@@ -69,7 +69,7 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain`
 
 | 指标 | 当前值 | 解释 |
 |---|---:|---|
-| 宿主 Python 模块 / 内部依赖边 | 848 / 6,924 | `dependency-baseline.json` 当前快照 |
+| 宿主 Python 模块 / 内部依赖边 | 849 / 6,940 | `dependency-baseline.json` 当前快照 |
 | 非平凡 SCC | 2 | 新增 Chain 包根环；另一个是隔离的 29 模块 TMDB 移植包环 |
 | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 |
 | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |
@@ -78,7 +78,7 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain`
 | Python 源码量 | 约 271,400 行 | 60 个文件超过 1,000 行，14 个超过 2,000 行 |
 | 长方法 | 281 个超过 80 行 | 67 个超过 150 行，23 个超过 250 行；大量是私有方法 |
 | 全量 mypy 历史债务 | 11,827 / 596 文件 | strict frontier 当前覆盖 41 个文件，本批迁移路径的类型债务已清零 |
-| Ruff 历史诊断 | 888 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
+| Ruff 历史诊断 | 885 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
 | 覆盖率低水位 | Application 78.63%，Domain 79.29% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 |
 
 ### 3.3 热点文件
diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md
index 11408e4d2..1cbd4efc5 100644
--- a/docs/architecture-overview.md
+++ b/docs/architecture-overview.md
@@ -704,8 +704,8 @@ flowchart LR
 
 | 指标 | 当前值 |
 |---|---:|
-| Python 模块 | 848 |
-| 内部导入边 | 6,924 |
+| Python 模块 | 849 |
+| 内部导入边 | 6,940 |
 | 非平凡 SCC | 2（`ARCH-107` 临时 Chain 包根环；精确 containment 的 TMDB 移植包环） |
 | Direct egress | 66（12 条待迁移债务，54 条精确 containment） |
 | Module Contract V2 spec | 217（其中 215 个进入 `run_module` 观察面） |
diff --git a/docs/architecture-refactor-roadmap.md b/docs/architecture-refactor-roadmap.md
index db7e58947..eaa9254b3 100644
--- a/docs/architecture-refactor-roadmap.md
+++ b/docs/architecture-refactor-roadmap.md
@@ -144,7 +144,7 @@ G-ARCH 只有在以下条件全部满足后才可完成：
 | S4-L2 Event strict contract | `PLANNED` | S0-L2.6,S1-L6 | 宿主事件输入/输出按风险 strict，诊断例外只属于第三方插件兼容 |
 | S4-L3 Complexity v2 | `PLANNED` | S3 | 私有方法、class/file、圈复杂度进入门禁；所有超限通过职责拆分归零 |
 | S4-L4 全量 mypy 清零 | `PLANNED` | S3,S4-L1,S4-L2 | `mypy-baseline.json` 归零并删除债务接受路径，全宿主 strict 类型通过 |
-| S4-L5 Ruff 治理债务清零 | `PLANNED` | S3 | 当前受控 888 条诊断归零，规则集扩展经过独立审查且新增诊断为零 |
+| S4-L5 Ruff 治理债务清零 | `PLANNED` | S3 | 当前受控 885 条诊断归零，规则集扩展经过独立审查且新增诊断为零 |
 | S4-L6 Coverage/并发/质量证据 | `PLANNED` | S3,S4-L1,S4-L2 | 高风险包纳入 coverage；raw concurrency 分类清零；Module Quality 有真实 evidence test |
 
 ### S5：Plugin、Agent、Domain、Startup 与最终收口
diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json
index 4a822d27b..ddbe5094b 100644
--- a/tests/fixtures/architecture/dependency-baseline.json
+++ b/tests/fixtures/architecture/dependency-baseline.json
@@ -1441,8 +1441,8 @@
       "runtime_only": true
     }
   },
-  "edge_count": 6924,
-  "edge_sha256": "d5471439086bcfdd8060b77670af6f821a34e3a864ba7e2cb7230c599a1cb3bb",
+  "edge_count": 6940,
+  "edge_sha256": "8ff91be099f1230655ceeb006bd1bf604063054ddb0721b3e887c956ea1251fb",
   "edges": [
     "app -> app.runtime",
     "app -> app.runtime.compat",
@@ -4008,10 +4008,6 @@
     "app.application.configuration -> app.schemas.types",
     "app.application.dashboard -> app.schemas",
     "app.application.dashboard -> app.schemas.dashboard",
-    "app.application.data_query -> app.application",
-    "app.application.data_query -> app.application.database",
-    "app.application.data_query -> app.schemas",
-    "app.application.data_query -> app.schemas.query",
     "app.application.directory -> app.adapters",
     "app.application.directory -> app.adapters.system",
     "app.application.directory -> app.adapters.system.host",
@@ -4264,6 +4260,10 @@
     "app.application.plugin.transaction -> app.application.database",
     "app.application.plugin.transaction -> app.application.plugin",
     "app.application.plugin.transaction -> app.application.plugin.identity",
+    "app.application.query -> app.application",
+    "app.application.query -> app.application.database",
+    "app.application.query -> app.schemas",
+    "app.application.query -> app.schemas.query",
     "app.application.recognition -> app.application",
     "app.application.recognition -> app.application.configuration",
     "app.application.recognition -> app.schemas",
@@ -5135,18 +5135,6 @@
     "app.db.adapters.chain -> app.db.oper.transferpending",
     "app.db.adapters.chain -> app.db.oper.transfersettlementreceipt",
     "app.db.adapters.chain -> app.db.uow",
-    "app.db.adapters.data_query -> app.application",
-    "app.db.adapters.data_query -> app.application.data_query",
-    "app.db.adapters.data_query -> app.db",
-    "app.db.adapters.data_query -> app.db.base",
-    "app.db.adapters.data_query -> app.db.models",
-    "app.db.adapters.data_query -> app.db.models.downloadhistory",
-    "app.db.adapters.data_query -> app.db.models.subscribe",
-    "app.db.adapters.data_query -> app.db.models.subscribehistory",
-    "app.db.adapters.data_query -> app.db.models.transferhistory",
-    "app.db.adapters.data_query -> app.schemas",
-    "app.db.adapters.data_query -> app.schemas.query",
-    "app.db.adapters.data_query -> app.schemas.types",
     "app.db.adapters.download -> app.db",
     "app.db.adapters.download -> app.db.oper",
     "app.db.adapters.download -> app.db.oper.downloadfailure",
@@ -5176,6 +5164,16 @@
     "app.db.adapters.plugininstallation -> app.db.models",
     "app.db.adapters.plugininstallation -> app.db.models.pluginidentity",
     "app.db.adapters.plugininstallation -> app.db.models.plugininstallation",
+    "app.db.adapters.query -> app.application",
+    "app.db.adapters.query -> app.application.query",
+    "app.db.adapters.query -> app.db",
+    "app.db.adapters.query -> app.db.oper",
+    "app.db.adapters.query -> app.db.oper.downloadhistory",
+    "app.db.adapters.query -> app.db.oper.subscribe",
+    "app.db.adapters.query -> app.db.oper.subscribehistory",
+    "app.db.adapters.query -> app.db.oper.transferhistory",
+    "app.db.adapters.query -> app.schemas",
+    "app.db.adapters.query -> app.schemas.query",
     "app.db.adapters.site -> app.db",
     "app.db.adapters.site -> app.db.oper",
     "app.db.adapters.site -> app.db.oper.site",
@@ -5355,7 +5353,10 @@
     "app.db.oper.downloadhistory -> app.db.base",
     "app.db.oper.downloadhistory -> app.db.models",
     "app.db.oper.downloadhistory -> app.db.models.downloadhistory",
+    "app.db.oper.downloadhistory -> app.db.oper",
+    "app.db.oper.downloadhistory -> app.db.oper.query",
     "app.db.oper.downloadhistory -> app.schemas",
+    "app.db.oper.downloadhistory -> app.schemas.query",
     "app.db.oper.downloadhistory -> app.schemas.types",
     "app.db.oper.mediaserver -> app.db",
     "app.db.oper.mediaserver -> app.db.base",
@@ -5380,6 +5381,11 @@
     "app.db.oper.pluginidentity -> app.db.base",
     "app.db.oper.pluginidentity -> app.db.models",
     "app.db.oper.pluginidentity -> app.db.models.pluginidentity",
+    "app.db.oper.query -> app.db",
+    "app.db.oper.query -> app.db.base",
+    "app.db.oper.query -> app.schemas",
+    "app.db.oper.query -> app.schemas.query",
+    "app.db.oper.query -> app.schemas.types",
     "app.db.oper.site -> app.db",
     "app.db.oper.site -> app.db.base",
     "app.db.oper.site -> app.db.models",
@@ -5395,12 +5401,19 @@
     "app.db.oper.subscribe -> app.db.models",
     "app.db.oper.subscribe -> app.db.models.subscribe",
     "app.db.oper.subscribe -> app.db.models.subscribehistory",
+    "app.db.oper.subscribe -> app.db.oper",
+    "app.db.oper.subscribe -> app.db.oper.query",
     "app.db.oper.subscribe -> app.schemas",
+    "app.db.oper.subscribe -> app.schemas.query",
     "app.db.oper.subscribe -> app.schemas.types",
     "app.db.oper.subscribehistory -> app.db",
     "app.db.oper.subscribehistory -> app.db.base",
     "app.db.oper.subscribehistory -> app.db.models",
     "app.db.oper.subscribehistory -> app.db.models.subscribehistory",
+    "app.db.oper.subscribehistory -> app.db.oper",
+    "app.db.oper.subscribehistory -> app.db.oper.query",
+    "app.db.oper.subscribehistory -> app.schemas",
+    "app.db.oper.subscribehistory -> app.schemas.query",
     "app.db.oper.systemconfig -> app.db",
     "app.db.oper.systemconfig -> app.db.base",
     "app.db.oper.systemconfig -> app.db.models",
@@ -5417,7 +5430,10 @@
     "app.db.oper.transferhistory -> app.db.base",
     "app.db.oper.transferhistory -> app.db.models",
     "app.db.oper.transferhistory -> app.db.models.transferhistory",
+    "app.db.oper.transferhistory -> app.db.oper",
+    "app.db.oper.transferhistory -> app.db.oper.query",
     "app.db.oper.transferhistory -> app.schemas",
+    "app.db.oper.transferhistory -> app.schemas.query",
     "app.db.oper.transferhistory -> app.schemas.types",
     "app.db.oper.transferpending -> app.db",
     "app.db.oper.transferpending -> app.db.base",
@@ -7769,7 +7785,7 @@
     "app.sdk.plugins -> app.runtime.extensions.module_manager",
     "app.sdk.plugins -> app.runtime.extensions.plugin_manager",
     "app.sdk.queries -> app.application",
-    "app.sdk.queries -> app.application.data_query",
+    "app.sdk.queries -> app.application.query",
     "app.sdk.queries -> app.schemas",
     "app.sdk.queries -> app.schemas.query",
     "app.sdk.security -> app.adapters",
@@ -7953,7 +7969,6 @@
     "app.startup.initializers.modules -> app.application.chain.data",
     "app.startup.initializers.modules -> app.application.chain.events",
     "app.startup.initializers.modules -> app.application.configuration",
-    "app.startup.initializers.modules -> app.application.data_query",
     "app.startup.initializers.modules -> app.application.database",
     "app.startup.initializers.modules -> app.application.history",
     "app.startup.initializers.modules -> app.application.image",
@@ -7966,6 +7981,7 @@
     "app.startup.initializers.modules -> app.application.plugin",
     "app.startup.initializers.modules -> app.application.plugin.runtime",
     "app.startup.initializers.modules -> app.application.plugin.transaction",
+    "app.startup.initializers.modules -> app.application.query",
     "app.startup.initializers.modules -> app.application.security",
     "app.startup.initializers.modules -> app.application.security.auth",
     "app.startup.initializers.modules -> app.application.security.passkey",
@@ -7996,11 +8012,11 @@
     "app.startup.initializers.modules -> app.db",
     "app.startup.initializers.modules -> app.db.adapters",
     "app.startup.initializers.modules -> app.db.adapters.chain",
-    "app.startup.initializers.modules -> app.db.adapters.data_query",
     "app.startup.initializers.modules -> app.db.adapters.download",
     "app.startup.initializers.modules -> app.db.adapters.outbox",
     "app.startup.initializers.modules -> app.db.adapters.pluginidentity",
     "app.startup.initializers.modules -> app.db.adapters.plugininstallation",
+    "app.startup.initializers.modules -> app.db.adapters.query",
     "app.startup.initializers.modules -> app.db.adapters.site",
     "app.startup.initializers.modules -> app.db.adapters.subscription",
     "app.startup.initializers.modules -> app.db.adapters.transaction",
@@ -8369,7 +8385,7 @@
     "app.workflow.actions.transfer_file -> app.workflow",
     "app.workflow.actions.transfer_file -> app.workflow.actions"
   ],
-  "module_count": 848,
+  "module_count": 849,
   "modules": [
     "app",
     "app.adapters",
@@ -8629,7 +8645,6 @@
     "app.application.commands",
     "app.application.configuration",
     "app.application.dashboard",
-    "app.application.data_query",
     "app.application.database",
     "app.application.directory",
     "app.application.download",
@@ -8678,6 +8693,7 @@
     "app.application.plugin.runtime",
     "app.application.plugin.source",
     "app.application.plugin.transaction",
+    "app.application.query",
     "app.application.recognition",
     "app.application.rss",
     "app.application.rules",
@@ -8762,11 +8778,11 @@
     "app.db",
     "app.db.adapters",
     "app.db.adapters.chain",
-    "app.db.adapters.data_query",
     "app.db.adapters.download",
     "app.db.adapters.outbox",
     "app.db.adapters.pluginidentity",
     "app.db.adapters.plugininstallation",
+    "app.db.adapters.query",
     "app.db.adapters.site",
     "app.db.adapters.subscription",
     "app.db.adapters.transaction",
@@ -8819,6 +8835,7 @@
     "app.db.oper.passkey",
     "app.db.oper.plugindata",
     "app.db.oper.pluginidentity",
+    "app.db.oper.query",
     "app.db.oper.site",
     "app.db.oper.subscribe",
     "app.db.oper.subscribehistory",
diff --git a/tests/fixtures/architecture/ruff-baseline.json b/tests/fixtures/architecture/ruff-baseline.json
index 6758b3e1d..438fbe92b 100644
--- a/tests/fixtures/architecture/ruff-baseline.json
+++ b/tests/fixtures/architecture/ruff-baseline.json
@@ -467,18 +467,12 @@
   "app/db/models/workflow.py": {
     "I001": 1
   },
-  "app/db/oper/downloadhistory.py": {
-    "I001": 1
-  },
   "app/db/oper/message.py": {
     "I001": 1
   },
   "app/db/oper/site.py": {
     "I001": 1
   },
-  "app/db/oper/subscribe.py": {
-    "I001": 1
-  },
   "app/db/oper/systemconfig.py": {
     "I001": 1
   },
@@ -1128,9 +1122,6 @@
   "tests/test_agent_lazy_runtime_boundary.py": {
     "F401": 1
   },
-  "tests/test_agent_lifecycle.py": {
-    "I001": 1
-  },
   "tests/test_agent_llm_capability.py": {
     "I001": 1
   },
diff --git a/tests/test_agent_lifecycle.py b/tests/test_agent_lifecycle.py
index 16b7eb94a..14a3c8167 100644
--- a/tests/test_agent_lifecycle.py
+++ b/tests/test_agent_lifecycle.py
@@ -4,18 +4,19 @@ from unittest.mock import AsyncMock, MagicMock
 import pytest
 
 import app.agent.orchestrator as agent_module
-from app.application.messaging.agent import (
-    create_web_agent_background_task,
-    shutdown_web_agent_background_tasks,
-)
+from app.agent.memory import MemoryManager
 from app.agent.orchestrator import (
     AGENT_SESSION_QUEUE_MAX_SIZE,
     AgentManager,
     AgentManagerQueueFullError,
     AgentManagerUnavailableError,
 )
-from app.agent.memory import MemoryManager
 from app.agent.tools.base import reopen_blocking_executors
+from app.application.messaging.agent import (
+    create_web_agent_background_task,
+    shutdown_web_agent_background_tasks,
+)
+from app.application.query import get_configured_data_query_service
 from app.startup.initializers import agent as agent_initializer
 from app.startup.initializers import modules as modules_initializer
 
@@ -202,6 +203,9 @@ async def test_agent_initialization_failure_does_not_stop_module_startup(
         assert runtime.workflow.system_config() is (
             modules_initializer.get_configured_system_config()
         )
+        query_page = get_configured_data_query_service().list_subscriptions({"ids": [-1]})
+        assert query_page.items == []
+        assert query_page.total == 0
     finally:
         await modules_initializer.stop_database_worker()
 
diff --git a/tests/test_sdk_queries.py b/tests/test_sdk_queries.py
index 191e494f8..33be4bff3 100644
--- a/tests/test_sdk_queries.py
+++ b/tests/test_sdk_queries.py
@@ -61,9 +61,9 @@ class _RecordingExecutor:
 @pytest.fixture
 def query_sdk(db, monkeypatch):
     """装配真实数据查询适配器，并把 SDK 绑定到本用例的 SQLite 服务。"""
-    from app.application import data_query as data_query_module
-    from app.application.data_query import DataQueryService
-    from app.db.adapters.data_query import SqlAlchemyDataQueryAdapter
+    from app.application import query as data_query_module
+    from app.application.query import DataQueryService
+    from app.db.adapters.query import SqlAlchemyDataQueryAdapter
     from app.db.session import SessionFactory
     from app.sdk import queries as sdk
 
@@ -180,7 +180,7 @@ def _transfer_history(
     episodes: str | None = "E01",
     download_hash: str | None = "hash-1",
     episode_group: str | None = None,
-    status: bool = True,
+    status: bool | None = True,
     date: str = "2026-08-27 10:00:00",
     mtype: str = MediaType.TV.value,
 ) -> TransferHistoryModel:
@@ -630,6 +630,67 @@ def test_snapshots_normalize_legacy_identity_and_transfer_status():
     assert snapshot.status is False
 
 
+def test_transfer_failure_filter_includes_legacy_null_status(db, query_sdk):
+    """失败筛选与快照语义一致，包含状态尚未回填的旧整理记录。"""
+    sdk, _executor = query_sdk
+    legacy_row = db.add(
+        _transfer_history(
+            "Legacy null status",
+            media_id="legacy-status",
+            src="/legacy/status-src.mkv",
+            dest="/legacy/status-dest.mkv",
+            status=None,
+        )
+    )
+    legacy_row.status = None
+    db.session.commit()
+
+    page = sdk.list_transfer_history(
+        TransferHistoryFilter(
+            media_source=TMDB,
+            media_id="legacy-status",
+            status=False,
+        )
+    )
+
+    assert [item.id for item in page.items] == [legacy_row.id]
+    assert page.items[0].status is False
+
+
+def test_transfer_required_identity_excludes_unparseable_legacy_source(db, query_sdk):
+    """有效身份筛选不得返回随后会被快照降级为空身份的脏来源。"""
+    sdk, _executor = query_sdk
+    valid_row = db.add(
+        _transfer_history(
+            "Valid dynamic source",
+            media_id="valid-dynamic",
+            src="/valid/source.mkv",
+            dest="/valid/dest.mkv",
+        )
+    )
+    invalid_row = db.add(
+        _transfer_history(
+            "Invalid legacy source",
+            media_id="invalid-source",
+            src="/invalid/source.mkv",
+            dest="/invalid/dest.mkv",
+        )
+    )
+    valid_row.media_source = "plugin-source"
+    invalid_row.media_source = "invalid!"
+    db.session.commit()
+
+    page = sdk.list_transfer_history(
+        TransferHistoryFilter(
+            ids=(valid_row.id, invalid_row.id),
+            require_media_identity=True,
+        )
+    )
+
+    assert [item.id for item in page.items] == [valid_row.id]
+    assert page.items[0].media_source == MediaSource("plugin-source")
+
+
 def test_query_snapshots_are_owned_by_the_sdk_contract_module():
     """公开查询返回值由独立快照定义，不复用宿主写入或 API 响应模型。"""
     snapshots = (