diff --git a/backend/app/api/v1/base/business_roles.py b/backend/app/api/v1/base/business_roles.py
index 6244ad7..ce3d0df 100644
--- a/backend/app/api/v1/base/business_roles.py
+++ b/backend/app/api/v1/base/business_roles.py
@@ -12,7 +12,8 @@ from fastapi import APIRouter, Depends
 from app.api.deps import DbSession, get_current_user
 from app.core.response import ApiResponse, ok
 from app.models import User
-from app.repositories import business_role_repository
+from app.schemas.business_role import BusinessRoleOptionOut
+from app.services import business_role_service
 
 router = APIRouter(prefix="/business-roles", tags=["人员管理"])
 
@@ -20,21 +21,14 @@ router = APIRouter(prefix="/business-roles", tags=["人员管理"])
 CurrentUser = Annotated[User, Depends(get_current_user)]
 
 
-@router.get("/options", response_model=ApiResponse[list[dict]], summary="岗位下拉选项")
-def list_options(session: DbSession, _: CurrentUser) -> ApiResponse[list[dict]]:
+@router.get(
+    "/options",
+    response_model=ApiResponse[list[BusinessRoleOptionOut]],
+    summary="岗位下拉选项",
+)
+def list_options(session: DbSession, _: CurrentUser) -> ApiResponse[list[BusinessRoleOptionOut]]:
     """岗位选项（登录即可；含停用——人员档案可能仍引用停用岗位）。
 
-    岗位规模有限（内置 + 自定义），一次取足够上限，不做分页。
+    岗位规模有限（内置 + 自定义），一次取全量，不做分页。
     """
-    roles = business_role_repository.list_roles(session, offset=0, limit=1000)
-    return ok(
-        [
-            {
-                "id": role.id,
-                "role_code": role.role_code,
-                "role_name": role.role_name,
-                "status": role.status,
-            }
-            for role in roles
-        ]
-    )
+    return ok(business_role_service.list_options(session))
diff --git a/backend/app/api/v1/dashboard.py b/backend/app/api/v1/dashboard.py
index b0f492a..70afe93 100644
--- a/backend/app/api/v1/dashboard.py
+++ b/backend/app/api/v1/dashboard.py
@@ -19,11 +19,7 @@ router = APIRouter(prefix="/dashboard", tags=["工作台"])
 
 _DashboardUser = Annotated[
     User,
-    Depends(
-        require_roles(
-            ROLE_ADMIN, ROLE_FINANCE, ROLE_LEADER, message="无权访问工作台"
-        )
-    ),
+    Depends(require_roles(ROLE_ADMIN, ROLE_FINANCE, ROLE_LEADER, message="无权访问工作台")),
 ]
 
 
