feat(calc): 二次分摊拆行与 #12 单笔侧开关——按方案比例拆给参与人、余额归负责人

- _personal_rate 的 TPL_MANAGER_SHARE 分支读 manager.personal.included：1=按业务员递减公式计提，0=单笔落 0（现状）
- _load_allocations 装载生效方案并防御拦截（合计超 100%、参与人含负责人）
- _persist_ledger 拆行落库：参与人=份额×比例(q2)、负责人=q2 全额−已分累计（行合计守恒防分漂）；快照带 alloc_plan/alloc_role/alloc_ratio
- _prepare_mode 重算改 list_active_by_receipt 作废全部拆分行；uk 三列版支持同回款按人多行
- business_service 保存侧同口径校验（超 100% 400、负责人自分 400、hint 文案定稿）
- 集成用例 +5（拆行守恒/叠加进度登记/重算全作废/双拦截）+保存侧 3 条

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

backend/tests/integration/test_businesses.py

@@ -412,6 +412,71 @@ class TestEdit:
         assert len(detail["items"]) == 1
         assert Decimal(detail["items"][0]["quantity"]) == Decimal("100")
 
+    def test_二次分配方案合计超100_保存400(self, client: TestClient, normal: dict) -> None:
+        """口径（澄清 #1/#2 答复 2026-10-19）：分配绩效、余额归合同负责人——
+        超 100% 会把负责人份额算成负数，保存侧先拦（核算侧另有防御）。"""
+        headers = normal["headers"]
+        customer = _customer(client, headers)
+        owner = _personnel(client, headers)
+        part = _personnel(client, headers)
+        data = _create(client, headers, customer["id"], owner_personnel_id=owner["id"])
+
+        resp = client.put(
+            f"{_BUSINESSES}/{data['id']}",
+            json=_payload(
+                customer["id"],
+                secondary_allocations=[
+                    {"personnel_id": part["id"], "ratio": "60"},
+                    {"personnel_id": _personnel(client, headers)["id"], "ratio": "50"},
+                ],
+            ),
+            headers=headers,
+        )
+        assert resp.status_code == 400, resp.text
+        assert "超过 100%" in resp.json()["message"]
+
+    def test_二次分配参与人含负责人_保存400(self, client: TestClient, normal: dict) -> None:
+        headers = normal["headers"]
+        customer = _customer(client, headers)
+        owner = _personnel(client, headers)
+        data = _create(client, headers, customer["id"], owner_personnel_id=owner["id"])
+
+        resp = client.put(
+            f"{_BUSINESSES}/{data['id']}",
+            json=_payload(
+                customer["id"],
+                secondary_allocations=[{"personnel_id": owner["id"], "ratio": "30"}],
+            ),
+            headers=headers,
+        )
+        assert resp.status_code == 400, resp.text
+        assert "含合同负责营销人员" in resp.json()["message"]
+
+    def test_二次分配方案合法比例_保存成功且回显(
+        self, client: TestClient, normal: dict
+    ) -> None:
+        headers = normal["headers"]
+        customer = _customer(client, headers)
+        owner = _personnel(client, headers)
+        part = _personnel(client, headers)
+        data = _create(client, headers, customer["id"], owner_personnel_id=owner["id"])
+
+        resp = client.put(
+            f"{_BUSINESSES}/{data['id']}",
+            json=_payload(
+                customer["id"],
+                secondary_allocations=[{"personnel_id": part["id"], "ratio": "40"}],
+            ),
+            headers=headers,
+        )
+        assert resp.status_code == 200, resp.text
+        # 方案在详情回包（更新回包为简版 Out，不含方案区）
+        detail = client.get(f"{_BUSINESSES}/{data['id']}", headers=headers).json()["data"]
+        assert len(detail["secondary_allocations"]) == 1
+        assert Decimal(detail["secondary_allocations"][0]["ratio"]) == Decimal("40")
+        # 余额 60% 归合同负责人（hint 文案 = #2 答复口径）
+        assert detail["allocation_hint"] is not None and "60" in detail["allocation_hint"]
+
 
 class TestStatus:
     """状态机：终态不可逆是终态单据禁写的前提；留痕从创建首条记起。"""

backend/tests/integration/test_calc.py

@@ -22,7 +22,7 @@ from fastapi.testclient import TestClient
 from sqlalchemy import select
 
 from app.core.db import get_session_factory
