feat(pl): 按装送段分别计量双证额度并保留历史单侧口径。

backend/app/api/pl/transport.py

@@ -885,6 +885,8 @@ def arrange_transport(
             selected_escort_id=transport.escort_id,
         )
         apply_resource_snapshot_json(plan=plan, transport=transport, snapshot=snapshot)
+    from app.services.pl_quota_planning import pin_measurement_version
+    pin_measurement_version(db, user, plan)
     plan.transport_status = "arranged"
     plan.follow_status = "arranged"
     plan.freight_amount = freight_amount or 0

backend/app/schemas/pl/quota_planning.py

@@ -31,6 +31,8 @@ class CertCondition(InputBody):
 
 
 class PlanQuotaConfiguration(InputBody):
+    # 可回传服务端值；不作为版本选择依据。
+    measurement_version: Optional[Literal['legacy_single_side', 'per_stop_v2']] = None
     source_mapping: Optional[List[SourcePair]] = None
     purchase_cert_ids: List[str] = Field(default_factory=list)
     transport_cert_ids: List[str] = Field(default_factory=list)

backend/app/services/pl_plan_service.py

@@ -1165,6 +1165,8 @@ def create_transport_for_plan(
         driver_id=driver_id,
         escort_id=escort_id,
     )
+    from app.services.pl_quota_planning import pin_measurement_version
+    pin_measurement_version(db, user, plan)
     plan.transport_status = "arranged"
     plan.follow_status = "arranged"
     plan.freight_amount = freight_amount or 0

backend/app/services/pl_quota_business.py

@@ -128,6 +128,8 @@ def apply_plan_quota(db, user, plan, action, payload=None):
                 [('transport', cid) for cid in config.transport_cert_ids])}
             validate_conditions(config, certs, plan_fingerprint(request.sources))
             plan.cert_configuration = config.model_dump(mode='json', by_alias=True)
+        plan.cert_configuration = {**(plan.cert_configuration or {}),
+                                   'measurementVersion': config.measurement_version}
         return result
     except ValueError as exc:
         raise HTTPException(409, str(exc)) from exc

backend/app/services/pl_quota_planning.py

@@ -45,16 +45,38 @@ def _plan(db, user, plan_id, lock=False):
     return row
 
 
+def resolve_measurement_version(db, user, plan):
+    """只信任持久化的服务端配置与账本，历史无版本记录保持原口径。"""
+    usages = db.query(PlPlanCertUsage).filter_by(tenant_no=user.tenant_no, plan_id=plan.id, del_flag=0).all()
+    if usages:
+        return ('per_stop_v2' if all((u.confirmation_snapshot or {}).get('measurementVersion') == 'per_stop_v2'
+                                   for u in usages) else 'legacy_single_side')
+    saved = (plan.cert_configuration or {}).get('measurementVersion')
+    if saved in ('per_stop_v2', 'legacy_single_side'):
+        return saved
+    if plan.transport_status == 'arranged' or plan.archive_status in ('archived', 'exception'):
+        return 'legacy_single_side'
+    return 'per_stop_v2'
+
+
+def pin_measurement_version(db, user, plan):
+    """安排改变状态前固化版本，与业务及额度在同一事务保存。"""
+    version = resolve_measurement_version(db, user, plan)
+    plan.cert_configuration = {**(plan.cert_configuration or {}), 'measurementVersion': version}
+    return version
+
+
 def build_usage_request(db, user, plan_id, operation, action, *, lock=False, check_conditions=True, preview_vehicle=None):
     """调用者先flush已授权的计划/实际量变更，再组装；不能将客户端UsageRequest直接传账本。"""
     op = operation if isinstance(operation, PlanQuotaOperation) else PlanQuotaOperation.model_validate(operation)
     plan = _plan(db, user, plan_id, lock)
     config = op.configuration or PlanQuotaConfiguration.model_validate(plan.cert_configuration or {})
+    config = config.model_copy(update={"measurement_version": resolve_measurement_version(db, user, plan)})
     mapping = None if config.source_mapping is None else [r.model_dump(by_alias=True) for r in config.source_mapping]
     if action == 'release':
         sources, actuals = [], {}
     else:
-        sources, actuals = extract_sources(db, user, plan_id, mapping, actual=action in ('settle', 'actual_adjust'), lock=lock, preview_vehicle=preview_vehicle)
+        sources, actuals = extract_sources(db, user, plan_id, mapping, actual=action in ('settle', 'actual_adjust'), lock=lock, preview_vehicle=preview_vehicle, measurement_version=config.measurement_version)
     keys = [('purchase', cid) for cid in config.purchase_cert_ids] + [('transport', cid) for cid in config.transport_cert_ids]
     if len(set(keys)) != len(keys):
         raise HTTPException(400, '证件选择重复')
@@ -82,8 +104,9 @@ def configuration_context(db, user, plan_id, config, preview_vehicle=None):
     config = config if isinstance(config, PlanQuotaConfiguration) else PlanQuotaConfiguration.model_validate(config)
     mapping = None if config.source_mapping is None else [r.model_dump(by_alias=True) for r in config.source_mapping]
     with db.no_autoflush:
-        sources, _ = extract_sources(db, user, plan_id, mapping, preview_vehicle=preview_vehicle)
-        return {'sourceFingerprint': plan_fingerprint(sources),
+        config = config.model_copy(update={"measurement_version": resolve_measurement_version(db, user, _plan(db, user, plan_id))})
+        sources, _ = extract_sources(db, user, plan_id, mapping, preview_vehicle=preview_vehicle, measurement_version=config.measurement_version)
+        return {'measurementVersion': config.measurement_version, 'sourceFingerprint': plan_fingerprint(sources),
                 'sources': [s.model_dump(mode='json') for s in sources]}
 
 
@@ -91,7 +114,7 @@ def preview_plan(db, user, plan_id, config, preview_vehicle=None):
     from app.services.pl_quota_usage_service import preview_usage
     config = config if isinstance(config, PlanQuotaConfiguration) else PlanQuotaConfiguration.model_validate(config)
     with db.no_autoflush:
-        request, _ = build_usage_request(db, user, plan_id, PlanQuotaOperation(operationId='preview', configuration=config),
+        request, config = build_usage_request(db, user, plan_id, PlanQuotaOperation(operationId='preview', configuration=config),
                                         'resize', check_conditions=False, preview_vehicle=preview_vehicle)
         result = preview_usage(db, user, request)
         fingerprint = plan_fingerprint(request.sources)
@@ -102,7 +125,7 @@ def preview_plan(db, user, plan_id, config, preview_vehicle=None):
             validate_conditions(config, certs, fingerprint)
         except HTTPException as error:
             reasons.append(error.detail)
-        return {**result, 'sourceFingerprint': fingerprint,
+        return {**result, 'measurementVersion': config.measurement_version, 'sourceFingerprint': fingerprint,
                 'sources': [s.model_dump(mode='json') for s in request.sources],
                 'status': 'needs_confirmation' if reasons else ('available' if request.sources else 'not_required'),
                 'reasons': reasons}
@@ -115,8 +138,9 @@ def save_configuration(db, user, plan_id, config):
     if plan.transport_status == 'arranged' or plan.archive_status == 'archived':
         raise HTTPException(409, '已安排计划请通过调整运输在同一事务重算用证')
     # 草稿允许缺少条件确认，但真实来源身份及选证权限必须有效。
+    config = config.model_copy(update={"measurement_version": resolve_measurement_version(db, user, plan)})
     mapping = None if config.source_mapping is None else [r.model_dump(by_alias=True) for r in config.source_mapping]
-    extract_sources(db, user, plan_id, mapping, lock=True)
+    extract_sources(db, user, plan_id, mapping, lock=True, measurement_version=config.measurement_version)
     for kind, ids in [('purchase', config.purchase_cert_ids), ('transport', config.transport_cert_ids)]:
         for cid in sorted(ids):
             _get_row(db, user, kind, cid)
@@ -132,18 +156,34 @@ def save_configuration(db, user, plan_id, config):
 def get_plan_quota(db, user, plan_id):
     with db.no_autoflush:
         plan, products, rows, stops = source_context(db, user, plan_id)
+        from app.services.pl_quota_contract_service import _contract
+        contract_numbers = {}
+        for cid in {r.contract_id for values in rows.values() for r in values.values() if r.contract_id}:
+            try:
+                contract_numbers[cid] = _contract(db, user, cid).contract_no
+            except HTTPException as exc:
+                if exc.status_code not in (403, 404):
+                    raise
+                contract_numbers[cid] = None
         options = []
         for side, values in rows.items():
             for row in values.values():
                 if products[row.plan_product_id].is_toxic == 1:
-                    options.append({'sourceStopType': side, 'sourceStopProductId': row.id,
+                    stop = stops[side][getattr(row, 'load_id' if side == 'load' else 'delivery_id')]
+                    options.append({'sourceStopId': stop.id, 'stopId': stop.id,
+                                    'stopOrder': getattr(stop, 'seq', None) or stop.order_num or 1,
+                                    'productId': row.product_id, 'productNo': row.product_no,
+                                    'sourceStopType': side, 'sourceStopProductId': row.id,
                                     'planProductId': row.plan_product_id, 'productName': row.product_name,
                                     'model': row.model, 'quantity': row.plan_qty, 'unit': row.unit,
-                                    'contractId': row.contract_id, 'contractProductId': row.contract_product_id,
-                                    'partyName': getattr(stops[side][getattr(row, 'load_id' if side == 'load' else 'delivery_id')], 'customer_name', None)})
+                                    'contractId': row.contract_id, 'contractNo': contract_numbers.get(row.contract_id),
+                                    'contractProductId': row.contract_product_id,
+                                    'partyName': stop.customer_name})
         usages = db.query(PlPlanCertUsage).filter_by(tenant_no=user.tenant_no, plan_id=plan_id, del_flag=0).order_by(PlPlanCertUsage.create_time.desc()).all()
+        version = resolve_measurement_version(db, user, plan)
         result = {'planId': plan.id, 'planNo': plan.plan_no, 'licensePlate': plan.license_plate,
-                  'configuration': plan.cert_configuration or {}, 'sourceOptions': options,
+                  'measurementVersion': version,
+                  'configuration': {**(plan.cert_configuration or {}), 'measurementVersion': version}, 'sourceOptions': options,
                   'canSaveConfiguration': plan.transport_status != 'arranged' and plan.archive_status != 'archived',
                   'usages': []}
         for usage in usages:

backend/app/services/pl_quota_sources.py

@@ -1,4 +1,4 @@
-"""从当前计划段行提取唯一用证计量侧，绝不按同 SKU 猜测装送配对。"""
+"""从真实装送段提取用证来源；保留历史单侧口径。"""
 import hashlib
 import json
 from decimal import Decimal
@@ -52,64 +52,72 @@ def _source_context(db, user, plan_id, lock=False):
     return plan, products, rows, stops
 
 
-def extract_sources(db, user, plan_id, mapping=None, *, actual=False, lock=False, preview_vehicle=None):
+def extract_sources(db, user, plan_id, mapping=None, *, actual=False, lock=False, preview_vehicle=None, measurement_version="legacy_single_side"):
     """mapping 仅选择真实段行及对侧追溯ID；数量/单位/合同/指纹全部由数据库生成。
 
-    单侧来源可直接确定；同一计划产品有装送两侧时，必须逐项显式声明计量侧与配对。
-    另一侧可关联多条计量行（例如一装多送），但不再次累计其数量。
+    per_stop_v2 自动提取每条装送段行，忽略旧客户端的单侧选择。
+    legacy_single_side 保留显式配对和唯一计量侧，避免历史账本静默重计。
     """
     plan, products, rows, stops = source_context(db, user, plan_id, lock)
     toxic = {pid for pid, row in products.items() if row.is_toxic == 1}
+    if measurement_version not in ("legacy_single_side", "per_stop_v2"):
+        raise HTTPException(400, "未知用证计量版本")
     if not toxic:
-        if mapping:
+        if mapping and measurement_version == "legacy_single_side":
             raise HTTPException(400, '该计划无需本次两证计量配置')
         return [], {}
     groups = {pid: {side: [r for r in values.values() if r.plan_product_id == pid]
                     for side, values in rows.items()} for pid in toxic}
-    if mapping is None:
-        mapping = []
-        for group in groups.values():
-            available = [side for side, values in group.items() if values]
-            if len(available) != 1:
-                raise HTTPException(409, '请明确装送产品配对及唯一用证计量侧')
-            side = available[0]
-            mapping.extend({'sourceStopType': side, 'sourceStopProductId': r.id,
-                            'relatedStopProductIds': []} for r in group[side])
-    chosen, seen, measured_sides = [], set(), {}
-    covered_opposite = {pid: set() for pid in toxic}
-    for entry in mapping:
-        side, rid = entry.get('sourceStopType'), entry.get('sourceStopProductId')
-        if side not in SIDE or (side, rid) in seen:
-            raise HTTPException(400, '计量来源无效或重复')
-        seen.add((side, rid))
-        row = rows[side].get(rid)
-        if row is None or row.plan_product_id not in toxic:
-            raise HTTPException(400, '计量来源不属于当前计划易制毒产品')
-        pid = row.plan_product_id
-        if pid in measured_sides and measured_sides[pid] != side:
-            raise HTTPException(400, '同一物理产品不能同时按装送两侧计量')
-        measured_sides[pid] = side
-        opposite = 'delivery' if side == 'load' else 'load'
-        ids = entry.get('relatedStopProductIds', [])
-        if not isinstance(ids, list) or len(ids) != len(set(ids)):
-            raise HTTPException(400, '对侧来源关联无效或重复')
-        related = []
-        for other_id in ids:
-            other = rows[opposite].get(other_id)
-            if other is None or other.plan_product_id != pid:
-                raise HTTPException(400, '对侧产品不属于当前计划产品明细')
-            related.append(other)
-        if groups[pid][opposite] and not related:
-            raise HTTPException(400, '存在对侧装送产品，请明确配对')
-        covered_opposite[pid].update(ids)
-        chosen.append((side, row, sorted(related, key=lambda r: r.id)))
-    for pid, group in groups.items():
-        side = measured_sides.get(pid)
-        if side is None or {r.id for r in group[side]} != {r.id for s, r, _ in chosen if s == side and r.plan_product_id == pid}:
-            raise HTTPException(400, '用证计量来源未覆盖全部计划产品段行')
-        other_side = 'delivery' if side == 'load' else 'load'
-        if covered_opposite[pid] != {r.id for r in group[other_side]}:
-            raise HTTPException(400, '装送来源配对未覆盖全部对侧段行')
+    if measurement_version == 'per_stop_v2':
+        chosen = [(side, row, []) for side, values in rows.items()
+                  for row in values.values() if row.plan_product_id in toxic]
+        if {row.plan_product_id for _, row, _ in chosen} != toxic:
+            raise HTTPException(409, '用证计量来源未覆盖全部计划产品段行')
+    else:
+        if mapping is None:
+            mapping = []
+            for group in groups.values():
+                available = [side for side, values in group.items() if values]
+                if len(available) != 1:
+                    raise HTTPException(409, '请明确装送产品配对及唯一用证计量侧')
+                side = available[0]
+                mapping.extend({'sourceStopType': side, 'sourceStopProductId': r.id,
+                                'relatedStopProductIds': []} for r in group[side])
+        chosen, seen, measured_sides = [], set(), {}
+        covered_opposite = {pid: set() for pid in toxic}
+        for entry in mapping:
+            side, rid = entry.get('sourceStopType'), entry.get('sourceStopProductId')
+            if side not in SIDE or (side, rid) in seen:
+                raise HTTPException(400, '计量来源无效或重复')
+            seen.add((side, rid))
+            row = rows[side].get(rid)
+            if row is None or row.plan_product_id not in toxic:
+                raise HTTPException(400, '计量来源不属于当前计划易制毒产品')
+            pid = row.plan_product_id
+            if pid in measured_sides and measured_sides[pid] != side:
+                raise HTTPException(400, '同一物理产品不能同时按装送两侧计量')
+            measured_sides[pid] = side
+            opposite = 'delivery' if side == 'load' else 'load'
+            ids = entry.get('relatedStopProductIds', [])
+            if not isinstance(ids, list) or len(ids) != len(set(ids)):
+                raise HTTPException(400, '对侧来源关联无效或重复')
+            related = []
+            for other_id in ids:
+                other = rows[opposite].get(other_id)
+                if other is None or other.plan_product_id != pid:
+                    raise HTTPException(400, '对侧产品不属于当前计划产品明细')
+                related.append(other)
+            if groups[pid][opposite] and not related:
+                raise HTTPException(400, '存在对侧装送产品，请明确配对')
+            covered_opposite[pid].update(ids)
+            chosen.append((side, row, sorted(related, key=lambda r: r.id)))
+        for pid, group in groups.items():
+            side = measured_sides.get(pid)
+            if side is None or {r.id for r in group[side]} != {r.id for s, r, _ in chosen if s == side and r.plan_product_id == pid}:
+                raise HTTPException(400, '用证计量来源未覆盖全部计划产品段行')
+            other_side = 'delivery' if side == 'load' else 'load'
+            if covered_opposite[pid] != {r.id for r in group[other_side]}:
+                raise HTTPException(400, '装送来源配对未覆盖全部对侧段行')
     sources, actuals = [], {}
     for side, row, related in sorted(chosen, key=lambda v: (v[0], v[1].id)):
         if row.plan_qty is None or row.plan_qty < 0 or not row.unit or not row.contract_id or not row.contract_product_id:
@@ -123,6 +131,8 @@ def extract_sources(db, user, plan_id, mapping=None, *, actual=False, lock=False
             stop_facts.append({key: getattr(parent, key, None) for key in ('id', 'supplier_id', 'supplier_name', 'customer_id', 'customer_name', 'address')})
         snapshot = {'source': _facts(row), 'related': [_facts(r) for r in related], 'stops': stop_facts,
                     'vehicle': plan.license_plate, 'trailer': plan.trailer_no, 'planType': plan.plan_type}
+        if measurement_version == 'per_stop_v2':
+            snapshot['measurementVersion'] = measurement_version
         if preview_vehicle is not None:
             snapshot['vehicle'] = preview_vehicle.get('licensePlate', plan.license_plate)
             snapshot['trailer'] = preview_vehicle.get('trailerNo', plan.trailer_no)

backend/app/services/pl_quota_usage_service.py

@@ -156,6 +156,9 @@ def _execute(db,user,r,digest,plan):
     else:
         supplement=_settle(db,user,r,certs,usages,items,plan)
     for k,u in usages.items():
+        u.confirmation_snapshot = {**(u.confirmation_snapshot or {}),
+            'measurementVersion': (u.confirmation_snapshot or {}).get('measurementVersion',
+                next((s.snapshot.get('measurementVersion', 'legacy_single_side') for s in r.sources), 'legacy_single_side'))}
         u.row_version=(u.row_version or 0)+1; u.update_by=user.id
         if certs[k].reserved_qty<0 or certs[k].used_qty<0 or certs[k].reserved_qty+certs[k].used_qty>certs[k].total_qty:
             raise ValueError('余额越界')
@@ -253,13 +256,14 @@ def _reserve(db,user,r,plan,certs,usages,items,selected,preview=False,lock_previ
         new_qty=sum((a.permit_quantity for a in allocations),ZERO)
         c.reserved_qty+=new_qty-u.reserved_qty; u.reserved_qty=new_qty; u.usage_status='reserved' if allocations else 'released'
         u.cert_revision=c.revision; u.confirmed_by=user.id; u.confirmed_time=datetime.now()
-        u.confirmation_snapshot={'executed':False,'vehicleNo':r.vehicle_no}
+        u.confirmation_snapshot={'executed':False,'vehicleNo':r.vehicle_no,
+                                 'measurementVersion': next((s.snapshot.get('measurementVersion', 'legacy_single_side') for s in r.sources), 'legacy_single_side')}
         for a in allocations:
             s=sources[a.source_id]; link,scope=scopes[(k,a.source_id)]
             item=previous.get(a.source_id)
             if item is None:
                 item=m.PlPlanCertUsageItem(**_audit(user),usage_id=u.id); db.add(item); items[k].append(item)
-            values=dict(contract_cert_link_id=link.id,contract_cert_product_id=scope.id,contract_scope_revision=link.scope_revision,contract_id=s.contract_id,contract_product_id=s.contract_product_id,plan_product_id=s.plan_product_id,source_stop_type=s.source_stop_type,source_stop_id=s.source_stop_id,source_stop_product_id=s.source_stop_product_id,related_stop_products=s.related_stop_products,allocation_seq=a.allocation_seq,source_unit=s.unit,planned_source_qty=s.quantity,allocated_source_qty=a.source_quantity,planned_cert_qty=a.permit_quantity,reserved_qty=a.permit_quantity,used_qty=ZERO,item_status='reserved',source_fingerprint=s.fingerprint,source_snapshot={**s.snapshot, 'productId': s.product_id},del_flag=0,reserve_input_mode='same_unit' if s.unit==c.quantity_unit else 'manual',reserve_conversion_note=notes.get((k[0],k[1],a.source_id)),reserve_confirmed_by=user.id,reserve_confirmed_time=datetime.now(),update_by=user.id)
+            values=dict(contract_cert_link_id=link.id,contract_cert_product_id=scope.id,contract_scope_revision=link.scope_revision,contract_id=s.contract_id,contract_product_id=s.contract_product_id,plan_product_id=s.plan_product_id,source_stop_type=s.source_stop_type,source_stop_id=s.source_stop_id,source_stop_product_id=s.source_stop_product_id,related_stop_products=s.related_stop_products,allocation_seq=a.allocation_seq,source_unit=s.unit,planned_source_qty=s.quantity,allocated_source_qty=a.source_quantity,planned_cert_qty=a.permit_quantity,reserved_qty=a.permit_quantity,used_qty=ZERO,item_status='reserved',source_fingerprint=s.fingerprint,source_snapshot={**s.snapshot, 'measurementVersion': s.snapshot.get('measurementVersion', 'legacy_single_side'), 'productId': s.product_id},del_flag=0,reserve_input_mode='same_unit' if s.unit==c.quantity_unit else 'manual',reserve_conversion_note=notes.get((k[0],k[1],a.source_id)),reserve_confirmed_by=user.id,reserve_confirmed_time=datetime.now(),update_by=user.id)
             for name,value in values.items(): setattr(item,name,value)
 
 
@@ -425,7 +429,8 @@ def _supplement_items(db,user,r,plan,certs,usages,items,active,item_map,suppleme
         if usage is None:
             usage=m.PlPlanCertUsage(**_audit(user),plan_id=plan.id,plan_no_snapshot=plan.plan_no,
                 cert_type=kind,cert_id=pid,cert_no_snapshot=cert.cert_no,cert_revision=cert.revision,
-                quantity_unit=cert.quantity_unit,reserved_qty=ZERO,used_qty=ZERO,del_flag=0)
+                quantity_unit=cert.quantity_unit,reserved_qty=ZERO,used_qty=ZERO,del_flag=0,
+                confirmation_snapshot={'measurementVersion': source.snapshot.get('measurementVersion', 'legacy_single_side')})
             db.add(usage); usages[k]=usage; items[k]=[]
         # 保留曾移除行的稳定ID（唯一键包括软删除行）。
         item=next((i for i in items[k] if _key(i)==sid),None)
@@ -439,6 +444,6 @@ def _supplement_items(db,user,r,plan,certs,usages,items,active,item_map,suppleme
             allocation_seq=max((i.allocation_seq for rows in active.values() for i in rows),default=0)+1,
             source_unit=source.unit,planned_source_qty=source.quantity,allocated_source_qty=ZERO,planned_cert_qty=ZERO,
             reserved_qty=ZERO,used_qty=ZERO,item_status='settled',source_fingerprint=source.fingerprint,
-            source_snapshot={**source.snapshot,'productId':source.product_id},del_flag=0)
+            source_snapshot={**source.snapshot,'measurementVersion':source.snapshot.get('measurementVersion', 'legacy_single_side'),'productId':source.product_id},del_flag=0)
         for name,value in values.items(): setattr(item,name,value)
         active.setdefault(k,[]).append(item); item_map[key]=item

backend/tests/test_pl_quota_business.py

@@ -58,7 +58,7 @@ def business(case, monkeypatch):
     case.db.add(m.PlPlanLoadProduct(id='sp',load_id='load',plan_id='plan',plan_product_id='pp',product_id='product',product_no='SKU',product_name='酸',plan_qty=29,unit='吨',contract_id='c',contract_product_id='cp',tenant_no='t',inbound_qty=0))
     case.db.commit()
     monkeypatch.setattr(pl_quota_planning,'plan_execution_lock_reason', lambda *a: None)
-    sources,_=extract_sources(case.db,case.user,'plan')
+    sources,_=extract_sources(case.db,case.user,'plan', measurement_version='per_stop_v2')
     fingerprint=plan_fingerprint(sources)
     payload=TransportArrange(vehicleId='v',operationId='business',quotaConfiguration={'purchaseCertIds':['p'],'transportCertIds':['t'],'conditions':[
         {'certType':kind,'certId':cid,'certRevision':1,'sourceFingerprint':fingerprint,'partiesConfirmed':True,'routeConfirmed':True,'note':'已核对'}
@@ -349,7 +349,7 @@ def test_single_plan_arrange_cannot_change_shared_transport_or_either_quota(appr
     c.db.add(c.models.PlPlanLoadProduct(id='spb',load_id='loadb',plan_id='other',plan_product_id='ppb',product_id='product',product_no='SKU',product_name='酸',plan_qty=29,unit='吨',contract_id='c',contract_product_id='cp',tenant_no='t'))
     c.db.flush()
     payload_b=c.payload.model_copy(deep=True);payload_b.operation_id='reserve-b'
-    fingerprint=plan_fingerprint(extract_sources(c.db,c.user,'other')[0])
+    fingerprint=plan_fingerprint(extract_sources(c.db,c.user,'other', measurement_version='per_stop_v2')[0])
     for condition in payload_b.quota_configuration.conditions: condition.source_fingerprint=fingerprint
     apply_plan_quota(c.db,c.user,c.plan,'reserve',c.payload)
     apply_plan_quota(c.db,c.user,other,'reserve',payload_b)
@@ -370,3 +370,70 @@ def test_single_plan_arrange_cannot_change_shared_transport_or_either_quota(appr
     assert error.value.status_code == 409
     assert '共享运输' in error.value.detail
     assert snapshot() == before
+
+
+def test_per_stop_reserves_both_certificates_and_settles_each_actual(business):
+    from app.services.pl_quota_planning import configuration_context
+    from app.services.pl_quota_business import apply_plan_quota, settle_plan_quota
+    c = business
+    c.db.get(c.models.PlPlanLoadProduct, 'sp').plan_qty = 100
+    c.db.get(c.models.PlPlanLoadProduct, 'sp').inbound_qty = 90
+    for index, qty in [(1, 40), (2, 60)]:
+        c.db.add(c.models.PlPlanDelivery(id=f'd{index}', plan_id='plan', tenant_no='t', seq=index))
+        c.db.add(c.models.PlPlanDeliveryProduct(id=f'dp{index}', delivery_id=f'd{index}', plan_id='plan', plan_product_id='pp', product_id='product', product_no='SKU', product_name='酸', plan_qty=qty, outbound_qty=qty-5, unit='吨', contract_id='c', contract_product_id='cp', tenant_no='t'))
+    for model, cid in [(c.m.PlPurchaseCert, 'p'), (c.m.PlTransportCert, 't')]:
+        c.db.get(model, cid).total_qty = 300
+    c.db.flush()
+    config = c.payload.quota_configuration
+    fingerprint = configuration_context(c.db, c.user, 'plan', config)['sourceFingerprint']
+    for condition in config.conditions: condition.source_fingerprint = fingerprint
+    apply_plan_quota(c.db, c.user, c.plan, 'reserve', c.payload)
+    assert c.plan.cert_configuration['measurementVersion'] == 'per_stop_v2'
+    assert all(u.confirmation_snapshot['measurementVersion'] == 'per_stop_v2' for u in c.db.query(c.m.PlPlanCertUsage))
+    assert all(i.source_snapshot['measurementVersion'] == 'per_stop_v2' for i in c.db.query(c.m.PlPlanCertUsageItem))
+    for model, cid in [(c.m.PlPurchaseCert, 'p'), (c.m.PlTransportCert, 't')]:
+        assert c.db.get(model, cid).reserved_qty == 200
+    assert settle_plan_quota(c.db, c.user, c.plan, c.payload)
+    for model, cid in [(c.m.PlPurchaseCert, 'p'), (c.m.PlTransportCert, 't')]:
+        assert c.db.get(model, cid).used_qty == 180
+
+
+def test_arrange_pins_v2_before_state_changes(approval_case, monkeypatch):
+    from app.services import pl_plan_execution_lock, pl_contract_plan_qty
+    c = approval_case
+    c.models.PlPlanTransport.__table__.create(c.db.bind)
+    monkeypatch.setattr(pl_plan_execution_lock, 'plan_execution_lock_reason', lambda *a: None)
+    monkeypatch.setattr(pl_contract_plan_qty, 'auto_start_execute_approved_contracts', lambda *a, **kw: None)
+    c.models.PlPlanContract.__table__.create(c.db.bind)
+    # 检查正式安排路径抵达额度调用时，pending 计划的版本已经固定。
+    def check_quota(db, user, plan, action, payload):
+        assert plan.transport_status == 'arranged'
+        assert plan.cert_configuration.get('measurementVersion') == 'per_stop_v2'
+        raise RuntimeError('checked quota boundary')
+    monkeypatch.setattr(c.api, 'apply_plan_quota', check_quota)
+    with pytest.raises(RuntimeError, match='checked quota boundary'):
+        c.api.arrange_transport('plan', c.payload, c.db, c.user)
+
+
+def test_supplement_new_certificate_keeps_server_measurement_version(business):
+    from datetime import date
+    from test_pl_quota_usage import _second_pair
+    from app.services.pl_quota_business import apply_plan_quota, settle_plan_quota
+    from app.services.pl_quota_planning import configuration_context
+    from app.schemas.pl.quota_planning import CertCondition
+    c = business
+    apply_plan_quota(c.db, c.user, c.plan, 'reserve', c.payload)
+    _second_pair(c)
+    for model, cid in [(c.m.PlPurchaseCert, 'p'), (c.m.PlTransportCert, 't')]:
+        c.db.get(model, cid).total_qty = 100
+        c.db.get(model, cid).valid_to = date(2000, 1, 1)
+    c.db.get(c.models.PlPlanLoadProduct, 'sp').inbound_qty = 35
+    c.db.flush()
+    config = c.payload.quota_configuration.model_copy(deep=True)
+    config.purchase_cert_ids = ['p2']; config.transport_cert_ids = ['t2']
+    fingerprint = configuration_context(c.db, c.user, 'plan', config)['sourceFingerprint']
+    config.conditions = [CertCondition(certType=kind, certId=cid, certRevision=1, sourceFingerprint=fingerprint, partiesConfirmed=True, routeConfirmed=True, note='核对补证') for kind, cid in [('purchase', 'p2'), ('transport', 't2')]]
+    payload = SimpleNamespace(operation_id='supplement', quota_allow_supplement=True, quota_configuration=config, quota_reason='实际超量', quota_source_note='磅单')
+    assert settle_plan_quota(c.db, c.user, c.plan, payload)
+    assert all(u.confirmation_snapshot.get('measurementVersion') == 'per_stop_v2' for u in c.db.query(c.m.PlPlanCertUsage))
+    assert configuration_context(c.db, c.user, 'plan', config)['measurementVersion'] == 'per_stop_v2'

backend/tests/test_pl_quota_sources.py

@@ -122,6 +122,8 @@ def test_planning_context_and_options_return_readonly_authoritative_facts(case):
     c = case
     for model in (PlPlanCertUsage, PlPlanCertUsageItem):
         model.__table__.create(c.db.bind)
+    from app.models.crm_models import CrmContract
+    CrmContract.__table__.create(c.db.bind)
     info = planning.get_plan_quota(c.db, c.user, 'plan')
     assert len(info['sourceOptions']) == 2
     context = planning.configuration_context(c.db, c.user, 'plan', {'sourceMapping': mapping()},
@@ -155,3 +157,53 @@ def test_overview_filters_real_usage_states_ignores_old_ready_and_other_tenant(c
     assert list_plans(c.db, other)['total'] == 0
     c.db.get(c.m.PlPlanProduct, 'pp').is_toxic = 0; c.db.flush()
     assert list_plans(c.db, c.user)['list'][0]['quotaStatus'] == 'not_required'
+
+
+def test_per_stop_counts_all_sides_ignoring_old_client_selection(case):
+    c = case
+    c.db.get(c.m.PlPlanLoadProduct, 'lp').plan_qty = 100
+    c.db.get(c.m.PlPlanDeliveryProduct, 'dp').plan_qty = 40
+    c.db.add(c.m.PlPlanDelivery(id='delivery2', plan_id='plan', tenant_no='t', seq=2))
+    c.db.add(c.m.PlPlanDeliveryProduct(id='dp2', delivery_id='delivery2', plan_id='plan',
+        plan_product_id='pp', product_id='sku', product_no='SKU', product_name='硫酸',
+        plan_qty=60, contract_id='sale2', contract_product_id='cs2', tenant_no='t'))
+    c.db.flush()
+    sources, _ = c.svc.extract_sources(c.db, c.user, 'plan', mapping(), measurement_version='per_stop_v2')
+    assert sum(s.quantity for s in sources) == 200
+    assert {(s.contract_id, s.contract_product_id) for s in sources} == {('purchase', 'cp'), ('sale', 'cs'), ('sale2', 'cs2')}
+    assert all(s.snapshot['measurementVersion'] == 'per_stop_v2' and not s.related_stop_products for s in sources)
+
+
+@pytest.mark.parametrize('history', ['draft', 'arranged', 'archived', 'ledger'])
+def test_server_version_cannot_be_overridden_by_client(case, history):
+    from app.models.pl_quota_models import PlPlanCertUsage
+    from app.services.pl_quota_planning import configuration_context
+    c = case
+    PlPlanCertUsage.__table__.create(c.db.bind)
+    plan = c.db.get(c.m.PlPlan, 'plan')
+    if history == 'arranged': plan.transport_status = 'arranged'
+    if history == 'archived': plan.archive_status = 'archived'
+    if history == 'ledger':
+        c.db.add(PlPlanCertUsage(id='u', tenant_no='t', plan_id='plan', plan_no_snapshot='PLAN', cert_type='purchase', cert_id='p', cert_no_snapshot='P', cert_revision=1, quantity_unit='吨'))
+    c.db.flush()
+    config = {'sourceMapping': mapping(), 'measurementVersion': 'legacy_single_side' if history == 'draft' else 'per_stop_v2'}
+    result = configuration_context(c.db, c.user, 'plan', config)
+    assert result['measurementVersion'] == ('per_stop_v2' if history == 'draft' else 'legacy_single_side')
+    assert sum(Decimal(s['quantity']) for s in result['sources']) == (60 if history == 'draft' else 30)
+
+
+def test_options_include_stop_contract_and_party_without_foreign_contract_leak(case):
+    from app.models.crm_models import CrmContract
+    from app.models.pl_quota_models import PlPlanCertUsage, PlPlanCertUsageItem
+    from app.services.pl_quota_planning import get_plan_quota
+    c = case
+    for model in (CrmContract, PlPlanCertUsage, PlPlanCertUsageItem): model.__table__.create(c.db.bind)
+    c.db.add(CrmContract(id='purchase', contract_no='PUR-1', contract_name='购买合同', customer_id='customer', contract_type='purchase', tenant_no='t'))
+    c.db.add(CrmContract(id='sale', contract_no='SECRET', contract_name='其他租户', customer_id='customer', contract_type='sale', tenant_no='other'))
+    c.db.get(c.m.PlPlanLoad, 'load').customer_name = '装货方'
+    c.db.flush()
+    options = {o['sourceStopType']: o for o in get_plan_quota(c.db, c.user, 'plan')['sourceOptions']}
+    assert options['load']['contractNo'] == 'PUR-1'
+    assert options['load']['partyName'] == '装货方'
+    assert options['load']['stopId'] == 'load' and options['load']['stopOrder'] == 1
+    assert options['delivery']['contractNo'] is None