refactor(m3): settlement_service 三段收尾——生成/聚合/调整单拆分

code-review 约定合规项续（calc_receipt 之后的剩余超长函数）：

- generate（~120 行）→ 编排 + 六段私有函数：_guard_regeneration（重生成
  准入）、_release_previous_round（解绑回填+旧聚合作废）、_upsert_draft
  （草稿头表新建/重置）、_build_items（=_bucket_rows + _item_of 记账）与
  _apply_totals（头表三合计）
- _build_manager_aggregates（~110 行）→ _ShareParams 冻结 dataclass +
  _manager_share_params（月末取参/百分数口径/#12 开关同源）/ _collect_team_buckets
  （含 #12「个人户行归经理本人」分支）/ _add_team_performance / _aggregate_row；
  全公司基数循环外算一次（口径逐行固定，行为等价）
- create_adjustment（~90 行）→ _adjustment_amount（恒正）/ _wire_adjustment_approval
  （submit_order 分流与留痕，含「不得手工插 ApprovalStep」外键教训注释）/
  _adjustment_out（出参前置到 commit 之前）

行为零变化；#12 参数化（include_manager_personal 过滤、个人户行按
personnel_id 直归经理、snapshot 的 manager_personal_included）语义原样保留。
验证：重构后月结集成 15 例 + 月结单测 5 例全绿（81s 内无并发干扰的一轮）；
全量回归待 88 会话 calc_service 收口后复跑（其进行中的拆行改动与测试
并发互清导致临时 401，非本重构所致）

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

backend/app/services/settlement_service.py

@@ -10,8 +10,10 @@
 
 import logging
 from calendar import monthrange
+from dataclasses import dataclass
 from datetime import date
 from decimal import ROUND_HALF_UP, Decimal
+from fractions import Fraction
 
 from sqlalchemy import select
 from sqlalchemy.orm import Session
