fix(approval): 版本号唯一键兜底 + 取号改当前读防并发重复

- V12 迁移：先重排存量重复 (order_type, ref_id) 分组的 version_no
  （V11 把存量一律置 1，而 V11 前重提是新建行），再加
  uk_approval_orders_order_type_ref_version；只重排真有重复的分组，
  避免改动前端已展示过的历史版本号
- _next_version 改锁读（select version_no ... FOR UPDATE，最大值在
  Python 侧算）：请求级鉴权的普通 SELECT 已把 REPEATABLE READ 快照
  锚定在请求开始时刻，原 MAX 快照读在并发下会算出同一版本号；
  聚合 + FOR UPDATE 为 MySQL 明确禁止，故不锁聚合结果
- 编辑与月结提交路径把 1062 收敛为 400「提交冲突（版本号生成重复），
  请重试」，其余完整性错误原样上抛；submit_order 内不 catch/rollback
- 驳回清空方案的过时注释改为 V11 起的 append-only 语义（置历史不删行）
- init-dev.sql 同步唯一键与 alembic_version 7f4f7b918bc4，导出 V12 SQL
- 测试：集成 +5（锁读 SQL 形态、两会话快照回归、唯一键 1062、冲突 400、
  正常递增）+ 单元 +1；工作台假审批单改用不同 ref_id 适配唯一键
- 全量回归 897 passed；ruff 与前端 type-check 绿

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

backend/README.md

@@ -231,6 +231,10 @@ bcrypt 5.0 起对超过 72 字节的密码**直接抛 ValueError**（此前是
 - **详情内容来源三态**：`content_source=snapshot`（送审快照）/ `live`（V11 前的历史单体无快照，只读回退当前内容并显式告警）/ `missing`（引用对象不存在且无快照）。**禁止回填历史单快照**——回填等于伪造提交时内容。
 - **二次分配方案 append-only**：`business_secondary_allocations` 行不物理删除，编辑全量替换与驳回清空只把旧行 `is_active=0`；唯一键为 `(business_id, version_no, personnel_id)`，同一版本内参与人唯一、跨版本可重复。`business_repository.list_allocations(active_only=True)` 是核算与详情回显的唯一取数口径；业务详情 `allocation_history` 返回全部版本供回溯。
 - **月结提交要锁行**：`settlement_service.submit` 取月结行必须 `for_update=True`——并发提交会让两次 `submit_order` 算出同一个审批单 `version_no`，送审次序失真。
+- **审批单 `version_no` 有唯一键兜底（V12）**：`uk_approval_orders_order_type_ref_version (order_type, ref_id, version_no)`——同一对象的第 N 次送审在库里有唯一一行。ORM 侧约束名必须与迁移 DDL 严格一致，否则 autogenerate 误判。
+- **取版本号必须是「当前读」（`approval_service._next_version`）**，两个禁忌：(a) 不能用普通 `MAX` 查询——请求级鉴权与业务共用同一 DbSession，鉴权的普通 SELECT 已把 REPEATABLE READ 快照锚定在请求开始时刻，父行锁只串行化写入时刻、**不刷新快照**，MAX 会读到旧值；(b) 不能写成 `select(func.max(...)).with_for_update()`——MySQL 官方明确「Locking clauses are not permitted in queries with aggregate functions or GROUP BY」，聚合上的 FOR UPDATE 语义无保证。正解：锁读该 ref 的现有版本行（`select(version_no)...with_for_update()`），最大值 Python 侧算（同 `no_generator` 套路）。
+- **版本号冲突的收敛口径**：`submit_order` 内不 catch、不 rollback（红线）。create 路径由外层编号重试循环整体重来（`business_service._NO_RETRY`）；编辑/月结提交路径由审计上下文统一回滚，再把 1062 翻译为 400「提交冲突（版本号生成重复），请重试」（同 `param_service` 版本号冲突模板）。唯一键是最后一道防线，不是取号正确性的前提。
+- **迁移前先修存量**：V11 把存量行 `version_no` 一律置 1，而 V11 前每次重提都新建行——同一 `(order_type, ref_id)` 存在多行 version_no=1，直接加唯一键必报 1062。V12 先重排「组内有重复」的分组（按 id 升序 1..n），其余分组不动（全量重排会改掉前端已展示过的历史版本号）。
 - **新增审批对象类型**须同时提供快照组装函数并接入 `_content_of` 的 live 回退分支，不要在各业务服务内联 JSON 结构。
 
 ### 工作台口径（M4 ⑧，改动前先读）

backend/alembic/versions/7f4f7b918bc4_v12_审批单版本号唯一键.py

@@ -0,0 +1,71 @@
+"""v12 审批单版本号唯一键
+
+Revision ID: 7f4f7b918bc4
+Revises: f3a8c2d61b47
+Create Date: 2026-10-21 10:15:00.000000
+
+给 ``approval_orders`` 加 ``(order_type, ref_id, version_no)`` 唯一键，兜住
+「同一引用对象的第几次送审」在并发下算重。
+
+**加键前必须先重排存量**：V11 迁移用 ``ADD COLUMN version_no INT NOT NULL
+DEFAULT 1`` 把存量行一律置 1，而 V11 之前的每次重提都是**新建行**（撤回
+只改旧单 status，行不删）。于是任何有 V11 前审批历史的库里，同一
+``(order_type, ref_id)`` 存在多行且 ``version_no`` 全为 1——直接
+``ADD CONSTRAINT`` 必报 1062，迁移会停在半途。
+
+重排只针对「组内有重复」的分组（``COUNT(*) > COUNT(DISTINCT version_no)``），
+按 id 升序重排为 1..n；其余分组一个字节都不动——全量重排会把前端已展示
+过的历史版本号（``V{{ version_no }}``）改掉，属于用户可见的历史数据篡改。
+"""
+
+from collections.abc import Sequence
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "7f4f7b918bc4"
+down_revision: str | Sequence[str] | None = "f3a8c2d61b47"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+_UK = "uk_approval_orders_order_type_ref_version"
+
+
+def upgrade() -> None:
+    """先重排存量重复的版本号，再加唯一键。"""
+    # 重排：只动「组内有重复版本号」的分组，按 id 升序编 1..n。
+    # 用派生表内联而非临时表——同一张临时表被 JOIN 两侧引用会报 1137
+    # "Can't reopen table"。
+    op.execute(
+        "UPDATE approval_orders ao "
+        "JOIN (SELECT id, ROW_NUMBER() OVER ("
+        "  PARTITION BY order_type, ref_id ORDER BY id) AS rn "
+        "  FROM approval_orders "
+        "  WHERE (order_type, ref_id) IN ("
+        "    SELECT order_type, ref_id FROM approval_orders "
+        "    GROUP BY order_type, ref_id "
+        "    HAVING COUNT(*) > COUNT(DISTINCT version_no))) t "
+        "ON ao.id = t.id "
+        "SET ao.version_no = t.rn"
+    )
+    # 幂等 guard：MySQL DDL 隐式提交，若加键成功而版本号未落盘，
+    # 重跑会报 1091（键已存在）。写法沿用 V9 先例。
+    op.execute(
+        "SET @uk_exists := (SELECT COUNT(*) FROM information_schema.TABLE_CONSTRAINTS "
+        "WHERE CONSTRAINT_SCHEMA = DATABASE() AND TABLE_NAME = 'approval_orders' "
+        f"AND CONSTRAINT_NAME = '{_UK}')"
+    )
+    op.execute(
+        "SET @ddl := IF(@uk_exists = 0, "
+        f"'ALTER TABLE approval_orders ADD CONSTRAINT {_UK} "
+        "UNIQUE (order_type, ref_id, version_no)', 'DO 0')"
+    )
+    op.execute("PREPARE stmt FROM @ddl")
+    op.execute("EXECUTE stmt")
+    op.execute("DEALLOCATE PREPARE stmt")
+
+
+def downgrade() -> None:
+    """删唯一键。唯一键本质是索引，DROP INDEX 不校验数据，存量是否已有
+    重复 version_no 与删索引无关；重排结果也不回滚（版本号保持 1..n 连续）。"""
+    op.execute(f"ALTER TABLE approval_orders DROP INDEX {_UK}")

backend/app/models/approval.py

@@ -15,7 +15,16 @@ ref_id 为**固化无外键**（同 business_status_logs 先例）：order_type=
 
 from datetime import datetime
 
-from sqlalchemy import JSON, DateTime, ForeignKey, Index, Integer, String, text
+from sqlalchemy import (
+    JSON,
+    DateTime,
+    ForeignKey,
+    Index,
+    Integer,
+    String,
+    UniqueConstraint,
+    text,
+)
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.models.base import BIGINT_UNSIGNED, TINYINT_TYPE, Base, TimestampMixin, mysql_table_args
@@ -54,6 +63,14 @@ class ApprovalOrder(Base, TimestampMixin):
     __tablename__ = "approval_orders"
     __table_args__ = mysql_table_args(
         "审批单表（营销后勤提交→财务复核→领导审核；开关关闭时自动通过留痕）",
+        # 同一引用对象的第 N 次送审在库里有唯一一行：并发提单算重版本号时
+        # 由 DB 兜住（应用层取号走锁读，见 approval_service._next_version）
+        UniqueConstraint(
+            "order_type",
+            "ref_id",
+            "version_no",
+            name="uk_approval_orders_order_type_ref_version",
+        ),
         Index("idx_approval_orders_type_status", "order_type", "status"),
         # 版本号按 (order_type, ref_id) 递增，须有对应索引避免全表 max 扫描
         Index("idx_approval_orders_type_ref", "order_type", "ref_id"),
@@ -82,7 +99,8 @@ class ApprovalOrder(Base, TimestampMixin):
         Integer,
         nullable=False,
         server_default="1",
-        comment="提交版本号（同一对象类型+引用对象从1递增；重提产生新版本，旧单快照不覆盖）",
+        comment="提交版本号（同一对象类型+引用对象从1递增；重提产生新版本，旧单快照不覆盖；"
+        "同一对象内唯一，由uk_approval_orders_order_type_ref_version强制）",
     )
     snapshot_json: Mapped[dict | None] = mapped_column(
         JSON,

backend/app/services/approval_service.py

@@ -296,11 +296,11 @@ def act(
                 operator_name=operator.username,
             )
         )
-        # 驳回=方案未获批准：清空该业务的二次分配行，防止被否决的方案以
-        # 「当前生效行」身份留在表里、被 M3 核算快照算进工资；方案事实由
-        # 审批单留痕与操作日志 request_data 承载（分摊语义落定后如需保留
-        # 历史版本，在 ④ 设计时演进）。审批单的清理由调用方（act 内联）
-        # 委托 business_repository 完成 —— service 层互调、不触 commit。
+        # 驳回=方案未获批准：把该业务的生效方案行置历史（is_active=0，
+        # **不删行**——V11 起 append-only 版本化），防止被否决的方案以
+        # 「当前生效行」身份被 M3 核算快照算进工资；历史方案仍可在业务
+        # 详情 allocation_history 回溯。委托 business_repository 完成
+        # —— service 层互调、不触 commit。
         if rejected and order.order_type == 1:
             from app.repositories import business_repository
 
@@ -414,18 +414,34 @@ def _withdraw_pending(session: Session, *, order_type: int, ref_id: int, operato
 
 
 def _next_version(session: Session, order_type: int, ref_id: int) -> int:
-    """同一对象类型+引用对象的下一个提交版本号（从 1 递增）。"""
-    return (
-        int(
-            session.execute(
-                select(func.coalesce(func.max(ApprovalOrder.version_no), 0)).where(
-                    ApprovalOrder.order_type == order_type,
-                    ApprovalOrder.ref_id == ref_id,
-                )
-            ).scalar_one()
+    """同一对象类型+引用对象的下一个提交版本号（从 1 递增）。
+
+    **必须用锁读（当前读），不能用普通 MAX 查询**：普通 SELECT 在
+    REPEATABLE READ 下是快照读，读到的是本事务首次一致读的快照；请求级
+    鉴权（deps.get_current_user 与业务共用同一 DbSession）那次 SELECT 已把
+    快照锚定在请求开始时刻，此后即使父行锁等到并发事务提交，MAX 仍停在
+    旧快照上——两个请求会算出同一个 version_no。锁读是当前读，永远读最新
+    已提交数据。
+
+    **不能写成 select(func.max(...)).with_for_update()**：MySQL 官方明确
+    「Locking clauses are not permitted in queries with aggregate functions
+    or GROUP BY」，聚合结果上的 FOR UPDATE 语义无保证。改锁「该 ref 的现有
+    版本行」，最大值在 Python 侧算——同 no_generator 的既有套路。只取
+    version_no 单列，不把 snapshot_json 拉进内存。
+    """
+    versions = (
+        session.execute(
+            select(ApprovalOrder.version_no)
+            .where(
+                ApprovalOrder.order_type == order_type,
+                ApprovalOrder.ref_id == ref_id,
+            )
+            .with_for_update()
         )
-        + 1
+        .scalars()
+        .all()
     )
+    return (max(versions) if versions else 0) + 1
 
 
 def allocation_content(session: Session, business: Business) -> dict:

backend/app/services/business_service.py

@@ -347,15 +347,14 @@ def _build_allocation_rows(
     return rows
 
 
-def _plan_changed(
-    old_rows: list[SecondaryAllocation], new_rows: list[SecondaryAllocation]
-) -> bool:
+def _plan_changed(old_rows: list[SecondaryAllocation], new_rows: list[SecondaryAllocation]) -> bool:
     """方案内容（参与人 + 比例）是否真的变了。
 
     比较口径是**集合**（人员 ID 与其比例的组合），与提交顺序无关——送审
     快照同样按库内生效行的顺序生成，若按顺序比较，用户只是把行拖了个
     序就会被判成「变了」，进而无谓地新增版本与重提审批单。
     """
+
     def _key(row: SecondaryAllocation) -> tuple[int, Decimal]:
         return (row.personnel_id, Decimal(row.ratio).quantize(Decimal("0.01")))
 
@@ -564,14 +563,22 @@ def update_business(
                 ref_id=business_id,
                 snapshot=snapshot,
             ):
-                approval_service.submit_order(
-                    session,
-                    order_type=APPROVAL_TYPE_SECONDARY_ALLOCATION,
-                    ref_id=business_id,
-                    refer_no=business.business_no,
-                    submitter=operator,
-                    snapshot=snapshot,
-                )
+                try:
+                    approval_service.submit_order(
+                        session,
+                        order_type=APPROVAL_TYPE_SECONDARY_ALLOCATION,
+                        ref_id=business_id,
+                        refer_no=business.business_no,
+                        submitter=operator,
+                        snapshot=snapshot,
+                    )
+                except IntegrityError as exc:
+                    # version_no 唯一键的并发冲突（取号是锁读后此项理论不可达，
+                    # 留作 DB 兜底）：审计上下文会统一回滚，这里只把 1062
+                    # 翻译成可重试的业务提示，其余完整性错误原样上抛
+                    if _is_duplicate_entry(exc):
+                        raise BadRequestError("提交冲突（版本号生成重复），请重试") from exc
+                    raise
         else:
             approval_service.withdraw_pending_orders(
                 session,

backend/app/services/settlement_service.py

@@ -13,11 +13,13 @@ from datetime import date
 from decimal import ROUND_HALF_UP, Decimal
 
 from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
 from app.core.constants import (
     MODULE_SETTLEMENT,
 )
+from app.core.db_errors import is_duplicate_entry
 from app.core.exceptions import BadRequestError, NotFoundError
 from app.models import (
     MonthlySettlement,
@@ -254,16 +256,24 @@ def submit(
         if settlement.status not in (SETTLEMENT_STATUS_DRAFT, SETTLEMENT_STATUS_REJECTED):
             raise BadRequestError("仅草稿或已驳回状态可提交")
 
-        order = approval_service.submit_order(
-            session,
-            order_type=_SETTLEMENT_ORDER_TYPE,
-            ref_id=settlement.id,
-            refer_no=f"JS{settlement.settlement_month.replace('-', '')}",
-            submitter=operator,
-            # 月结明细在生成时已固化；提交时快照让复核者看到的是送审版本，
-            # 而不是审批期间若发生重生成后的内容
-            snapshot=approval_service.settlement_content(session, settlement.id),
-        )
+        try:
+            order = approval_service.submit_order(
+                session,
+                order_type=_SETTLEMENT_ORDER_TYPE,
+                ref_id=settlement.id,
+                refer_no=f"JS{settlement.settlement_month.replace('-', '')}",
+                submitter=operator,
+                # 月结明细在生成时已固化；提交时快照让复核者看到的是送审版本，
+                # 而不是审批期间若发生重生成后的内容
+                snapshot=approval_service.settlement_content(session, settlement.id),
+            )
+        except IntegrityError as exc:
+            # version_no 唯一键的并发冲突（提交前已锁月结行，此项理论不可达，
+            # 留作 DB 兜底）：审计上下文会统一回滚，这里只把 1062 翻译成
+            # 可重试的业务提示，其余完整性错误原样上抛
+            if is_duplicate_entry(exc):
+                raise BadRequestError("提交冲突（版本号生成重复），请重试") from exc
+            raise
         settlement.approval_order_no = order.order_no
         if order.status == SETTLEMENT_STATUS_APPROVED:
             mark_locked(session, settlement, operator=operator)

backend/tests/README.md

@@ -169,7 +169,14 @@ uv run pytest tests/integration -q   # 仅集成
 | `tests/integration/test_secondary_allocation.py`（扩） | **同一会话内旧方案行随置历史同步刷新**（Core UPDATE 的会话同步策略退化为不刷新时，行对象会停留在 is_active=1 的过期内存值） | +1 |
 | `tests/integration/test_approval_flow.py`（扩） | **送审内容未变不重提**：方案与金额都没动 → 方案不新增版本、审批单不重提（旧单仍待财务复核）；**只改金额** → 方案版本不变但审批单 version_no+1 且新快照金额为改后值；**改方案比例** → 方案版本+1、审批单 version_no+1 且旧单撤回；**无在途单时保存仍提新单**（保守规则）；`pending_matches_snapshot` 四态（一致/不一致/无在途单/快照为 None） | +5 |
 
-运行：`uv run pytest tests/integration/test_approval_flow.py tests/integration/test_secondary_allocation.py tests/integration/test_schema_sync.py -q` → **37 passed**（2026-10-20）；`tests/integration/test_settlements.py` → **16 passed**；`tests/integration/test_calc.py` → **20 passed**；`tests/unit` → **525 passed**（含新增 `test_approval_helpers.py` 10 例）；**全量 `uv run pytest -q` → 891 passed（10:23）**。ruff、前端 `pnpm type-check` 与 `pnpm build` 均绿。
+### 审批单版本号并发防线（2026-10-21 完成：V12 唯一键 + 锁读取号）
+
+| 文件 | 守什么 | 用例数 |
+| --- | --- | --- |
+| `tests/integration/test_approval_version.py`（新） | **取号读最新已提交值**（两会话法：s1 锚定快照 + s2 提交 version_no=2 → s1 必须取到 3，守住「快照读 vs 当前读」根因）；**取号查询带 FOR UPDATE 且不含聚合**（事件钩子抓真实 SQL——MySQL 官方不允许聚合函数携带锁子句）；**唯一键生效**（重复 (order_type, ref_id, version_no) 直插撞 1062）；**编辑路径冲突收敛为 400** 且业务数据不写坏（monkeypatch `_next_version` 恒返回 1）；**正常路径版本号仍递增**（1→2→3） | +5 |
+| `tests/unit/test_models.py`（扩） | 审批单唯一键覆盖 (order_type, ref_id, version_no) 三列且约束名与迁移 DDL 一致 | +1 |
+
+运行（2026-10-22，容器恢复后）：`tests/integration/test_approval_version.py` + `test_dashboard.py` → **16 passed**；全量 `uv run pytest -q` → **897 passed（12:02）**；`uv run ruff check .` 绿、前端 `pnpm type-check` 绿。首次全量跑出两处**测试代码**适配（非产品缺陷）：① 工作台 `_insert_approval` 用同一 `(order_type=2, ref_id=0)` 连插两张单，撞 V12 新唯一键 → 每张单改给不同 `ref_id`；② 新用例跨模块引用 `switched_on` 夹具（pytest 夹具只在定义模块与 conftest 可见，报 fixture not found）→ 用 `_set_switch` 本地重建。
 
 ## 三、运行记录
 
@@ -217,6 +224,9 @@ uv run pytest tests/integration -q   # 仅集成
 | 2026-10-20 | 每次保存带方案的业务都刷出版本与审批留痕 | 只改备注、方案一字未动，也会「置历史+新增版本+撤回旧单+提新单」，方案历史与审批留痕被噪音淹没，两个 `version_no` 语义混同 | 判据统一为**送审内容**：方案未变不新增版本行；在途单送审内容一致则不重提（`pending_matches_snapshot`，只比 `data`） |
 | 2026-10-20 | 三处列表查询整列拉快照 | 审批单列表、工作台待办、审批超时预警都用 `select(ApprovalOrder)` 取全列，快照 JSON 可达上百 KB，翻页与工作台负担随快照体积线性放大 | 三处加 `defer(ApprovalOrder.snapshot_json)`（沿用操作日志列表排除 `request_data` 的既有模式） |
 | 2026-10-20 | 月结提交未锁行 | 并发/双击提交同一月结时两次 `_next_version` 可能算出同一 `version_no`，送审次序失真 | `settlement_repository.get_by_id` 加 `for_update` 参数，提交路径锁行 |
+| 2026-10-21 | 审批单 `version_no` 无唯一约束 | 应用层取号即使正确，并发下 DB 层无兜底；且唯一键是「一个对象至多一张在途单」红线之外版本语义的最后防线 | V12 迁移加 `uk_approval_orders_order_type_ref_version (order_type, ref_id, version_no)`；加键前先重排存量重复分组（V11 把存量一律置 1 而 V11 前重提是新建行，直接加键必报 1062） |
+| 2026-10-21 | `_next_version` 是快照读 | 请求级鉴权与业务共用同一 DbSession，鉴权的普通 SELECT 已锚定 REPEATABLE READ 快照——父行锁只串行化写入时刻、**不刷新快照**，并发提单的 MAX 停在请求开始时刻 → 版本号重复 | 取号改锁读（`select(version_no)...with_for_update()`，当前读），最大值 Python 侧算；**不能用聚合 + FOR UPDATE**——MySQL 官方不允许聚合函数携带锁子句 |
+| 2026-10-21 | `act()` 驳回清空方案的注释与行为相反 | 注释写「清空该业务的二次分配行…如需保留历史版本，在 ④ 设计时演进」，而 V11 已实现 append-only（只置 `is_active=0` 行不删）——误导后来者按旧语义改代码 | 注释改为「置历史不删行，历史方案可回溯」 |
 | 2026-10-19 | 前端报表五处状态缺陷 | 打印超限判定恒假（静默截断）、部门下钻 query 未读取、回款多桶共用一个 drill 状态（面板串行）、summary 在请求序号守卫前写入（合计串号）、e2e 全页行数选择器必失败 | 分别修正判定/读取 query/按桶独立状态+数组参数/`onData` 回调/限定主桶表；补 e2e 多桶展开、下钻、截断、部门打印用例 |
 | 2026-10-19 | 工作台权限收紧后守卫仍回退 `/dashboard` | 普通用户登录/直连受限页被反复重定向到同一受限页 → **无法进入系统**（导航被判定无限重定向） | `router/access.ts` 统一准入与落地页解析（工作台 → 业务数据 → 回款）；登录回跳经 `canAccessPath` 校验；菜单/面包屑复用同一判定 |
 | 2026-10-19 | 部门汇总应发只计一次全公司 5% | 月结每位有聚合行的经理各计一次 5%，签字稿少计 N-1 份；**导出/打印金额小于月结应发** | 团队行增 `has_aggregate/aggregate_amount`（优先冻结聚合行）；对账改按聚合额求和，新增「聚合行舍入差」吸收分列 q2 尾差；补「与 `total_payable` 对拍」用例 |

backend/tests/integration/test_approval_version.py

@@ -0,0 +1,183 @@
+"""审批单版本号并发防线（V12 唯一键 + 锁读取号）。
+
+守的是：
+- **取号必须读最新已提交值**：请求级鉴权与业务共用同一 DbSession，鉴权
+  的普通 SELECT 已把 REPEATABLE READ 快照锚定在请求开始时刻；父行锁只
+  串行化写入时刻、不刷新快照——取号若是快照读，并发提单会算出同一个
+  version_no（两会话法做行为级回归锁，不依赖真多线程）
+- **取号查询带 FOR UPDATE 且不用聚合**：MySQL 官方不允许聚合函数携带
+  锁子句（Locking clauses are not permitted in queries with aggregate
+  functions or GROUP BY），聚合上的 FOR UPDATE 语义无保证
+- **DB 层唯一键生效**：即使应用层取号缺陷被绕过，重复
+  (order_type, ref_id, version_no) 也必须在库上被拒（1062）
+- **冲突收敛为 400**：编辑路径撞唯一键 → 审计上下文回滚 → 业务提示
+  可重试，业务数据不得写坏
+- **正常路径版本号仍递增**：锁读改写不破坏既有 1→2→3 行为
+"""
+
+import uuid
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import event, text
+
+from app.core.db import get_engine, get_session_factory
+from app.core.db_errors import is_duplicate_entry
+from app.models import ApprovalOrder
+from tests.integration.factories import business_payload
+from tests.integration.test_approval_flow import (
+    _BUSINESSES,
+    _create_with_plan,
+    _latest_order,
+    _person,
+    _set_switch,
+)
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture
+def switched_on(client: TestClient, admin: dict) -> None:
+    """把二次分配审批开关置 1（用例结束由 conftest 恢复种子基线）。
+
+    pytest 夹具只在定义它的模块与 conftest 中可见，跨模块直接引用会报
+    ``fixture 'switched_on' not found``；这里用 ``_set_switch`` 本地重建。
+    """
+    _set_switch(client, admin["headers"], "1")
+
+
+class TestNextVersionLockRead:
+    """取号查询的锁读口径（真实 SQL 形态 + 行为级快照回归）。"""
+
+    def test_取号查询带FOR_UPDATE_不含聚合(self, client: TestClient, admin: dict) -> None:
+        """抓 _next_version 真实发出的 SQL（比本地重造语句再编译更贴真）：
+        必须含 FOR UPDATE、不含 MAX( —— 聚合上的锁子句 MySQL 不保证语义。"""
+
+        captured: list[str] = []
+
+        def _spy(_conn, _cursor, statement, *_rest):
+            if "approval_orders" in statement and "version_no" in statement:
+                captured.append(statement)
+
+        engine = get_engine()
+        event.listen(engine, "before_cursor_execute", _spy)
+        try:
+            p = _person(client, admin["headers"])
+            _create_with_plan(client, admin["headers"], p["id"])  # 触发一次真实提单
+        finally:
+            event.remove(engine, "before_cursor_execute", _spy)
+
+        assert captured, "未捕获到取号查询"
+        assert any("FOR UPDATE" in sql.upper() for sql in captured), captured
+        assert not any("MAX(" in sql.upper() for sql in captured), captured
+
+    def test_版本号读最新已提交值_不受事务旧快照影响(self) -> None:
+        """两会话法（无需线程）：s1 先锚定快照 → s2 提交新版本 →
+        s1 取号必须读到 3 而非旧快照里的 2——守住「快照读 vs 当前读」根因。"""
+        from app.services import approval_service
+
+        factory = get_session_factory()
+        with factory() as s1:
+            # 锚定快照（模拟请求级鉴权读先于业务写发生）
+            s1.execute(text("SELECT order_no FROM approval_orders LIMIT 1")).all()
+
+            with factory() as s2:
+                order = ApprovalOrder(
+                    order_no=f"AP-{uuid.uuid4().hex[:12]}",
+                    order_type=1,
+                    ref_id=987654321,
+                    refer_no="YW-SNAPSHOT-TEST",
+                    version_no=2,
+                    status=5,
+                    submitter_name="t_snapshot_probe",
+                )
+                s2.add(order)
+                s2.commit()
+
+            # s1 的普通读仍停在旧快照（该分组在快照里不存在）
+            stale = s1.execute(
+                text(
+                    "SELECT COALESCE(MAX(version_no),0) FROM approval_orders "
+                    "WHERE order_type=1 AND ref_id=987654321"
+                )
+            ).scalar_one()
+            assert stale == 0, "前置失效：快照读应停在锚定时刻，先修隔离级别再跑本用例"
+
+            assert approval_service._next_version(s1, 1, 987654321) == 3, (
+                "取号必须读到最新已提交值（2+1=3），读到 2 说明退化为快照读"
+            )
+            s1.rollback()
+
+    def test_唯一键生效_重复版本号被拒(self) -> None:
+        """重复 (order_type, ref_id, version_no) 直插必须撞 1062——
+        应用层取号即使被绕过，DB 也是最后一道防线。"""
+        from sqlalchemy.exc import IntegrityError
+
+        factory = get_session_factory()
+        with factory() as s:
+            common = {
+                "order_type": 1,
+                "ref_id": 987654322,
+                "refer_no": "YW-UK-TEST",
+                "status": 5,
+                "submitter_name": "t_uk_probe",
+            }
+            s.add(ApprovalOrder(order_no=f"AP-{uuid.uuid4().hex[:12]}", version_no=1, **common))
+            s.flush()
+            s.add(ApprovalOrder(order_no=f"AP-{uuid.uuid4().hex[:12]}", version_no=1, **common))
+            with pytest.raises(IntegrityError) as exc_info:
+                s.flush()
+            assert is_duplicate_entry(exc_info.value), "应为唯一键冲突（1062）而非其他完整性错误"
+            s.rollback()
+
+    def test_正常路径版本号仍递增(self, client: TestClient, admin: dict, switched_on) -> None:
+        """常规编辑两次：送审版本号 1→2→3——锁读改写不得破坏既有行为。"""
+        p = _person(client, admin["headers"])
+        data = _create_with_plan(client, admin["headers"], p["id"])
+
+        first = _latest_order(client, admin["headers"], data["business_no"])
+        assert first["version_no"] == 1
+
+        for ratio in ("60.00", "70.00"):
+            detail = client.get(f"{_BUSINESSES}/{data['id']}", headers=admin["headers"]).json()[
+                "data"
+            ]
+            payload = business_payload(
+                detail["customer_id"],
+                secondary_allocations=[{"personnel_id": p["id"], "ratio": ratio}],
+            )
+            resp = client.put(f"{_BUSINESSES}/{data['id']}", json=payload, headers=admin["headers"])
+            assert resp.status_code == 200, resp.text
+
+        third = _latest_order(client, admin["headers"], data["business_no"])
+        assert third["version_no"] == 3
+
+
+class TestConflictConvergence:
+    """编辑路径撞版本号唯一键 → 收敛为可重试的 400，业务数据不写坏。"""
+
+    def test_编辑路径版本号冲突_收敛为400(
+        self, client: TestClient, admin: dict, switched_on, monkeypatch
+    ) -> None:
+        from app.services import approval_service
+
+        p = _person(client, admin["headers"])
+        data = _create_with_plan(client, admin["headers"], p["id"])
+        before = client.get(f"{_BUSINESSES}/{data['id']}", headers=admin["headers"]).json()["data"]
+
+        # 模拟取号读到过期快照：恒返回 1（库中 V1 已存在）→ 撞唯一键
+        monkeypatch.setattr(approval_service, "_next_version", lambda *a, **k: 1)
+
+        payload = business_payload(
+            before["customer_id"],
+            secondary_allocations=[{"personnel_id": p["id"], "ratio": "80.00"}],
+        )
+        resp = client.put(f"{_BUSINESSES}/{data['id']}", json=payload, headers=admin["headers"])
+        assert resp.status_code == 400, resp.text
+        assert "版本号生成重复" in resp.json()["message"]
+
+        monkeypatch.undo()
+        # 业务数据不得写坏：整个事务回滚，方案仍是 50.00（V1 生效行）
+        after = client.get(f"{_BUSINESSES}/{data['id']}", headers=admin["headers"]).json()["data"]
+        assert [a["ratio"] for a in after["secondary_allocations"]] == ["50.00"]
+        assert [h["version_no"] for h in after["allocation_history"]] == [1]

backend/tests/integration/test_dashboard.py

@@ -250,7 +250,13 @@ def leader(make_user, auth_headers) -> dict:
 
 
 def _insert_approval(status: int) -> None:
-    """直插在途审批单（避免依赖两次流转，专注工作台过滤口径）。"""
+    """直插在途审批单（避免依赖两次流转，专注工作台过滤口径）。
+
+    ``ref_id`` 仅作对象分组锚点、本用例不校验：**每张单必须给不同的
+    ref_id**——V12 的 ``uk_approval_orders_order_type_ref_version`` 要求
+    (order_type, ref_id, version_no) 唯一，同一假对象连插两张 version_no=1
+    会撞 1062。
+    """
     from app.core.db import get_session_factory
     from app.models import ApprovalOrder
 
@@ -260,7 +266,7 @@ def _insert_approval(status: int) -> None:
             ApprovalOrder(
                 order_no=f"AP{uuid.uuid4().hex[:12]}",
                 order_type=2,
-                ref_id=0,
+                ref_id=status,
                 refer_no="t_工作台",
                 status=status,
                 submitter_name="t_admin",

backend/tests/unit/test_models.py

@@ -96,6 +96,20 @@ class TestNamingConvention:
             for fk in table.foreign_keys:
                 assert fk.constraint.name.startswith(f"fk_{table.name}_"), fk.constraint.name
 
+    def test_审批单版本号唯一键覆盖三列(self) -> None:
+        """并发提单算重版本号时由 DB 兜底（应用层取号走锁读，见
+        approval_service._next_version）；约束名必须与迁移 DDL 严格一致，
+        否则 alembic autogenerate 会误判成「约束缺失/多余」。"""
+        table = Base.metadata.tables["approval_orders"]
+        uks = [c for c in table.constraints if c.__class__.__name__ == "UniqueConstraint"]
+        hit = [
+            c
+            for c in uks
+            if {col.name for col in c.columns} == {"order_type", "ref_id", "version_no"}
+        ]
+        assert hit, "审批单缺 (order_type, ref_id, version_no) 唯一键"
+        assert hit[0].name == "uk_approval_orders_order_type_ref_version"
+
 
 class TestComments:
     """规范红线要求中文 COMMENT。"""

db/init/init-dev.sql

@@ -722,6 +722,7 @@ CREATE TABLE approval_orders (
     update_time DATETIME NOT NULL COMMENT '更新时间' DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
     CONSTRAINT pk_approval_orders PRIMARY KEY (id),
     CONSTRAINT uk_approval_orders_order_no UNIQUE (order_no),
+    CONSTRAINT uk_approval_orders_order_type_ref_version UNIQUE (order_type, ref_id, version_no),
     CONSTRAINT fk_approval_orders_submitter_id FOREIGN KEY(submitter_id) REFERENCES users (id) ON DELETE SET NULL
 )CHARSET=utf8mb4 ENGINE=InnoDB COMMENT='审批单表（营销后勤提交→财务复核→领导审核；开关关闭时自动通过留痕）' COLLATE utf8mb4_general_ci;
 
@@ -890,4 +891,4 @@ CREATE TABLE alembic_version (
     CONSTRAINT alembic_version_pkc PRIMARY KEY (version_num)
 )CHARSET=utf8mb4 ENGINE=InnoDB COLLATE utf8mb4_general_ci;
 
-INSERT INTO alembic_version (version_num) VALUES ('f3a8c2d61b47');
+INSERT INTO alembic_version (version_num) VALUES ('7f4f7b918bc4');

db/migration/V12__approval_order_version_unique.sql

@@ -0,0 +1,16 @@
+-- Running upgrade f3a8c2d61b47 -> 7f4f7b918bc4
+
+UPDATE approval_orders ao JOIN (SELECT id, ROW_NUMBER() OVER (  PARTITION BY order_type, ref_id ORDER BY id) AS rn   FROM approval_orders   WHERE (order_type, ref_id) IN (    SELECT order_type, ref_id FROM approval_orders     GROUP BY order_type, ref_id     HAVING COUNT(*) > COUNT(DISTINCT version_no))) t ON ao.id = t.id SET ao.version_no = t.rn;
+
+SET @uk_exists := (SELECT COUNT(*) FROM information_schema.TABLE_CONSTRAINTS WHERE CONSTRAINT_SCHEMA = DATABASE() AND TABLE_NAME = 'approval_orders' AND CONSTRAINT_NAME = 'uk_approval_orders_order_type_ref_version');
+
+SET @ddl := IF(@uk_exists = 0, 'ALTER TABLE approval_orders ADD CONSTRAINT uk_approval_orders_order_type_ref_version UNIQUE (order_type, ref_id, version_no)', 'DO 0');
+
+PREPARE stmt FROM @ddl;
+
+EXECUTE stmt;
+
+DEALLOCATE PREPARE stmt;
+
+UPDATE alembic_version SET version_num='7f4f7b918bc4' WHERE alembic_version.version_num = 'f3a8c2d61b47';
+