-from app.models import PerformanceLedger, Receipt
+from app.models import PerformanceLedger, Personnel, Receipt, SecondaryAllocation
 from app.models.receipt import SETTLEMENT_DONE
 from tests.integration.factories import business_payload, customer
 
@@ -584,3 +584,207 @@ class TestBatch:
         assert by_id[r1["id"]]["ok"] is True and by_id[r1["id"]]["repeated"] is False
         assert by_id[r2["id"]]["ok"] is True and by_id[r2["id"]]["repeated"] is True
         assert by_id[999999]["ok"] is False
+
+
+class TestSecondaryAllocation:
+    """二次分摊拆行（澄清 #1 答复 2026-10-19：分配绩效——算后分）。
+
+    方案行直插入库（跳过保存侧校验，聚焦核算引擎；保存侧校验用例在
+    test_businesses）。金标准：双类别混合业务绩效 39.11 / 折算 4000.00，
+    拆 30%+40% 后参与人 11.73/15.64、负责人余额 11.74，行合计守恒。
+    """
+
+    def _add_plan(self, business_id: int, rows: list[tuple[int, str]]) -> None:
+        """直插方案行（测试隔离造数；核算引擎按「当前落库方案」消费）。"""
+        factory = get_session_factory()
+        with factory() as session:
+            for pid, ratio in rows:
+                person = session.get(Personnel, pid)
+                session.add(
+                    SecondaryAllocation(
+                        business_id=business_id,
+                        personnel_id=pid,
+                        personnel_name=person.personnel_name,
+                        ratio=Decimal(ratio),
+                    )
+                )
+            session.commit()
+
+    def test_分摊拆行_参与人比例与负责人余额守恒(self, client: TestClient, admin: dict) -> None:
+        biz, receipt = _create_full_chain(
+            client,
+            admin["headers"],
+            items=_GOLD_ITEMS,
+            ship_date="2026-10-01",
+            receipt_date="2026-10-20",
+            amount="20000.00",
+        )
+        parts = [_person(client, admin["headers"]) for _ in range(2)]
+        self._add_plan(biz["id"], [(parts[0]["id"], "30"), (parts[1]["id"], "40")])
+
+        resp = _calc(client, admin["headers"], receipt["id"])
+        assert resp.status_code == 200, resp.text
+
+        factory = get_session_factory()
+        with factory() as session:
+            rows = (
+                session.execute(
+                    select(PerformanceLedger).where(
+                        PerformanceLedger.receipt_id == receipt["id"],
+                        PerformanceLedger.is_active == 1,
+                    )
+                )
+                .scalars()
+                .all()
+            )
+            assert len(rows) == 3, "负责人+两参与人各一行"
+            by_pid = {r.personnel_id: r for r in rows}
+            a, b, o = (
+                by_pid[parts[0]["id"]],
+                by_pid[parts[1]["id"]],
+                by_pid[biz["owner_personnel_id"]],
+            )
+            # 绩效份额：A=39.11×30%=11.73、B=39.11×40%=15.64、负责人=守恒余额 11.74
+            assert (a.performance_amount, b.performance_amount, o.performance_amount) == (
+                Decimal("11.73"),
+                Decimal("15.64"),
+                Decimal("11.74"),
+            )
+            # 行合计 == 业务级全额（守恒是分摊红线：多一分少一分都是错账）
+            assert a.payable_amount + b.payable_amount + o.payable_amount == Decimal("39.11")
+            # 折算额同比例拆分：4000 的 30%/40%/余额
+            assert (a.converted_profit, b.converted_profit, o.converted_profit) == (
+                Decimal("1200.00"),
+                Decimal("1600.00"),
+                Decimal("1200.00"),
+            )
+            # 行级身份固化：参与人各自岗位；快照带方案与角色标签
+            # （ratio 为 NUMERIC(5,2) 读回两位 scale，断言用数值比较）
+            assert a.role_code == b.role_code == "salesperson"
+            assert a.snapshot_json["alloc_role"] == "participant"
+            assert Decimal(a.snapshot_json["alloc_ratio"]) == Decimal("30")
+            assert o.snapshot_json["alloc_role"] == "owner"
+            assert Decimal(o.snapshot_json["alloc_plan"][0]["ratio"]) == Decimal("30")
+
+    def test_分摊后第二笔回款仅登记进度(self, client: TestClient, admin: dict) -> None:
+        """一次性计提（#25）与分摊叠加：第二笔不新增行、不重复拆分。"""
+        biz, first = _create_full_chain(
+            client,
+            admin["headers"],
+            items=_GOLD_ITEMS,
+            ship_date="2026-10-01",
+            receipt_date="2026-10-20",
+            amount="10000.00",
+        )
+        part = _person(client, admin["headers"])
+        self._add_plan(biz["id"], [(part["id"], "50")])
+        assert _calc(client, admin["headers"], first["id"]).status_code == 200
+
+        resp = client.post(
+            _RECEIPTS,
+            json={
+                "business_id": biz["id"],
+                "receipt_date": "2026-10-25",
+                "amount": "5000.00",
+            },
+            headers=admin["headers"],
+        )
+        assert resp.status_code == 200, resp.text
+        second = resp.json()["data"]
+        out = _calc(client, admin["headers"], second["id"]).json()["data"]
+        assert out["repeated"] is True
+
+        factory = get_session_factory()
+        with factory() as session:
+            rows = (
+                session.execute(
+                    select(PerformanceLedger).where(
+                        PerformanceLedger.business_id == biz["id"],
+                        PerformanceLedger.is_active == 1,
+                    )
+                )
+                .scalars()
+                .all()
+            )
+            assert len(rows) == 2, "第二笔回款不得新增拆分行（一次性计提）"
+
+    def test_重算作废全部拆分行并落新版本(
+        self, client: TestClient, admin: dict, finance: dict
+    ) -> None:
+        biz, receipt = _create_full_chain(
+            client,
+            admin["headers"],
+            items=_GOLD_ITEMS,
+            ship_date="2026-10-01",
+            receipt_date="2026-10-20",
+            amount="20000.00",
+        )
+        parts = [_person(client, admin["headers"]) for _ in range(2)]
+        self._add_plan(biz["id"], [(parts[0]["id"], "30"), (parts[1]["id"], "40")])
+        assert _calc(client, admin["headers"], receipt["id"]).status_code == 200
+
+        resp = client.post(
+            f"{_CALC}/receipts/{receipt['id']}/recalc",
+            json={"reason": "用例：分摊重算"},
+            headers=finance["headers"],
+        )
+        assert resp.status_code == 200, resp.text
+
+        factory = get_session_factory()
+        with factory() as session:
+            rows = (
+                session.execute(
+                    select(PerformanceLedger).where(PerformanceLedger.receipt_id == receipt["id"])
+                )
+                .scalars()
+                .all()
+            )
+            active = [r for r in rows if r.is_active == 1]
+            inactive = [r for r in rows if r.is_active == 0]
+            assert len(active) == 3 and len(inactive) == 3
+            assert {r.calc_version for r in active} == {2}, "新版本三行"
+            assert {r.calc_version for r in inactive} == {1}, "旧版本三行全部作废"
+
+    def test_合计超100核算侧拦截(self, client: TestClient, admin: dict) -> None:
+        biz, receipt = _create_full_chain(
+            client,
+            admin["headers"],
+            items=_GOLD_ITEMS,
+            ship_date="2026-10-01",
+            receipt_date="2026-10-20",
+            amount="100.00",
+        )
+        part = _person(client, admin["headers"])
+        self._add_plan(biz["id"], [(part["id"], "60")])
+        # 再直插另一行绕过保存校验（合计 110），核算侧防御必须拦
+        other = _person(client, admin["headers"])
+        factory = get_session_factory()
+        with factory() as session:
+            person = session.get(Personnel, other["id"])
+            session.add(
+                SecondaryAllocation(
+                    business_id=biz["id"],
+                    personnel_id=other["id"],
+                    personnel_name=person.personnel_name,
+                    ratio=Decimal("50"),
+                )
+            )
+            session.commit()
+
+        resp = _calc(client, admin["headers"], receipt["id"])
+        assert resp.status_code == 400
+        assert "超过 100%" in resp.json()["message"]
+
+    def test_参与人含负责人核算侧拦截(self, client: TestClient, admin: dict) -> None:
+        biz, receipt = _create_full_chain(
+            client,
+            admin["headers"],
+            items=_GOLD_ITEMS,
+            ship_date="2026-10-01",
+            receipt_date="2026-10-20",
+            amount="100.00",
+        )
+        self._add_plan(biz["id"], [(biz["owner_personnel_id"], "30")])
+        resp = _calc(client, admin["headers"], receipt["id"])
+        assert resp.status_code == 400
+        assert "含合同负责营销人员" in resp.json()["message"]