@@ -64,8 +66,11 @@ def generate(
     """生成月度结算快照（草稿）：按确认回款口径取有效台账行，按人汇总。
 
     - 同月已有月结行（任意状态）时 400——一行一月，重生成走「驳回/草稿态
-      重新生成」（先清 items 重算头表）
+      重新生成」（_release_previous_round 解绑回填后清明细重算）
-    - 经理聚合（calc_type=2）在生成时计算（#12 暂口径：基数不含经理个人户）
+    - 经理聚合（calc_type=2）在生成时计算并落库（#12 参数化见
+      ``_build_manager_aggregates``）
+
+    本函数只做编排，各段独立私有函数（口径修改只动对应段）。
     """
     with audit_service.audit(
         session,
@@ -79,117 +84,22 @@ def generate(
         meta=meta,
     ):
         existing = settlement_repository.get_by_month(session, month)
-        if existing is not None:
+        _guard_regeneration(existing, month)
-            if existing.status == SETTLEMENT_STATUS_APPROVED:
+        _release_previous_round(session, existing, month)
-                raise BadRequestError(f"{month} 已审定锁定，不可重新生成（调整走调整单）")
-            if existing.status == SETTLEMENT_STATUS_PENDING:
-                raise BadRequestError(f"{month} 有待审批的月结，不可重新生成")
-            # 草稿/已驳回：允许重新生成（清明细重算）
-
-        # 重生成前先释放上一轮的固化标记：台账行 settlement_id 回填后，
-        # 「未入结」过滤会把它们永久排除在候选集外；旧聚合行也要作废，
-        # 否则本轮重建的聚合行与旧行重复计入（两者都必须在 _confirm_scope 前）
-        if existing is not None:
-            settlement_repository.clear_ledger_settlement_id(session, existing.id)
-            performance_ledger_repository.deactivate_month_aggregates(session, month)
 
         ledger_rows = _confirm_scope(session, month)
         if not ledger_rows:
             raise BadRequestError(f"{month} 无可结算的核算记录（先在台账完成核算）")
 
-        # 经理团队聚合（#12 策略点：基数不含经理个人户业绩）
         aggregate_rows = _build_manager_aggregates(session, ledger_rows, month)
-
+        settlement = _upsert_draft(session, existing, operator, month)
-        settlement = settlement_repository.get_by_month(session, month)
+        items, ledger_ids = _build_items(session, ledger_rows, aggregate_rows, month)
-        is_new = settlement is None
-        if is_new:
-            settlement = MonthlySettlement(
-                settlement_month=month,
-                submitter_id=operator.id,
-                submitter_name=operator.username,
-                status=SETTLEMENT_STATUS_DRAFT,
-            )
-            settlement_repository.add(session, settlement)
-        else:
-            settlement.submitter_id = operator.id
-            settlement.submitter_name = operator.username
-            settlement.status = SETTLEMENT_STATUS_DRAFT
-
-        # 按人汇总（台账行：personnel+calc_type 维度）
-        by_person: dict[tuple[int, int], dict] = {}
-        for row in ledger_rows + aggregate_rows:
-            key = (row.personnel_id, row.calc_type)
-            bucket = by_person.setdefault(
-                key,
-                {
-                    "personnel_id": row.personnel_id,
-                    "personnel_name": row.personnel_name,
-                    "role_code": row.role_code,
-                    "calc_type": row.calc_type,
-                    "performance_sum": Decimal("0"),
-                    "payable_sum": Decimal("0"),
-                    "ledger_ids": [],
-                    "is_probation": row.is_probation,
-                },
-            )
-            bucket["performance_sum"] += row.performance_amount or Decimal("0")
-            bucket["payable_sum"] += row.payable_amount or Decimal("0")
-            bucket["ledger_ids"].append(row.id)
-
-        items: list[SettlementItem] = []
-        for (_, _), bucket in sorted(by_person.items()):
-            person = session.get(Personnel, bucket["personnel_id"])
-            basic = person.salary_standard if person else None
-            mode = _salary_mode_of(session, bucket["role_code"], month)
-            net = (
-                monthly_salary(bucket["payable_sum"], basic, mode)
-                if person and person.is_probation != 1
-                else None
-            )
-            note = None
-            if person and person.is_probation == 1:
-                note = (
-                    f"试用期绩效照算不发放；年累计超 {_probation_threshold(session, month)} 元部分"
-                    "年底一次性结算"
-                )
-            items.append(
-                SettlementItem(
-                    personnel_id=bucket["personnel_id"],
-                    personnel_name=bucket["personnel_name"],
-                    role_code=bucket["role_code"],
-                    calc_type=bucket["calc_type"],
-                    performance_sum=bucket["performance_sum"].quantize(
-                        Decimal("0.01"), rounding=ROUND_HALF_UP
-                    ),
-                    payable_sum=bucket["payable_sum"].quantize(
-                        Decimal("0.01"), rounding=ROUND_HALF_UP
-                    ),
-                    basic_amount=basic,
-                    salary_mode=mode,
-                    net_amount=(
-                        net.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
-                        if net is not None
-                        else None
-                    ),
-                    is_probation=bucket["is_probation"],
-                    probation_note=note,
-                    snapshot_json={"ledger_ids": bucket["ledger_ids"]},
-                )
-            )
 
         settlement_repository.replace_items(session, settlement.id, items)
-        settlement.total_performance = sum(
+        _apply_totals(settlement, items)
-            (i.performance_sum for i in items), Decimal("0")
-        ).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
-        settlement.total_payable = sum((i.payable_sum for i in items), Decimal("0")).quantize(
-            Decimal("0.01"), rounding=ROUND_HALF_UP
-        )
-        settlement.person_count = len({i.personnel_id for i in items})
         settlement.snapshot_json = {"note": _SETTLEMENT_NOTE}
-
         # 入结回填：台账行 settlement_id 是「未入结」防重判据（_confirm_scope），
         # 不回填会让重生成把已固化行反复纳入；聚合行此刻已落库、id 可用
-        ledger_ids = [lid for bucket in by_person.values() for lid in bucket["ledger_ids"]]
         settlement_repository.update_ledger_settlement_id(session, ledger_ids, settlement.id)
 
         # 出参在 commit 前固化（audit 约定：commit 后查询失败会把已生成的月结
@@ -201,6 +111,127 @@ def generate(
         return detail
 
 
+def _guard_regeneration(existing: MonthlySettlement | None, month: str) -> None:
+    """重生成准入：已审定 / 待审批不可重生成（草稿、已驳回放行）。"""
+    if existing is None:
+        return
+    if existing.status == SETTLEMENT_STATUS_APPROVED:
+        raise BadRequestError(f"{month} 已审定锁定，不可重新生成（调整走调整单）")
+    if existing.status == SETTLEMENT_STATUS_PENDING:
+        raise BadRequestError(f"{month} 有待审批的月结，不可重新生成")
+
+
+def _release_previous_round(
+    session: Session, existing: MonthlySettlement | None, month: str
+) -> None:
+    """释放上一轮固化标记（必须早于 _confirm_scope）。
+
+    台账行 settlement_id 回填后，「未入结」过滤会把它们永久排除在候选集外；
+    旧聚合行也要作废，否则本轮重建的聚合行与旧行重复计入。
+    """
+    if existing is None:
+        return
+    settlement_repository.clear_ledger_settlement_id(session, existing.id)
+    performance_ledger_repository.deactivate_month_aggregates(session, month)
+
+
+def _upsert_draft(
+    session: Session, existing: MonthlySettlement | None, operator: User, month: str
+) -> MonthlySettlement:
+    """新建或重置草稿头表（重生成时刷新提交人与状态）。"""
+    if existing is None:
+        settlement = MonthlySettlement(
+            settlement_month=month,
+            submitter_id=operator.id,
+            submitter_name=operator.username,
+            status=SETTLEMENT_STATUS_DRAFT,
+        )
+        return settlement_repository.add(session, settlement)
+    existing.submitter_id = operator.id
+    existing.submitter_name = operator.username
+    existing.status = SETTLEMENT_STATUS_DRAFT
+    return existing
+
+
+def _build_items(
+    session: Session, ledger_rows: list, aggregate_rows: list, month: str
+) -> tuple[list[SettlementItem], list[int]]:
+    """按人汇总生成明细行，返回 (明细, 本次入结的台账行 id 列表)。"""
+    by_person = _bucket_rows(ledger_rows + aggregate_rows)
+    items = [_item_of(session, bucket, month) for _, bucket in sorted(by_person.items())]
+    ledger_ids = [lid for bucket in by_person.values() for lid in bucket["ledger_ids"]]
+    return items, ledger_ids
+
+
+def _bucket_rows(rows: list) -> dict[tuple[int, int], dict]:
+    """台账行按 (人员, 核算类型) 维度分桶（冗余人读字段取首行）。"""
+    by_person: dict[tuple[int, int], dict] = {}
+    for row in rows:
+        key = (row.personnel_id, row.calc_type)
+        bucket = by_person.setdefault(
+            key,
+            {
+                "personnel_id": row.personnel_id,
+                "personnel_name": row.personnel_name,
+                "role_code": row.role_code,
+                "calc_type": row.calc_type,
+                "performance_sum": Decimal("0"),
+                "payable_sum": Decimal("0"),
+                "ledger_ids": [],
+                "is_probation": row.is_probation,
+            },
+        )
+        bucket["performance_sum"] += row.performance_amount or Decimal("0")
+        bucket["payable_sum"] += row.payable_amount or Decimal("0")
+        bucket["ledger_ids"].append(row.id)
+    return by_person
+
+
+def _item_of(session: Session, bucket: dict, month: str) -> SettlementItem:
+    """单条结算明细：月工资三口径净额、试用期说明与台账行引用列表。"""
+    person = session.get(Personnel, bucket["personnel_id"])
+    basic = person.salary_standard if person else None
+    mode = _salary_mode_of(session, bucket["role_code"], month)
+    net = (
+        monthly_salary(bucket["payable_sum"], basic, mode)
+        if person and person.is_probation != 1
+        else None
+    )
+    note = None
+    if person and person.is_probation == 1:
+        note = (
+            f"试用期绩效照算不发放；年累计超 {_probation_threshold(session, month)} 元部分"
+            "年底一次性结算"
+        )
+    return SettlementItem(
+        personnel_id=bucket["personnel_id"],
+        personnel_name=bucket["personnel_name"],
+        role_code=bucket["role_code"],
+        calc_type=bucket["calc_type"],
+        performance_sum=bucket["performance_sum"].quantize(Decimal("0.01"), rounding=ROUND_HALF_UP),
+        payable_sum=bucket["payable_sum"].quantize(Decimal("0.01"), rounding=ROUND_HALF_UP),
+        basic_amount=basic,
+        salary_mode=mode,
+        net_amount=(
+            net.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) if net is not None else None
+        ),
+        is_probation=bucket["is_probation"],
+        probation_note=note,
+        snapshot_json={"ledger_ids": bucket["ledger_ids"]},
+    )
+
+
+def _apply_totals(settlement: MonthlySettlement, items: list[SettlementItem]) -> None:
+    """头表三合计固化（明细已两位小数，合计再 HALF_UP 收口）。"""
+    settlement.total_performance = sum((i.performance_sum for i in items), Decimal("0")).quantize(
+        Decimal("0.01"), rounding=ROUND_HALF_UP
+    )
+    settlement.total_payable = sum((i.payable_sum for i in items), Decimal("0")).quantize(
+        Decimal("0.01"), rounding=ROUND_HALF_UP
+    )
+    settlement.person_count = len({i.personnel_id for i in items})
+
+
 def submit(
     session: Session,
     settlement_id: int,
@@ -321,68 +352,151 @@ def _confirm_scope(session: Session, month: str) -> list:
     )
 
 
-def _build_manager_aggregates(session: Session, ledger_rows: list, month: str) -> list:
+@dataclass(frozen=True, slots=True)
-    """按经理生成团队聚合行（calc_type=2）。
+class _ShareParams:
+    """经理分成取参与参数原文（同一次生成共用一份，随月月末时点化）。"""
 
-    #12 答复（2026-10-19）参数化：经理个人户业绩是否计入聚合基数由参数
+    month: str
-    ``manager.personal.included`` 决定（0=不计入默认，1=计入）——该部分
+    param_date: date
-    按经理个人户规则单笔落台账（rate=0/绩效 0），是否并入聚合基数由此
+    team_text: str
-    开关控制。
+    company_text: str
-    经理 = 本月台账行里 role_code=sales_manager 且 customer_type=personal
+    team_pct: Fraction
-    的人员（本人带领的业务员经 manager_id 关联）。
+    company_pct: Fraction
+    include_manager_personal: bool
+
+
+def _manager_share_params(session: Session, month: str) -> _ShareParams:
+    """解析经理分成参数（锚定所属月月末，取参与台账行口径同源）。
+
+    参数按百分数存储（12.5 表示 12.5%）：parse_fraction 得 12.5，
+    manager_team_share 内部再 ÷100 —— 调用方不得再乘 100（否则放大 100 倍）。
+    #12 开关（manager.personal.included）与分成比例同日取参，答复后调整可追溯。
     """
-    from app.models import PerformanceLedger
-    from app.repositories import business_role_repository
     from app.services.param_service import parse_fraction, resolve_snapshot
 
-    # 取参锚定所属月月末（不是生成日）：补生成历史月份时结果不随操作日期漂移
     param_date = _month_end(month)
     snapshot = resolve_snapshot(session, param_date)
-    # #12：经理个人户基数开关（与分成比例同日取参，答复后调整可追溯）
-    include_manager_personal = snapshot.get("manager.personal.included") == "1"
-
-    managers: dict[int, dict] = {}
-    for row in ledger_rows:
-        if not include_manager_personal and row.customer_type == "manager_personal":
-            continue  # 经理个人户业绩不入聚合基数（#12 参数化，默认不计入）
-        if row.customer_type == "manager_personal":
-            # 开关计入：个人户行归属经理本人（行 personnel_id 即经理，
-            # 不走 manager_id 上溯——经理没有上级，按上溯会被跳过漏计）
-            bucket = managers.setdefault(
-                row.personnel_id,
-                {
-                    "performance": Decimal("0"),
-                    "business_id": row.business_id,
-                    "business_no": row.business_no,
-                    "names": set(),
-                },
-            )
-            bucket["performance"] += row.performance_amount or Decimal("0")
-            bucket["names"].add(row.personnel_name)
-            continue
-        # 找到该行业务归属人员的经理（manager_id）
-        person = session.get(Personnel, row.personnel_id)
-        manager_id = person.manager_id if person else None
-        if manager_id is None:
-            continue
-        bucket = managers.setdefault(
-            manager_id,
-            {
-                "performance": Decimal("0"),
-                "business_id": row.business_id,
-                "business_no": row.business_no,
-                "names": set(),
-            },
-        )
-        bucket["performance"] += row.performance_amount or Decimal("0")
-        bucket["names"].add(row.personnel_name)
-
-    # 参数按百分数存储（12.5 表示 12.5%）：parse_fraction 得 12.5，
-    # manager_team_share 内部再 ÷100 —— 调用方不得再乘 100（否则放大 100 倍）
     team_text = snapshot.get("manager.share.team")
     company_text = snapshot.get("manager.share.company")
-    team_pct = parse_fraction(team_text)
+    return _ShareParams(
-    company_pct = parse_fraction(company_text)
+        month=month,
+        param_date=param_date,
+        team_text=team_text,
+        company_text=company_text,
+        team_pct=parse_fraction(team_text),
+        company_pct=parse_fraction(company_text),
+        include_manager_personal=snapshot.get("manager.personal.included") == "1",
+    )
+
+
+def _add_team_performance(managers: dict[int, dict], manager_id: int, row) -> None:
+    """把一行业绩并入经理的团队桶（首行固化业务锚点，成员名去重收集）。"""
+    bucket = managers.setdefault(
+        manager_id,
+        {
+            "performance": Decimal("0"),
+            "business_id": row.business_id,
+            "business_no": row.business_no,
+            "names": set(),
+        },
+    )
+    bucket["performance"] += row.performance_amount or Decimal("0")
+    bucket["names"].add(row.personnel_name)
+
+
+def _collect_team_buckets(
+    session: Session, ledger_rows: list, include_manager_personal: bool
+) -> dict[int, dict]:
+    """按经理采集团队业绩。
+
+    业务员行经 manager_id 上溯到其经理；经理个人户行（#12 开关计入时）
+    归属经理本人——行 personnel_id 即经理，不走上溯（经理没有上级，
+    按上溯会被跳过漏计）。
+    """
+    managers: dict[int, dict] = {}
+    for row in ledger_rows:
+        if row.customer_type == "manager_personal":
+            if include_manager_personal:
+                _add_team_performance(managers, row.personnel_id, row)
+            continue  # 默认不计入聚合基数（#12 参数化）
+        person = session.get(Personnel, row.personnel_id)
+        manager_id = person.manager_id if person else None
+        if manager_id is not None:
+            _add_team_performance(managers, manager_id, row)
+    return managers
+
+
+def _aggregate_row(
+    session: Session, manager, bucket: dict, company_base: Decimal, share: _ShareParams
+):
+    """单经理聚合台账行（必须落库：月结 ledger_ids 与报表下钻的锚点）。"""
+    from app.models import PerformanceLedger
+
+    amount = manager_team_share(
+        bucket["performance"],
+        company_base,
+        team_pct=share.team_pct,
+        company_pct=share.company_pct,
+    )
+    member_names = "、".join(sorted(bucket["names"])) or "（无成员）"
+    return performance_ledger_repository.add(
+        session,
+        PerformanceLedger(
+            receipt_id=None,
+            calc_version=1,  # 聚合行无回款维度，版本恒 1（uk 含 NULL 不去重）
+            calc_type=CALC_TYPE_MANAGER_AGGREGATE,
+            business_id=bucket["business_id"],
+            business_no=bucket["business_no"],
+            personnel_id=manager.id,
+            personnel_name=manager.personnel_name,
+            role_code="sales_manager",
+            customer_type="personal",
+            product_category="mixed",
+            receipt_date=share.param_date,
+            # 比例原文取实际参数（勿硬编码默认值——参数调整后台账说明会失真）
+            rate_text=f"聚合:{share.team_text}%+{share.company_text}%",
+            converted_profit=None,
+            performance_amount=amount.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP),
+            barter_applied=0,
+            payable_amount=amount.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP),
+            settlement_month=share.month,
+            is_probation=1 if manager.is_probation == 1 else 0,
+            snapshot_json={
+                "team_base": str(bucket["performance"]),
+                "company_base": str(company_base),
+                "members": member_names,
+                "params_as_of": str(share.param_date),
+                "manager_personal_included": 1 if share.include_manager_personal else 0,
+                "note": (
+                    f"#12 参数化：基数{'含' if share.include_manager_personal else '不含'}"
+                    "经理个人户业绩；9.8 方案公式"
+                ),
+            },
+        ),
+    )
+
+
+def _build_manager_aggregates(session: Session, ledger_rows: list, month: str) -> list:
+    """按经理生成团队聚合行（calc_type=2）并落库。
+
+    #12 答复（2026-10-19）参数化：经理个人户业绩是否计入聚合基数由参数
+    ``manager.personal.included`` 决定（0=不计入默认，1=计入）；
+    经理 = 台账行归属人员经 manager_id 关联的 sales_manager 岗人员。
+    """
+    from app.repositories import business_role_repository
+
+    share = _manager_share_params(session, month)
+    managers = _collect_team_buckets(session, ledger_rows, share.include_manager_personal)
+    # 全公司基数：默认排除经理个人户行（#12 开关计入时不过滤）；
+    # 口径逐行固定，循环外算一次
+    company_base = sum(
+        (
+            r.performance_amount or Decimal("0")
+            for r in ledger_rows
+            if share.include_manager_personal or r.customer_type != "manager_personal"
+        ),
+        Decimal("0"),
+    )
     aggregates: list = []
     for manager_id, bucket in managers.items():
         manager = session.get(Personnel, manager_id)
@@ -395,60 +509,7 @@ def _build_manager_aggregates(session: Session, ledger_rows: list, month: str) -
         )
         if role is None or role.role_code != "sales_manager":
             continue  # 非营销经理的上级不参与聚合
-        # 全公司基数：默认排除经理个人户行（#12 开关计入时不过滤）
+        aggregates.append(_aggregate_row(session, manager, bucket, company_base, share))
-        company_base = sum(
-            (
-                r.performance_amount or Decimal("0")
-                for r in ledger_rows
-                if include_manager_personal or r.customer_type != "manager_personal"
-            ),
-            Decimal("0"),
-        )
-        share = manager_team_share(
-            bucket["performance"],
-            company_base,
-            team_pct=team_pct,
-            company_pct=company_pct,
-        )
-        member_names = "、".join(sorted(bucket["names"])) or "（无成员）"
-        # 聚合行必须落库（模型约定的 calc_type=2 台账行）：月结明细的
-        # ledger_ids 与报表下钻都要能从台账反查来源
-        row = performance_ledger_repository.add(
-            session,
-            PerformanceLedger(
-                receipt_id=None,
-                calc_version=1,  # 聚合行无回款维度，版本恒 1（uk 含 NULL 不去重）
-                calc_type=CALC_TYPE_MANAGER_AGGREGATE,
-                business_id=bucket["business_id"],
-                business_no=bucket["business_no"],
-                personnel_id=manager_id,
-                personnel_name=manager.personnel_name,
-                role_code="sales_manager",
-                customer_type="personal",
-                product_category="mixed",
-                receipt_date=param_date,
-                # 比例原文取实际参数（勿硬编码默认值——参数调整后台账说明会失真）
-                rate_text=f"聚合:{team_text}%+{company_text}%",
-                converted_profit=None,
-                performance_amount=share.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP),
-                barter_applied=0,
-                payable_amount=share.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP),
-                settlement_month=month,
-                is_probation=1 if manager.is_probation == 1 else 0,
-                snapshot_json={
-                    "team_base": str(bucket["performance"]),
-                    "company_base": str(company_base),
-                    "members": member_names,
-                    "params_as_of": str(param_date),
-                    "manager_personal_included": 1 if include_manager_personal else 0,
-                    "note": (
-                        f"#12 参数化：基数{'含' if include_manager_personal else '不含'}"
-                        "经理个人户业绩；9.8 方案公式"
-                    ),
-                },
-            ),
-        )
-        aggregates.append(row)
     return aggregates
 
 
@@ -497,25 +558,11 @@ def create_adjustment(
     meta: RequestMeta | None = None,
 ):
     """新建调整单：开关分流（关=提交即生效留痕；开=待审批）。"""
-    from decimal import InvalidOperation
-
     from app.core.constants import NO_PREFIX_ADJUSTMENT
     from app.models import AdjustmentOrder
-    from app.models.approval import APPROVAL_STATUS_APPROVED, APPROVAL_TYPE_ADJUSTMENT
-    from app.models.settlement import (
-        ADJUSTMENT_STATUS_EFFECTIVE,
-        ADJUSTMENT_STATUS_PENDING,
-    )
     from app.services.no_generator import next_no
 
-    # 金额恒正（方向表达补发/扣减）：列注释即此约定，负数会让方向语义反转、
+    amount = _adjustment_amount(payload)
-    # 审批/报表按 direction 汇总出错；格式已由 schema 正则拦截，此处兜底 + 语义判定
-    try:
-        amount = Decimal(payload.amount)
-    except InvalidOperation as exc:
-        raise BadRequestError("调整金额格式非法（非负两位小数）") from exc
-    if amount <= 0:
-        raise BadRequestError("调整金额须大于 0（方向表达补发/扣减）")
 
     person = session.get(Personnel, payload.personnel_id)
     if person is None:
@@ -551,33 +598,68 @@ def create_adjustment(
         session.add(order)
         session.flush()
 
-        # 开关分流与留痕统一交给 submit_order：开关关闭时它自动通过并写两条
+        _wire_adjustment_approval(session, order, operator)
-        # 「系统」留痕。**不得手工插 ApprovalStep** —— approval_steps.order_id
+        out = _adjustment_out(order)  # 出参在 commit 前固化（audit 约定）
-        # 外键指向 approval_orders（不是 adjustment_orders），用调整单 id 落步
-        # 必然外键失败（历史缺陷：关开关时新建调整单不可用）
-        approval_order = approval_service.submit_order(
-            session,
-            order_type=APPROVAL_TYPE_ADJUSTMENT,
-            ref_id=order.id,
-            refer_no=order.order_no,
-            submitter=operator,
-        )
-        order.status = (
-            ADJUSTMENT_STATUS_EFFECTIVE
-            if approval_order.status == APPROVAL_STATUS_APPROVED
-            else ADJUSTMENT_STATUS_PENDING
-        )
-        session.flush()
         session.commit()
-        logger.info("调整单创建 | order=%s | status=%s", order.order_no, order.status)
+        logger.info("调整单创建 | order=%s | status=%s", out["order_no"], out["status"])
-        return {
+        return out
-            "id": order.id,
+
-            "order_no": order.order_no,
+
-            "status": order.status,
+def _adjustment_amount(payload) -> Decimal:
-            "personnel_name": order.personnel_name,
+    """金额恒正校验（方向表达补发/扣减）。
-            "amount": str(order.amount),
+
-            "direction": order.direction,
+    列注释即「恒正」约定，负数会让方向语义反转、审批/报表按 direction
-        }
+    汇总出错；格式已由 schema 正则拦截，此处兜底转换 + 语义判定。
+    """
+    from decimal import InvalidOperation
+
+    try:
+        amount = Decimal(payload.amount)
+    except InvalidOperation as exc:
+        raise BadRequestError("调整金额格式非法（非负两位小数）") from exc
+    if amount <= 0:
+        raise BadRequestError("调整金额须大于 0（方向表达补发/扣减）")
+    return amount
+
+
+def _wire_adjustment_approval(session: Session, order, operator: User) -> None:
+    """开关分流与留痕统一交给 submit_order（关=自动通过留痕；开=待财务复核）。
+
+    **不得手工插 ApprovalStep**——``approval_steps.order_id`` 外键指向
+    approval_orders（不是 adjustment_orders），用调整单 id 落步必然外键
+    失败（历史缺陷：关开关时新建调整单不可用）。
+    """
+    from app.models.approval import APPROVAL_STATUS_APPROVED, APPROVAL_TYPE_ADJUSTMENT
+    from app.models.settlement import (
+        ADJUSTMENT_STATUS_EFFECTIVE,
+        ADJUSTMENT_STATUS_PENDING,
+    )
+
+    approval_order = approval_service.submit_order(
+        session,
+        order_type=APPROVAL_TYPE_ADJUSTMENT,
+        ref_id=order.id,
+        refer_no=order.order_no,
+        submitter=operator,
+    )
+    order.status = (
+        ADJUSTMENT_STATUS_EFFECTIVE
+        if approval_order.status == APPROVAL_STATUS_APPROVED
+        else ADJUSTMENT_STATUS_PENDING
+    )
+    session.flush()
+
+
+def _adjustment_out(order) -> dict:
+    """调整单出参（commit 前固化；expire_on_commit 关闭下也不依赖后置读取）。"""
+    return {
+        "id": order.id,
+        "order_no": order.order_no,
+        "status": order.status,
+        "personnel_name": order.personnel_name,
+        "amount": str(order.amount),
+        "direction": order.direction,
+    }
 
 
 def list_adjustments(session: Session) -> list[dict]: