diff --git a/app/application/data_query.py b/app/application/data_query.py
new file mode 100644
index 000000000..6b83a9b2f
--- /dev/null
+++ b/app/application/data_query.py
@@ -0,0 +1,350 @@
+"""插件只读数据查询的应用服务与领域端口。"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from functools import partial
+from typing import Any, Callable, Generic, Protocol, TypeVar, cast
+
+from pydantic import BaseModel
+
+from app.application.database import AsyncDatabaseExecutor
+from app.schemas.history import DownloadHistory, TransferHistory
+from app.schemas.query import (
+    DownloadHistoryFilter,
+    QueryPage,
+    QueryPageRequest,
+    SubscribeHistory,
+    SubscriptionFilter,
+    SubscriptionHistoryFilter,
+    TransferHistoryFilter,
+)
+from app.schemas.subscribe import Subscribe
+
+RecordT = TypeVar("RecordT")
+DtoT = TypeVar("DtoT", bound=BaseModel)
+ResultT = TypeVar("ResultT")
+
+
+@dataclass(frozen=True, slots=True)
+class QueryRows(Generic[RecordT]):
+    """查询端口返回的短生命周期结果，不跨越应用服务边界。"""
+
+    items: Sequence[RecordT]
+    total: int
+
+
+class SubscriptionQueryPort(Protocol):
+    """订阅领域查询端口，包含订阅和订阅完成历史两个只读切片。"""
+
+    def list_subscriptions(
+        self,
+        *,
+        filters: SubscriptionFilter,
+        page: QueryPageRequest,
+    ) -> QueryRows[object]:
+        """按组合筛选和稳定分页读取订阅记录。"""
+        ...
+
+    def get_subscription(self, subscription_id: int) -> object | None:
+        """按 ID 读取一条订阅记录。"""
+        ...
+
+    def list_subscription_history(
+        self,
+        *,
+        filters: SubscriptionHistoryFilter,
+        page: QueryPageRequest,
+    ) -> QueryRows[object]:
+        """按组合筛选和稳定分页读取订阅完成历史。"""
+        ...
+
+    def get_subscription_history(self, history_id: int) -> object | None:
+        """按 ID 读取一条订阅完成历史。"""
+        ...
+
+
+class HistoryQueryPort(Protocol):
+    """下载与整理历史查询端口。"""
+
+    def list_download_history(
+        self,
+        *,
+        filters: DownloadHistoryFilter,
+        page: QueryPageRequest,
+    ) -> QueryRows[object]:
+        """按组合筛选和稳定分页读取下载历史。"""
+        ...
+
+    def get_download_history(self, history_id: int) -> object | None:
+        """按 ID 读取一条下载历史。"""
+        ...
+
+    def list_transfer_history(
+        self,
+        *,
+        filters: TransferHistoryFilter,
+        page: QueryPageRequest,
+    ) -> QueryRows[object]:
+        """按组合筛选和稳定分页读取整理历史。"""
+        ...
+
+    def get_transfer_history(self, history_id: int) -> object | None:
+        """按 ID 读取一条整理历史。"""
+        ...
+
+
+class DataQueryService:
+    """把订阅和历史查询统一投影为不含持久化实现的 DTO。"""
+
+    def __init__(
+        self,
+        *,
+        subscriptions: SubscriptionQueryPort,
+        histories: HistoryQueryPort,
+        async_executor: AsyncDatabaseExecutor,
+    ) -> None:
+        """保存两个领域查询端口和异步数据库执行边界。"""
+        self._subscriptions = subscriptions
+        self._histories = histories
+        self._async_executor = async_executor
+
+    @staticmethod
+    def _page_request(page: QueryPageRequest | dict[str, Any] | None) -> QueryPageRequest:
+        """把调用方输入规范化为有界分页合同。"""
+        if page is None:
+            return QueryPageRequest()
+        if isinstance(page, QueryPageRequest):
+            return page
+        return cast(QueryPageRequest, QueryPageRequest.model_validate(page))
+
+    @staticmethod
+    def _filter(
+        value: Any,
+        filter_type: type[SubscriptionFilter]
+        | type[SubscriptionHistoryFilter]
+        | type[DownloadHistoryFilter]
+        | type[TransferHistoryFilter],
+    ) -> Any:
+        """把字典筛选条件转换为带媒体身份校验的模型。"""
+        if value is None:
+            return filter_type()
+        if isinstance(value, filter_type):
+            return value
+        return filter_type.model_validate(value)
+
+    @staticmethod
+    def _rows(value: QueryRows[object]) -> QueryRows[object]:
+        """校验端口结果的总数，避免负数污染分页合同。"""
+        if not isinstance(value, QueryRows):
+            raise TypeError("查询端口必须返回 QueryRows")
+        return QueryRows(
+            items=value.items,
+            total=max(int(value.total), 0),
+        )
+
+    @classmethod
+    def _to_page(
+        cls,
+        dto_type: type[DtoT],
+        page: QueryPageRequest,
+        rows: QueryRows[object],
+    ) -> QueryPage[DtoT]:
+        """在查询端口结果离开应用层前完成 DTO 投影。"""
+        normalized_rows = cls._rows(rows)
+        return cast(
+            QueryPage[DtoT],
+            QueryPage(
+                items=[dto_type.model_validate(item) for item in normalized_rows.items],
+                total=normalized_rows.total,
+                page=page.page,
+                count=page.count,
+            ),
+        )
+
+    @staticmethod
+    def _to_item(dto_type: type[DtoT], item: object | None) -> DtoT | None:
+        """把单条端口记录投影为稳定 DTO。"""
+        return dto_type.model_validate(item) if item is not None else None
+
+    async def _async_run(self, operation: Callable[[], ResultT]) -> ResultT:
+        """通过统一数据库执行器运行一个同步查询用例。"""
+        return cast(ResultT, await self._async_executor.run(operation))
+
+    def list_subscriptions(
+        self,
+        filters: SubscriptionFilter | dict[str, Any] | None = None,
+        page: QueryPageRequest | dict[str, Any] | None = None,
+    ) -> QueryPage[Subscribe]:
+        """同步分页查询订阅。"""
+        normalized_page = self._page_request(page)
+        normalized_filters = self._filter(filters, SubscriptionFilter)
+        rows = self._subscriptions.list_subscriptions(
+            filters=normalized_filters,
+            page=normalized_page,
+        )
+        return self._to_page(Subscribe, normalized_page, rows)
+
+    async def async_list_subscriptions(
+        self,
+        filters: SubscriptionFilter | dict[str, Any] | None = None,
+        page: QueryPageRequest | dict[str, Any] | None = None,
+    ) -> QueryPage[Subscribe]:
+        """异步分页查询订阅，业务规则在数据库 worker 中复用同步入口。"""
+        return await self._async_run(
+            partial(self.list_subscriptions, filters, page)
+        )
+
+    def get_subscription(self, subscription_id: int) -> Subscribe | None:
+        """同步按 ID 查询订阅。"""
+        return self._to_item(
+            Subscribe,
+            self._subscriptions.get_subscription(subscription_id),
+        )
+
+    async def async_get_subscription(self, subscription_id: int) -> Subscribe | None:
+        """异步按 ID 查询订阅。"""
+        return await self._async_run(
+            partial(self.get_subscription, subscription_id)
+        )
+
+    def list_subscription_history(
+        self,
+        filters: SubscriptionHistoryFilter | dict[str, Any] | None = None,
+        page: QueryPageRequest | dict[str, Any] | None = None,
+    ) -> QueryPage[SubscribeHistory]:
+        """同步分页查询订阅完成历史。"""
+        normalized_page = self._page_request(page)
+        normalized_filters = self._filter(filters, SubscriptionHistoryFilter)
+        rows = self._subscriptions.list_subscription_history(
+            filters=normalized_filters,
+            page=normalized_page,
+        )
+        return self._to_page(SubscribeHistory, normalized_page, rows)
+
+    async def async_list_subscription_history(
+        self,
+        filters: SubscriptionHistoryFilter | dict[str, Any] | None = None,
+        page: QueryPageRequest | dict[str, Any] | None = None,
+    ) -> QueryPage[SubscribeHistory]:
+        """异步分页查询订阅完成历史。"""
+        return await self._async_run(
+            partial(self.list_subscription_history, filters, page)
+        )
+
+    def get_subscription_history(self, history_id: int) -> SubscribeHistory | None:
+        """同步按 ID 查询订阅完成历史。"""
+        return self._to_item(
+            SubscribeHistory,
+            self._subscriptions.get_subscription_history(history_id),
+        )
+
+    async def async_get_subscription_history(
+        self,
+        history_id: int,
+    ) -> SubscribeHistory | None:
+        """异步按 ID 查询订阅完成历史。"""
+        return await self._async_run(
+            partial(self.get_subscription_history, history_id)
+        )
+
+    def list_download_history(
+        self,
+        filters: DownloadHistoryFilter | dict[str, Any] | None = None,
+        page: QueryPageRequest | dict[str, Any] | None = None,
+    ) -> QueryPage[DownloadHistory]:
+        """同步分页查询下载历史。"""
+        normalized_page = self._page_request(page)
+        normalized_filters = self._filter(filters, DownloadHistoryFilter)
+        rows = self._histories.list_download_history(
+            filters=normalized_filters,
+            page=normalized_page,
+        )
+        return self._to_page(DownloadHistory, normalized_page, rows)
+
+    async def async_list_download_history(
+        self,
+        filters: DownloadHistoryFilter | dict[str, Any] | None = None,
+        page: QueryPageRequest | dict[str, Any] | None = None,
+    ) -> QueryPage[DownloadHistory]:
+        """异步分页查询下载历史。"""
+        return await self._async_run(
+            partial(self.list_download_history, filters, page)
+        )
+
+    def get_download_history(self, history_id: int) -> DownloadHistory | None:
+        """同步按 ID 查询下载历史。"""
+        return self._to_item(
+            DownloadHistory,
+            self._histories.get_download_history(history_id),
+        )
+
+    async def async_get_download_history(self, history_id: int) -> DownloadHistory | None:
+        """异步按 ID 查询下载历史。"""
+        return await self._async_run(
+            partial(self.get_download_history, history_id)
+        )
+
+    def list_transfer_history(
+        self,
+        filters: TransferHistoryFilter | dict[str, Any] | None = None,
+        page: QueryPageRequest | dict[str, Any] | None = None,
+    ) -> QueryPage[TransferHistory]:
+        """同步分页查询整理历史。"""
+        normalized_page = self._page_request(page)
+        normalized_filters = self._filter(filters, TransferHistoryFilter)
+        rows = self._histories.list_transfer_history(
+            filters=normalized_filters,
+            page=normalized_page,
+        )
+        return self._to_page(TransferHistory, normalized_page, rows)
+
+    async def async_list_transfer_history(
+        self,
+        filters: TransferHistoryFilter | dict[str, Any] | None = None,
+        page: QueryPageRequest | dict[str, Any] | None = None,
+    ) -> QueryPage[TransferHistory]:
+        """异步分页查询整理历史。"""
+        return await self._async_run(
+            partial(self.list_transfer_history, filters, page)
+        )
+
+    def get_transfer_history(self, history_id: int) -> TransferHistory | None:
+        """同步按 ID 查询整理历史。"""
+        return self._to_item(
+            TransferHistory,
+            self._histories.get_transfer_history(history_id),
+        )
+
+    async def async_get_transfer_history(self, history_id: int) -> TransferHistory | None:
+        """异步按 ID 查询整理历史。"""
+        return await self._async_run(
+            partial(self.get_transfer_history, history_id)
+        )
+
+
+_configured_data_query_service: DataQueryService | None = None
+
+
+def configure_data_query_service(service: DataQueryService) -> None:
+    """由启动组合根登记插件数据查询服务。"""
+    global _configured_data_query_service
+    _configured_data_query_service = service
+
+
+def get_configured_data_query_service() -> DataQueryService:
+    """返回启动阶段登记的插件数据查询服务。"""
+    if _configured_data_query_service is None:
+        raise RuntimeError("插件数据查询服务尚未配置")
+    return _configured_data_query_service
+
+
+__all__ = [
+    "DataQueryService",
+    "HistoryQueryPort",
+    "QueryRows",
+    "SubscriptionQueryPort",
+    "configure_data_query_service",
+    "get_configured_data_query_service",
+]
diff --git a/app/db/adapters/data_query.py b/app/db/adapters/data_query.py
new file mode 100644
index 000000000..41d3769bd
--- /dev/null
+++ b/app/db/adapters/data_query.py
@@ -0,0 +1,375 @@
+"""插件只读数据查询的 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.history import (
+    DownloadHistory as DownloadHistoryView,
+)
+from app.schemas.history import (
+    TransferHistory as TransferHistoryView,
+)
+from app.schemas.query import (
+    DownloadHistoryFilter,
+    QueryPageRequest,
+    QuerySortDirection,
+    QuerySortField,
+    SubscriptionFilter,
+    SubscriptionHistoryFilter,
+    TransferHistoryFilter,
+)
+from app.schemas.query import SubscribeHistory as SubscribeHistoryView
+from app.schemas.subscribe import Subscribe as SubscribeView
+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[SubscribeView]:
+        """按受控组合条件分页查询当前订阅。"""
+        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=SubscribeView,
+            conditions=conditions,
+            page=page,
+        )
+
+    def get_subscription(self, subscription_id: int) -> SubscribeView | None:
+        """按主键查询订阅并返回脱离 Session 的 DTO。"""
+        return self._get(
+            model=Subscribe,
+            view_model=SubscribeView,
+            record_id=subscription_id,
+        )
+
+    def list_subscription_history(
+        self,
+        *,
+        filters: SubscriptionHistoryFilter,
+        page: QueryPageRequest,
+    ) -> QueryRows[SubscribeHistoryView]:
+        """按受控组合条件分页查询订阅完成历史。"""
+        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=SubscribeHistoryView,
+            conditions=conditions,
+            page=page,
+        )
+
+    def get_subscription_history(
+        self,
+        history_id: int,
+    ) -> SubscribeHistoryView | None:
+        """按主键查询订阅完成历史并返回稳定 DTO。"""
+        return self._get(
+            model=SubscribeHistoryModel,
+            view_model=SubscribeHistoryView,
+            record_id=history_id,
+        )
+
+    def list_download_history(
+        self,
+        *,
+        filters: DownloadHistoryFilter,
+        page: QueryPageRequest,
+    ) -> QueryRows[DownloadHistoryView]:
+        """按受控组合条件分页查询下载历史。"""
+        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.path, query.path),
+        ):
+            if value:
+                conditions.append(_contains(column, value))
+        for column, value in (
+            (DownloadHistory.year, query.year),
+            (DownloadHistory.seasons, query.seasons),
+            (DownloadHistory.episodes, query.episodes),
+            (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 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=DownloadHistoryView,
+            conditions=conditions,
+            page=page,
+        )
+
+    def get_download_history(self, history_id: int) -> DownloadHistoryView | None:
+        """按主键查询下载历史并返回稳定 DTO。"""
+        return self._get(
+            model=DownloadHistory,
+            view_model=DownloadHistoryView,
+            record_id=history_id,
+        )
+
+    def list_transfer_history(
+        self,
+        *,
+        filters: TransferHistoryFilter,
+        page: QueryPageRequest,
+    ) -> QueryRows[TransferHistoryView]:
+        """按受控组合条件分页查询整理历史。"""
+        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(_contains(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=TransferHistoryView,
+            conditions=conditions,
+            page=page,
+        )
+
+    def get_transfer_history(self, history_id: int) -> TransferHistoryView | None:
+        """按主键查询整理历史并返回稳定 DTO。"""
+        return self._get(
+            model=TransferHistory,
+            view_model=TransferHistoryView,
+            record_id=history_id,
+        )
+
+
+__all__ = ["SqlAlchemyDataQueryAdapter"]
diff --git a/app/schemas/query.py b/app/schemas/query.py
new file mode 100644
index 000000000..684e366cf
--- /dev/null
+++ b/app/schemas/query.py
@@ -0,0 +1,215 @@
+"""插件只读数据查询使用的稳定筛选、分页与数据投影合同。"""
+
+from enum import Enum
+from typing import Generic, Optional, TypeVar
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from app.schemas.common import JsonData
+from app.schemas.history import DownloadHistory, TransferHistory
+from app.schemas.media import OptionalMediaIdentityMixin
+from app.schemas.subscribe import Subscribe
+from app.schemas.types import MediaSource, MediaType
+
+T = TypeVar("T")
+
+DEFAULT_QUERY_PAGE_SIZE = 50
+MAX_QUERY_PAGE_SIZE = 200
+
+
+class _QueryInput(BaseModel):
+    """查询输入共同的严格字段合同。"""
+
+    model_config = ConfigDict(extra="forbid")
+
+
+class QuerySortField(str, Enum):
+    """所有公开数据查询共同支持的稳定排序字段。"""
+
+    DATE = "date"
+    ID = "id"
+
+
+class QuerySortDirection(str, Enum):
+    """公开查询的排序方向。"""
+
+    ASC = "asc"
+    DESC = "desc"
+
+
+class QuerySort(_QueryInput):
+    """声明公开查询的稳定排序字段与方向。"""
+
+    field: QuerySortField = QuerySortField.DATE
+    direction: QuerySortDirection = QuerySortDirection.DESC
+
+
+class QueryPageRequest(_QueryInput):
+    """限制插件单次读取规模，并为跨页扫描提供稳定顺序。"""
+
+    page: int = Field(default=1, ge=1)
+    count: int = Field(
+        default=DEFAULT_QUERY_PAGE_SIZE,
+        ge=1,
+        le=MAX_QUERY_PAGE_SIZE,
+    )
+    sort: QuerySort = Field(default_factory=QuerySort)
+
+
+class QueryPage(BaseModel, Generic[T]):
+    """公开查询返回的分页 DTO。"""
+
+    items: list[T] = Field(default_factory=list)
+    total: int = Field(default=0, ge=0)
+    page: int = Field(default=1, ge=1)
+    count: int = Field(
+        default=DEFAULT_QUERY_PAGE_SIZE,
+        ge=1,
+        le=MAX_QUERY_PAGE_SIZE,
+    )
+
+    @property
+    def has_next(self) -> bool:
+        """返回当前分页后是否仍有记录。"""
+        return self.page * self.count < self.total
+
+
+class MediaIdentityQuery(OptionalMediaIdentityMixin, _QueryInput):
+    """允许省略身份，但显式筛选时要求来源与原生 ID 成对有效。"""
+
+    media_source: Optional[MediaSource] = None
+    media_id: Optional[str] = None
+
+
+class SubscribeHistory(OptionalMediaIdentityMixin, BaseModel):
+    """订阅完成历史的稳定只读投影。"""
+
+    id: int
+    name: Optional[str] = None
+    year: Optional[str] = None
+    type: Optional[str] = None
+    keyword: Optional[str] = None
+    media_source: Optional[MediaSource] = None
+    media_id: Optional[str] = None
+    music_type: Optional[str] = None
+    total_tracks: Optional[int] = None
+    season: Optional[int] = None
+    poster: Optional[str] = None
+    backdrop: Optional[str] = None
+    vote: Optional[float] = None
+    description: Optional[str] = None
+    filter: Optional[str] = None
+    include: Optional[str] = None
+    exclude: Optional[str] = None
+    quality: Optional[str] = None
+    resolution: Optional[str] = None
+    effect: Optional[str] = None
+    audio_quality: Optional[str] = None
+    audio_format: Optional[str] = None
+    min_bitrate: Optional[int] = None
+    min_bit_depth: Optional[int] = None
+    min_sample_rate: Optional[int] = None
+    total_episode: Optional[int] = None
+    start_episode: Optional[int] = None
+    date: Optional[str] = None
+    username: Optional[str] = None
+    sites: Optional[list[int]] = None
+    best_version: Optional[int] = None
+    best_version_full: Optional[int] = None
+    current_priority: Optional[int] = None
+    current_audio_format: Optional[str] = None
+    current_bitrate: Optional[int] = None
+    current_bit_depth: Optional[int] = None
+    current_sample_rate: Optional[int] = None
+    episode_priority: Optional[dict[str, int]] = None
+    save_path: Optional[str] = None
+    search_imdbid: Optional[int] = None
+    note: Optional[JsonData] = None
+    custom_words: Optional[str] = None
+    media_category: Optional[str] = None
+    filter_groups: Optional[list[str]] = None
+    episode_group: Optional[str] = None
+
+    model_config = ConfigDict(from_attributes=True)
+
+
+class SubscriptionFilter(MediaIdentityQuery):
+    """订阅查询允许组合的稳定业务字段。"""
+
+    ids: tuple[int, ...] = ()
+    names: tuple[str, ...] = ()
+    states: tuple[str, ...] = ()
+    usernames: tuple[str, ...] = ()
+    media_types: tuple[MediaType, ...] = ()
+    season: Optional[int] = None
+    episode_group: Optional[str] = None
+    music_type: Optional[str] = None
+
+
+class SubscriptionHistoryFilter(MediaIdentityQuery):
+    """订阅完成历史查询允许组合的稳定业务字段。"""
+
+    ids: tuple[int, ...] = ()
+    names: tuple[str, ...] = ()
+    usernames: tuple[str, ...] = ()
+    media_types: tuple[MediaType, ...] = ()
+    season: Optional[int] = None
+    episode_group: Optional[str] = None
+    music_type: Optional[str] = None
+
+
+class DownloadHistoryFilter(MediaIdentityQuery):
+    """下载历史查询允许组合的稳定业务字段。"""
+
+    ids: tuple[int, ...] = ()
+    media_types: tuple[MediaType, ...] = ()
+    title: Optional[str] = None
+    year: Optional[str] = None
+    seasons: Optional[str] = None
+    episodes: Optional[str] = None
+    path: Optional[str] = None
+    download_hash: Optional[str] = None
+    username: Optional[str] = None
+    usernames: tuple[str, ...] = ()
+    music_type: Optional[str] = None
+    episode_group: Optional[str] = None
+
+
+class TransferHistoryFilter(MediaIdentityQuery):
+    """整理历史查询允许组合的稳定业务字段。"""
+
+    ids: tuple[int, ...] = ()
+    media_types: tuple[MediaType, ...] = ()
+    media_sources: tuple[MediaSource, ...] = ()
+    require_media_identity: bool = False
+    title: Optional[str] = None
+    text: Optional[str] = None
+    year: Optional[str] = None
+    seasons: Optional[str] = None
+    episodes: Optional[str] = None
+    src: Optional[str] = None
+    dest: Optional[str] = None
+    status: Optional[bool] = None
+    download_hash: Optional[str] = None
+    music_type: Optional[str] = None
+    episode_group: Optional[str] = None
+
+
+__all__ = [
+    "DEFAULT_QUERY_PAGE_SIZE",
+    "MAX_QUERY_PAGE_SIZE",
+    "DownloadHistory",
+    "DownloadHistoryFilter",
+    "MediaIdentityQuery",
+    "QueryPage",
+    "QueryPageRequest",
+    "QuerySort",
+    "QuerySortDirection",
+    "QuerySortField",
+    "Subscribe",
+    "SubscribeHistory",
+    "SubscriptionFilter",
+    "SubscriptionHistoryFilter",
+    "TransferHistory",
+    "TransferHistoryFilter",
+]
diff --git a/app/sdk/queries.py b/app/sdk/queries.py
new file mode 100644
index 000000000..145a633f0
--- /dev/null
+++ b/app/sdk/queries.py
@@ -0,0 +1,173 @@
+"""插件可使用的订阅与历史只读查询门面。"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from app.schemas.query import (
+    DEFAULT_QUERY_PAGE_SIZE,
+    MAX_QUERY_PAGE_SIZE,
+    DownloadHistory,
+    DownloadHistoryFilter,
+    MediaIdentityQuery,
+    QueryPage,
+    QueryPageRequest,
+    QuerySort,
+    QuerySortDirection,
+    QuerySortField,
+    Subscribe,
+    SubscribeHistory,
+    SubscriptionFilter,
+    SubscriptionHistoryFilter,
+    TransferHistory,
+    TransferHistoryFilter,
+)
+
+
+def _service() -> Any:
+    """获取启动阶段登记的查询服务，避免把应用服务暴露为 SDK 合同。"""
+    from app.application.data_query import (
+        get_configured_data_query_service,
+    )
+
+    return get_configured_data_query_service()
+
+
+def list_subscriptions(
+    filters: SubscriptionFilter | dict[str, Any] | None = None,
+    page: QueryPageRequest | dict[str, Any] | None = None,
+) -> QueryPage[Subscribe]:
+    """同步分页读取订阅 DTO。"""
+    return _service().list_subscriptions(filters, page)
+
+
+async def async_list_subscriptions(
+    filters: SubscriptionFilter | dict[str, Any] | None = None,
+    page: QueryPageRequest | dict[str, Any] | None = None,
+) -> QueryPage[Subscribe]:
+    """异步分页读取订阅 DTO。"""
+    return await _service().async_list_subscriptions(filters, page)
+
+
+def get_subscription(subscription_id: int) -> Subscribe | None:
+    """同步按 ID 读取订阅 DTO。"""
+    return _service().get_subscription(subscription_id)
+
+
+async def async_get_subscription(subscription_id: int) -> Subscribe | None:
+    """异步按 ID 读取订阅 DTO。"""
+    return await _service().async_get_subscription(subscription_id)
+
+
+def list_subscription_history(
+    filters: SubscriptionHistoryFilter | dict[str, Any] | None = None,
+    page: QueryPageRequest | dict[str, Any] | None = None,
+) -> QueryPage[SubscribeHistory]:
+    """同步分页读取订阅完成历史 DTO。"""
+    return _service().list_subscription_history(filters, page)
+
+
+async def async_list_subscription_history(
+    filters: SubscriptionHistoryFilter | dict[str, Any] | None = None,
+    page: QueryPageRequest | dict[str, Any] | None = None,
+) -> QueryPage[SubscribeHistory]:
+    """异步分页读取订阅完成历史 DTO。"""
+    return await _service().async_list_subscription_history(filters, page)
+
+
+def get_subscription_history(history_id: int) -> SubscribeHistory | None:
+    """同步按 ID 读取订阅完成历史 DTO。"""
+    return _service().get_subscription_history(history_id)
+
+
+async def async_get_subscription_history(history_id: int) -> SubscribeHistory | None:
+    """异步按 ID 读取订阅完成历史 DTO。"""
+    return await _service().async_get_subscription_history(history_id)
+
+
+def list_download_history(
+    filters: DownloadHistoryFilter | dict[str, Any] | None = None,
+    page: QueryPageRequest | dict[str, Any] | None = None,
+) -> QueryPage[DownloadHistory]:
+    """同步分页读取下载历史 DTO。"""
+    return _service().list_download_history(filters, page)
+
+
+async def async_list_download_history(
+    filters: DownloadHistoryFilter | dict[str, Any] | None = None,
+    page: QueryPageRequest | dict[str, Any] | None = None,
+) -> QueryPage[DownloadHistory]:
+    """异步分页读取下载历史 DTO。"""
+    return await _service().async_list_download_history(filters, page)
+
+
+def get_download_history(history_id: int) -> DownloadHistory | None:
+    """同步按 ID 读取下载历史 DTO。"""
+    return _service().get_download_history(history_id)
+
+
+async def async_get_download_history(history_id: int) -> DownloadHistory | None:
+    """异步按 ID 读取下载历史 DTO。"""
+    return await _service().async_get_download_history(history_id)
+
+
+def list_transfer_history(
+    filters: TransferHistoryFilter | dict[str, Any] | None = None,
+    page: QueryPageRequest | dict[str, Any] | None = None,
+) -> QueryPage[TransferHistory]:
+    """同步分页读取整理历史 DTO。"""
+    return _service().list_transfer_history(filters, page)
+
+
+async def async_list_transfer_history(
+    filters: TransferHistoryFilter | dict[str, Any] | None = None,
+    page: QueryPageRequest | dict[str, Any] | None = None,
+) -> QueryPage[TransferHistory]:
+    """异步分页读取整理历史 DTO。"""
+    return await _service().async_list_transfer_history(filters, page)
+
+
+def get_transfer_history(history_id: int) -> TransferHistory | None:
+    """同步按 ID 读取整理历史 DTO。"""
+    return _service().get_transfer_history(history_id)
+
+
+async def async_get_transfer_history(history_id: int) -> TransferHistory | None:
+    """异步按 ID 读取整理历史 DTO。"""
+    return await _service().async_get_transfer_history(history_id)
+
+
+__all__ = [
+    "DEFAULT_QUERY_PAGE_SIZE",
+    "MAX_QUERY_PAGE_SIZE",
+    "DownloadHistory",
+    "DownloadHistoryFilter",
+    "MediaIdentityQuery",
+    "QueryPage",
+    "QueryPageRequest",
+    "QuerySort",
+    "QuerySortDirection",
+    "QuerySortField",
+    "Subscribe",
+    "SubscribeHistory",
+    "SubscriptionFilter",
+    "SubscriptionHistoryFilter",
+    "TransferHistory",
+    "TransferHistoryFilter",
+    "async_get_download_history",
+    "async_get_subscription",
+    "async_get_subscription_history",
+    "async_get_transfer_history",
+    "async_list_download_history",
+    "async_list_subscription_history",
+    "async_list_subscriptions",
+    "async_list_transfer_history",
+    "get_download_history",
+    "get_subscription",
+    "get_subscription_history",
+    "get_transfer_history",
+    "list_download_history",
+    "list_subscription_history",
+    "list_subscriptions",
+    "list_transfer_history",
+]
diff --git a/app/startup/initializers/modules.py b/app/startup/initializers/modules.py
index 9ebe1256b..65405992c 100644
--- a/app/startup/initializers/modules.py
+++ b/app/startup/initializers/modules.py
@@ -60,6 +60,10 @@ 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
@@ -105,6 +109,7 @@ 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.site import TransactionalSiteRepository
@@ -757,6 +762,14 @@ async def init_modules() -> HostRuntime:
         except Exception as cleanup_error:  # noqa: BLE001  保留原始启动异常
             logger.error(f"启动失败后的数据库任务清理失败：{cleanup_error}")
         raise
+    data_query_adapter = SqlAlchemyDataQueryAdapter(SessionFactory)
+    configure_data_query_service(
+        DataQueryService(
+            subscriptions=data_query_adapter,
+            histories=data_query_adapter,
+            async_executor=database_worker,
+        )
+    )
     configure_plugin_persistence(
         PluginPersistenceService(
             executor=database_worker,
diff --git a/tests/test_sdk_queries.py b/tests/test_sdk_queries.py
new file mode 100644
index 000000000..dd51f6757
--- /dev/null
+++ b/tests/test_sdk_queries.py
@@ -0,0 +1,705 @@
+"""统一插件只读查询 SDK 的真实 SQLite 合同测试。
+
+这些用例从真实 SQLAlchemy 表读取，再经查询服务和 ``app.sdk.queries`` 返回，
+因此同时约束筛选、分页、DTO 投影以及异步数据库执行边界。所有数据都由共享的
+隔离 SQLite harness 写入；测试不调用外部服务。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import inspect
+import threading
+from collections.abc import Callable
+from typing import Any
+
+import pytest
+
+from app.db.models.downloadhistory import DownloadHistory as DownloadHistoryModel
+from app.db.models.subscribe import Subscribe as SubscribeModel
+from app.db.models.subscribehistory import SubscribeHistory as SubscribeHistoryModel
+from app.db.models.transferhistory import TransferHistory as TransferHistoryModel
+from app.schemas.history import (
+    DownloadHistory as DownloadHistoryDTO,
+    TransferHistory as TransferHistoryDTO,
+)
+from app.schemas.query import (
+    DownloadHistoryFilter,
+    QueryPage,
+    QueryPageRequest,
+    QuerySort,
+    QuerySortDirection,
+    QuerySortField,
+    SubscribeHistory as SubscribeHistoryDTO,
+    SubscriptionFilter,
+    SubscriptionHistoryFilter,
+    TransferHistoryFilter,
+)
+from app.schemas.subscribe import Subscribe as SubscribeDTO
+from app.schemas.types import MediaSource, MediaType
+
+
+TMDB = MediaSource.TMDB.value
+
+
+class _RecordingExecutor:
+    """记录服务是否把异步查询提交到独立执行线程。"""
+
+    def __init__(self) -> None:
+        self.calls = 0
+        self.worker_thread_ids: list[int] = []
+
+    async def run(self, operation: Callable[[], Any]) -> Any:
+        """在线程中执行同步查询，模拟宿主数据库 worker 的调用合同。"""
+        self.calls += 1
+
+        def invoke() -> Any:
+            self.worker_thread_ids.append(threading.get_ident())
+            return operation()
+
+        return await asyncio.to_thread(invoke)
+
+
+@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.db.session import SessionFactory
+    from app.sdk import queries as sdk
+
+    adapter = SqlAlchemyDataQueryAdapter(SessionFactory)
+    executor = _RecordingExecutor()
+    service = DataQueryService(
+        subscriptions=adapter,
+        histories=adapter,
+        async_executor=executor,
+    )
+    monkeypatch.setattr(
+        data_query_module,
+        "_configured_data_query_service",
+        service,
+    )
+    return sdk, executor
+
+
+def _subscribe(
+    name: str,
+    *,
+    media_id: str,
+    state: str = "N",
+    username: str = "alice",
+    mtype: str = MediaType.TV.value,
+    season: int | None = 1,
+    episode_group: str | None = None,
+    date: str = "2026-08-27 10:00:00",
+    music_type: str | None = None,
+) -> SubscribeModel:
+    """构造隔离用例使用的订阅行。"""
+    return SubscribeModel(
+        name=name,
+        type=mtype,
+        state=state,
+        media_source=TMDB,
+        media_id=media_id,
+        season=season,
+        episode_group=episode_group,
+        username=username,
+        date=date,
+        music_type=music_type,
+    )
+
+
+def _subscribe_history(
+    name: str,
+    *,
+    media_id: str,
+    username: str = "alice",
+    mtype: str = MediaType.TV.value,
+    season: int | None = 1,
+    episode_group: str | None = None,
+    date: str = "2026-08-27 10:00:00",
+    music_type: str | None = None,
+) -> SubscribeHistoryModel:
+    """构造隔离用例使用的订阅完成历史行。"""
+    return SubscribeHistoryModel(
+        name=name,
+        type=mtype,
+        media_source=TMDB,
+        media_id=media_id,
+        season=season,
+        episode_group=episode_group,
+        username=username,
+        date=date,
+        music_type=music_type,
+    )
+
+
+def _download_history(
+    title: str,
+    *,
+    media_id: str,
+    path: str,
+    year: str = "2026",
+    seasons: str | None = "S01",
+    episodes: str | None = "E01",
+    username: str = "alice",
+    download_hash: str | None = "hash-1",
+    episode_group: str | None = None,
+    date: str = "2026-08-27 10:00:00",
+    mtype: str = MediaType.TV.value,
+    music_type: str | None = None,
+) -> DownloadHistoryModel:
+    """构造隔离用例使用的下载历史行。"""
+    return DownloadHistoryModel(
+        path=path,
+        type=mtype,
+        title=title,
+        year=year,
+        media_source=TMDB,
+        media_id=media_id,
+        music_type=music_type,
+        seasons=seasons,
+        episodes=episodes,
+        username=username,
+        download_hash=download_hash,
+        episode_group=episode_group,
+        date=date,
+    )
+
+
+def _transfer_history(
+    title: str,
+    *,
+    media_id: str | None,
+    src: str,
+    dest: str,
+    year: str = "2026",
+    seasons: str | None = "S01",
+    episodes: str | None = "E01",
+    download_hash: str | None = "hash-1",
+    episode_group: str | None = None,
+    status: bool = True,
+    date: str = "2026-08-27 10:00:00",
+    mtype: str = MediaType.TV.value,
+) -> TransferHistoryModel:
+    """构造隔离用例使用的整理历史行。"""
+    return TransferHistoryModel(
+        src=src,
+        src_storage="local",
+        dest=dest,
+        type=mtype,
+        title=title,
+        year=year,
+        media_source=TMDB if media_id is not None else None,
+        media_id=media_id,
+        seasons=seasons,
+        episodes=episodes,
+        download_hash=download_hash,
+        episode_group=episode_group,
+        status=status,
+        date=date,
+    )
+
+
+def _assert_projected_page(page: QueryPage, dto_type: type[Any], model_type: type[Any], row_id: int) -> None:
+    """断言分页结果是稳定 DTO，而不是 ORM 行或延迟加载对象。"""
+    assert isinstance(page, QueryPage)
+    assert page.total == 1
+    assert len(page.items) == 1
+    item = page.items[0]
+    assert isinstance(item, dto_type)
+    assert not isinstance(item, model_type)
+    assert item.id == row_id
+    assert item.media_source == MediaSource.TMDB
+
+
+def test_sdk_projects_subscription_and_three_history_domains_to_dtos(db, query_sdk):
+    """订阅、订阅完成历史、下载历史、整理历史均只返回 Pydantic 投影。"""
+    sdk, _executor = query_sdk
+    subscribe = db.add(_subscribe("订阅 DTO", media_id="dto-sub"))
+    subscribe_history = db.add(
+        _subscribe_history("订阅历史 DTO", media_id="dto-sub-history")
+    )
+    download_history = db.add(
+        _download_history("下载历史 DTO", media_id="dto-download", path="/dto/download")
+    )
+    transfer_history = db.add(
+        _transfer_history(
+            "整理历史 DTO",
+            media_id="dto-transfer",
+            src="/dto/src",
+            dest="/dto/dest",
+        )
+    )
+
+    _assert_projected_page(
+        sdk.list_subscriptions(
+            SubscriptionFilter(media_source=TMDB, media_id="dto-sub")
+        ),
+        SubscribeDTO,
+        SubscribeModel,
+        subscribe.id,
+    )
+    _assert_projected_page(
+        sdk.list_subscription_history(
+            SubscriptionHistoryFilter(
+                media_source=TMDB,
+                media_id="dto-sub-history",
+            )
+        ),
+        SubscribeHistoryDTO,
+        SubscribeHistoryModel,
+        subscribe_history.id,
+    )
+    _assert_projected_page(
+        sdk.list_download_history(
+            DownloadHistoryFilter(media_source=TMDB, media_id="dto-download")
+        ),
+        DownloadHistoryDTO,
+        DownloadHistoryModel,
+        download_history.id,
+    )
+    _assert_projected_page(
+        sdk.list_transfer_history(
+            TransferHistoryFilter(media_source=TMDB, media_id="dto-transfer")
+        ),
+        TransferHistoryDTO,
+        TransferHistoryModel,
+        transfer_history.id,
+    )
+
+
+def test_sdk_applies_combined_filters_in_each_query_domain(db, query_sdk):
+    """多个筛选条件必须按 AND 组合，不能只应用媒体身份或单个主条件。"""
+    sdk, _executor = query_sdk
+
+    db.add(
+        _subscribe(
+            "订阅命中",
+            media_id="combo-sub",
+            state="N",
+            username="alice",
+            season=2,
+            episode_group="eg-a",
+        ),
+        _subscribe(
+            "状态不符",
+            media_id="combo-sub",
+            state="R",
+            username="alice",
+            season=2,
+            episode_group="eg-a",
+        ),
+        _subscribe(
+            "用户不符",
+            media_id="combo-sub",
+            state="N",
+            username="bob",
+            season=2,
+            episode_group="eg-a",
+        ),
+        _subscribe(
+            "季不符",
+            media_id="combo-sub",
+            state="N",
+            username="alice",
+            season=1,
+            episode_group="eg-a",
+        ),
+        _subscribe(
+            "剧集组不符",
+            media_id="combo-sub",
+            state="N",
+            username="alice",
+            season=2,
+            episode_group="eg-b",
+        ),
+        _subscribe(
+            "类型不符",
+            media_id="combo-sub",
+            state="N",
+            username="alice",
+            mtype=MediaType.MOVIE.value,
+            season=2,
+            episode_group="eg-a",
+        ),
+    )
+    subscribe_page = sdk.list_subscriptions(
+        SubscriptionFilter(
+            media_source=TMDB,
+            media_id="combo-sub",
+            states=("N",),
+            usernames=("alice",),
+            media_types=(MediaType.TV,),
+            season=2,
+            episode_group="eg-a",
+        )
+    )
+    assert [item.name for item in subscribe_page.items] == ["订阅命中"]
+
+    db.add(
+        _subscribe_history(
+            "历史命中",
+            media_id="combo-sub-history",
+            username="alice",
+            season=2,
+            episode_group="eg-a",
+        ),
+        _subscribe_history(
+            "历史用户不符",
+            media_id="combo-sub-history",
+            username="bob",
+            season=2,
+            episode_group="eg-a",
+        ),
+        _subscribe_history(
+            "历史季不符",
+            media_id="combo-sub-history",
+            username="alice",
+            season=1,
+            episode_group="eg-a",
+        ),
+        _subscribe_history(
+            "历史类型不符",
+            media_id="combo-sub-history",
+            username="alice",
+            mtype=MediaType.MOVIE.value,
+            season=2,
+            episode_group="eg-a",
+        ),
+    )
+    history_page = sdk.list_subscription_history(
+        SubscriptionHistoryFilter(
+            media_source=TMDB,
+            media_id="combo-sub-history",
+            usernames=("alice",),
+            media_types=(MediaType.TV,),
+            season=2,
+            episode_group="eg-a",
+        )
+    )
+    assert [item.name for item in history_page.items] == ["历史命中"]
+
+    db.add(
+        _download_history(
+            "Combo Film",
+            media_id="combo-download",
+            path="/combo/match.mkv",
+            year="2026",
+            seasons="S02",
+            episodes="E03",
+            username="alice",
+            download_hash="combo-hash",
+            episode_group="eg-a",
+        ),
+        _download_history(
+            "Combo Film",
+            media_id="combo-download",
+            path="/combo/wrong-user.mkv",
+            year="2026",
+            seasons="S02",
+            episodes="E03",
+            username="bob",
+            download_hash="combo-hash",
+            episode_group="eg-a",
+        ),
+        _download_history(
+            "Combo Film",
+            media_id="combo-download",
+            path="/combo/wrong-episode.mkv",
+            year="2026",
+            seasons="S02",
+            episodes="E04",
+            username="alice",
+            download_hash="combo-hash",
+            episode_group="eg-a",
+        ),
+    )
+    download_page = sdk.list_download_history(
+        DownloadHistoryFilter(
+            media_source=TMDB,
+            media_id="combo-download",
+            media_types=(MediaType.TV,),
+            title="Combo",
+            year="2026",
+            seasons="S02",
+            episodes="E03",
+            path="match",
+            download_hash="combo-hash",
+            username="alice",
+            episode_group="eg-a",
+        )
+    )
+    assert [item.path for item in download_page.items] == ["/combo/match.mkv"]
+
+    db.add(
+        _transfer_history(
+            "Transfer Combo",
+            media_id="combo-transfer",
+            src="/combo/src-match.mkv",
+            dest="/combo/dest-match.mkv",
+            year="2026",
+            seasons="S02",
+            episodes="E03",
+            download_hash="transfer-hash",
+            episode_group="eg-a",
+        ),
+        _transfer_history(
+            "Transfer Combo",
+            media_id="combo-transfer",
+            src="/combo/src-failed.mkv",
+            dest="/combo/dest-failed.mkv",
+            year="2026",
+            seasons="S02",
+            episodes="E03",
+            download_hash="transfer-hash",
+            episode_group="eg-a",
+            status=False,
+        ),
+        _transfer_history(
+            "Transfer No Identity",
+            media_id=None,
+            src="/combo/src-no-id.mkv",
+            dest="/combo/dest-no-id.mkv",
+            year="2026",
+            seasons="S02",
+            episodes="E03",
+            download_hash="transfer-hash",
+            episode_group="eg-a",
+        ),
+    )
+    transfer_page = sdk.list_transfer_history(
+        TransferHistoryFilter(
+            media_types=(MediaType.TV,),
+            media_sources=(MediaSource.TMDB,),
+            require_media_identity=True,
+            title="Transfer",
+            text="dest-match",
+            year="2026",
+            seasons="S02",
+            episodes="E03",
+            src="/combo/src-match.mkv",
+            dest="/combo/dest-match.mkv",
+            status=True,
+            download_hash="transfer-hash",
+            episode_group="eg-a",
+        )
+    )
+    assert [item.src for item in transfer_page.items] == ["/combo/src-match.mkv"]
+
+
+def test_sdk_pagination_reports_total_and_stable_date_id_order(db, query_sdk):
+    """同日期记录按 ID 打破平局，跨页总数和结果顺序保持稳定。"""
+    sdk, _executor = query_sdk
+    rows = db.add(
+        _download_history(
+            "排序一",
+            media_id="sort-download",
+            path="/sort/one",
+            date="2026-08-27 10:00:00",
+            download_hash="sort-1",
+        ),
+        _download_history(
+            "排序二",
+            media_id="sort-download",
+            path="/sort/two",
+            date="2026-08-27 10:00:00",
+            download_hash="sort-2",
+        ),
+        _download_history(
+            "排序三",
+            media_id="sort-download",
+            path="/sort/three",
+            date="2026-08-26 10:00:00",
+            download_hash="sort-3",
+        ),
+    )
+    rows = list(rows)
+    expected_desc = sorted(rows, key=lambda row: (row.date, row.id), reverse=True)
+    filters = DownloadHistoryFilter(media_source=TMDB, media_id="sort-download")
+
+    page_one = sdk.list_download_history(
+        filters,
+        QueryPageRequest(page=1, count=2),
+    )
+    page_two = sdk.list_download_history(
+        filters,
+        QueryPageRequest(page=2, count=2),
+    )
+
+    assert page_one.total == 3
+    assert page_one.page == 1
+    assert page_one.count == 2
+    assert page_one.has_next is True
+    assert [item.id for item in page_one.items] == [
+        row.id for row in expected_desc[:2]
+    ]
+    assert page_two.total == 3
+    assert page_two.has_next is False
+    assert [item.id for item in page_two.items] == [expected_desc[2].id]
+
+    asc_page = sdk.list_download_history(
+        filters,
+        QueryPageRequest(
+            page=1,
+            count=3,
+            sort=QuerySort(
+                field=QuerySortField.DATE,
+                direction=QuerySortDirection.ASC,
+            ),
+        ),
+    )
+    expected_asc = sorted(rows, key=lambda row: (row.date, row.id))
+    assert [item.id for item in asc_page.items] == [
+        row.id for row in expected_asc
+    ]
+
+
+@pytest.mark.parametrize(
+    "payload",
+    [
+        {"media_source": TMDB},
+        {"media_id": "half-id"},
+        {"media_source": TMDB, "media_id": ""},
+        {"media_source": TMDB, "media_id": "   "},
+        {"media_source": TMDB, "media_id": "0"},
+        {"media_source": "not valid!", "media_id": "id"},
+    ],
+)
+def test_sdk_rejects_invalid_or_half_media_identity_fail_closed(query_sdk, payload):
+    """媒体身份不完整、空白、零值或未知来源不得退化成全表查询。"""
+    sdk, _executor = query_sdk
+
+    with pytest.raises(ValueError):
+        sdk.list_download_history(payload)
+
+
+def test_sdk_get_returns_none_for_missing_records(query_sdk):
+    """四个查询领域的按 ID 读取在未命中时统一返回 None。"""
+    sdk, _executor = query_sdk
+    missing_id = 2_147_483_647
+
+    assert sdk.get_subscription(missing_id) is None
+    assert sdk.get_subscription_history(missing_id) is None
+    assert sdk.get_download_history(missing_id) is None
+    assert sdk.get_transfer_history(missing_id) is None
+
+
+def test_sdk_sync_async_semantics_match_and_async_uses_executor(db, query_sdk):
+    """四个查询门面的异步结果与同步一致，并交给 executor 线程。"""
+    sdk, executor = query_sdk
+    subscribe = db.add(
+        _subscribe("订阅同步异步", media_id="async-subscription")
+    )
+    subscribe_history = db.add(
+        _subscribe_history(
+            "订阅历史同步异步",
+            media_id="async-subscription-history",
+        )
+    )
+    download = db.add(
+        _download_history(
+            "同步异步一致",
+            media_id="async-download",
+            path="/async/download",
+        )
+    )
+    transfer = db.add(
+        _transfer_history(
+            "整理历史同步异步",
+            media_id="async-transfer",
+            src="/async/src",
+            dest="/async/dest",
+        )
+    )
+    request = QueryPageRequest(page=1, count=10)
+    caller_thread_id = threading.get_ident()
+
+    cases = (
+        (
+            sdk.list_subscriptions,
+            sdk.async_list_subscriptions,
+            SubscriptionFilter(
+                media_source=TMDB,
+                media_id="async-subscription",
+            ),
+            subscribe.id,
+        ),
+        (
+            sdk.list_subscription_history,
+            sdk.async_list_subscription_history,
+            SubscriptionHistoryFilter(
+                media_source=TMDB,
+                media_id="async-subscription-history",
+            ),
+            subscribe_history.id,
+        ),
+        (
+            sdk.list_download_history,
+            sdk.async_list_download_history,
+            DownloadHistoryFilter(
+                media_source=TMDB,
+                media_id="async-download",
+            ),
+            download.id,
+        ),
+        (
+            sdk.list_transfer_history,
+            sdk.async_list_transfer_history,
+            TransferHistoryFilter(
+                media_source=TMDB,
+                media_id="async-transfer",
+            ),
+            transfer.id,
+        ),
+    )
+    for sync_call, async_call, filters, expected_id in cases:
+        sync_page = sync_call(filters, request)
+        calls_before = executor.calls
+        async_page = asyncio.run(async_call(filters, request))
+
+        assert async_page == sync_page
+        assert [item.id for item in async_page.items] == [expected_id]
+        assert executor.calls == calls_before + 1
+
+    assert executor.worker_thread_ids
+    assert all(thread_id != caller_thread_id for thread_id in executor.worker_thread_ids)
+
+
+def test_sdk_public_exports_do_not_leak_persistence_implementation(query_sdk):
+    """SDK 的机器可读公开合同只允许 DTO、筛选模型和查询函数。"""
+    sdk, _executor = query_sdk
+    forbidden_tokens = ("session", "oper", "provider", "configure")
+    forbidden_modules = (
+        "app.db.models",
+        "app.db.oper",
+        "app.db.session",
+        "app.application",
+    )
+
+    assert sdk.__all__
+    for name in sdk.__all__:
+        lowered = name.casefold()
+        assert not any(token in lowered for token in forbidden_tokens), name
+        exported = getattr(sdk, name)
+        module_name = getattr(exported, "__module__", "")
+        assert not module_name.startswith(forbidden_modules), (
+            name,
+            module_name,
+        )
+        if inspect.isfunction(exported):
+            assert module_name == "app.sdk.queries"
+
+    assert not hasattr(sdk, "TransactionalDataQueryRepository")
+    assert not hasattr(sdk, "DataQueryService")
+    assert not hasattr(sdk, "get_configured_data_query_service")
+
+
+def test_query_tests_use_local_sqlite_backend(query_sdk):
+    """查询 focused tests 固定在隔离 SQLite 上，不依赖外部数据库或网络。"""
+    from app.db.engine import get_engine
+
+    engine = get_engine()
+    assert engine.url.get_backend_name() == "sqlite"