@@ -39,6 +35,4 @@ def get_dashboard(
 
     待办按当前用户角色只返回其可操作节点（admin 可查看全部在途单）。
     """
-    return ok(
-        dashboard_service.get_dashboard(session, month, current_user=current_user)
-    )
+    return ok(dashboard_service.get_dashboard(session, month, current_user=current_user))
diff --git a/backend/app/models/business.py b/backend/app/models/business.py
index 18e9947..72fa6f2 100644
--- a/backend/app/models/business.py
+++ b/backend/app/models/business.py
@@ -79,9 +79,7 @@ class SecondaryAllocation(Base, TimestampMixin):
         Index("idx_business_secondary_allocations_business_id", "business_id"),
         Index("idx_business_secondary_allocations_personnel_id", "personnel_id"),
         # 取当前生效行的唯一查询形态：WHERE business_id=? AND is_active=1
-        Index(
-            "idx_business_secondary_allocations_business_active", "business_id", "is_active"
-        ),
+        Index("idx_business_secondary_allocations_business_active", "business_id", "is_active"),
     )
 
     id: Mapped[int] = mapped_column(
diff --git a/backend/app/repositories/business_repository.py b/backend/app/repositories/business_repository.py
index 9fa9456..ef75415 100644
--- a/backend/app/repositories/business_repository.py
+++ b/backend/app/repositories/business_repository.py
@@ -5,8 +5,6 @@
 models/customer.py 定义）。
 """
 
-from datetime import date
-
 from sqlalchemy import Row, Select, delete, exists, func, or_, select, update
 from sqlalchemy.orm import Session, aliased
 
@@ -23,6 +21,8 @@ from app.models import (
 )
 from app.models.customer import OWNER_TYPE_BUSINESS
 from app.repositories import ownership_log_repository
+from app.repositories.ownership_log_repository import OwnershipLogRef
+from app.schemas.business import BusinessQuery
 
 
 def get_by_id(session: Session, business_id: int, *, for_update: bool = False) -> Business | None:
@@ -77,50 +77,32 @@ def get_with_names(
     return session.execute(stmt).first()
 
 
-def _apply_filters(
-    stmt: Select,
-    *,
-    keyword: str | None,
-    status: int | None,
-    customer_id: int | None,
-    owner_personnel_id: int | None,
-    date_from: date | None,
-    date_to: date | None,
-) -> Select:
+def _apply_filters(stmt: Select, query: BusinessQuery) -> Select:
     """业务列表的筛选条件（调用方须已 join Customer）。"""
-    if keyword:
+    if query.keyword:
         # 业务编号/公司名称双列 OR 模糊；contains(autoescape) 同 query_utils.fuzzy
         # 的转义口径（fuzzy 为单列 where 语义，双列 OR 在此直接组合）
         stmt = stmt.where(
             or_(
-                Business.business_no.contains(keyword, autoescape=True),
-                Customer.company_name.contains(keyword, autoescape=True),
+                Business.business_no.contains(query.keyword, autoescape=True),
+                Customer.company_name.contains(query.keyword, autoescape=True),
             )
         )
-    if status is not None:
-        stmt = stmt.where(Business.status == status)
-    if customer_id is not None:
-        stmt = stmt.where(Business.customer_id == customer_id)
-    if owner_personnel_id is not None:
-        stmt = stmt.where(Business.owner_personnel_id == owner_personnel_id)
-    if date_from is not None:
-        stmt = stmt.where(Business.business_date >= date_from)
-    if date_to is not None:
-        stmt = stmt.where(Business.business_date <= date_to)
+    if query.status is not None:
+        stmt = stmt.where(Business.status == query.status)
+    if query.customer_id is not None:
+        stmt = stmt.where(Business.customer_id == query.customer_id)
+    if query.owner_personnel_id is not None:
+        stmt = stmt.where(Business.owner_personnel_id == query.owner_personnel_id)
+    if query.date_from is not None:
+        stmt = stmt.where(Business.business_date >= query.date_from)
+    if query.date_to is not None:
+        stmt = stmt.where(Business.business_date <= query.date_to)
     return stmt
 
 
 def list_businesses(
-    session: Session,
-    *,
-    offset: int,
-    limit: int,
-    keyword: str | None,
-    status: int | None,
-    customer_id: int | None,
-    owner_personnel_id: int | None,
-    date_from: date | None,
-    date_to: date | None,
+    session: Session, query: BusinessQuery
 ) -> list[tuple[Business, str, str | None, str | None]]:
     """分页查询业务（附客户名/负责人名/协助人名），按 id 倒序。"""
     owner = aliased(Personnel)
@@ -128,49 +110,24 @@ def list_businesses(
     stmt = select(
         Business, Customer.company_name, owner.personnel_name, assistant.personnel_name
     ).join(Customer, Customer.id == Business.customer_id)
-    stmt = _apply_filters(
-        stmt,
-        keyword=keyword,
-        status=status,
-        customer_id=customer_id,
-        owner_personnel_id=owner_personnel_id,
-        date_from=date_from,
-        date_to=date_to,
-    )
+    stmt = _apply_filters(stmt, query)
     stmt = (
         stmt.outerjoin(owner, owner.id == Business.owner_personnel_id)
         .outerjoin(assistant, assistant.id == Business.assistant_personnel_id)
         .order_by(Business.id.desc())
-        .offset(offset)
-        .limit(limit)
+        .offset(query.offset)
+        .limit(query.page_size)
     )
     return list(session.execute(stmt).all())
 
 
-def count_businesses(
-    session: Session,
-    *,
-    keyword: str | None,
-    status: int | None,
-    customer_id: int | None,
-    owner_personnel_id: int | None,
-    date_from: date | None,
-    date_to: date | None,
-) -> int:
+def count_businesses(session: Session, query: BusinessQuery) -> int:
     stmt = (
         select(func.count())
         .select_from(Business)
         .join(Customer, Customer.id == Business.customer_id)
     )
-    stmt = _apply_filters(
-        stmt,
-        keyword=keyword,
-        status=status,
-        customer_id=customer_id,
-        owner_personnel_id=owner_personnel_id,
-        date_from=date_from,
-        date_to=date_to,
-    )
+    stmt = _apply_filters(stmt, query)
     return int(session.execute(stmt).scalar_one())
 
 
@@ -313,18 +270,17 @@ def list_ownership_logs(
 ) -> list[Row]:
     """某业务的归属变更留痕（共享实现，owner_type/business_id 参数化）。"""
     return ownership_log_repository.list_by_ref(
-        session,
-        owner_type=OWNER_TYPE_BUSINESS,
-        ref_column=OwnershipChangeLog.business_id,
-        ref_id=business_id,
-        offset=offset,
-        limit=limit,
+        session, _business_log_ref(business_id), offset=offset, limit=limit
     )
 
 
 def count_ownership_logs(session: Session, business_id: int) -> int:
-    return ownership_log_repository.count_by_ref(
-        session,
+    return ownership_log_repository.count_by_ref(session, _business_log_ref(business_id))
+
+
+def _business_log_ref(business_id: int) -> OwnershipLogRef:
+    """业务维度的归属留痕锚点。"""
+    return OwnershipLogRef(
         owner_type=OWNER_TYPE_BUSINESS,
         ref_column=OwnershipChangeLog.business_id,
         ref_id=business_id,
diff --git a/backend/app/repositories/business_role_repository.py b/backend/app/repositories/business_role_repository.py
index a45f7af..92a0984 100644
--- a/backend/app/repositories/business_role_repository.py
+++ b/backend/app/repositories/business_role_repository.py
@@ -7,6 +7,7 @@ from sqlalchemy import Select, func, or_, select
 from sqlalchemy.orm import Session
 
 from app.models import BusinessRole
+from app.schemas.business_role import BusinessRoleQuery
 
 # ==================== 读取 ====================
 
@@ -39,41 +40,48 @@ def list_by_codes(session: Session, role_codes: list[str]) -> dict[str, Business
     return {role.role_code: role for role in rows}
 
 
-def list_roles(
-    session: Session,
-    *,
-    offset: int,
-    limit: int,
-    keyword: str | None = None,
-    status: int | None = None,
-) -> list[BusinessRole]:
+def list_roles(session: Session, query: BusinessRoleQuery) -> list[BusinessRole]:
     """分页查询岗位（含停用）。排序 (sort_order, id)，与系统角色口径一致。"""
-    stmt = _apply_filters(select(BusinessRole), keyword=keyword, status=status)
-    stmt = stmt.order_by(BusinessRole.sort_order, BusinessRole.id).offset(offset).limit(limit)
+    stmt = _apply_filters(select(BusinessRole), query)
+    stmt = (
+        stmt.order_by(BusinessRole.sort_order, BusinessRole.id)
+        .offset(query.offset)
+        .limit(query.page_size)
+    )
     return list(session.execute(stmt).scalars())
 
 
-def count_roles(session: Session, *, keyword: str | None = None, status: int | None = None) -> int:
+def count_roles(session: Session, query: BusinessRoleQuery) -> int:
     """统计符合条件的岗位总数，供分页使用。"""
-    stmt = _apply_filters(select(func.count(BusinessRole.id)), keyword=keyword, status=status)
+    stmt = _apply_filters(select(func.count(BusinessRole.id)), query)
     return int(session.execute(stmt).scalar_one())
 
 
-def _apply_filters(stmt: Select, *, keyword: str | None, status: int | None) -> Select:
+def list_all(session: Session) -> list[BusinessRole]:
+    """全部岗位（含停用，不分页）——岗位下拉选项的取数口径。
+
+    岗位规模有限（内置 + 自定义），一次取全量，不受分页上限约束；
+    排序与 list_roles 一致（sort_order, id），下拉与列表相对次序相同。
+    """
+    stmt = select(BusinessRole).order_by(BusinessRole.sort_order, BusinessRole.id)
+    return list(session.execute(stmt).scalars())
+
+
+def _apply_filters(stmt: Select, query: BusinessRoleQuery) -> Select:
     """应用通用筛选条件。列表与计数共用，避免两处口径不一致。
 
     关键词走 ``contains(autoescape=True)``（转义 LIKE 通配符）——岗位编码含下划线，
     不转义时 ``_`` 会当单字符通配执行。
     """
-    if keyword:
+    if query.keyword:
         stmt = stmt.where(
             or_(
-                BusinessRole.role_code.contains(keyword, autoescape=True),
-                BusinessRole.role_name.contains(keyword, autoescape=True),
+                BusinessRole.role_code.contains(query.keyword, autoescape=True),
+                BusinessRole.role_name.contains(query.keyword, autoescape=True),
             )
         )
-    if status is not None:
-        stmt = stmt.where(BusinessRole.status == status)
+    if query.status is not None:
+        stmt = stmt.where(BusinessRole.status == query.status)
     return stmt
 
 
diff --git a/backend/app/repositories/customer_repository.py b/backend/app/repositories/customer_repository.py
index 9e10511..498a94e 100644
--- a/backend/app/repositories/customer_repository.py
+++ b/backend/app/repositories/customer_repository.py
@@ -9,7 +9,9 @@ from sqlalchemy.orm import Session
 from app.models import Customer, OwnershipChangeLog, Personnel
 from app.models.customer import OWNER_TYPE_CUSTOMER
 from app.repositories import ownership_log_repository
+from app.repositories.ownership_log_repository import OwnershipLogRef
 from app.repositories.query_utils import fuzzy
+from app.schemas.customer import CustomerQuery
 
 # ==================== 读取 ====================
 
@@ -45,49 +47,25 @@ def get_by_company_owner(session: Session, company_name: str, owner_name: str) -
     return session.execute(stmt).scalar_one_or_none()
 
 
-def list_customers(
-    session: Session,
-    *,
-    offset: int,
-    limit: int,
-    keyword: str | None = None,
-    customer_type: str | None = None,
-    owner_personnel_id: int | None = None,
-) -> list[Row]:
+def list_customers(session: Session, query: CustomerQuery) -> list[Row]:
     """分页查询客户，行内容为 (客户, 归属人员姓名)。
 
     归属人员外键非空但统一用 outerjoin：人员行不可考（被物理删除的脏数据）
     不应让客户行整个查不出来（同 personnel_repository 的韧性口径）。
     排序按 id 倒序：客户无排序字段，新增的靠前符合列表页习惯。
     """
-    stmt = _apply_filters(
-        _base_select(),
-        keyword=keyword,
-        customer_type=customer_type,
-        owner_personnel_id=owner_personnel_id,
-    )
-    stmt = stmt.order_by(Customer.id.desc()).offset(offset).limit(limit)
+    stmt = _apply_filters(_base_select(), query)
+    stmt = stmt.order_by(Customer.id.desc()).offset(query.offset).limit(query.page_size)
     return list(session.execute(stmt).all())
 
 
-def count_customers(
-    session: Session,
-    *,
-    keyword: str | None = None,
-    customer_type: str | None = None,
-    owner_personnel_id: int | None = None,
-) -> int:
+def count_customers(session: Session, query: CustomerQuery) -> int:
     """统计符合条件的客户总数，供分页使用。
 
     筛选条件全部落在主表上，计数无需 join（列表与计数的筛选口径由
     ``_apply_filters`` 单点保证）。
     """
-    stmt = _apply_filters(
-        select(func.count(Customer.id)),
-        keyword=keyword,
-        customer_type=customer_type,
-        owner_personnel_id=owner_personnel_id,
-    )
+    stmt = _apply_filters(select(func.count(Customer.id)), query)
     return int(session.execute(stmt).scalar_one())
 
 
@@ -101,9 +79,7 @@ def list_ownership_logs(
     """查某客户的归属变更留痕（共享实现，owner_type/customer_id 参数化）。"""
     return ownership_log_repository.list_by_ref(
         session,
-        owner_type=OWNER_TYPE_CUSTOMER,
-        ref_column=OwnershipChangeLog.customer_id,
-        ref_id=customer_id,
+        _customer_log_ref(customer_id),
         offset=offset,
         limit=limit,
     )
@@ -111,8 +87,12 @@ def list_ownership_logs(
 
 def count_ownership_logs(session: Session, *, customer_id: int) -> int:
     """统计某客户的归属留痕总数，供分页使用。"""
-    return ownership_log_repository.count_by_ref(
-        session,
+    return ownership_log_repository.count_by_ref(session, _customer_log_ref(customer_id))
+
+
+def _customer_log_ref(customer_id: int) -> OwnershipLogRef:
+    """客户维度的归属留痕锚点。"""
+    return OwnershipLogRef(
         owner_type=OWNER_TYPE_CUSTOMER,
         ref_column=OwnershipChangeLog.customer_id,
         ref_id=customer_id,
@@ -126,20 +106,14 @@ def _base_select() -> Select:
     )
 
 
-def _apply_filters(
-    stmt: Select,
-    *,
-    keyword: str | None,
-    customer_type: str | None,
-    owner_personnel_id: int | None,
-) -> Select:
+def _apply_filters(stmt: Select, query: CustomerQuery) -> Select:
     """应用通用筛选条件。列表与计数共用，避免两处口径不一致。"""
-    if keyword:
-        stmt = stmt.where(_keyword_clause(keyword))
-    if customer_type:
-        stmt = stmt.where(Customer.customer_type == customer_type)
-    if owner_personnel_id is not None:
-        stmt = stmt.where(Customer.owner_personnel_id == owner_personnel_id)
+    if query.keyword:
+        stmt = stmt.where(_keyword_clause(query.keyword))
+    if query.customer_type:
+        stmt = stmt.where(Customer.customer_type == query.customer_type)
+    if query.owner_personnel_id is not None:
+        stmt = stmt.where(Customer.owner_personnel_id == query.owner_personnel_id)
     return stmt
 
 
diff --git a/backend/app/repositories/dashboard_repository.py b/backend/app/repositories/dashboard_repository.py
index 4c0e46c..931734c 100644
--- a/backend/app/repositories/dashboard_repository.py
+++ b/backend/app/repositories/dashboard_repository.py
@@ -34,20 +34,25 @@ _PENDING_STATUSES = (
 
 # ==== 经营绩效总览 ====
 
+
 def sum_month_performance(session: Session, month: str) -> dict:
     """当月单笔生效行合计与人员数（拆行求和不失真，人数需 DISTINCT）。"""
-    return session.execute(
-        select(
-            func.coalesce(func.sum(PerformanceLedger.performance_amount), 0).label(
-                "performance_total"
-            ),
-            func.coalesce(func.sum(PerformanceLedger.payable_amount), 0).label("payable_total"),
-            func.count(func.distinct(PerformanceLedger.personnel_id)).label("person_count"),
-        ).where(
-            PerformanceLedger.settlement_month == month,
-            *_ACTIVE_SINGLE,
+    return (
+        session.execute(
+            select(
+                func.coalesce(func.sum(PerformanceLedger.performance_amount), 0).label(
+                    "performance_total"
+                ),
+                func.coalesce(func.sum(PerformanceLedger.payable_amount), 0).label("payable_total"),
+                func.count(func.distinct(PerformanceLedger.personnel_id)).label("person_count"),
+            ).where(
+                PerformanceLedger.settlement_month == month,
+                *_ACTIVE_SINGLE,
+            )
         )
-    ).mappings().one()
+        .mappings()
+        .one()
+    )
 
 
 def performance_trend(session: Session, months: list[str]) -> dict[str, object]:
@@ -66,9 +71,7 @@ def performance_trend(session: Session, months: list[str]) -> dict[str, object]:
     return dict(rows)
 
 
-def receipt_trend(
-    session: Session, *, start_date: date, end_date: date
-) -> dict[str, object]:
+def receipt_trend(session: Session, *, start_date: date, end_date: date) -> dict[str, object]:
     """近 N 个月回款合计（按到账月，含未核算回款——与台账口径不同，页面注明）。
 
     过滤走 ``receipt_date`` 原始列范围（可用索引）；``DATE_FORMAT`` 只用于
@@ -107,6 +110,7 @@ def get_month_settlement(session: Session, month: str) -> MonthlySettlement | No
 
 # ==== 待办事项 ====
 
+
 def list_pending_approvals(
     session: Session, *, statuses: tuple[int, ...], limit: int
 ) -> list[ApprovalOrder]:
@@ -160,15 +164,14 @@ def list_unassigned_receipts(session: Session, *, limit: int) -> list:
 def count_unassigned_receipts(session: Session) -> int:
     return int(
         session.execute(
-            select(func.count())
-            .select_from(Receipt)
-            .where(Receipt.shipment_id.is_(None))
+            select(func.count()).select_from(Receipt).where(Receipt.shipment_id.is_(None))
         ).scalar_one()
     )
 
 
 # ==== 风险预警 ====
 
+
 def list_overdue_shipments(session: Session, *, cutoff: date, limit: int) -> list:
     """发货后超过阈值且该批次无任何回款的发货单（回款逾期规则）。
 
diff --git a/backend/app/repositories/login_log_repository.py b/backend/app/repositories/login_log_repository.py
index 122e9e6..07ad4d5 100644
--- a/backend/app/repositories/login_log_repository.py
+++ b/backend/app/repositories/login_log_repository.py
@@ -1,5 +1,7 @@
 """登录日志数据访问。"""
 
+from dataclasses import dataclass
+
 from sqlalchemy import Select, func, select
 from sqlalchemy.orm import Session
 
@@ -10,68 +12,73 @@ from app.repositories.query_utils import LogFilters, apply_log_filters
 _USER_AGENT_MAX_LEN = 255
 
 
-def add(
-    session: Session,
-    *,
-    user_id: int | None,
-    username: str,
-    ip: str | None,
-    user_agent: str | None,
-    success: bool,
-    fail_reason: str | None = None,
-) -> None:
+@dataclass(frozen=True, slots=True)
+class LoginLogEntry:
+    """一条待写入的登录日志（全字段 DTO）。
+
+    与 ``OperationLogEntry`` 同款做法：写入字段平铺传递时，任何一处漏传
+    都只表现为日志字段缺失、不报错（success 默认 0 的语义尤其隐蔽）。
+    """
+
+    user_id: int | None
+    username: str
+    success: bool
+    ip: str | None = None
+    user_agent: str | None = None
+    fail_reason: str | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class LoginLogListQuery:
+    """登录日志列表的查询条件（筛选 + 分页）。"""
+
+    filters: LogFilters
+    offset: int = 0
+    limit: int = 20
+    ip: str | None = None
+
+
+def add(session: Session, entry: LoginLogEntry) -> None:
     """写入一条登录日志。只 flush 不 commit。
 
     **绝不记录密码** —— 明文与哈希都不写。日志是最容易泄漏凭据的地方。
 
-    :param user_id: 账号不存在时传 None（列本身可空）
-    :param username: 用户名原样记录，便于账号不存在时也能追溯
+    :param entry.user_id: 账号不存在时为 None（列本身可空）
+    :param entry.username: 用户名原样记录，便于账号不存在时也能追溯
     """
     session.add(
         LoginLog(
-            user_id=user_id,
-            username=username,
-            ip=ip,
-            user_agent=user_agent[:_USER_AGENT_MAX_LEN] if user_agent else None,
-            success=1 if success else 0,
-            fail_reason=fail_reason,
+            user_id=entry.user_id,
+            username=entry.username,
+            ip=entry.ip,
+            user_agent=entry.user_agent[:_USER_AGENT_MAX_LEN] if entry.user_agent else None,
+            success=1 if entry.success else 0,
+            fail_reason=entry.fail_reason,
         )
     )
     session.flush()
 
 
-def list_logs(
-    session: Session,
-    *,
-    offset: int,
-    limit: int,
-    filters: LogFilters,
-    ip: str | None = None,
-) -> list[LoginLog]:
+def list_logs(session: Session, query: LoginLogListQuery) -> list[LoginLog]:
     """分页查询登录日志。
 
     **按 id 倒序而非 create_time**：后者是秒级精度，同一秒内的多条记录排序
     不确定，翻页会重复或遗漏（与用户列表同一理由）。
     """
-    stmt = _apply_filters(select(LoginLog), filters=filters, ip=ip)
-    stmt = stmt.order_by(LoginLog.id.desc()).offset(offset).limit(limit)
+    stmt = _apply_filters(select(LoginLog), query)
+    stmt = stmt.order_by(LoginLog.id.desc()).offset(query.offset).limit(query.limit)
     return list(session.execute(stmt).scalars())
 
 
-def count_logs(
-    session: Session,
-    *,
-    filters: LogFilters,
-    ip: str | None = None,
-) -> int:
+def count_logs(session: Session, query: LoginLogListQuery) -> int:
     """统计符合条件的登录日志总数，供分页使用。"""
-    stmt = _apply_filters(select(func.count(LoginLog.id)), filters=filters, ip=ip)
+    stmt = _apply_filters(select(func.count(LoginLog.id)), query)
     return int(session.execute(stmt).scalar_one())
 
 
-def _apply_filters(stmt: Select, *, filters: LogFilters, ip: str | None) -> Select:
+def _apply_filters(stmt: Select, query: LoginLogListQuery) -> Select:
     """应用筛选条件（公共部分复用 query_utils，与操作日志同口径）。"""
-    stmt = apply_log_filters(stmt, LoginLog, filters)
-    if ip:
-        stmt = stmt.where(LoginLog.ip == ip)
+    stmt = apply_log_filters(stmt, LoginLog, query.filters)
+    if query.ip:
+        stmt = stmt.where(LoginLog.ip == query.ip)
     return stmt
diff --git a/backend/app/repositories/operation_log_repository.py b/backend/app/repositories/operation_log_repository.py
index 20b74c3..0aba518 100644
--- a/backend/app/repositories/operation_log_repository.py
+++ b/backend/app/repositories/operation_log_repository.py
@@ -107,15 +107,18 @@ def get_by_id(session: Session, log_id: int) -> OperationLog | None:
     return session.get(OperationLog, log_id)
 
 
-def list_logs(
-    session: Session,
-    *,
-    offset: int,
-    limit: int,
-    filters: LogFilters,
-    module: str | None = None,
-    action: str | None = None,
-) -> list[OperationLog]:
+@dataclass(frozen=True, slots=True)
+class OperationLogListQuery:
+    """操作日志列表的查询条件（筛选 + 分页）。"""
+
+    filters: LogFilters
+    offset: int = 0
+    limit: int = 20
+    module: str | None = None
+    action: str | None = None
+
+
+def list_logs(session: Session, query: OperationLogListQuery) -> list[OperationLog]:
     """分页查询操作日志。
 
     **按 id 倒序而非 create_time**：后者是秒级精度，同一秒内的多条记录排序
@@ -127,41 +130,25 @@ def list_logs(
     """
     stmt = _apply_filters(
         select(OperationLog).options(defer(OperationLog.request_data)),
-        filters=filters,
-        module=module,
-        action=action,
+        query,
     )
-    stmt = stmt.order_by(OperationLog.id.desc()).offset(offset).limit(limit)
+    stmt = stmt.order_by(OperationLog.id.desc()).offset(query.offset).limit(query.limit)
     return list(session.execute(stmt).scalars())
 
 
-def count_logs(
-    session: Session,
-    *,
-    filters: LogFilters,
-    module: str | None = None,
-    action: str | None = None,
-) -> int:
+def count_logs(session: Session, query: OperationLogListQuery) -> int:
     """统计符合条件的日志总数，供分页使用。"""
-    stmt = _apply_filters(
-        select(func.count(OperationLog.id)), filters=filters, module=module, action=action
-    )
+    stmt = _apply_filters(select(func.count(OperationLog.id)), query)
     return int(session.execute(stmt).scalar_one())
 
 
-def _apply_filters(
-    stmt: Select,
-    *,
-    filters: LogFilters,
-    module: str | None,
-    action: str | None,
-) -> Select:
+def _apply_filters(stmt: Select, query: OperationLogListQuery) -> Select:
     """应用筛选条件（公共部分复用 query_utils，与登录日志同口径）。"""
-    stmt = apply_log_filters(stmt, OperationLog, filters)
-    if module:
-        stmt = stmt.where(OperationLog.module == module)
-    if action:
-        stmt = stmt.where(OperationLog.action == action)
+    stmt = apply_log_filters(stmt, OperationLog, query.filters)
+    if query.module:
+        stmt = stmt.where(OperationLog.module == query.module)
+    if query.action:
+        stmt = stmt.where(OperationLog.action == query.action)
     return stmt
 
 
diff --git a/backend/app/repositories/ownership_log_repository.py b/backend/app/repositories/ownership_log_repository.py
index c5bb7ab..03bfb70 100644
--- a/backend/app/repositories/ownership_log_repository.py
+++ b/backend/app/repositories/ownership_log_repository.py
@@ -5,18 +5,32 @@ owner_type 与引用列由调用方参数化 —— 此前客户、业务两侧
 「被删人员留痕仍可读」「翻页稳定双键倒序」等口径单点维护。
 """
 
+from dataclasses import dataclass
+from typing import Any
+
 from sqlalchemy import Row, func, select
 from sqlalchemy.orm import Session, aliased
 
 from app.models import OwnershipChangeLog, Personnel
 
 
+@dataclass(frozen=True, slots=True)
+class OwnershipLogRef:
+    """归属留痕的引用锚点（维度类型 + 引用列 + 引用 ID）。
+
+    引用列是 SQLAlchemy 列表达式（如 ``OwnershipChangeLog.customer_id``），
+    不是可序列化值 —— 故用冻结 dataclass 而非 Pydantic 模型。
+    """
+
+    owner_type: int
+    ref_column: Any
+    ref_id: int
+
+
 def list_by_ref(
     session: Session,
+    ref: OwnershipLogRef,
     *,
-    owner_type: int,
-    ref_column,
-    ref_id: int,
     offset: int,
     limit: int,
 ) -> list[Row]:
@@ -32,7 +46,7 @@ def list_by_ref(
         select(OwnershipChangeLog, old_person.personnel_name, new_person.personnel_name)
         .outerjoin(old_person, OwnershipChangeLog.old_owner_personnel_id == old_person.id)
         .outerjoin(new_person, OwnershipChangeLog.new_owner_personnel_id == new_person.id)
-        .where(OwnershipChangeLog.owner_type == owner_type, ref_column == ref_id)
+        .where(OwnershipChangeLog.owner_type == ref.owner_type, ref.ref_column == ref.ref_id)
         .order_by(OwnershipChangeLog.create_time.desc(), OwnershipChangeLog.id.desc())
         .offset(offset)
         .limit(limit)
@@ -40,11 +54,11 @@ def list_by_ref(
     return list(session.execute(stmt).all())
 
 
-def count_by_ref(session: Session, *, owner_type: int, ref_column, ref_id: int) -> int:
+def count_by_ref(session: Session, ref: OwnershipLogRef) -> int:
     """统计某引用对象的归属留痕总数，供分页使用。"""
     stmt = select(func.count(OwnershipChangeLog.id)).where(
-        OwnershipChangeLog.owner_type == owner_type,
-        ref_column == ref_id,
+        OwnershipChangeLog.owner_type == ref.owner_type,
+        ref.ref_column == ref.ref_id,
     )
     return int(session.execute(stmt).scalar_one())
 
diff --git a/backend/app/repositories/param_repository.py b/backend/app/repositories/param_repository.py
index 270b9e8..9abdf16 100644
--- a/backend/app/repositories/param_repository.py
+++ b/backend/app/repositories/param_repository.py
@@ -10,6 +10,7 @@ from sqlalchemy import Select, delete, func, select
 from sqlalchemy.orm import Session
 
 from app.models import ParamChangeBatch, ParamDefinition, ParamValue
+from app.schemas.param import ParamBatchQuery, ParamHistoryQuery
 
 # ==================== 读取 ====================
 
@@ -97,27 +98,27 @@ def list_definitions_by_codes(
 
 
 def list_history(
-    session: Session,
-    param_id: int,
-    *,
-    offset: int,
-    limit: int,
-    dimension_value: str | None = None,
+    session: Session, param_id: int, query: ParamHistoryQuery
 ) -> list[tuple[ParamValue, ParamChangeBatch]]:
-    """单参数的版本历史（含批次信息），最新在前。"""
-    stmt = _history_stmt(session, param_id, dimension_value=dimension_value)
-    return [(value, batch) for value, batch in session.execute(stmt.offset(offset).limit(limit))]
+    """单参数的版本历史（含批次信息），最新在前。
+
+    ``param_id`` 是「查哪个参数」的主锚点（由路径参数 param_code 解析而来），
+    不属于列表筛选项，故仍以位置参数传入；筛选与分页走 ``query``。
+    """
+    stmt = _history_stmt(param_id, query)
+    return [
+        (value, batch)
+        for value, batch in session.execute(stmt.offset(query.offset).limit(query.page_size))
+    ]
 
 
-def count_history(session: Session, param_id: int, *, dimension_value: str | None = None) -> int:
+def count_history(session: Session, param_id: int, query: ParamHistoryQuery) -> int:
     """统计单参数的历史条目数。"""
-    stmt = select(func.count()).select_from(
-        _history_stmt(session, param_id, dimension_value=dimension_value).subquery()
-    )
+    stmt = select(func.count()).select_from(_history_stmt(param_id, query).subquery())
     return int(session.execute(stmt).scalar_one())
 
 
-def _history_stmt(session: Session, param_id: int, *, dimension_value: str | None) -> Select:
+def _history_stmt(param_id: int, query: ParamHistoryQuery) -> Select:
     """历史查询基语句：条目 join 批次（版本号/操作人），最新在前。"""
     stmt = (
         select(ParamValue, ParamChangeBatch)
@@ -125,8 +126,8 @@ def _history_stmt(session: Session, param_id: int, *, dimension_value: str | Non
         .where(ParamValue.param_id == param_id)
         .order_by(ParamValue.effective_date.desc(), ParamValue.id.desc())
     )
-    if dimension_value is not None:
-        stmt = stmt.where(ParamValue.dimension_value == dimension_value)
+    if query.dimension_value is not None:
+        stmt = stmt.where(ParamValue.dimension_value == query.dimension_value)
     return stmt
 
 
@@ -150,41 +151,30 @@ def get_batch_by_id(
     return session.get(ParamChangeBatch, batch_id)
 
 
-def list_batches(
-    session: Session,
-    *,
-    offset: int,
-    limit: int,
-    start_date: date | None = None,
-    end_date: date | None = None,
-) -> list[ParamChangeBatch]:
+def list_batches(session: Session, query: ParamBatchQuery) -> list[ParamChangeBatch]:
     """分页查询变更批次（按创建时间倒序）。"""
-    stmt = _batch_filters(select(ParamChangeBatch), start_date=start_date, end_date=end_date)
+    stmt = _batch_filters(select(ParamChangeBatch), query)
     return list(
         session.execute(
             stmt.order_by(ParamChangeBatch.create_time.desc(), ParamChangeBatch.id.desc())
-            .offset(offset)
-            .limit(limit)
+            .offset(query.offset)
+            .limit(query.page_size)
         ).scalars()
     )
 
 
-def count_batches(
-    session: Session, *, start_date: date | None = None, end_date: date | None = None
-) -> int:
+def count_batches(session: Session, query: ParamBatchQuery) -> int:
     """统计符合条件的批次总数。"""
-    stmt = _batch_filters(
-        select(func.count(ParamChangeBatch.id)), start_date=start_date, end_date=end_date
-    )
+    stmt = _batch_filters(select(func.count(ParamChangeBatch.id)), query)
     return int(session.execute(stmt).scalar_one())
 
 
-def _batch_filters(stmt: Select, *, start_date: date | None, end_date: date | None) -> Select:
+def _batch_filters(stmt: Select, query: ParamBatchQuery) -> Select:
     """批次筛选（生效日期区间，含边界）。列表与计数共用。"""
-    if start_date is not None:
-        stmt = stmt.where(ParamChangeBatch.effective_date >= start_date)
-    if end_date is not None:
-        stmt = stmt.where(ParamChangeBatch.effective_date <= end_date)
+    if query.start_date is not None:
+        stmt = stmt.where(ParamChangeBatch.effective_date >= query.start_date)
+    if query.end_date is not None:
+        stmt = stmt.where(ParamChangeBatch.effective_date <= query.end_date)
     return stmt
 
 
diff --git a/backend/app/repositories/personnel_repository.py b/backend/app/repositories/personnel_repository.py
index 3f25451..5f50fb9 100644
--- a/backend/app/repositories/personnel_repository.py
+++ b/backend/app/repositories/personnel_repository.py
@@ -9,6 +9,7 @@ from sqlalchemy.orm import Session, aliased
 
 from app.models import BusinessRole, Customer, Personnel, PersonnelPlan, PersonnelRegion
 from app.repositories.query_utils import fuzzy
+from app.schemas.personnel import PersonnelQuery
 
 # 自引用别名：manager_id 指向 personnel.id，同一张表自联取「直属经理姓名」
 _manager = aliased(Personnel)
@@ -30,44 +31,23 @@ def get_with_names(session: Session, personnel_id: int) -> Row | None:
     return session.execute(stmt).first()
 
 
-def list_personnel(
-    session: Session,
-    *,
-    offset: int,
-    limit: int,
-    keyword: str | None = None,
-    status: int | None = None,
-    business_role_id: int | None = None,
-) -> list[Row]:
+def list_personnel(session: Session, query: PersonnelQuery) -> list[Row]:
     """分页查询人员，行内容为 (人员, 岗位名称, 直属经理姓名)。
 
     排序按 id 倒序：人员无排序字段，新增的靠前符合列表页习惯。
     """
-    stmt = _apply_filters(
-        _base_select(), keyword=keyword, status=status, business_role_id=business_role_id
-    )
-    stmt = stmt.order_by(Personnel.id.desc()).offset(offset).limit(limit)
+    stmt = _apply_filters(_base_select(), query)
+    stmt = stmt.order_by(Personnel.id.desc()).offset(query.offset).limit(query.page_size)
     return list(session.execute(stmt).all())
 
 
-def count_personnel(
-    session: Session,
-    *,
-    keyword: str | None = None,
-    status: int | None = None,
-    business_role_id: int | None = None,
-) -> int:
+def count_personnel(session: Session, query: PersonnelQuery) -> int:
     """统计符合条件的人员总数，供分页使用。
 
     筛选条件全部落在主表上，计数无需 join（列表与计数的筛选口径由
     ``_apply_filters`` 单点保证）。
     """
-    stmt = _apply_filters(
-        select(func.count(Personnel.id)),
-        keyword=keyword,
-        status=status,
-        business_role_id=business_role_id,
-    )
+    stmt = _apply_filters(select(func.count(Personnel.id)), query)
     return int(session.execute(stmt).scalar_one())
 
 
@@ -154,20 +134,14 @@ def _base_select() -> Select:
     )
 
 
-def _apply_filters(
-    stmt: Select,
-    *,
-    keyword: str | None,
-    status: int | None,
-    business_role_id: int | None,
-) -> Select:
+def _apply_filters(stmt: Select, query: PersonnelQuery) -> Select:
     """应用通用筛选条件。列表与计数共用，避免两处口径不一致。"""
-    if keyword:
-        stmt = fuzzy(stmt, Personnel.personnel_name, keyword)
-    if status is not None:
-        stmt = stmt.where(Personnel.status == status)
-    if business_role_id is not None:
-        stmt = stmt.where(Personnel.business_role_id == business_role_id)
+    if query.keyword:
+        stmt = fuzzy(stmt, Personnel.personnel_name, query.keyword)
+    if query.status is not None:
+        stmt = stmt.where(Personnel.status == query.status)
+    if query.business_role_id is not None:
+        stmt = stmt.where(Personnel.business_role_id == query.business_role_id)
     return stmt
 
 
diff --git a/backend/app/repositories/product_repository.py b/backend/app/repositories/product_repository.py
index 1dbaf61..3e8f8b0 100644
--- a/backend/app/repositories/product_repository.py
+++ b/backend/app/repositories/product_repository.py
@@ -7,6 +7,7 @@ from sqlalchemy import Select, func, or_, select
 from sqlalchemy.orm import Session
 
 from app.models import BusinessItem, Product, ShipmentItem
+from app.schemas.product import ProductQuery
 
 # ==================== 读取 ====================
 
@@ -34,50 +35,20 @@ def get_by_name_spec(session: Session, product_name: str, spec_model: str) -> Pr
     return session.execute(stmt).scalar_one_or_none()
 
 
-def list_products(
-    session: Session,
-    *,
-    offset: int,
-    limit: int,
-    keyword: str | None = None,
-    product_category: str | None = None,
-    status: int | None = None,
-) -> list[Product]:
+def list_products(session: Session, query: ProductQuery) -> list[Product]:
     """分页查询产品（含停用）。按 id 升序，与 M1 岗位列表的次序口径一致。"""
-    stmt = _apply_filters(
-        select(Product),
-        keyword=keyword,
-        product_category=product_category,
-        status=status,
-    )
-    stmt = stmt.order_by(Product.id).offset(offset).limit(limit)
+    stmt = _apply_filters(select(Product), query)
+    stmt = stmt.order_by(Product.id).offset(query.offset).limit(query.page_size)
     return list(session.execute(stmt).scalars())
 
 
-def count_products(
-    session: Session,
-    *,
-    keyword: str | None = None,
-    product_category: str | None = None,
-    status: int | None = None,
-) -> int:
+def count_products(session: Session, query: ProductQuery) -> int:
     """统计符合条件的产品总数，供分页使用。"""
-    stmt = _apply_filters(
-        select(func.count(Product.id)),
-        keyword=keyword,
-        product_category=product_category,
-        status=status,
-    )
+    stmt = _apply_filters(select(func.count(Product.id)), query)
     return int(session.execute(stmt).scalar_one())
 
 
-def _apply_filters(
-    stmt: Select,
-    *,
-    keyword: str | None,
-    product_category: str | None,
-    status: int | None,
-) -> Select:
+def _apply_filters(stmt: Select, query: ProductQuery) -> Select:
     """应用通用筛选条件。列表与计数共用，避免两处口径不一致。
 
     关键词对名称/规格两列取 OR 命中。转义与 query_utils.fuzzy 同一机制
@@ -86,17 +57,17 @@ def _apply_filters(
     写 contains —— M1 business_role_repository 的双列关键词同款写法。
     规格型号常含 ``_``（如 DN300_II），不转义会当单字符通配执行。
     """
-    if keyword:
+    if query.keyword:
         stmt = stmt.where(
             or_(
-                Product.product_name.contains(keyword, autoescape=True),
-                Product.spec_model.contains(keyword, autoescape=True),
+                Product.product_name.contains(query.keyword, autoescape=True),
+                Product.spec_model.contains(query.keyword, autoescape=True),
             )
         )
-    if product_category:
-        stmt = stmt.where(Product.product_category == product_category)
-    if status is not None:
-        stmt = stmt.where(Product.status == status)
+    if query.product_category:
+        stmt = stmt.where(Product.product_category == query.product_category)
+    if query.status is not None:
+        stmt = stmt.where(Product.status == query.status)
     return stmt
 
 
diff --git a/backend/app/repositories/receipt_repository.py b/backend/app/repositories/receipt_repository.py
index 9804f70..36b70da 100644
--- a/backend/app/repositories/receipt_repository.py
+++ b/backend/app/repositories/receipt_repository.py
@@ -14,6 +14,7 @@ from sqlalchemy import Select, func, select
 from sqlalchemy.orm import Session
 
 from app.models import Business, Receipt, Shipment
+from app.schemas.receipt import ReceiptQuery
 
 
 def get_by_id(session: Session, receipt_id: int) -> Receipt | None:
@@ -43,72 +44,42 @@ def get_shipment(session: Session, shipment_id: int) -> Shipment | None:
     return session.get(Shipment, shipment_id)
 
 
-def _apply_filters(
-    stmt: Select,
-    *,
-    business_id: int | None,
-    date_from: date | None,
-    date_to: date | None,
-    is_barter: int | None,
-    settlement_status: int | None,
-) -> Select:
+def _apply_filters(stmt: Select, query: ReceiptQuery) -> Select:
     """回款列表的筛选条件（全部落在 Receipt 自身列上）。"""
-    if business_id is not None:
-        stmt = stmt.where(Receipt.business_id == business_id)
-    if date_from is not None:
-        stmt = stmt.where(Receipt.receipt_date >= date_from)
-    if date_to is not None:
-        stmt = stmt.where(Receipt.receipt_date <= date_to)
-    if is_barter is not None:
-        stmt = stmt.where(Receipt.is_barter == is_barter)
-    if settlement_status is not None:
-        stmt = stmt.where(Receipt.settlement_status == settlement_status)
+    if query.business_id is not None:
+        stmt = stmt.where(Receipt.business_id == query.business_id)
+    if query.date_from is not None:
+        stmt = stmt.where(Receipt.receipt_date >= query.date_from)
+    if query.date_to is not None:
+        stmt = stmt.where(Receipt.receipt_date <= query.date_to)
+    if query.is_barter is not None:
+        stmt = stmt.where(Receipt.is_barter == query.is_barter)
+    # settlement_status 是 M3 核算防重锚点（需规 2.3.3），必须保持具名字段
+    if query.settlement_status is not None:
+        stmt = stmt.where(Receipt.settlement_status == query.settlement_status)
     return stmt
 
 
 def list_receipts(
-    session: Session,
-    *,
-    offset: int,
-    limit: int,
-    business_id: int | None,
-    date_from: date | None,
-    date_to: date | None,
-    is_barter: int | None,
-    settlement_status: int | None,
+    session: Session, query: ReceiptQuery
 ) -> list[tuple[Receipt, str, str | None, date | None]]:
     """分页查询回款（附业务号/发货单号/发货日），按 id 倒序。"""
     stmt = _apply_filters(
         select(Receipt, Business.business_no, Shipment.shipment_no, Shipment.ship_date)
         .join(Business, Business.id == Receipt.business_id)
         .outerjoin(Shipment, Shipment.id == Receipt.shipment_id),
-        business_id=business_id,
-        date_from=date_from,
-        date_to=date_to,
-        is_barter=is_barter,
-        settlement_status=settlement_status,
+        query,
+    )
+    return list(
+        session.execute(
+            stmt.order_by(Receipt.id.desc()).offset(query.offset).limit(query.page_size)
+        ).all()
     )
-    return list(session.execute(stmt.order_by(Receipt.id.desc()).offset(offset).limit(limit)).all())
 
 
-def count_receipts(
-    session: Session,
-    *,
-    business_id: int | None,
-    date_from: date | None,
-    date_to: date | None,
-    is_barter: int | None,
-    settlement_status: int | None,
-) -> int:
+def count_receipts(session: Session, query: ReceiptQuery) -> int:
     """回款总数（筛选条件与列表共用同一 apply_filters，防两侧口径漂移）。"""
-    stmt = _apply_filters(
-        select(func.count()).select_from(Receipt),
-        business_id=business_id,
-        date_from=date_from,
-        date_to=date_to,
-        is_barter=is_barter,
-        settlement_status=settlement_status,
-    )
+    stmt = _apply_filters(select(func.count()).select_from(Receipt), query)
     return int(session.execute(stmt).scalar_one())
 
 
diff --git a/backend/app/repositories/report_repository.py b/backend/app/repositories/report_repository.py
index 021592f..c221abe 100644
--- a/backend/app/repositories/report_repository.py
+++ b/backend/app/repositories/report_repository.py
@@ -389,9 +389,7 @@ def _bucket_condition(bucket: DayBucket) -> tuple:
     return tuple(conds)
 
 
-def list_payment_detail(
-    session: Session, query: PaymentDetailQuery, bucket: DayBucket, *, offset: int, limit: int
-) -> list:
+def list_payment_detail(session: Session, query: PaymentDetailQuery, bucket: DayBucket) -> list:
     """区间下钻：单笔生效行联回款表（行级展示回款金额与批次号）。
 
     bucket 由服务层解析（未注册编码 400），仓储只按区间拼条件。
@@ -408,7 +406,7 @@ def list_payment_detail(
         .where(*_payment_filters(query), *_bucket_condition(bucket))
     )
     stmt = stmt.order_by(PerformanceLedger.receipt_date.desc(), PerformanceLedger.id.desc())
-    stmt = stmt.offset(offset).limit(limit)
+    stmt = stmt.offset(query.offset).limit(query.page_size)
     return list(session.execute(stmt).all())
 
 
diff --git a/backend/app/repositories/role_repository.py b/backend/app/repositories/role_repository.py
index cdd2c13..e12b029 100644
--- a/backend/app/repositories/role_repository.py
+++ b/backend/app/repositories/role_repository.py
@@ -10,6 +10,7 @@ from sqlalchemy import Select, func, or_, select
 from sqlalchemy.orm import Session
 
 from app.models import Role, UserRole
+from app.schemas.role import RoleQuery
 
 # ==================== 读取 ====================
 
@@ -47,44 +48,37 @@ def get_all_roles(session: Session) -> list[Role]:
     return list(session.execute(stmt).scalars())
 
 
-def list_roles(
-    session: Session,
-    *,
-    offset: int,
-    limit: int,
-    keyword: str | None = None,
-    status: int | None = None,
-) -> list[Role]:
+def list_roles(session: Session, query: RoleQuery) -> list[Role]:
     """分页查询角色（管理列表：含停用角色）。
 
     排序与下拉口径一致（sort_order, id），保持同一套相对次序。
     """
-    stmt = _apply_filters(select(Role), keyword=keyword, status=status)
-    stmt = stmt.order_by(Role.sort_order, Role.id).offset(offset).limit(limit)
+    stmt = _apply_filters(select(Role), query)
+    stmt = stmt.order_by(Role.sort_order, Role.id).offset(query.offset).limit(query.page_size)
     return list(session.execute(stmt).scalars())
 
 
-def count_roles(session: Session, *, keyword: str | None = None, status: int | None = None) -> int:
+def count_roles(session: Session, query: RoleQuery) -> int:
     """统计符合条件的角色总数，供分页使用。"""
-    stmt = _apply_filters(select(func.count(Role.id)), keyword=keyword, status=status)
+    stmt = _apply_filters(select(func.count(Role.id)), query)
     return int(session.execute(stmt).scalar_one())
 
 
-def _apply_filters(stmt: Select, *, keyword: str | None, status: int | None) -> Select:
+def _apply_filters(stmt: Select, query: RoleQuery) -> Select:
     """应用通用筛选条件。列表与计数共用，避免两处口径不一致。
 
     关键词匹配走 `contains(autoescape=True)`（转义 LIKE 通配符）——
     角色编码普遍含下划线，直接拼接会让 `_` 当单字符通配执行。
     """
-    if keyword:
+    if query.keyword:
         stmt = stmt.where(
             or_(
-                Role.role_code.contains(keyword, autoescape=True),
-                Role.role_name.contains(keyword, autoescape=True),
+                Role.role_code.contains(query.keyword, autoescape=True),
+                Role.role_name.contains(query.keyword, autoescape=True),
             )
         )
-    if status is not None:
-        stmt = stmt.where(Role.status == status)
+    if query.status is not None:
+        stmt = stmt.where(Role.status == query.status)
     return stmt
 
 
diff --git a/backend/app/repositories/shipment_repository.py b/backend/app/repositories/shipment_repository.py
index b463270..64a26ae 100644
--- a/backend/app/repositories/shipment_repository.py
+++ b/backend/app/repositories/shipment_repository.py
@@ -8,13 +8,13 @@
   业务级汇总仍收口 business_service._recalc_totals
 """
 
-from datetime import date
 from decimal import Decimal
 
 from sqlalchemy import Select, delete, func, select
 from sqlalchemy.orm import Session
 
 from app.models import Business, Receipt, Shipment, ShipmentItem
+from app.schemas.shipment import ShipmentQuery
 
 
 def get_by_id(session: Session, shipment_id: int) -> Shipment | None:
@@ -31,59 +31,37 @@ def get_with_business_no(session: Session, shipment_id: int) -> tuple[Shipment,
     ).first()
 
 
-def _apply_filters(
-    stmt: Select,
-    *,
-    business_id: int | None,
-    ship_date_from: date | None,
-    ship_date_to: date | None,
-) -> Select:
+def _apply_filters(stmt: Select, query: ShipmentQuery) -> Select:
     """发货列表的筛选条件（business_id 空=全量，一期按业务维度查看为主）。"""
-    if business_id is not None:
-        stmt = stmt.where(Shipment.business_id == business_id)
-    if ship_date_from is not None:
-        stmt = stmt.where(Shipment.ship_date >= ship_date_from)
-    if ship_date_to is not None:
-        stmt = stmt.where(Shipment.ship_date <= ship_date_to)
+    if query.business_id is not None:
+        stmt = stmt.where(Shipment.business_id == query.business_id)
+    if query.ship_date_from is not None:
+        stmt = stmt.where(Shipment.ship_date >= query.ship_date_from)
+    if query.ship_date_to is not None:
+        stmt = stmt.where(Shipment.ship_date <= query.ship_date_to)
     return stmt
 
 
-def list_shipments(
-    session: Session,
-    *,
-    offset: int,
-    limit: int,
-    business_id: int | None,
-    ship_date_from: date | None,
-    ship_date_to: date | None,
-) -> list[tuple[Shipment, str]]:
+def list_shipments(session: Session, query: ShipmentQuery) -> list[tuple[Shipment, str]]:
     """分页查询发货单（附业务号），按 id 倒序。"""
     stmt = _apply_filters(
         select(Shipment, Business.business_no).join(Business, Business.id == Shipment.business_id),
-        business_id=business_id,
-        ship_date_from=ship_date_from,
-        ship_date_to=ship_date_to,
+        query,
     )
     return list(
-        session.execute(stmt.order_by(Shipment.id.desc()).offset(offset).limit(limit)).all()
+        session.execute(
+            stmt.order_by(Shipment.id.desc()).offset(query.offset).limit(query.page_size)
+        ).all()
     )
 
 
-def count_shipments(
-    session: Session,
-    *,
-    business_id: int | None,
-    ship_date_from: date | None,
-    ship_date_to: date | None,
-) -> int:
+def count_shipments(session: Session, query: ShipmentQuery) -> int:
     """发货单总数（筛选条件与列表共用同一 apply_filters，防两侧口径漂移）。"""
     stmt = _apply_filters(
         select(func.count())
         .select_from(Shipment)
         .join(Business, Business.id == Shipment.business_id),
-        business_id=business_id,
-        ship_date_from=ship_date_from,
-        ship_date_to=ship_date_to,
+        query,
     )
     return int(session.execute(stmt).scalar_one())
 
diff --git a/backend/app/repositories/user_repository.py b/backend/app/repositories/user_repository.py
index 6b18d5b..7aba47b 100644
--- a/backend/app/repositories/user_repository.py
+++ b/backend/app/repositories/user_repository.py
@@ -10,6 +10,7 @@ from sqlalchemy.orm import Session, selectinload
 
 from app.models import Role, User, UserRole
 from app.repositories.query_utils import fuzzy
+from app.schemas.user import UserQuery
 
 
 def _base_query() -> Select[tuple[User]]:
@@ -48,16 +49,7 @@ def touch_last_login(session: Session, user: User, moment: datetime) -> None:
 # ==================== 列表查询 ====================
 
 
-def list_users(
-    session: Session,
-    *,
-    offset: int,
-    limit: int,
-    username: str | None = None,
-    real_name: str | None = None,
-    status: int | None = None,
-    role_code: str | None = None,
-) -> list[User]:
+def list_users(session: Session, query: UserQuery) -> list[User]:
     """分页查询用户列表。
 
     **按 id 倒序而非 create_time**：create_time 是秒级精度，批量导入的用户
@@ -65,48 +57,35 @@ def list_users(
     id 是自增主键，天然唯一且与创建顺序一致。
     """
     stmt = _base_query()
-    stmt = _apply_filters(stmt, username=username, real_name=real_name, status=status)
-    if role_code:
+    stmt = _apply_filters(stmt, query)
+    if query.role_code:
         # 按角色筛选需 join 关联表；角色是字典数据，数量极少，join 代价可忽略
-        stmt = stmt.join(UserRole).join(Role).where(Role.role_code == role_code)
-    stmt = stmt.order_by(User.id.desc()).offset(offset).limit(limit)
+        stmt = stmt.join(UserRole).join(Role).where(Role.role_code == query.role_code)
+    stmt = stmt.order_by(User.id.desc()).offset(query.offset).limit(query.page_size)
     return list(session.execute(stmt).scalars().unique())
 
 
-def count_users(
-    session: Session,
-    *,
-    username: str | None = None,
-    real_name: str | None = None,
-    status: int | None = None,
-    role_code: str | None = None,
-) -> int:
+def count_users(session: Session, query: UserQuery) -> int:
     """统计符合条件的用户总数，供分页使用。"""
     stmt = select(func.count(User.id))
-    stmt = _apply_filters(stmt, username=username, real_name=real_name, status=status)
-    if role_code:
-        stmt = stmt.join(UserRole).join(Role).where(Role.role_code == role_code)
+    stmt = _apply_filters(stmt, query)
+    if query.role_code:
+        stmt = stmt.join(UserRole).join(Role).where(Role.role_code == query.role_code)
     return int(session.execute(stmt).scalar_one())
 
 
-def _apply_filters(
-    stmt: Select,
-    *,
-    username: str | None,
-    real_name: str | None,
-    status: int | None,
-) -> Select:
+def _apply_filters(stmt: Select, query: UserQuery) -> Select:
     """应用通用筛选条件。列表与计数共用，避免两处口径不一致。
 
     模糊匹配统一走 `query_utils.fuzzy`（转义 LIKE 通配符）—— 用户名普遍
     含下划线，直接拼接会让 `_` 当单字符通配执行、误配无关账号。
     """
-    if username:
-        stmt = fuzzy(stmt, User.username, username)
-    if real_name:
-        stmt = fuzzy(stmt, User.real_name, real_name)
-    if status is not None:
-        stmt = stmt.where(User.status == status)
+    if query.username:
+        stmt = fuzzy(stmt, User.username, query.username)
+    if query.real_name:
+        stmt = fuzzy(stmt, User.real_name, query.real_name)
+    if query.status is not None:
+        stmt = stmt.where(User.status == query.status)
     return stmt
 
 
diff --git a/backend/app/schemas/approval.py b/backend/app/schemas/approval.py
index 7a3eac9..04c0ff5 100644
--- a/backend/app/schemas/approval.py
+++ b/backend/app/schemas/approval.py
@@ -49,7 +49,8 @@ class ApprovalOrderDetailOut(ApprovalOrderOut):
 
     steps: list[ApprovalStepOut] = Field(default_factory=list)
     snapshot: dict[str, Any] | None = Field(
-        default=None, description="提交时内容快照（信封：schema_version/order_type/captured_at/data）"
+        default=None,
+        description="提交时内容快照（信封：schema_version/order_type/captured_at/data）",
     )
     content_source: Literal["snapshot", "live", "missing"] = Field(
         default="missing", description="审批内容来源：提交时快照/当前内容回退/缺失"
diff --git a/backend/app/schemas/business_role.py b/backend/app/schemas/business_role.py
index fe6a0ab..e3f1b2a 100644
--- a/backend/app/schemas/business_role.py
+++ b/backend/app/schemas/business_role.py
@@ -27,6 +27,19 @@ class BusinessRoleOut(BaseModel):
     is_builtin: int = 0
 
 
+class BusinessRoleOptionOut(BaseModel):
+    """岗位下拉选项（人员档案与报表筛选共用）。
+
+    与 ``BusinessRoleOut`` 分开：下拉只需渲染用的最小字段集，不必顺带
+    暴露计提模板等管理字段。字段名直出蛇形（前端按同名取值）。
+    """
+
+    id: int = Field(description="岗位ID")
+    role_code: str = Field(description="岗位编码")
+    role_name: str = Field(description="岗位名称")
+    status: int = Field(description="状态：1启用 0禁用")
+
+
 class TemplateParamOut(BaseModel):
     """模板声明的参数（供岗位表单展示，含当前口径说明）。"""
 
diff --git a/backend/app/schemas/report.py b/backend/app/schemas/report.py
index 542fcd0..b457f4b 100644
--- a/backend/app/schemas/report.py
+++ b/backend/app/schemas/report.py
@@ -264,7 +264,9 @@ class DepartmentReconciliation(BaseModel):
         description="聚合额 − 团队分成合计 − 全公司分成合计（分列舍入差）",
     )
     adjust_net_total: Decimal = Field(description="明细层调整净额合计（成员 + 未分配）")
-    payable_total: Decimal = Field(description="应发合计 = 成员应付 + 未分配应付 + 聚合行 + 调整净额")
+    payable_total: Decimal = Field(
+        description="应发合计 = 成员应付 + 未分配应付 + 聚合行 + 调整净额"
+    )
     team_base_total: Decimal = Field(description="展示团队基数合计")
     unassigned_performance_total: Decimal = Field(description="展示未分配人员绩效合计")
     other_team_base: Decimal = Field(
diff --git a/backend/app/services/auth_service.py b/backend/app/services/auth_service.py
index cc51f2a..db5d148 100644
--- a/backend/app/services/auth_service.py
+++ b/backend/app/services/auth_service.py
@@ -37,12 +37,14 @@ def _record(
     """写登录日志并立即提交。单独提交是刻意为之，见模块 docstring。"""
     login_log_repository.add(
         session,
-        user_id=user_id,
-        username=username,
-        ip=ip,
-        user_agent=user_agent,
-        success=success,
-        fail_reason=fail_reason,
+        login_log_repository.LoginLogEntry(
+            user_id=user_id,
+            username=username,
+            ip=ip,
+            user_agent=user_agent,
+            success=success,
+            fail_reason=fail_reason,
+        ),
     )
     session.commit()
 
diff --git a/backend/app/services/business_role_service.py b/backend/app/services/business_role_service.py
index d89037c..43ae38b 100644
--- a/backend/app/services/business_role_service.py
+++ b/backend/app/services/business_role_service.py
@@ -31,6 +31,7 @@ from app.models import BusinessRole, User
 from app.repositories import business_role_repository
 from app.schemas.business_role import (
     BusinessRoleCreateIn,
+    BusinessRoleOptionOut,
     BusinessRoleOut,
     BusinessRoleQuery,
     BusinessRoleUpdateIn,
@@ -67,16 +68,8 @@ def _ensure_template(calc_template: str) -> None:
 
 def list_roles(session: Session, query: BusinessRoleQuery) -> PageResult[BusinessRoleOut]:
     """分页查询岗位（含停用）。"""
-    roles = business_role_repository.list_roles(
-        session,
-        offset=query.offset,
-        limit=query.page_size,
-        keyword=query.keyword,
-        status=query.status,
-    )
-    total = business_role_repository.count_roles(
-        session, keyword=query.keyword, status=query.status
-    )
+    roles = business_role_repository.list_roles(session, query)
+    total = business_role_repository.count_roles(session, query)
     return PageResult[BusinessRoleOut](
         records=[_to_out(role) for role in roles],
         total=total,
@@ -85,6 +78,23 @@ def list_roles(session: Session, query: BusinessRoleQuery) -> PageResult[Busines
     )
 
 
+def list_options(session: Session) -> list[BusinessRoleOptionOut]:
+    """岗位下拉选项（含停用，不分页）。
+
+    含停用是因为人员档案可能仍引用停用岗位（停用只是禁止新增引用，
+    不改历史归属）；岗位规模有限（内置 + 自定义），一次取全量。
+    """
+    return [
+        BusinessRoleOptionOut(
+            id=role.id,
+            role_code=role.role_code,
+            role_name=role.role_name,
+            status=role.status,
+        )
+        for role in business_role_repository.list_all(session)
+    ]
+
+
 def get_role(session: Session, role_id: int) -> BusinessRoleOut:
     """岗位详情。
 
diff --git a/backend/app/services/business_service.py b/backend/app/services/business_service.py
index a13b2fa..86d604d 100644
--- a/backend/app/services/business_service.py
+++ b/backend/app/services/business_service.py
@@ -15,6 +15,8 @@
 """
 
 import logging
+from collections.abc import Callable
+from dataclasses import dataclass
 from datetime import date
 from decimal import ROUND_HALF_UP, Decimal
 from fractions import Fraction
@@ -234,26 +236,8 @@ def _is_duplicate_entry(exc: IntegrityError) -> bool:
 
 def list_businesses(session: Session, query: BusinessQuery) -> PageResult[BusinessOut]:
     """分页查询业务（附客户名与归属人名）。"""
-    rows = business_repository.list_businesses(
-        session,
-        offset=query.offset,
-        limit=query.page_size,
-        keyword=query.keyword,
-        status=query.status,
-        customer_id=query.customer_id,
-        owner_personnel_id=query.owner_personnel_id,
-        date_from=query.date_from,
-        date_to=query.date_to,
-    )
-    total = business_repository.count_businesses(
-        session,
-        keyword=query.keyword,
-        status=query.status,
-        customer_id=query.customer_id,
-        owner_personnel_id=query.owner_personnel_id,
-        date_from=query.date_from,
-        date_to=query.date_to,
-    )
+    rows = business_repository.list_businesses(session, query)
+    total = business_repository.count_businesses(session, query)
     return PageResult[BusinessOut](
         records=[_to_out(row) for row in rows],
         total=total,
@@ -748,6 +732,67 @@ def change_status(
         return out
 
 
+@dataclass(frozen=True, slots=True)
+class _BusinessOwnershipResolution:
+    """业务归属变更的解析结果（新归属人行 + 变更前固化 ID）。"""
+
+    owner: Personnel | None
+    assistant: Personnel | None
+    old_owner_id: int | None
+    old_assistant_id: int | None
+
+
+def _resolve_business_ownership(
+    session: Session, business: Business, payload: BusinessOwnershipIn
+) -> _BusinessOwnershipResolution:
+    """校验新归属人（须在职）并判定归属是否真的变化。
+
+    校验在审计上下文内执行 —— 失败也留痕。两者均未变化 → 400「归属未变化」。
+    """
+    # 归属变更的出参姓名来自校验函数的返回行（锁定读）：首查姓名是变更前
+    # 固化标量，新归属人名必须按新 ID 重取
+    owner = (
+        _ensure_personnel_active(session, payload.new_owner_personnel_id, "负责")
+        if payload.new_owner_personnel_id is not None
+        else None
+    )
+    assistant = (
+        _ensure_personnel_active(session, payload.new_assistant_personnel_id, "协助")
+        if payload.new_assistant_personnel_id is not None
+        else None
+    )
+    owner_changed = business.owner_personnel_id != payload.new_owner_personnel_id
+    assistant_changed = business.assistant_personnel_id != payload.new_assistant_personnel_id
+    if not owner_changed and not assistant_changed:
+        raise BadRequestError("归属未变化")
+    return _BusinessOwnershipResolution(
+        owner=owner,
+        assistant=assistant,
+        old_owner_id=business.owner_personnel_id,
+        old_assistant_id=business.assistant_personnel_id,
+    )
+
+
+def _business_ownership_log(
+    business: Business,
+    payload: BusinessOwnershipIn,
+    resolution: _BusinessOwnershipResolution,
+    operator: User,
+) -> OwnershipChangeLog:
+    """归属变更留痕行（负责/协助两个字段的前后值都记）。"""
+    return OwnershipChangeLog(
+        owner_type=OWNER_TYPE_BUSINESS,
+        business_id=business.id,
+        old_owner_personnel_id=resolution.old_owner_id,
+        new_owner_personnel_id=payload.new_owner_personnel_id,
+        old_assistant_personnel_id=resolution.old_assistant_id,
+        new_assistant_personnel_id=payload.new_assistant_personnel_id,
+        change_reason=payload.reason,
+        operator_id=operator.id,
+        operator_name=operator.username,
+    )
+
+
 def change_ownership(
     session: Session,
     business_id: int,
@@ -779,41 +824,12 @@ def change_ownership(
         if row is None:
             raise NotFoundError("业务不存在")
         business, _, old_owner_name, old_assistant_name = row
-        # 归属变更的出参姓名来自校验函数的返回行（锁定读）：首查姓名是变更前
-        # 固化标量，新归属人名必须按新 ID 重取
-        new_owner = (
-            _ensure_personnel_active(session, payload.new_owner_personnel_id, "负责")
-            if payload.new_owner_personnel_id is not None
-            else None
-        )
-        new_assistant = (
-            _ensure_personnel_active(session, payload.new_assistant_personnel_id, "协助")
-            if payload.new_assistant_personnel_id is not None
-            else None
-        )
+        resolution = _resolve_business_ownership(session, business, payload)
 
-        owner_changed = business.owner_personnel_id != payload.new_owner_personnel_id
-        assistant_changed = business.assistant_personnel_id != payload.new_assistant_personnel_id
-        if not owner_changed and not assistant_changed:
-            raise BadRequestError("归属未变化")
-
-        old_owner_id = business.owner_personnel_id
-        old_assistant_id = business.assistant_personnel_id
         business.owner_personnel_id = payload.new_owner_personnel_id
         business.assistant_personnel_id = payload.new_assistant_personnel_id
         business_repository.add_ownership_log(
-            session,
-            OwnershipChangeLog(
-                owner_type=OWNER_TYPE_BUSINESS,
-                business_id=business_id,
-                old_owner_personnel_id=old_owner_id,
-                new_owner_personnel_id=payload.new_owner_personnel_id,
-                old_assistant_personnel_id=old_assistant_id,
-                new_assistant_personnel_id=payload.new_assistant_personnel_id,
-                change_reason=payload.reason,
-                operator_id=operator.id,
-                operator_name=operator.username,
-            ),
+            session, _business_ownership_log(business, payload, resolution, operator)
         )
         # 出参组装：ID 在锁定行上已赋值（_to_out 读内存值即反映新值），姓名
         # 一侧「变更项用新 ID 的人名、未变项沿用首查名」——首查名是变更前快照
@@ -821,17 +837,19 @@ def change_ownership(
             (
                 business,
                 row[1],
-                new_owner.personnel_name if new_owner is not None else old_owner_name,
-                new_assistant.personnel_name if new_assistant is not None else old_assistant_name,
+                resolution.owner.personnel_name if resolution.owner is not None else old_owner_name,
+                resolution.assistant.personnel_name
+                if resolution.assistant is not None
+                else old_assistant_name,
             )
         )
         session.commit()
         logger.info(
             "调整业务归属 | business_id=%s | owner %s->%s | assistant %s->%s",
             business_id,
-            old_owner_id,
+            resolution.old_owner_id,
             payload.new_owner_personnel_id,
-            old_assistant_id,
+            resolution.old_assistant_id,
             payload.new_assistant_personnel_id,
         )
         return out
@@ -897,6 +915,94 @@ def list_status_logs(
     )
 
 
+@dataclass(frozen=True, slots=True)
+class _GrossProfitScan:
+    """毛利润逐行扫描结果（行 DTO、类别聚合桶、成本缺失计数）。"""
+
+    rows: list[GrossProfitRowOut]
+    category_agg: dict[str, list[Decimal]]
+    missing_count: int
+
+
+def _gross_profit_row(
+    item: BusinessItem, threshold: Fraction
+) -> tuple[GrossProfitRowOut, tuple[Decimal, Decimal] | None]:
+    """单行毛利润派生值（行 DTO + 类别聚合桶）。
+
+    聚合桶为 None 表示成本缺失行 —— 这类行不计入类别汇总，但仍要输出
+    派生列为 None 的行与缺失原因。阈值对两类行都要取（成本缺失也展示阈值）。
+    """
+    row_kwargs: dict = {
+        "item_id": item.id,
+        "product_category": item.product_category,
+        "product_name": item.product_name,
+        "spec_model": item.spec_model,
+        "quantity": str(item.quantity),
+        "sale_amount": str(item.sale_amount),
+        "threshold": str(threshold),
+    }
+    if item.cost_amount is None:
+        row_kwargs.update(
+            cost_amount=None,
+            profit_amount=None,
+            profit_rate=None,
+            converted_profit=None,
+            missing_reason="成本缺失，无法计算毛利润",
+        )
+        return GrossProfitRowOut(**row_kwargs), None
+
+    profit = item.sale_amount - item.cost_amount
+    rate = profit_rate(profit, item.sale_amount)
+    conv = converted_profit(item.sale_amount, rate, threshold)
+    row_kwargs.update(
+        cost_amount=str(item.cost_amount),
+        profit_amount=str(profit),
+        profit_rate=str(rate),
+        converted_profit=str(conv),
+        missing_reason=None,
+    )
+    return GrossProfitRowOut(**row_kwargs), (item.sale_amount, item.cost_amount)
+
+
+def _scan_gross_profit_rows(
+    session: Session, business_id: int, threshold_of: Callable[[str], Fraction]
+) -> _GrossProfitScan:
+    """逐行扫描业务明细：出行 DTO、类别聚合桶与成本缺失计数。
+
+    类别顺序沿用明细行的首次出现顺序（不排序）—— 前端汇总展示依赖该顺序。
+    """
+    rows: list[GrossProfitRowOut] = []
+    agg: dict[str, list[Decimal]] = {}
+    missing_count = 0
+    for item in business_repository.list_items(session, business_id):
+        row, bucket = _gross_profit_row(item, threshold_of(item.product_category))
+        rows.append(row)
+        if bucket is None:
+            missing_count += 1
+            continue
+        current = agg.setdefault(item.product_category, [Decimal("0"), Decimal("0")])
+        current[0] += bucket[0]
+        current[1] += bucket[1]
+    return _GrossProfitScan(rows=rows, category_agg=agg, missing_count=missing_count)
+
+
+def _gross_profit_category(
+    category: str, sale: Decimal, cost: Decimal, threshold: Fraction
+) -> GrossProfitCategoryOut:
+    """类别汇总行（派生口径与行级一致：利润率/折算复用 formula 纯函数）。"""
+    profit = sale - cost
+    rate = profit_rate(profit, sale)
+    return GrossProfitCategoryOut(
+        product_category=category,
+        sale_amount=str(sale),
+        cost_amount=str(cost),
+        profit_amount=str(profit),
+        profit_rate=str(rate),
+        threshold=str(threshold),
+        converted_profit=str(converted_profit(sale, rate, threshold)),
+    )
+
+
 def gross_profit(session: Session, business_id: int, as_of: date | None) -> GrossProfitOut:
     """毛利润数据视图（需求 2.2.4）：行级派生值 + 类别汇总，实时计算不落库。
 
@@ -915,64 +1021,15 @@ def gross_profit(session: Session, business_id: int, as_of: date | None) -> Gros
     def threshold_of(category: str) -> Fraction:
         return snapshot.fraction("product.threshold", product_category=category)
 
-    rows_out: list[GrossProfitRowOut] = []
-    missing_count = 0
-    # 类别聚合容器：{类别: [sale, cost]}，成本缺失行不计入该类别聚合
-    agg: dict[str, list[Decimal]] = {}
-    for item in business_repository.list_items(session, business_id):
-        threshold = threshold_of(item.product_category)
-        row_kwargs: dict = {
-            "item_id": item.id,
-            "product_category": item.product_category,
-            "product_name": item.product_name,
-            "spec_model": item.spec_model,
-            "quantity": str(item.quantity),
-            "sale_amount": str(item.sale_amount),
-            "threshold": str(threshold),
-        }
-        if item.cost_amount is None:
-            missing_count += 1
-            row_kwargs.update(
-                cost_amount=None,
-                profit_amount=None,
-                profit_rate=None,
-                converted_profit=None,
-                missing_reason="成本缺失，无法计算毛利润",
-            )
-        else:
-            profit = item.sale_amount - item.cost_amount
-            rate = profit_rate(profit, item.sale_amount)
-            conv = converted_profit(item.sale_amount, rate, threshold)
-            row_kwargs.update(
-                cost_amount=str(item.cost_amount),
-                profit_amount=str(profit),
-                profit_rate=str(rate),
-                converted_profit=str(conv),
-                missing_reason=None,
-            )
-            bucket = agg.setdefault(item.product_category, [Decimal("0"), Decimal("0")])
-            bucket[0] += item.sale_amount
-            bucket[1] += item.cost_amount
-        rows_out.append(GrossProfitRowOut(**row_kwargs))
-
+    scan = _scan_gross_profit_rows(session, business_id, threshold_of)
     categories = [
-        GrossProfitCategoryOut(
-            product_category=category,
-            sale_amount=str(sale),
-            cost_amount=str(cost),
-            profit_amount=str(sale - cost),
-            profit_rate=str(profit_rate(sale - cost, sale)),
-            threshold=str(threshold_of(category)),
-            converted_profit=str(
-                converted_profit(sale, profit_rate(sale - cost, sale), threshold_of(category))
-            ),
-        )
-        for category, [sale, cost] in agg.items()
+        _gross_profit_category(category, sale, cost, threshold_of(category))
+        for category, [sale, cost] in scan.category_agg.items()
     ]
     return GrossProfitOut(
         business_id=business_id,
         as_of=as_of,
-        rows=rows_out,
+        rows=scan.rows,
         categories=categories,
-        missing_count=missing_count,
+        missing_count=scan.missing_count,
     )
diff --git a/backend/app/services/customer_service.py b/backend/app/services/customer_service.py
index 0d4879d..e102a19 100644
--- a/backend/app/services/customer_service.py
+++ b/backend/app/services/customer_service.py
@@ -14,6 +14,7 @@
 """
 
 import logging
+from dataclasses import dataclass
 
 from sqlalchemy import Row
 from sqlalchemy.exc import IntegrityError
@@ -107,20 +108,8 @@ def _is_duplicate_entry(exc: IntegrityError) -> bool:
 
 def list_customers(session: Session, query: CustomerQuery) -> PageResult[CustomerOut]:
     """分页查询客户（附归属人员姓名）。"""
-    rows = customer_repository.list_customers(
-        session,
-        offset=query.offset,
-        limit=query.page_size,
-        keyword=query.keyword,
-        customer_type=query.customer_type,
-        owner_personnel_id=query.owner_personnel_id,
-    )
-    total = customer_repository.count_customers(
-        session,
-        keyword=query.keyword,
-        customer_type=query.customer_type,
-        owner_personnel_id=query.owner_personnel_id,
-    )
+    rows = customer_repository.list_customers(session, query)
+    total = customer_repository.count_customers(session, query)
     return PageResult[CustomerOut](
         records=[_to_out(customer, owner_name) for customer, owner_name in rows],
         total=total,
@@ -317,6 +306,71 @@ def delete_customer(
         logger.info("删除客户 | customer_id=%s | company=%s", customer_id, company_name)
 
 
+@dataclass(frozen=True, slots=True)
+class _CustomerOwnershipPrefetch:
+    """客户归属变更的审计前只读预取（摘要要客户名与新旧归属人名）。"""
+
+    row: Row | None
+    new_person: Personnel | None
+    old_label: str
+    new_label: str
+    customer_name: str
+
+
+def _prefetch_customer_ownership(
+    session: Session, customer_id: int, payload: CustomerOwnershipIn
+) -> _CustomerOwnershipPrefetch:
+    """审计上下文之前的只读预取（预取不写库，失败路径也要有可读摘要）。"""
+    row = customer_repository.get_with_owner_name(session, customer_id)
+    old_person = (
+        personnel_repository.get_by_id(session, row[0].owner_personnel_id)
+        if row is not None and row[0].owner_personnel_id is not None
+        else None
+    )
+    new_person = (
+        personnel_repository.get_by_id(session, payload.new_owner_personnel_id)
+        if payload.new_owner_personnel_id is not None
+        else None
+    )
+    return _CustomerOwnershipPrefetch(
+        row=row,
+        new_person=new_person,
+        old_label=old_person.personnel_name if old_person else "无",
+        new_label=new_person.personnel_name if new_person else "无",
+        customer_name=row[0].company_name if row else f"#{customer_id}",
+    )
+
+
+def _validate_customer_ownership(
+    session: Session, customer: Customer, payload: CustomerOwnershipIn
+) -> int | None:
+    """校验新归属人（须在职）并判定归属是否变化，返回变更前归属人 ID。"""
+    if payload.new_owner_personnel_id is not None:
+        _ensure_owner_personnel(session, payload.new_owner_personnel_id)
+    old_owner_id = customer.owner_personnel_id
+    if old_owner_id == payload.new_owner_personnel_id:
+        raise BadRequestError("归属未变化")
+    return old_owner_id
+
+
+def _customer_ownership_log(
+    customer: Customer,
+    payload: CustomerOwnershipIn,
+    old_owner_id: int | None,
+    operator: User,
+) -> OwnershipChangeLog:
+    """客户归属变更留痕行。"""
+    return OwnershipChangeLog(
+        owner_type=OWNER_TYPE_CUSTOMER,
+        customer_id=customer.id,
+        old_owner_personnel_id=old_owner_id,
+        new_owner_personnel_id=payload.new_owner_personnel_id,
+        change_reason=payload.reason,
+        operator_id=operator.id,
+        operator_name=operator.username,
+    )
+
+
 def change_ownership(
     session: Session,
     customer_id: int,
@@ -335,57 +389,34 @@ def change_ownership(
     """
     # 只读预取在审计上下文之前：摘要需要客户名与新旧归属人名；校验仍在
     # 上下文内进行，失败同样要留下可追溯的记录
-    row = customer_repository.get_with_owner_name(session, customer_id)
-    old_person = (
-        personnel_repository.get_by_id(session, row[0].owner_personnel_id)
-        if row is not None and row[0].owner_personnel_id is not None
-        else None
-    )
-    new_person = (
-        personnel_repository.get_by_id(session, payload.new_owner_personnel_id)
-        if payload.new_owner_personnel_id is not None
-        else None
-    )
-    old_label = old_person.personnel_name if old_person else "无"
-    new_label = new_person.personnel_name if new_person else "无"
-    customer_name = row[0].company_name if row else f"#{customer_id}"
+    prefetch = _prefetch_customer_ownership(session, customer_id, payload)
     with audit_service.audit(
         session,
         operator=operator,
         record=AuditRecord(
             module=MODULE_CUSTOMER,
             action=ACTION_UPDATE,
-            summary=f"调整客户归属 {customer_name}：{old_label} → {new_label}",
+            summary=(
+                f"调整客户归属 {prefetch.customer_name}："
+                f"{prefetch.old_label} → {prefetch.new_label}"
+            ),
             request_data=payload.model_dump(exclude_none=True),
         ),
         meta=meta,
     ):
-        if row is None:
+        if prefetch.row is None:
             raise NotFoundError("客户不存在")
-        customer, _ = row
+        customer, _ = prefetch.row
 
-        if payload.new_owner_personnel_id is not None:
-            _ensure_owner_personnel(session, payload.new_owner_personnel_id)
-        old_owner_id = customer.owner_personnel_id
-        if old_owner_id == payload.new_owner_personnel_id:
-            raise BadRequestError("归属未变化")
+        old_owner_id = _validate_customer_ownership(session, customer, payload)
 
         # 主表更新与留痕插入同事务：只在本方法内一次 commit（见模块 docstring）
         customer.owner_personnel_id = payload.new_owner_personnel_id
         customer_repository.add_ownership_log(
-            session,
-            OwnershipChangeLog(
-                owner_type=OWNER_TYPE_CUSTOMER,
-                customer_id=customer.id,
-                old_owner_personnel_id=old_owner_id,
-                new_owner_personnel_id=payload.new_owner_personnel_id,
-                change_reason=payload.reason,
-                operator_id=operator.id,
-                operator_name=operator.username,
-            ),
+            session, _customer_ownership_log(customer, payload, old_owner_id, operator)
         )
         # 响应在 commit 之前构造（同 user_service 顶部说明）；归属人名以新值为准
-        out = _to_out(customer, new_person.personnel_name if new_person else None)
+        out = _to_out(customer, prefetch.new_person.personnel_name if prefetch.new_person else None)
         session.commit()
         logger.info(
             "调整客户归属 | customer_id=%s | old=%s | new=%s",
diff --git a/backend/app/services/dashboard_service.py b/backend/app/services/dashboard_service.py
index e96fe89..546416f 100644
--- a/backend/app/services/dashboard_service.py
+++ b/backend/app/services/dashboard_service.py
@@ -168,9 +168,7 @@ def _warning_params(session: Session) -> dict:
     return {
         "enabled": snapshot.get("warning.enabled") == "1",
         "receipt_overdue_days": int(parse_fraction(snapshot.get("warning.receipt.overdue.days"))),
-        "approval_timeout_days": int(
-            parse_fraction(snapshot.get("warning.approval.timeout.days"))
-        ),
+        "approval_timeout_days": int(parse_fraction(snapshot.get("warning.approval.timeout.days"))),
     }
 
 
@@ -253,9 +251,11 @@ def _activities(session: Session) -> list[dict]:
     """最近动态（FR-036）：以操作日志为唯一来源（写操作即留痕）。"""
     logs = operation_log_repository.list_logs(
         session,
-        offset=0,
-        limit=_ACTIVITY_LIMIT,
-        filters=LogFilters(),
+        operation_log_repository.OperationLogListQuery(
+            filters=LogFilters(),
+            offset=0,
+            limit=_ACTIVITY_LIMIT,
+        ),
     )
     return [
         {
diff --git a/backend/app/services/log_service.py b/backend/app/services/log_service.py
index 3ff945f..59b7201 100644
--- a/backend/app/services/log_service.py
+++ b/backend/app/services/log_service.py
@@ -22,18 +22,15 @@ from app.schemas.log import (
 
 def list_operation_logs(session: Session, query: OperationLogQuery) -> PageResult[OperationLogOut]:
     """分页查询操作日志（列表不含 request_data，详情走 `get_operation_log`）。"""
-    filters = _filters_from(query)
-    logs = operation_log_repository.list_logs(
-        session,
+    repo_query = operation_log_repository.OperationLogListQuery(
+        filters=_filters_from(query),
         offset=query.offset,
         limit=query.page_size,
-        filters=filters,
         module=query.module,
         action=query.action,
     )
-    total = operation_log_repository.count_logs(
-        session, filters=filters, module=query.module, action=query.action
-    )
+    logs = operation_log_repository.list_logs(session, repo_query)
+    total = operation_log_repository.count_logs(session, repo_query)
     return PageResult[OperationLogOut](
         records=[OperationLogOut.model_validate(log) for log in logs],
         total=total,
@@ -55,15 +52,14 @@ def get_operation_log(session: Session, log_id: int) -> OperationLogDetailOut:
 
 def list_login_logs(session: Session, query: LoginLogQuery) -> PageResult[LoginLogOut]:
     """分页查询登录日志。"""
-    filters = _filters_from(query)
-    logs = login_log_repository.list_logs(
-        session,
+    repo_query = login_log_repository.LoginLogListQuery(
+        filters=_filters_from(query),
         offset=query.offset,
         limit=query.page_size,
-        filters=filters,
         ip=query.ip,
     )
-    total = login_log_repository.count_logs(session, filters=filters, ip=query.ip)
+    logs = login_log_repository.list_logs(session, repo_query)
+    total = login_log_repository.count_logs(session, repo_query)
     return PageResult[LoginLogOut](
         records=[LoginLogOut.model_validate(log) for log in logs],
         total=total,
diff --git a/backend/app/services/param_registry.py b/backend/app/services/param_registry.py
index 4184fb8..103d720 100644
--- a/backend/app/services/param_registry.py
+++ b/backend/app/services/param_registry.py
@@ -341,9 +341,7 @@ PARAM_REGISTRY: dict[str, ParamSpec] = {
             default_value="90",
             value_min="1",
             value_max=None,
-            description=(
-                "发货后超过该天数仍未回款的发货批次触发预警（澄清 #10 建议默认 90 天）"
-            ),
+            description=("发货后超过该天数仍未回款的发货批次触发预警（澄清 #10 建议默认 90 天）"),
             sort_order=22,
         ),
         ParamSpec(
@@ -357,9 +355,7 @@ PARAM_REGISTRY: dict[str, ParamSpec] = {
             default_value="3",
             value_min="1",
             value_max=None,
-            description=(
-                "待审批单提交后超过该天数未处理触发预警（澄清 #10 建议默认 3 天）"
-            ),
+            description=("待审批单提交后超过该天数未处理触发预警（澄清 #10 建议默认 3 天）"),
             sort_order=23,
         ),
     ]
diff --git a/backend/app/services/param_service.py b/backend/app/services/param_service.py
index 5cdcabc..d82c3df 100644
--- a/backend/app/services/param_service.py
+++ b/backend/app/services/param_service.py
@@ -6,7 +6,7 @@ ParamSnapshot 是 M3 核算与计算预览共用的输入契约：按业务日
 """
 
 import logging
-from collections.abc import Iterable
+from collections.abc import Callable, Iterable
 from dataclasses import dataclass
 from datetime import date
 from decimal import Decimal
@@ -379,16 +379,8 @@ def get_history(
     definition = param_repository.get_definition_by_code(session, param_code)
     if definition is None:
         raise NotFoundError("参数不存在")
-    rows = param_repository.list_history(
-        session,
-        definition.id,
-        offset=query.offset,
-        limit=query.page_size,
-        dimension_value=query.dimension_value,
-    )
-    total = param_repository.count_history(
-        session, definition.id, dimension_value=query.dimension_value
-    )
+    rows = param_repository.list_history(session, definition.id, query)
+    total = param_repository.count_history(session, definition.id, query)
     return PageResult[ParamHistoryOut](
         records=[
             ParamHistoryOut(
@@ -410,16 +402,8 @@ def get_history(
 
 def list_batches(session: Session, query: ParamBatchQuery) -> PageResult[ParamBatchOut]:
     """变更批次分页（按创建时间倒序）。"""
-    batches = param_repository.list_batches(
-        session,
-        offset=query.offset,
-        limit=query.page_size,
-        start_date=query.start_date,
-        end_date=query.end_date,
-    )
-    total = param_repository.count_batches(
-        session, start_date=query.start_date, end_date=query.end_date
-    )
+    batches = param_repository.list_batches(session, query)
+    total = param_repository.count_batches(session, query)
     return PageResult[ParamBatchOut](
         records=[
             ParamBatchOut(
@@ -509,19 +493,66 @@ def compare(session: Session, *, from_date: date, to_date: date) -> list[ParamDi
     return out
 
 
-def _preview_rate(
-    session: Session,
-    payload: ParamPreviewIn,
-    snapshot: ParamSnapshot,
-    days: int,
-    take,
-) -> tuple[Fraction, str, Decimal | None]:
+@dataclass(frozen=True, slots=True)
+class _PreviewContext:
+    """预览上下文（会话、入参、参数取值闭包）。
+
+    ``take`` 闭包同时负责取数与记录「用了哪些参数」——**调用顺序即
+    ``params_used`` 的展示顺序**，拆分时不得调换各段的先后。
+    """
+
+    session: Session
+    payload: ParamPreviewIn
+    take: Callable[[str], Fraction]
+
+
+@dataclass(frozen=True, slots=True)
+class _PreviewRateDecision:
+    """计提比例判定（回款天数、纯比例、人读说明、绩效基数覆盖值）。"""
+
+    days: int
+    rate: Fraction
+    rate_text: str
+    base_override: Decimal | None
+
+
+@dataclass(frozen=True, slots=True)
+class _PreviewEarning:
+    """折算与绩效（毛利润率、折算额、绩效额）。"""
+
+    profit_rate: Fraction
+    converted: Decimal
+    performance: Decimal
+
+
+@dataclass(frozen=True, slots=True)
+class _PreviewBarter:
+    """以物抵顶（是否适用、折算后绩效、计发基数）。"""
+
+    applied: bool
+    after_barter: Decimal | None
+    base_amount: Decimal
+
+
+@dataclass(frozen=True, slots=True)
+class _PreviewPayment:
+    """支付环节（起付判定、R 权重、权重、应付金额）。"""
+
+    start_ok: bool
+    weight_r: int
+    weight: Fraction
+    payable: Decimal
+
+
+def _preview_rate(ctx: _PreviewContext, days: int) -> tuple[Fraction, str, Decimal | None]:
     """按岗位模板分派计提比例，返回（纯比例, 人读说明, 绩效基数覆盖值）。
 
     基数覆盖值非 None 表示该模板的绩效基数不是折算毛利润额（如信息员按
     「本人销售额」计提），预览层直接采用并跳过折算与阈值。
     聚合模板（营销经理 12.5%+5%）依赖月度团队数据，不支持单笔预览。
     """
+    payload = ctx.payload
+    take = ctx.take
     if payload.customer_type == "enterprise":
         # 存储口径为 ‰ 数（如 "3"），除以 1000 转纯比例 —— performance_amount
         # 按纯比例相乘，漏换算会把绩效金额放大 1000 倍（与业务员分支同防线）
@@ -531,7 +562,7 @@ def _preview_rate(
     role_code = payload.business_role
     template = "TPL_SALESPERSON_DECAY"
     if role_code:
-        role = business_role_repository.get_by_code(session, role_code)
+        role = business_role_repository.get_by_code(ctx.session, role_code)
         if role is None:
             raise BadRequestError(f"岗位 {role_code!r} 不存在")
         template = role.calc_template
@@ -561,12 +592,73 @@ def _preview_rate(
     raise BadRequestError(f"公式模板 {template} 暂不支持预览")
 
 
+def _preview_rate_decision(ctx: _PreviewContext) -> _PreviewRateDecision:
+    """产品类别校验 + 回款天数 + 计提比例判定。"""
+    payload = ctx.payload
+    if payload.product_category and payload.product_category not in PRODUCT_CATEGORIES:
+        raise BadRequestError(
+            f"产品类别 {payload.product_category!r} 不存在，可选：municipal_water / rail_transit"
+        )
+    days = formula.payment_days(payload.ship_date, payload.receipt_date)
+    rate, rate_text, base_override = _preview_rate(ctx, days)
+    return _PreviewRateDecision(
+        days=days, rate=rate, rate_text=rate_text, base_override=base_override
+    )
+
+
+def _preview_earning(ctx: _PreviewContext, decision: _PreviewRateDecision) -> _PreviewEarning:
+    """毛利润率 → 折算额（阈值）→ 绩效额（比例）。"""
+    payload = ctx.payload
+    revenue = Decimal(payload.revenue)
+    profit = Decimal(payload.profit)
+    profit_rate = formula.profit_rate(profit, revenue)
+    if decision.base_override is not None:
+        # 基数覆盖（信息员按本人销售额计提）：不走折算与阈值
+        converted = decision.base_override
+    else:
+        threshold = ctx.take("product.threshold")
+        converted = formula.converted_profit(revenue, profit_rate, threshold)
+    performance = formula.performance_amount(converted, decision.rate)
+    return _PreviewEarning(profit_rate=profit_rate, converted=converted, performance=performance)
+
+
+def _preview_barter(ctx: _PreviewContext, performance: Decimal) -> _PreviewBarter:
+    """以物抵顶折算（按应发绩效的固定比例折算，比例来自参数）。"""
+    applied = ctx.payload.barter
+    after_barter: Decimal | None = None
+    if applied:
+        discount = ctx.take("barter.discount")
+        after_barter = formula.apply_barter(performance, discount)
+    base_amount = after_barter if applied else performance
+    return _PreviewBarter(applied=applied, after_barter=after_barter, base_amount=base_amount)
+
+
+def _preview_payment(
+    ctx: _PreviewContext, received: Decimal, revenue: Decimal, base_amount: Decimal
+) -> _PreviewPayment:
+    """起付比例判定 + R 权重 + 应付金额。"""
+    start_ratio = ctx.take("payment.start.ratio")
+    start_ok = formula.start_payment_ok(received, revenue, start_ratio)
+    # R 钳制已提取为纯函数 clamp_weight_r（④ 核算与预览共用同一实现）
+    weight_r = formula.clamp_weight_r(received, revenue)
+    weight = formula.payment_weight(weight_r)
+    payable = to_decimal(Fraction(base_amount) * weight)
+    return _PreviewPayment(start_ok=start_ok, weight_r=weight_r, weight=weight, payable=payable)
+
+
+def _preview_probation_note(ctx: _PreviewContext) -> str | None:
+    """试用期说明（绩效照算不发放，年累计超阈值的部分年底一次性结算）。"""
+    if not ctx.payload.probation:
+        return None
+    annual = ctx.take("probation.annual.threshold")
+    return f"试用期绩效照算不发放；年累计超 {to_decimal(annual)} 元部分年底一次性结算"
+
+
 def preview(session: Session, payload: ParamPreviewIn) -> ParamPreviewOut:
     """计算预览：入参（发货/回款/金额/岗位）→ 公式纯函数 → 出参（含所用参数）。"""
     # 参数按回款到账日取值（对照表 #11）
     snapshot = resolve_snapshot(session, payload.receipt_date)
     revenue = Decimal(payload.revenue)
-    profit = Decimal(payload.profit)
     received = Decimal(payload.received_amount)
     used: list[ParamUsedOut] = []
 
@@ -589,55 +681,29 @@ def preview(session: Session, payload: ParamPreviewIn) -> ParamPreviewOut:
         )
         return parse_fraction(raw)
 
-    if payload.product_category and payload.product_category not in PRODUCT_CATEGORIES:
-        raise BadRequestError(
-            f"产品类别 {payload.product_category!r} 不存在，可选：municipal_water / rail_transit"
-        )
-
-    days = formula.payment_days(payload.ship_date, payload.receipt_date)
-    rate, rate_text, base_override = _preview_rate(session, payload, snapshot, days, take)
-
-    profit_rate = formula.profit_rate(profit, revenue)
-    if base_override is not None:
-        # 基数覆盖（信息员按本人销售额计提）：不走折算与阈值
-        converted = base_override
-    else:
-        threshold = take("product.threshold")
-        converted = formula.converted_profit(revenue, profit_rate, threshold)
-    performance = formula.performance_amount(converted, rate)
-
-    barter_applied = payload.barter
-    after_barter: Decimal | None = None
-    if barter_applied:
-        discount = take("barter.discount")
-        after_barter = formula.apply_barter(performance, discount)
-
-    start_ratio = take("payment.start.ratio")
-    start_ok = formula.start_payment_ok(received, revenue, start_ratio)
-    # R 钳制已提取为纯函数 clamp_weight_r（④ 核算与预览共用同一实现）
-    weight_r = formula.clamp_weight_r(received, revenue)
-    weight = formula.payment_weight(weight_r)
-    base_amount = after_barter if barter_applied else performance
-    payable = to_decimal(Fraction(base_amount) * weight)
-
-    probation_note: str | None = None
-    if payload.probation:
-        annual = take("probation.annual.threshold")
-        probation_note = f"试用期绩效照算不发放；年累计超 {to_decimal(annual)} 元部分年底一次性结算"
+    ctx = _PreviewContext(session=session, payload=payload, take=take)
+    # 各段顺序固定 —— take 的调用顺序即 params_used 的展示顺序
+    decision = _preview_rate_decision(ctx)
+    earning = _preview_earning(ctx, decision)
+    barter = _preview_barter(ctx, earning.performance)
+    payment = _preview_payment(ctx, received, revenue, barter.base_amount)
+    probation_note = _preview_probation_note(ctx)
 
     return ParamPreviewOut(
-        payment_days=days,
-        rate=str(rate),
-        rate_text=rate_text,
-        profit_rate=str(profit_rate),
-        converted_profit=str(converted),
-        performance_amount=str(performance),
-        barter_applied=barter_applied,
-        performance_after_barter=str(after_barter) if after_barter is not None else None,
-        start_payment_ok=start_ok,
-        payment_weight_r=weight_r,
-        payment_weight=str(weight),
-        payable_amount=str(payable),
+        payment_days=decision.days,
+        rate=str(decision.rate),
+        rate_text=decision.rate_text,
+        profit_rate=str(earning.profit_rate),
+        converted_profit=str(earning.converted),
+        performance_amount=str(earning.performance),
+        barter_applied=barter.applied,
+        performance_after_barter=(
+            str(barter.after_barter) if barter.after_barter is not None else None
+        ),
+        start_payment_ok=payment.start_ok,
+        payment_weight_r=payment.weight_r,
+        payment_weight=str(payment.weight),
+        payable_amount=str(payment.payable),
         probation_note=probation_note,
         params_used=used,
     )
diff --git a/backend/app/services/personnel_service.py b/backend/app/services/personnel_service.py
index c88df6f..a5809b6 100644
--- a/backend/app/services/personnel_service.py
+++ b/backend/app/services/personnel_service.py
@@ -70,20 +70,8 @@ def ensure_active(session: Session, personnel_id: int, *, label: str) -> Personn
 
 def list_personnel(session: Session, query: PersonnelQuery) -> PageResult[PersonnelOut]:
     """分页查询人员（含离职）。"""
-    rows = personnel_repository.list_personnel(
-        session,
-        offset=query.offset,
-        limit=query.page_size,
-        keyword=query.keyword,
-        status=query.status,
-        business_role_id=query.business_role_id,
-    )
-    total = personnel_repository.count_personnel(
-        session,
-        keyword=query.keyword,
-        status=query.status,
-        business_role_id=query.business_role_id,
-    )
+    rows = personnel_repository.list_personnel(session, query)
+    total = personnel_repository.count_personnel(session, query)
     return PageResult[PersonnelOut](
         records=[
             _to_out(personnel, role_name, manager_name)
diff --git a/backend/app/services/product_service.py b/backend/app/services/product_service.py
index 38e70bb..7b9fa71 100644
--- a/backend/app/services/product_service.py
+++ b/backend/app/services/product_service.py
@@ -99,20 +99,8 @@ def _raise_if_duplicate(exc: IntegrityError) -> None:
 
 def list_products(session: Session, query: ProductQuery) -> PageResult[ProductOut]:
     """分页查询产品（含停用）。"""
-    products = product_repository.list_products(
-        session,
-        offset=query.offset,
-        limit=query.page_size,
-        keyword=query.keyword,
-        product_category=query.product_category,
-        status=query.status,
-    )
-    total = product_repository.count_products(
-        session,
-        keyword=query.keyword,
-        product_category=query.product_category,
-        status=query.status,
-    )
+    products = product_repository.list_products(session, query)
+    total = product_repository.count_products(session, query)
     return PageResult[ProductOut](
         records=[_to_out(product) for product in products],
         total=total,
diff --git a/backend/app/services/receipt_service.py b/backend/app/services/receipt_service.py
index 57f15d5..e6e580b 100644
--- a/backend/app/services/receipt_service.py
+++ b/backend/app/services/receipt_service.py
@@ -23,6 +23,7 @@
 """
 
 import logging
+from dataclasses import dataclass
 from datetime import date
 from decimal import Decimal
 
@@ -39,7 +40,7 @@ from app.core.constants import (
 from app.core.db_errors import is_duplicate_entry
 from app.core.exceptions import BadRequestError, NotFoundError
 from app.core.response import PageResult
-from app.models import Receipt
+from app.models import Business, Receipt
 from app.models.business import BUSINESS_STATUS_ACTIVE
 from app.models.receipt import SETTLEMENT_DONE, SETTLEMENT_PENDING
 from app.repositories import business_repository, receipt_repository
@@ -129,24 +130,8 @@ def _ensure_business_writable(session: Session, business_id: int, action: str):
 
 def list_receipts(session: Session, query: ReceiptQuery) -> PageResult[ReceiptOut]:
     """分页查询回款（附业务号/发货单号与 payment_days 派生值）。"""
-    rows = receipt_repository.list_receipts(
-        session,
-        offset=query.offset,
-        limit=query.page_size,
-        business_id=query.business_id,
-        date_from=query.date_from,
-        date_to=query.date_to,
-        is_barter=query.is_barter,
-        settlement_status=query.settlement_status,
-    )
-    total = receipt_repository.count_receipts(
-        session,
-        business_id=query.business_id,
-        date_from=query.date_from,
-        date_to=query.date_to,
-        is_barter=query.is_barter,
-        settlement_status=query.settlement_status,
-    )
+    rows = receipt_repository.list_receipts(session, query)
+    total = receipt_repository.count_receipts(session, query)
     return PageResult[ReceiptOut](
         records=[_to_out(row) for row in rows],
         total=total,
@@ -166,6 +151,57 @@ def get_detail(session: Session, receipt_id: int) -> ReceiptOut:
     return _to_out(row)
 
 
+@dataclass(frozen=True, slots=True)
+class _CreatedReceipt:
+    """建行结果（出参 + 固化的 ID/批次号，commit 后不再触碰 ORM 属性）。"""
+
+    out: ReceiptOut
+    receipt_id: int
+    batch_no: str
+
+
+def _create_receipt_once(
+    session: Session,
+    payload: ReceiptCreateIn,
+    business: Business | None,
+    batch_no: str,
+) -> _CreatedReceipt:
+    """一次建行尝试：准入校验 → 建行 → 重算合计 → 出参。
+
+    **不 commit**：提交与回滚由调用方（重试循环）统一控制，保证「一次尝试
+    = 一个事务边界」。
+    """
+    if business is None:
+        raise NotFoundError("业务不存在")
+    if business.status != BUSINESS_STATUS_ACTIVE:
+        raise BadRequestError("业务已完成或已终止，禁止新增回款")
+    # 月结锁定拦截（2026-10-15 补充需求）：比需求原文「提示」取硬，
+    # 一期安全侧（策略点注释）；未指定日期时以当天月近似
+    from app.services import settlement_service
+
+    _lock_month = (payload.receipt_date or date.today()).strftime("%Y-%m")
+    settlement_service.ensure_month_not_locked(session, _lock_month)
+    shipment_no, ship_date = _ensure_shipment_of_business(
+        session, payload.shipment_id, payload.business_id
+    )
+    receipt = receipt_repository.create(
+        session,
+        Receipt(
+            receipt_batch_no=batch_no,
+            business_id=payload.business_id,
+            shipment_id=payload.shipment_id,
+            receipt_date=payload.receipt_date,
+            amount=Decimal(payload.amount),
+            is_barter=payload.is_barter,
+            settlement_status=SETTLEMENT_PENDING,
+            remark=payload.remark,
+        ),
+    )
+    _recalc_totals(session, business)
+    out = _to_out((receipt, business.business_no, shipment_no, ship_date))
+    return _CreatedReceipt(out=out, receipt_id=receipt.id, batch_no=batch_no)
+
+
 def create_receipt(
     session: Session,
     payload: ReceiptCreateIn,
@@ -206,35 +242,7 @@ def create_receipt(
         # 锁业务行与生成批次号——成功/最终失败各只记一条审计，无递归
         for attempt in range(_NO_RETRY + 1):
             try:
-                if business is None:
-                    raise NotFoundError("业务不存在")
-                if business.status != BUSINESS_STATUS_ACTIVE:
-                    raise BadRequestError("业务已完成或已终止，禁止新增回款")
-                # 月结锁定拦截（2026-10-15 补充需求）：比需求原文「提示」取硬，
-                # 一期安全侧（策略点注释）；未指定日期时以当天月近似
-                from app.services import settlement_service
-
-                _lock_month = (payload.receipt_date or date.today()).strftime("%Y-%m")
-                settlement_service.ensure_month_not_locked(session, _lock_month)
-                shipment_no, ship_date = _ensure_shipment_of_business(
-                    session, payload.shipment_id, payload.business_id
-                )
-                receipt = receipt_repository.create(
-                    session,
-                    Receipt(
-                        receipt_batch_no=batch_no,
-                        business_id=payload.business_id,
-                        shipment_id=payload.shipment_id,
-                        receipt_date=payload.receipt_date,
-                        amount=Decimal(payload.amount),
-                        is_barter=payload.is_barter,
-                        settlement_status=SETTLEMENT_PENDING,
-                        remark=payload.remark,
-                    ),
-                )
-                _recalc_totals(session, business)
-                out = _to_out((receipt, business.business_no, shipment_no, ship_date))
-                new_id, new_no = receipt.id, batch_no  # 固化：commit 后不碰 ORM
+                created = _create_receipt_once(session, payload, business, batch_no)
                 session.commit()
                 break
             except IntegrityError as exc:
@@ -249,11 +257,11 @@ def create_receipt(
 
         logger.info(
             "新增回款 | receipt_id=%s | no=%s | business_id=%s",
-            new_id,
-            new_no,
+            created.receipt_id,
+            created.batch_no,
             payload.business_id,
         )
-        return out
+        return created.out
 
 
 def update_receipt(
diff --git a/backend/app/services/report_export.py b/backend/app/services/report_export.py
index a99af0f..b6059cf 100644
--- a/backend/app/services/report_export.py
+++ b/backend/app/services/report_export.py
@@ -36,7 +36,6 @@ _DATE_FORMAT = "YYYY-MM-DD"
 _FORMULA_PREFIXES = ("=", "+", "-", "@", "\t")
 
 
-
 @dataclass(frozen=True, slots=True)
 class _Column:
     """列定义：key 取行 dict 字段，kind 决定单元格类型与格式。"""
@@ -403,7 +402,13 @@ def _write_department_document(
     company_share_count = int(company.get("company_share_count") or 0)
 
     _write_document_header(
-        sheet, len(_DEPARTMENT_WIDTHS), company_name, title, filters, template_version, operator_name
+        sheet,
+        len(_DEPARTMENT_WIDTHS),
+        company_name,
+        title,
+        filters,
+        template_version,
+        operator_name,
     )
 
     _write_block(
@@ -439,16 +444,12 @@ def _write_department_document(
                 "manager_name": "合计",
                 "member_count": sum(team["member_count"] for team in teams),
                 "team_base": _sum_amount(team["team_base"] for team in teams),
-                "team_share_amount": _sum_amount(
-                    team["team_share_amount"] for team in teams
-                ),
+                "team_share_amount": _sum_amount(team["team_share_amount"] for team in teams),
                 "team_subtotal": _sum_amount(team["team_subtotal"] for team in teams),
                 # 未生成月结时全部为 None，合计也留空（不能显示 0.00 误导）
                 "frozen_amount": (
                     _sum_amount(
-                        team["frozen_amount"]
-                        for team in teams
-                        if team["frozen_amount"] is not None
+                        team["frozen_amount"] for team in teams if team["frozen_amount"] is not None
                     )
                     if any(team["frozen_amount"] is not None for team in teams)
                     else None
diff --git a/backend/app/services/report_service.py b/backend/app/services/report_service.py
index b4b35a5..ea44a74 100644
--- a/backend/app/services/report_service.py
+++ b/backend/app/services/report_service.py
@@ -103,10 +103,7 @@ def _person_row(row: PerformanceLedger, role_names: dict[str, str]) -> dict:
 
 def _role_names(session) -> dict[str, str]:
     """岗位编码 → 名称一次查询（明细与部门行共用，避免 N+1）。"""
-    return {
-        r.role_code: r.role_name
-        for r in session.execute(select(BusinessRole)).scalars()
-    }
+    return {r.role_code: r.role_name for r in session.execute(select(BusinessRole)).scalars()}
 
 
 def _dec(value) -> Decimal | None:
@@ -178,11 +175,7 @@ def department_summary(session, query) -> dict:
         extra_ids = {
             target
             for pid in adjustments
-            if (
-                target := _adjustment_target(
-                    pid, member_map, manager_of, manager_role, include_mp
-                )
-            )
+            if (target := _adjustment_target(pid, member_map, manager_of, manager_role, include_mp))
             is not None
         } - set(manager_ids)
         if extra_ids:
@@ -193,9 +186,7 @@ def department_summary(session, query) -> dict:
         for team in teams_all
         if query.manager_id is None or team["manager_id"] == query.manager_id
     ]
-    unassigned_all = _unassigned_rows(
-        member_map, manager_of, manager_role, adjustments, include_mp
-    )
+    unassigned_all = _unassigned_rows(member_map, manager_of, manager_role, adjustments, include_mp)
     # 「仅看某经理团队」时未分配桶不随展示：全公司无经理人员与该团队无关
     unassigned = [] if query.manager_id is not None else unassigned_all
     company_share_amount = _company_share(company_base, share.company_pct)
@@ -335,9 +326,7 @@ def _reconciliation(
         (m["adjust_net"] for team in teams for m in team["members"]), Decimal("0")
     ) + sum((row["adjust_net"] for row in unassigned), Decimal("0"))
     team_base_total = sum((team["team_base"] for team in teams), Decimal("0"))
-    unassigned_performance = sum(
-        (row["performance_sum"] for row in unassigned), Decimal("0")
-    )
+    unassigned_performance = sum((row["performance_sum"] for row in unassigned), Decimal("0"))
     payable_total = (
         team_member_payable + unassigned_payable + aggregate_payable_total + adjust_net_total
     )
@@ -378,9 +367,7 @@ def _reconciliation(
 
 def _team_share(team_base: Decimal, team_pct: Fraction) -> Decimal:
     """12.5% 团队分成：全公司部分传 0，仅取 team_pct 项。"""
-    return manager_team_share(
-        team_base, Decimal("0"), team_pct=team_pct, company_pct=Fraction(0)
-    )
+    return manager_team_share(team_base, Decimal("0"), team_pct=team_pct, company_pct=Fraction(0))
 
 
 def _company_share(company_base: Decimal, company_pct: Fraction) -> Decimal:
@@ -483,12 +470,8 @@ def _team_relations(
     referenced = {p.manager_id for p in people if p.manager_id is not None}
     missing = referenced - seed
     if missing:
-        people.extend(
-            session.execute(select(Personnel).where(Personnel.id.in_(missing))).scalars()
-        )
-    role_by_id = {
-        r.id: r.role_code for r in session.execute(select(BusinessRole)).scalars()
-    }
+        people.extend(session.execute(select(Personnel).where(Personnel.id.in_(missing))).scalars())
+    role_by_id = {r.id: r.role_code for r in session.execute(select(BusinessRole)).scalars()}
     manager_of = {p.id: p.manager_id for p in people}
     manager_role = {p.id: role_by_id.get(p.business_role_id, "") for p in people}
     person_info = {
@@ -619,9 +602,7 @@ def _member_rows(
                 )
             )
         personal_in_team = (
-            pid == manager_id
-            and include_manager_personal
-            and member["has_manager_personal"]
+            pid == manager_id and include_manager_personal and member["has_manager_personal"]
         )
         if personal_in_team:
             rows.append(
@@ -893,9 +874,7 @@ def _group_rows(rows) -> list[dict]:
 def payment_detail(session, query) -> dict:
     """区间下钻分页（bucket 未注册 → 400）。"""
     bucket = _bucket_by_code(query.bucket)
-    rows = report_repository.list_payment_detail(
-        session, query, bucket, offset=query.offset, limit=query.page_size
-    )
+    rows = report_repository.list_payment_detail(session, query, bucket)
     return {
         "list": [_payment_row(r) for r in rows],
         "total": report_repository.count_payment_detail(session, query, bucket),
diff --git a/backend/app/services/role_service.py b/backend/app/services/role_service.py
index f86fbb2..71c2065 100644
--- a/backend/app/services/role_service.py
+++ b/backend/app/services/role_service.py
@@ -68,14 +68,8 @@ def list_roles(session: Session, query: RoleQuery) -> PageResult[RoleManageOut]:
     与用户表单下拉（`/users/meta/roles`）口径不同：**管理列表含停用角色**，
     下拉只给启用项。
     """
-    roles = role_repository.list_roles(
-        session,
-        offset=query.offset,
-        limit=query.page_size,
-        keyword=query.keyword,
-        status=query.status,
-    )
-    total = role_repository.count_roles(session, keyword=query.keyword, status=query.status)
+    roles = role_repository.list_roles(session, query)
+    total = role_repository.count_roles(session, query)
     counts = role_repository.count_users_by_role(session, [role.id for role in roles])
     return PageResult[RoleManageOut](
         records=[_to_out(role, counts.get(role.id, 0)) for role in roles],
diff --git a/backend/app/services/shipment_service.py b/backend/app/services/shipment_service.py
index 04eebf0..1ce99dc 100644
--- a/backend/app/services/shipment_service.py
+++ b/backend/app/services/shipment_service.py
@@ -157,20 +157,8 @@ def _is_duplicate_entry(exc: IntegrityError) -> bool:
 
 def list_shipments(session: Session, query: ShipmentQuery) -> PageResult[ShipmentOut]:
     """分页查询发货单（附业务号；business_id 可空=全量，一期按业务维度查看为主）。"""
-    rows = shipment_repository.list_shipments(
-        session,
-        offset=query.offset,
-        limit=query.page_size,
-        business_id=query.business_id,
-        ship_date_from=query.ship_date_from,
-        ship_date_to=query.ship_date_to,
-    )
-    total = shipment_repository.count_shipments(
-        session,
-        business_id=query.business_id,
-        ship_date_from=query.ship_date_from,
-        ship_date_to=query.ship_date_to,
-    )
+    rows = shipment_repository.list_shipments(session, query)
+    total = shipment_repository.count_shipments(session, query)
     return PageResult[ShipmentOut](
         records=[_to_out(row) for row in rows],
         total=total,
diff --git a/backend/app/services/user_service.py b/backend/app/services/user_service.py
index f679e00..e0405b5 100644
--- a/backend/app/services/user_service.py
+++ b/backend/app/services/user_service.py
@@ -68,22 +68,8 @@ def _to_out(user: User) -> UserOut:
 
 def list_users(session: Session, query: UserQuery) -> PageResult[UserOut]:
     """分页查询用户列表。"""
-    users = user_repository.list_users(
-        session,
-        offset=query.offset,
-        limit=query.page_size,
-        username=query.username,
-        real_name=query.real_name,
-        status=query.status,
-        role_code=query.role_code,
-    )
-    total = user_repository.count_users(
-        session,
-        username=query.username,
-        real_name=query.real_name,
-        status=query.status,
-        role_code=query.role_code,
-    )
+    users = user_repository.list_users(session, query)
+    total = user_repository.count_users(session, query)
     return PageResult[UserOut](
         records=[_to_out(user) for user in users],
         total=total,
diff --git a/backend/tests/README.md b/backend/tests/README.md
index 7b96360..35609df 100644
--- a/backend/tests/README.md
+++ b/backend/tests/README.md
@@ -205,6 +205,7 @@ uv run pytest tests/integration -q   # 仅集成
 | 2026-10-20 | 全量回归 | 862 | ✅ 862 passed（11:32） | **工作台/报表修复提交后的全量复核**；e2e 三脚本复跑：role_home 4/4、report 12/12、calc_settlement 8/8；calc 审批开关已还原为 1，无 `e2e_role_*` 残留 |
 | 2026-10-20 | code-review 遗留 2 项 | 全量 875 | ✅ 875 passed（12:13） | **方案版本化 + 审批内容快照**（V11 迁移）：审批单 `version_no`/`snapshot_json`，方案行 `is_active` 版本化；编辑/驳回不再物理删方案；历史单无快照时 live 回退并显式标注。新增集成 7 + 单元 6；定点审批/月结/核算/业务/报表全绿；ruff 与 type-check/build 绿 |
 | 2026-10-20 | 遗留项自评审收口 | 全量 891 | ✅ 891 passed（10:23） | **列表 defer 快照列**（审批列表/工作台待办/超时预警）+ **送审内容未变不重提、方案未变不新增版本**（两个 version_no 语义解耦）+ **月结提交加行锁**。新增单元 10 + 集成 5；单元 525、集成定点 96 全绿 |
+| 2026-10-22 | 清理批次（散参数 DTO 化） | 全量 900 | ✅ 900 passed（11:08） | **仓储层散参数收敛为查询 DTO**（`list_*(session, query)` 取代 keyword/status/offset/limit 四参数），同步调整服务层与测试调用；单元 526 绿、ruff 绿、前端 type-check 与 build 绿 |
 | 2026-10-19 | 单元全量 + 报表定点 | 484 | ✅ 484 passed（45.94s）；报表单元 11 例 0.70s | **M4 报表模块收尾**（前端四页/打印视图/导出与单元测试补齐）。ruff 绿；前端 `pnpm type-check` 绿。报表集成 7 例记录见上节 |
 | 2026-10-19 | 全部 | 806 | ✅ 806 passed（7:59） | **M4 ⑤ 报表统计全量回归**（报表新增 18 例：集成 7 + 单元 11；另含并行会话并入的 M3 前端修复与字体合规）。ruff 绿。e2e：`calc_settlement_flow` 8 步由并行会话跑通；`report_flow.py` 待 `E2E_ADMIN_PASSWORD` 运行 |
 
diff --git a/backend/tests/integration/test_approval_flow.py b/backend/tests/integration/test_approval_flow.py
index b0be800..414aa70 100644
--- a/backend/tests/integration/test_approval_flow.py
+++ b/backend/tests/integration/test_approval_flow.py
@@ -451,7 +451,9 @@ class TestContentSnapshot:
             row.snapshot_json = None
             session.commit()
 
-        detail = client.get(f"{_APPROVALS}/{order['id']}", headers=finance["headers"]).json()["data"]
+        detail = client.get(f"{_APPROVALS}/{order['id']}", headers=finance["headers"]).json()[
+            "data"
+        ]
         assert detail["content_source"] == "live"
         assert detail["snapshot"]["captured_at"] is None
         assert detail["snapshot"]["data"]["business_no"] == data["business_no"]
@@ -474,7 +476,9 @@ class TestContentSnapshot:
             row.ref_id = 999_999_999
             session.commit()
 
-        detail = client.get(f"{_APPROVALS}/{order['id']}", headers=finance["headers"]).json()["data"]
+        detail = client.get(f"{_APPROVALS}/{order['id']}", headers=finance["headers"]).json()[
+            "data"
+        ]
         assert detail["content_source"] == "missing"
         assert detail["snapshot"] is None
 
@@ -679,9 +683,7 @@ class TestAllocationVersions:
         data = _create_with_plan(client, admin["headers"], p["id"])
 
         detail = client.get(f"{_BUSINESSES}/{data['id']}", headers=admin["headers"]).json()["data"]
-        assert [(h["version_no"], h["is_active"]) for h in detail["allocation_history"]] == [
-            (1, 1)
-        ]
+        assert [(h["version_no"], h["is_active"]) for h in detail["allocation_history"]] == [(1, 1)]
 
         payload = business_payload(
             detail["customer_id"],
@@ -691,7 +693,9 @@ class TestAllocationVersions:
         assert resp.status_code == 200, resp.text
 
         detail = client.get(f"{_BUSINESSES}/{data['id']}", headers=admin["headers"]).json()["data"]
-        history = [(h["version_no"], h["is_active"], h["ratio"]) for h in detail["allocation_history"]]
+        history = [
+            (h["version_no"], h["is_active"], h["ratio"]) for h in detail["allocation_history"]
+        ]
         assert history == [(1, 0, "50.00"), (2, 1, "80.00")]
         assert [a["ratio"] for a in detail["secondary_allocations"]] == ["80.00"]
 
diff --git a/backend/tests/integration/test_audit_retry.py b/backend/tests/integration/test_audit_retry.py
index 7b2c5ba..98b31c8 100644
--- a/backend/tests/integration/test_audit_retry.py
+++ b/backend/tests/integration/test_audit_retry.py
@@ -13,21 +13,24 @@ audit_service.py「上下文内不得吞掉异常后正常退出」的约定相
 """
 
 import uuid
+from datetime import date
 
 import pytest
 from fastapi.testclient import TestClient
 from sqlalchemy import func, select
 
-from app.core.constants import ACTION_CREATE, MODULE_BUSINESS
+from app.core.constants import ACTION_CREATE, MODULE_BUSINESS, MODULE_RECEIPT
 from app.core.db import get_session_factory
 from app.main import app
 from app.models import OperationLog
-from app.services import business_service
+from app.services import business_service, receipt_service
+from tests.integration.factories import create_business
 
 pytestmark = pytest.mark.integration
 
 _BUSINESSES = "/api/v1/business/businesses"
 _CUSTOMERS = "/api/v1/base/customers"
+_RECEIPTS = "/api/v1/business/receipts"
 
 
 @pytest.fixture
@@ -160,3 +163,94 @@ class TestAuditRetry:
         # 每层各写一条记录；修复后重试有上限,最终失败仅一条失败日志
         assert _log_count(normal["username"], success=False) == base_fail + 1
         assert _log_count(normal["username"], success=True) == base_ok
+
+
+def _receipt_log_count(username: str, *, success: bool) -> int:
+    """按操作人统计「新增回款」操作日志条数。"""
+    factory = get_session_factory()
+    with factory() as session:
+        stmt = (
+            select(func.count())
+            .select_from(OperationLog)
+            .where(
+                OperationLog.username == username,
+                OperationLog.module == MODULE_RECEIPT,
+                OperationLog.action == ACTION_CREATE,
+                OperationLog.summary.like("新增回款%"),
+                OperationLog.success == int(success),
+            )
+        )
+        return int(session.execute(stmt).scalar_one())
+
+
+def _receipt_payload(business_id: int) -> dict:
+    return {
+        "business_id": business_id,
+        "receipt_date": str(date.today()),
+        "amount": "1200.00",
+    }
+
+
+class TestReceiptAuditRetry:
+    """回款新增的 1062 重试路径（回款与业务同构，但走独立实现）。
+
+    回款侧的重试循环本次从 create 函数体内移入 helper 调用，重试边界、
+    审计条数、批次号重取都必须保持原样 —— 用例会盯住这三条。
+    """
+
+    def test_首次撞号重试成功_审计恰一条真成功(
+        self, quiet_client: TestClient, normal: dict, monkeypatch
+    ) -> None:
+        client = quiet_client
+        business = create_business(client, normal["headers"])
+        first = client.post(
+            _RECEIPTS, json=_receipt_payload(business["id"]), headers=normal["headers"]
+        )
+        assert first.status_code == 200, first.text
+        existing_no = first.json()["data"]["receipt_batch_no"]
+
+        real_generate = receipt_service._generate_no
+        calls = {"n": 0}
+
+        def flaky_generate(session):
+            calls["n"] += 1
+            if calls["n"] == 1:
+                return existing_no  # 首次返回库内已存在批次号 → flush 撞 1062
+            return real_generate(session)
+
+        monkeypatch.setattr(receipt_service, "_generate_no", flaky_generate)
+        base_ok = _receipt_log_count(normal["username"], success=True)
+        base_fail = _receipt_log_count(normal["username"], success=False)
+
+        second = client.post(
+            _RECEIPTS, json=_receipt_payload(business["id"]), headers=normal["headers"]
+        )
+        assert second.status_code == 200, second.text
+        assert calls["n"] == 2, "重试应恰好再取一次号"
+        assert _receipt_log_count(normal["username"], success=True) == base_ok + 1
+        assert _receipt_log_count(normal["username"], success=False) == base_fail
+
+    def test_持续冲突_最终失败且审计恰一条失败(
+        self, quiet_client: TestClient, normal: dict, monkeypatch
+    ) -> None:
+        client = quiet_client
+        business = create_business(client, normal["headers"])
+        first = client.post(
+            _RECEIPTS, json=_receipt_payload(business["id"]), headers=normal["headers"]
+        )
+        assert first.status_code == 200, first.text
+        existing_no = first.json()["data"]["receipt_batch_no"]
+
+        def always_conflict(session):
+            return existing_no
+
+        monkeypatch.setattr(receipt_service, "_generate_no", always_conflict)
+        base_ok = _receipt_log_count(normal["username"], success=True)
+        base_fail = _receipt_log_count(normal["username"], success=False)
+
+        second = client.post(
+            _RECEIPTS, json=_receipt_payload(business["id"]), headers=normal["headers"]
+        )
+        assert second.status_code == 400, second.text
+        assert _receipt_log_count(normal["username"], success=False) == base_fail + 1
+        assert _receipt_log_count(normal["username"], success=True) == base_ok
diff --git a/backend/tests/integration/test_businesses.py b/backend/tests/integration/test_businesses.py
index 842ecb0..df59880 100644
--- a/backend/tests/integration/test_businesses.py
+++ b/backend/tests/integration/test_businesses.py
@@ -452,9 +452,7 @@ class TestEdit:
         assert resp.status_code == 400, resp.text
         assert "含合同负责营销人员" in resp.json()["message"]
 
-    def test_二次分配方案合法比例_保存成功且回显(
-        self, client: TestClient, normal: dict
-    ) -> None:
+    def test_二次分配方案合法比例_保存成功且回显(self, client: TestClient, normal: dict) -> None:
         headers = normal["headers"]
         customer = _customer(client, headers)
         owner = _personnel(client, headers)
diff --git a/backend/tests/integration/test_dashboard.py b/backend/tests/integration/test_dashboard.py
index eb7b509..e8c27cf 100644
--- a/backend/tests/integration/test_dashboard.py
+++ b/backend/tests/integration/test_dashboard.py
@@ -138,9 +138,7 @@ def test_总览_绩效合计与台账一致且非法月份400(client: TestClient
         expected = Decimal(
             str(
                 session.execute(
-                    select(
-                        func.coalesce(func.sum(PerformanceLedger.performance_amount), 0)
-                    ).where(
+                    select(func.coalesce(func.sum(PerformanceLedger.performance_amount), 0)).where(
                         PerformanceLedger.settlement_month == _month(),
                         PerformanceLedger.calc_type == 1,
                         PerformanceLedger.is_active == 1,
@@ -166,16 +164,12 @@ def test_待办_待审批与未指定批次与同步降级(client: TestClient, a
     )
     assert resp.status_code == 200, resp.text
     settlement_id = resp.json()["data"]["id"]
-    resp = client.post(
-        f"{_SETTLEMENTS}/{settlement_id}/submit", headers=admin["headers"]
-    )
+    resp = client.post(f"{_SETTLEMENTS}/{settlement_id}/submit", headers=admin["headers"])
     assert resp.status_code == 200, resp.text
 
     data = _dashboard(client, admin["headers"])
     assert data["todos"]["pending_approval_total"] >= 1
-    assert any(
-        item["order_type"] == 2 for item in data["todos"]["pending_approvals"]
-    )
+    assert any(item["order_type"] == 2 for item in data["todos"]["pending_approvals"])
     assert data["todos"]["unassigned_receipt_total"] >= 1
     assert data["todos"]["unassigned_receipts"][0]["receipt_id"] == ctx["receipt"]["id"]
     # ⑥ 未建：显式降级
@@ -335,17 +329,17 @@ def test_待办_按角色仅显示可操作节点(
     assert _dashboard(client, both["headers"])["todos"]["pending_approval_total"] == 2
 
     finance_orders = {
-        item["order_no"] for item in _dashboard(client, finance["headers"])["todos"]["pending_approvals"]
+        item["order_no"]
+        for item in _dashboard(client, finance["headers"])["todos"]["pending_approvals"]
     }
     leader_orders = {
-        item["order_no"] for item in _dashboard(client, leader["headers"])["todos"]["pending_approvals"]
+        item["order_no"]
+        for item in _dashboard(client, leader["headers"])["todos"]["pending_approvals"]
     }
     assert finance_orders.isdisjoint(leader_orders)
 
 
-def test_待办_财务通过后领导出现待办(
-    client: TestClient, admin: dict, finance: dict, leader: dict
-):
+def test_待办_财务通过后领导出现待办(client: TestClient, admin: dict, finance: dict, leader: dict):
     """真实流转：财务节点通过后，条目从财务待办消失并进入领导待办。"""
     _chain(client, admin["headers"])
     resp = client.post(
@@ -353,9 +347,7 @@ def test_待办_财务通过后领导出现待办(
     )
     assert resp.status_code == 200, resp.text
     settlement_id = resp.json()["data"]["id"]
-    resp = client.post(
-        f"{_SETTLEMENTS}/{settlement_id}/submit", headers=admin["headers"]
-    )
+    resp = client.post(f"{_SETTLEMENTS}/{settlement_id}/submit", headers=admin["headers"])
     assert resp.status_code == 200, resp.text
 
     order = _dashboard(client, admin["headers"])["todos"]["pending_approvals"][0]
diff --git a/backend/tests/integration/test_param_management.py b/backend/tests/integration/test_param_management.py
index abca972..022d4b0 100644
--- a/backend/tests/integration/test_param_management.py
+++ b/backend/tests/integration/test_param_management.py
@@ -326,6 +326,38 @@ class TestPreview:
         # Fraction 自动约分：4/1000 存储为 1/250，按数值断言
         assert Fraction(data["rate"]) == Fraction(4, 1000)
 
+    def test_预览参数使用顺序稳定(self, client: TestClient, admin: dict) -> None:
+        """params_used 的顺序即前端「用了哪些参数」的展示顺序。
+
+        预览实现拆成多段后，各段的调用先后决定参数读取先后；顺序漂移会让
+        复核者看到的参数清单与核算口径对不上（不报错，但会误导判读）。
+        """
+        payload = {
+            "business_role": "salesperson",
+            "customer_type": "personal",
+            "product_category": "municipal_water",
+            "ship_date": "2025-12-15",
+            "receipt_date": "2026-01-10",
+            "revenue": "100000",
+            "profit": "45000",
+            "received_amount": "60000",
+            "barter": True,
+            "probation": True,
+        }
+        data = client.post(f"{_PARAMS}/preview", json=payload, headers=admin["headers"]).json()[
+            "data"
+        ]
+        assert [p["param_code"] for p in data["params_used"]] == [
+            "salesperson.rate.start",
+            "salesperson.rate.slope",
+            "salesperson.rate.floor",
+            "salesperson.days.boundary",
+            "product.threshold",
+            "barter.discount",
+            "payment.start.ratio",
+            "probation.annual.threshold",
+        ]
+
     def test_经理聚合口径_预览拒绝(self, client: TestClient, admin: dict) -> None:
         payload = {
             "business_role": "sales_manager",
diff --git a/backend/tests/integration/test_reports.py b/backend/tests/integration/test_reports.py
index 076bb8b..4da0726 100644
--- a/backend/tests/integration/test_reports.py
+++ b/backend/tests/integration/test_reports.py
@@ -44,7 +44,9 @@ def _role_id(code: str) -> int:
         )
 
 
-def _person(client: TestClient, headers: dict, *, role: str = "salesperson", manager_id=None) -> dict:
+def _person(
+    client: TestClient, headers: dict, *, role: str = "salesperson", manager_id=None
+) -> dict:
     body = {
         "personnel_name": f"t_报表_{uuid.uuid4().hex[:8]}",
         "business_role_id": _role_id(role),
@@ -194,9 +196,7 @@ def test_部门汇总_无成员经理出零值行(admin, client: TestClient):
         headers=admin["headers"],
     )
     assert resp.status_code == 200, resp.text
-    team = next(
-        (t for t in resp.json()["data"]["teams"] if t["manager_id"] == manager["id"]), None
-    )
+    team = next((t for t in resp.json()["data"]["teams"] if t["manager_id"] == manager["id"]), None)
     assert team is not None
     assert Decimal(str(team["team_base"])) == Decimal("0")
     assert team["member_count"] == 0
@@ -224,7 +224,14 @@ def test_回款统计_空桶恒在且分摊拆行金额只计一次(admin, clien
 
     assert len(data["buckets"]) == 8
     assert [b["code"] for b in data["buckets"]] == [
-        "le0", "1-30", "31-60", "61-90", "91-180", "181-360", "361+", "none",
+        "le0",
+        "1-30",
+        "31-60",
+        "61-90",
+        "91-180",
+        "181-360",
+        "361+",
+        "none",
     ]
     # 60 天前发货 → 落在 61-90 桶（发货日首日计第 1 天）
     bucket = next(b for b in data["buckets"] if b["code"] == "61-90")
@@ -293,11 +300,15 @@ def test_导出_excel含签字栏且入操作日志(admin, client: TestClient):
 
     factory = get_session_factory()
     with factory() as session:
-        log = session.execute(
-            select(OperationLog)
-            .where(OperationLog.module == "报表统计", OperationLog.action == "导出")
-            .order_by(OperationLog.id.desc())
-        ).scalars().first()
+        log = (
+            session.execute(
+                select(OperationLog)
+                .where(OperationLog.module == "报表统计", OperationLog.action == "导出")
+                .order_by(OperationLog.id.desc())
+            )
+            .scalars()
+            .first()
+        )
     assert log is not None
 
 
@@ -389,19 +400,21 @@ def _latest_export_log():
 
     factory = get_session_factory()
     with factory() as session:
-        return session.execute(
-            select(OperationLog)
-            .where(OperationLog.module == "报表统计", OperationLog.action == "导出")
-            .order_by(OperationLog.id.desc())
-        ).scalars().first()
+        return (
+            session.execute(
+                select(OperationLog)
+                .where(OperationLog.module == "报表统计", OperationLog.action == "导出")
+                .order_by(OperationLog.id.desc())
+            )
+            .scalars()
+            .first()
+        )
 
 
 def test_部门汇总_无台账经理挂上级不报500(admin, client: TestClient):
     """经理 A 无台账且上级是经理 B：不能因 member_map 直接索引而 KeyError。"""
     manager_b = _person(client, admin["headers"], role="sales_manager")
-    manager_a = _person(
-        client, admin["headers"], role="sales_manager", manager_id=manager_b["id"]
-    )
+    manager_a = _person(client, admin["headers"], role="sales_manager", manager_id=manager_b["id"])
 
     resp = client.get(
         f"{_REPORTS}/department-summary",
@@ -467,9 +480,7 @@ def test_部门汇总_经理个人户计入时基数等于成员合计(admin, cl
             headers=admin["headers"],
         )
         assert resp.status_code == 200, resp.text
-        team = next(
-            t for t in resp.json()["data"]["teams"] if t["manager_id"] == manager["id"]
-        )
+        team = next(t for t in resp.json()["data"]["teams"] if t["manager_id"] == manager["id"])
         assert Decimal(str(team["team_base"])) == Decimal("100.00")
         assert sum(
             (Decimal(str(m["performance_sum"])) for m in team["members"]), Decimal("0")
@@ -516,9 +527,7 @@ def test_部门汇总_按经理筛选不误判其他团队下属(admin, client:
     assert [t["manager_id"] for t in data["teams"]] == [manager_1["id"]]
     assert data["unassigned"] == []
     assert all(
-        m["personnel_id"] != salesperson_2["id"]
-        for team in data["teams"]
-        for m in team["members"]
+        m["personnel_id"] != salesperson_2["id"] for team in data["teams"] for m in team["members"]
     )
     assert Decimal(str(data["reconciliation"]["other_team_base"])) > 0
 
@@ -557,9 +566,7 @@ def test_部门汇总_对账恒等式(admin, client: TestClient):
     )
 
 
-def _two_manager_scene(
-    client: TestClient, headers: dict
-) -> tuple[dict, dict, dict, dict, dict]:
+def _two_manager_scene(client: TestClient, headers: dict) -> tuple[dict, dict, dict, dict, dict]:
     """两位经理各有下属与业务 + 一位零成员经理（无聚合行）。"""
     manager_1 = _person(client, headers, role="sales_manager")
     manager_2 = _person(client, headers, role="sales_manager")
@@ -777,9 +784,7 @@ def test_导出取数异常_失败审计留痕(admin, client: TestClient, monkey
     assert "导出个人绩效工资明细表" in (log.summary or "")
 
 
-@pytest.mark.parametrize(
-    "path", ["/personal-detail", "/department-summary", "/payment-stats"]
-)
+@pytest.mark.parametrize("path", ["/personal-detail", "/department-summary", "/payment-stats"])
 def test_非法月份_查询返回400(admin, client: TestClient, path: str):
     """非法月份必须是参数错误 400，不能落 500。"""
     resp = client.get(
@@ -790,9 +795,7 @@ def test_非法月份_查询返回400(admin, client: TestClient, path: str):
     assert resp.status_code == 400, resp.text
 
 
-@pytest.mark.parametrize(
-    "code", ["personal-detail", "department-summary", "payment-stats"]
-)
+@pytest.mark.parametrize("code", ["personal-detail", "department-summary", "payment-stats"])
 def test_非法月份_导出返回400(admin, client: TestClient, code: str):
     """导出手工构造查询模型同样要转 400。"""
     resp = client.get(
diff --git a/backend/tests/unit/test_report_helpers.py b/backend/tests/unit/test_report_helpers.py
index 6af9708..0b40ad2 100644
--- a/backend/tests/unit/test_report_helpers.py
+++ b/backend/tests/unit/test_report_helpers.py
@@ -127,8 +127,7 @@ class TestExcel模板:
             cell
             for row in sheet.iter_rows()
             for cell in row
-            if isinstance(cell.value, (int, float))
-            and Decimal(str(cell.value)) == Decimal("39.11")
+            if isinstance(cell.value, (int, float)) and Decimal(str(cell.value)) == Decimal("39.11")
         ]
         assert amount_cells, "未找到金额数值单元格（Decimal 被转成了文本？）"
         assert amount_cells[0].number_format == "#,##0.00"
@@ -142,16 +141,14 @@ class TestExcel模板:
         workbook = load_workbook(io.BytesIO(content))
         cells = [cell for row in workbook.active.iter_rows() for cell in row]
         values = [
-            Decimal(str(cell.value))
-            for cell in cells
-            if isinstance(cell.value, (int, float))
+            Decimal(str(cell.value)) for cell in cells if isinstance(cell.value, (int, float))
         ]
         assert Decimal("15.50") in values
         # 「合计」标签只写一次（写满每个文本列是回归缺陷）
         assert sum(1 for cell in cells if cell.value == "合计") == 1
 
     def test_文本公式注入被前置单引号(self):
-        evil = "=HYPERLINK(\"http://evil\")"
+        evil = '=HYPERLINK("http://evil")'
         content = _build("personal-detail", [_personal_row(business_no=evil)])
         workbook = load_workbook(io.BytesIO(content))
         values = [