feat(approval): 审批单内容快照与二次分配方案版本化

补齐 code-review 遗留 2 项：审批单此前只固化 ref_id/refer_no 指针，复核者看到
的是业务当前内容；二次分配方案在编辑替换与驳回清空时一律物理删除，方案事实
无处回溯。

- 审批单加 version_no 与 snapshot_json（信封 schema_version/order_type/
  captured_at/data）；二次分配、月结、调整单三类提交均固化送审内容，开关关闭
  的自动通过路径同样写快照；详情按 snapshot/live/missing 三态回显，历史单
  只读回退当前内容并显式告警，不回填历史快照
- 二次分配方案行改 append-only 版本化（version_no/is_active，唯一键扩为
  business+version+personnel）：编辑替换与驳回清空只置历史，核算与详情默认
  只取生效行，业务详情新增 allocation_history
- 送审内容未变不重提、方案未变不新增版本：判据统一为送审内容，两个
  version_no 语义解耦（方案版本=内容第几版，审批单版本=第几次送审）
- 审批单列表、工作台待办、审批超时预警三处 defer 快照列，避免整列拉取
- 月结提交取行加 for_update，防并发提交算出重复的审批单版本号
- 修复审批类型筛选上界 le=2（调整单无法筛选）、快照金额把空值写成 0.00
- V11 迁移三件套同步（Alembic / db/migration / init-dev.sql）

测试：单元 +16（方案内容比较 10、审批入参 6）、集成 +12；全量 891 passed。

Co-Authored-By: Claude Code <noreply@anthropic.com>

backend/app/repositories/business_repository.py

@@ -7,7 +7,7 @@ models/customer.py 定义）。
 
 from datetime import date
 
-from sqlalchemy import Row, Select, delete, exists, func, or_, select
+from sqlalchemy import Row, Select, delete, exists, func, or_, select, update
 from sqlalchemy.orm import Session, aliased
 
 from app.models import (
@@ -199,43 +199,75 @@ def replace_items(session: Session, business_id: int, rows: list[BusinessItem])
     session.flush()
 
 
-def list_allocations(session: Session, business_id: int) -> list[SecondaryAllocation]:
-    """二次分配方案（详情回显；空列表=未设置方案）。"""
+def list_allocations(
+    session: Session, business_id: int, *, active_only: bool = True
+) -> list[SecondaryAllocation]:
+    """二次分配方案（详情回显与核算取数；默认仅当前生效版本）。"""
+    stmt = select(SecondaryAllocation).where(SecondaryAllocation.business_id == business_id)
+    if active_only:
+        stmt = stmt.where(SecondaryAllocation.is_active == 1)
     return list(
         session.execute(
-            select(SecondaryAllocation)
-            .where(SecondaryAllocation.business_id == business_id)
-            .order_by(SecondaryAllocation.id)
+            stmt.order_by(SecondaryAllocation.version_no, SecondaryAllocation.id)
         ).scalars()
     )
 
 
+def list_allocation_versions(session: Session, business_id: int) -> list[SecondaryAllocation]:
+    """方案全部版本行（含历史；业务详情「方案历史」用，按版本正序）。"""
+    return list_allocations(session, business_id, active_only=False)
+
+
 def replace_allocations(
     session: Session, business_id: int, rows: list[SecondaryAllocation]
 ) -> None:
-    """全量替换二次分配方案（不 commit，事务边界在服务层）。
+    """全量替换二次分配方案：旧生效行置历史，新行按下一版本号插入。
 
-    本表有 uk(business_id, personnel_id)，**必须**沿用先删后 flush 再插的
-    防御写法：同 flush 内 INSERT 先于 DELETE 处理，直接删加插会撞唯一键。
+    不 commit（事务边界在服务层）。版本行 append-only——编辑与清空都不
+    物理删除，驳回后仍可回溯被否决方案的参与人与比例。空 ``rows`` 表示
+    只清空生效态（不产生新版本）。
     """
-    session.execute(
-        delete(SecondaryAllocation).where(SecondaryAllocation.business_id == business_id)
+    deactivate_allocations(session, business_id)
+    if not rows:
+        return
+    next_version = (
+        session.execute(
+            select(func.coalesce(func.max(SecondaryAllocation.version_no), 0)).where(
+                SecondaryAllocation.business_id == business_id
+            )
+        ).scalar_one()
+        + 1
     )
-    session.flush()
     for row in rows:
         row.business_id = business_id
+        row.version_no = next_version
+        row.is_active = 1
         session.add(row)
     session.flush()
 
 
-def delete_allocations(session: Session, business_id: int) -> None:
-    """清空业务的二次分配方案（审批驳回/撤回时由 approval_service 调用）。
+def deactivate_allocations(session: Session, business_id: int) -> None:
+    """把业务的当前生效方案置为历史（审批驳回/编辑替换时调用；行保留）。
 
-    被否决的方案不得以「当前生效行」身份留在表里（M3 核算快照会取到）；
-    方案事实由审批单留痕与操作日志 request_data 承载。
+    被否决的方案不得以「当前生效行」身份继续被核算取到；``is_active=0``
+    后核算侧自然读不到，而历史版本仍可在业务详情回溯。不 commit。
+
+    显式声明 ``synchronize_session="fetch"``：Core UPDATE 默认走 auto 策略
+    （优先 evaluate，本例的条件恰好能在 Python 侧求值，故实测也会刷新在途
+    对象）。显式写 fetch 是为了不把「在途对象被刷新」这件事押在 evaluate
+    能否解析条件上——条件一旦复杂化，auto 会退化为不刷新，同一会话里已
+    加载的方案行就会停留在 ``is_active=1`` 的过期内存值（查询虽有
+    is_active 过滤，但行对象本身仍是「生效态」，被序列化或参与判断就是
+    错的）。
     """
     session.execute(
-        delete(SecondaryAllocation).where(SecondaryAllocation.business_id == business_id)
+        update(SecondaryAllocation)
+        .where(
+            SecondaryAllocation.business_id == business_id,
+            SecondaryAllocation.is_active == 1,
+        )
+        .values(is_active=0)
+        .execution_options(synchronize_session="fetch")
     )
     session.flush()
 

backend/app/repositories/dashboard_repository.py

@@ -8,7 +8,7 @@
 from datetime import date, datetime
 
 from sqlalchemy import exists, func, or_, select
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import Session, defer
 
 from app.models import (
     ApprovalOrder,
@@ -113,12 +113,15 @@ def list_pending_approvals(
     """待处理审批单（按提交时间升序——最久的排前面）。
 
     ``statuses`` 由调用方按当前用户角色解析（待我审批），空元组返回空列表。
+
+    排除 snapshot_json 列（快照可达上百 KB，待办只取单号/类型/提交人/时间）。
     """
     if not statuses:
         return []
     return list(
         session.execute(
             select(ApprovalOrder)
+            .options(defer(ApprovalOrder.snapshot_json))
             .where(ApprovalOrder.status.in_(statuses))
             .order_by(ApprovalOrder.create_time.asc(), ApprovalOrder.id.asc())
             .limit(limit)
@@ -210,10 +213,14 @@ def count_overdue_shipments(session: Session, *, cutoff: date) -> int:
 def list_overdue_approvals(
     session: Session, *, cutoff: datetime, limit: int
 ) -> list[ApprovalOrder]:
-    """提交后超过阈值仍未处理的审批单（审批超时规则）。"""
+    """提交后超过阈值仍未处理的审批单（审批超时规则）。
+
+    排除 snapshot_json 列（同待办：预警只用单号/类型/提交人/时间）。
+    """
     return list(
         session.execute(
             select(ApprovalOrder)
+            .options(defer(ApprovalOrder.snapshot_json))
             .where(
                 ApprovalOrder.status.in_(_PENDING_STATUSES),
                 ApprovalOrder.create_time < cutoff,

backend/app/repositories/settlement_repository.py

@@ -7,7 +7,16 @@ from app.models import MonthlySettlement, SettlementItem
 from app.schemas.settlement import SettlementQuery
 
 
-def get_by_id(session: Session, settlement_id: int) -> MonthlySettlement | None:
+def get_by_id(
+    session: Session, settlement_id: int, *, for_update: bool = False
+) -> MonthlySettlement | None:
+    """月结行；``for_update`` 供提交审批路径加行锁用。
+
+    提交要算审批单版本号（同一月结的第几次送审），并发提交会让两次
+    算出同一个版本号；锁住月结行使并发提交串行化。
+    """
+    if for_update:
+        return session.get(MonthlySettlement, settlement_id, with_for_update=True)
     return session.get(MonthlySettlement, settlement_id)