diff --git a/backend/app/api/pl/quota_cert.py b/backend/app/api/pl/quota_cert.py
index c9f7b29a4..32191dfb0 100644
--- a/backend/app/api/pl/quota_cert.py
+++ b/backend/app/api/pl/quota_cert.py
@@ -19,6 +19,7 @@ from app.schemas.pl.quota_scope import ContractCertBindingsBody
 from app.schemas.pl.legal_use_document import LegalDocumentsBody, StopLegalDocumentsBody
 from app.schemas.pl.quota_planning import PlanQuotaConfiguration, PlanQuotaPreview
 from app.services import pl_quota_cert_service as certs
+from app.services import pl_quota_cert_bindings as cert_bindings
 from app.services import pl_quota_contract_service as scopes
 from app.services import pl_quota_contract_query as contract_query
 from app.services import pl_legal_use_document as documents
@@ -161,13 +162,13 @@ def recognize(cert_type: CertType, body: RecognizeBody,
 @oper_log('证件档案', business_type=1)
 def create_cert(cert_type: CertType, body: dict = Body(...),
                 db: Session = Depends(get_db), user: DbUserContext = Depends(current_user)):
-    return _write(db, lambda: certs.save_cert(db, user, cert_type, body))
+    return _write(db, lambda: cert_bindings.save_cert(db, user, cert_type, body))
 
 
 @router.get('/{cert_type}/{cert_id}')
 def get_cert(cert_type: CertType, cert_id: str,
              db: Session = Depends(get_db), user: DbUserContext = Depends(current_user)):
-    return _wire(certs.get_cert(db, user, cert_type, cert_id))
+    return _wire(cert_bindings.get_cert(db, user, cert_type, cert_id))
 
 
 @router.get('/{cert_type}/{cert_id}/contracts')
@@ -180,7 +181,7 @@ def certificate_contracts(cert_type: CertType, cert_id: str, db: Session = Depen
 @oper_log('证件档案', business_type=2, path_param='cert_id')
 def update_cert(cert_type: CertType, cert_id: str, body: dict = Body(...),
                 db: Session = Depends(get_db), user: DbUserContext = Depends(current_user)):
-    return _write(db, lambda: certs.save_cert(db, user, cert_type, body, cert_id))
+    return _write(db, lambda: cert_bindings.save_cert(db, user, cert_type, body, cert_id))
 
 
 @router.delete('/{cert_type}/{cert_id}', dependencies=[Depends(require_menu_code('pl:quota-cert:edit'))])
diff --git a/backend/app/services/pl_quota_cert_bindings.py b/backend/app/services/pl_quota_cert_bindings.py
new file mode 100644
index 000000000..30ce68fb3
--- /dev/null
+++ b/backend/app/services/pl_quota_cert_bindings.py
@@ -0,0 +1,145 @@
+"""证面、购买证引用与多合同范围先整批校验，再在调用方事务内差量写入。"""
+from typing import List, Literal, Optional
+from uuid import uuid4
+
+from fastapi import HTTPException
+from pydantic import BaseModel, Field
+from app.crud.crm.data_scope_helpers import apply_crm_owner_scope
+from app.models.crm_models import CrmContract
+from app.models.pl_quota_models import PlContractCertLink, PlContractCertProduct, PlTransportCertPurchaseRef
+from app.schemas.pl.quota_cert import InputBody
+from app.schemas.pl.quota_scope import ScopeProductBody
+from app.services import pl_quota_cert_service as certs
+from app.services import pl_quota_contract_service as scopes
+
+
+class CertificateContractBinding(InputBody):
+    id: Optional[str] = Field(None, max_length=64)
+    contract_id: str = Field(min_length=1, max_length=64)
+    link_status: Optional[Literal['active', 'inactive']] = None
+    products: List[ScopeProductBody]
+    remark: Optional[str] = Field(None, max_length=1000)
+    order_num: Optional[int] = None
+
+
+class CertificateBindings(InputBody):
+    contract_bindings: Optional[List[CertificateContractBinding]] = None
+
+
+def _target_links(db, user, cert_type, cert_id, current=False):
+    query = db.query(PlContractCertLink).filter_by(tenant_no=user.tenant_no,
+        cert_type=cert_type, cert_id=cert_id, del_flag=0)
+    if current:
+        query = query.with_for_update().populate_existing()
+    return query.all()
+
+
+def _visible_contract_ids(db, user):
+    query = db.query(CrmContract.id).filter_by(tenant_no=user.tenant_no, del_flag=0)
+    return apply_crm_owner_scope(query, CrmContract, user.to_user_info(), db, user.tenant_no)
+
+
+def get_cert(db, user, cert_type, cert_id):
+    result = certs.get_cert(db, user, cert_type, cert_id)
+    links = db.query(PlContractCertLink).filter(PlContractCertLink.tenant_no == user.tenant_no,
+        PlContractCertLink.cert_type == cert_type, PlContractCertLink.cert_id == cert_id,
+        PlContractCertLink.del_flag == 0,
+        PlContractCertLink.contract_id.in_(_visible_contract_ids(db, user))).order_by(
+            PlContractCertLink.create_time.desc(), PlContractCertLink.id).all()
+    products = db.query(PlContractCertProduct).filter(
+        PlContractCertProduct.tenant_no == user.tenant_no, PlContractCertProduct.del_flag == 0,
+        PlContractCertProduct.contract_cert_link_id.in_([r.id for r in links])).order_by(
+            PlContractCertProduct.create_time.desc(), PlContractCertProduct.id).all() if links else []
+    total_links = db.query(PlContractCertLink.id).filter_by(tenant_no=user.tenant_no,
+        cert_type=cert_type, cert_id=cert_id, del_flag=0).count()
+    result['bindingsEditable'] = total_links == len(links)
+    result['contractBindings'] = [{
+        'id': link.id, 'contractId': link.contract_id, 'linkStatus': link.link_status,
+        'remark': link.remark, 'orderNum': link.order_num,
+        'products': [{'id': p.id, 'contractProductId': p.contract_product_id,
+                      'scopeStatus': p.scope_status, 'remark': p.remark, 'orderNum': p.order_num}
+                     for p in products if p.contract_cert_link_id == link.id],
+    } for link in links]
+    return result
+
+
+def _reference_ids(db, user, cert_id, submitted, current=False):
+    query = db.query(PlTransportCertPurchaseRef).filter_by(tenant_no=user.tenant_no,
+                                                         transport_cert_id=cert_id)
+    if current:
+        query = query.with_for_update().populate_existing()
+    restored = {r['reference_no']: r for r in (submitted or [])}
+    ids = {r.purchase_cert_id for r in query.all() if r.purchase_cert_id and (
+        not r.del_flag or (r.reference_no in restored and 'purchase_cert_id' not in restored[r.reference_no]))}
+    ids.update(r['purchase_cert_id'] for r in (submitted or []) if r.get('purchase_cert_id'))
+    return ids
+
+
+def save_cert(db, user, cert_type, data, cert_id=None):
+    payload = data.model_dump(exclude_unset=True) if isinstance(data, BaseModel) else dict(data)
+    present = 'contractBindings' in payload or 'contract_bindings' in payload
+    if not present:
+        result = certs.save_cert(db, user, cert_type, payload, cert_id)
+        return get_cert(db, user, cert_type, result['id'])
+    bindings_payload = {key: payload.pop(key) for key in ('contractBindings', 'contract_bindings') if key in payload}
+    body = CertificateBindings(**bindings_payload)
+    if body.contract_bindings is None:
+        raise HTTPException(400, '合同集合不可为null；清空请提交[]')
+    # 证面独立严格解析，未识别字段不会因拆分而被丢弃。
+    certs._model(cert_type)
+    face = certs.SCHEMAS[cert_type][1 if cert_id else 0](**payload)
+    values = body.model_dump(exclude_unset=True)['contract_bindings']
+    ids = [r['contract_id'] for r in values]
+    if len(ids) != len(set(ids)):
+        raise HTTPException(400, '合同重复提交')
+    effective_id = cert_id or str(uuid4())
+    target = (cert_type, effective_id)
+    initial = _target_links(db, user, *target)
+    contract_ids = set(ids) | {r.contract_id for r in initial}
+    # 无权合同即拒绝整个集合替换；省略集合仍允许修改证面。
+    for cid in sorted(contract_ids):
+        scopes._contract(db, user, cid)
+    keys = {target}
+    for cid in sorted(contract_ids):
+        keys.update((r.cert_type, r.cert_id) for r in scopes._links(db, user, cid) if r.del_flag == 0)
+    refs = face.model_dump(exclude_unset=True).get('purchase_refs')
+    if cert_type == 'transport':
+        keys.update(('purchase', pid) for pid in _reference_ids(db, user, effective_id, refs))
+    locked = {key: certs._get_row(db, user, *key, lock=True) for key in sorted(keys)
+              if cert_id or key != target}
+    # 新证仅构建临时对象，不 add/flush；全部准备成功后才能入会话。
+    if not cert_id:
+        locked[target] = certs.MODELS[cert_type](id=effective_id, status=face.status or 'draft')
+    contracts = {cid: scopes._contract(db, user, cid, lock=True) for cid in sorted(contract_ids)}
+    current = _target_links(db, user, *target, current=True)
+    if any(r.contract_id not in contract_ids for r in current):
+        raise HTTPException(409, '证件合同范围已变更，请重试')
+    if cert_type == 'transport' and any(('purchase', pid) not in keys for pid in
+            _reference_ids(db, user, effective_id, refs, current=True)):
+        raise HTTPException(409, '证面引用已变更，请重试')
+    prepared_face = certs.prepare_cert(db, user, cert_type, face, cert_id,
+                                      locked=locked, effective_id=effective_id)
+    # 用最终证面状态校验作废证不可扩大，不修改持久对象。
+    effective_certs = dict(locked)
+    effective_certs[target] = certs.MODELS[cert_type](id=effective_id,
+        status=prepared_face['values'].get('status', locked[target].status))
+    requested = {item['contract_id']: item for item in values}
+    prepared = []
+    for cid in sorted(contract_ids):
+        existing = scopes._links(db, user, cid, lock=True)
+        if any((r.cert_type, r.cert_id) not in keys for r in existing if r.del_flag == 0):
+            raise HTTPException(409, '合同用证范围已变更，请重试')
+        entries = [{'id': r.id, 'cert_type': r.cert_type, 'cert_id': r.cert_id}
+                   for r in existing if r.del_flag == 0 and (r.cert_type, r.cert_id) != target]
+        if cid in requested:
+            item = {k: v for k, v in requested[cid].items() if k != 'contract_id'}
+            item.update(cert_type=cert_type, cert_id=effective_id)
+            entries.append(item)
+        normalized = scopes.ContractCertBindingsBody(bindings=entries).model_dump(exclude_unset=True)['bindings']
+        prepared.append(scopes.prepare_contract_bindings(db, user, contracts[cid], normalized,
+                                                         effective_certs, keys))
+    row = certs.apply_cert(db, user, prepared_face)
+    for scope in prepared:
+        scopes.apply_contract_bindings(db, user, scope)
+    db.flush()
+    return get_cert(db, user, cert_type, row.id)
diff --git a/backend/app/services/pl_quota_cert_service.py b/backend/app/services/pl_quota_cert_service.py
index e74bc597a..b7ea7e5a2 100644
--- a/backend/app/services/pl_quota_cert_service.py
+++ b/backend/app/services/pl_quota_cert_service.py
@@ -141,7 +141,7 @@ def _validate_refs(db, user, cert_id, submitted, historical, purchases, current=
     return pending, removed
 
 
-def save_cert(db, user, cert_type, data, cert_id=None):
+def prepare_cert(db, user, cert_type, data, cert_id=None, *, locked=None, effective_id=None):
     model = _model(cert_type)
     payload = data.model_dump(exclude_unset=True) if isinstance(data, BaseModel) else data
     body = SCHEMAS[cert_type][1 if cert_id else 0](**payload)
@@ -158,8 +158,12 @@ def save_cert(db, user, cert_type, data, cert_id=None):
             not r.del_flag or (r.reference_no in restored and 'purchase_cert_id' not in restored[r.reference_no]))}
         purchase_ids.update(r['purchase_cert_id'] for r in (refs or []) if r.get('purchase_cert_id'))
         # 与账本统一购买证→运输证顺序；当前引用若出现未锁证件，拒绝而不倒序补锁。
-        purchases = {pid: _get_row(db, user, 'purchase', pid, lock=True) for pid in sorted(purchase_ids)}
-    row = _get_row(db, user, cert_type, cert_id, lock=True) if cert_id else None
+        purchases = {pid: (locked.get(('purchase', pid)) if locked is not None else
+                          _get_row(db, user, 'purchase', pid, lock=True)) for pid in sorted(purchase_ids)}
+        if any(row is None for row in purchases.values()):
+            raise HTTPException(409, '证面引用已变更，请重试')
+    row = (locked[(cert_type, cert_id)] if locked is not None else
+           _get_row(db, user, cert_type, cert_id, lock=True)) if cert_id else None
     historical = bool(row and _has_history(db,user,cert_type,cert_id))
     for key, value in values.items():
         column = model.__table__.columns[key]
@@ -185,10 +189,18 @@ def save_cert(db, user, cert_type, data, cert_id=None):
     duplicate = db.query(model).filter_by(tenant_no=user.tenant_no, cert_no=no).first()
     if duplicate and duplicate.id != cert_id:
         raise HTTPException(400, '证号已存在，删除后也不得重建额度池')
-    effective_id = cert_id or str(uuid4())
+    effective_id = cert_id or effective_id or str(uuid4())
     pending, removed = _validate_refs(db,user,effective_id,refs,historical,purchases,current=bool(cert_id)) if refs_present else ([],[])
     changed = {key for key, value in values.items() if row is None or getattr(row, key) != value}
     refs_changed = bool(removed) or any(previous is None or previous.del_flag == 1 or any(getattr(previous, key) != value for key, value in item.items() if key != 'id') for previous, item in pending)
+    return dict(model=model, row=row, values=values, cert_id=cert_id, effective_id=effective_id,
+                pending=pending, removed=removed, changed=changed, refs_changed=refs_changed)
+
+
+def apply_cert(db, user, prepared):
+    model, row, values, cert_id, effective_id, pending, removed, changed, refs_changed = (
+        prepared[key] for key in ('model', 'row', 'values', 'cert_id', 'effective_id',
+                                  'pending', 'removed', 'changed', 'refs_changed'))
     # 整批校验通过后才开始修改。
     if not row:
         audit = _audit(user); audit['id'] = effective_id
@@ -215,7 +227,12 @@ def save_cert(db, user, cert_type, data, cert_id=None):
         previous.del_flag = 1
         previous.update_by = user.id
     db.flush()
-    return get_cert(db,user,cert_type,row.id)
+    return row
+
+
+def save_cert(db, user, cert_type, data, cert_id=None):
+    row = apply_cert(db, user, prepare_cert(db, user, cert_type, data, cert_id))
+    return get_cert(db, user, cert_type, row.id)
 
 
 def delete_cert(db, user, cert_type, cert_id):
diff --git a/backend/app/services/pl_quota_contract_service.py b/backend/app/services/pl_quota_contract_service.py
index bd627536d..98e7622ab 100644
--- a/backend/app/services/pl_quota_contract_service.py
+++ b/backend/app/services/pl_quota_contract_service.py
@@ -115,6 +115,14 @@ def save_contract_bindings(db, user, contract_id, data):
     keys.update((item['cert_type'], item['cert_id']) for item in values)
     certs = {key: _get_row(db, user, *key, lock=True) for key in sorted(keys)}
     contract = _contract(db, user, contract_id, lock=True)
+    prepared = prepare_contract_bindings(db, user, contract, values, certs, keys)
+    apply_contract_bindings(db, user, prepared)
+    db.flush()
+    return get_contract_bindings(db, user, contract_id, current=True)
+
+
+def prepare_contract_bindings(db, user, contract, values, certs, keys):
+    contract_id = contract.id
     if contract.status not in ('approved', 'executing'):
         raise HTTPException(400, '仅已审批或执行中的合同可配置用证范围')
     if contract.contract_type not in ('sale', 'purchase', 'framework_sale', 'framework_purchase'):
@@ -156,6 +164,12 @@ def save_contract_bindings(db, user, contract_id, data):
     removed = [r for r in existing if r.del_flag == 0 and (r.cert_type, r.cert_id) not in seen]
     for row in removed:
         _assert_unreferenced(db, user, link_id=row.id)
+    return contract, pending, removed
+
+
+def apply_contract_bindings(db, user, prepared):
+    contract, pending, removed = prepared
+    contract_id = contract.id
     # 所有主子身份与引用校验通过，才开始写入；关系以唯一业务组合差量复用。
     for link, item, products in pending:
         is_new = link is None
@@ -195,5 +209,3 @@ def save_contract_bindings(db, user, contract_id, data):
         for product in _products(db, user, row.id):
             if product.del_flag == 0:
                 product.del_flag = 1; product.update_by = user.id
-    db.flush()
-    return get_contract_bindings(db, user, contract_id, current=True)
diff --git a/backend/tests/test_pl_quota_api.py b/backend/tests/test_pl_quota_api.py
index 76bb8d15b..42c0246e9 100644
--- a/backend/tests/test_pl_quota_api.py
+++ b/backend/tests/test_pl_quota_api.py
@@ -133,3 +133,18 @@ def test_total_adjust_requires_provenance_and_replays_without_duplicate_log(api)
     assert client.post(path, json={**body,'totalQty':'40'}).status_code == 409
     assert client.post(path, json={**body,'operationId':'adjust-2','sourceNote':''}).status_code == 409
     assert client.get(base + '/' + row['id']).json()['totalQty'] == '35.000000'
+
+
+def test_cert_bindings_api_accepts_collection_and_roundtrips(api):
+    client, app, auth, permissions = api
+    app.dependency_overrides[auth] = lambda: {'role': 'super_admin', 'user_id': 'u', 'tenant_no': 't'}
+    path = '/api/transport/quota-certs/purchase'
+    saved = client.post(path, json={**payload(), 'contractBindings': []})
+    assert saved.status_code == 200
+    row = saved.json()
+    assert row['contractBindings'] == []
+    detail = path + '/' + row['id']
+    assert client.get(detail).json()['contractBindings'] == []
+    assert client.put(detail, json={'contractBindings': [], 'remark': 'edit'}).status_code == 200
+    assert client.put(detail, json={'contractBindings': [], 'unknownField': 'bad'}).status_code == 422
+    assert client.get(detail).json()['remark'] == 'edit'
diff --git a/backend/tests/test_pl_quota_cert_bindings.py b/backend/tests/test_pl_quota_cert_bindings.py
new file mode 100644
index 000000000..5679b517d
--- /dev/null
+++ b/backend/tests/test_pl_quota_cert_bindings.py
@@ -0,0 +1,203 @@
+"""证件建档与多合同范围原子保存，SQLite 验证。"""
+import pytest
+from fastapi import HTTPException
+from test_pl_quota_contract_scope import case, binding
+
+
+def save(c, data, cert_id='g1', cert_type='purchase'):
+    from app.services import pl_quota_cert_bindings as svc
+    return svc.save_cert(c.db, c.user, cert_type, data, cert_id)
+
+
+def scopes():
+    return [{'contractId': 'c1', 'products': [{'contractProductId': 'p1'}]},
+            {'contractId': 'c2', 'products': [{'contractProductId': 'p3'}]}]
+
+
+def test_roundtrip_omission_empty_and_other_cert_preserved(case):
+    c = case
+    c.svc.save_contract_bindings(c.db, c.user, 'c1', {'bindings': [binding('g2')]})
+    result = save(c, {'contractBindings': scopes()})
+    original = result['contractBindings']
+    assert len(original) == 2
+    payload = [{k: v for k, v in x.items() if k in ('id', 'contractId', 'linkStatus', 'products', 'remark', 'orderNum')} for x in original]
+    for item in payload:
+        item['products'] = [{k: v for k, v in p.items() if k in ('id', 'contractProductId', 'scopeStatus')} for p in item['products']]
+    assert save(c, {'contractBindings': payload})['contractBindings'] == original
+    assert save(c, {'remark': 'only face'})['contractBindings'] == original
+    assert save(c, {'contractBindings': []})['contractBindings'] == []
+    assert c.svc.get_contract_bindings(c.db, c.user, 'c1')[0]['certId'] == 'g2'
+
+
+def test_second_invalid_contract_does_not_mutate_face_or_first(case):
+    c = case
+    data = scopes()
+    data[1]['products'][0]['contractProductId'] = 'p1'
+    with pytest.raises(HTTPException):
+        save(c, {'remark': 'must not write', 'contractBindings': data})
+    assert not c.db.new and not c.db.dirty
+    assert c.db.query(c.m.PlContractCertLink).count() == 0
+    assert c.db.query(c.m.PlPurchaseCert).filter_by(id='g1').one().remark is None
+
+
+@pytest.mark.parametrize('change', ['duplicate', 'foreign_id', 'foreign_product', 'null'])
+def test_bad_identity_rejected_before_writes(case, change):
+    data = scopes()
+    if change == 'duplicate': data.append(data[0])
+    if change == 'foreign_id': data[0]['id'] = 'foreign'
+    if change == 'foreign_product': data[0]['products'][0]['id'] = 'foreign'
+    if change == 'null': data = None
+    with pytest.raises(HTTPException):
+        save(case, {'contractBindings': data})
+    assert case.db.query(case.m.PlContractCertLink).count() == 0
+
+
+def test_create_with_bindings_and_bad_create_has_no_pending_row(case):
+    c = case
+    body = dict(certNo='new', productNameOnCert='硫酸', totalQty='300', quantityUnit='吨',
+                validUseText='多次有效', buyerName='买方', sellerName='卖方', contractBindings=scopes())
+    broken = {**body, 'contractBindings': [{'contractId': 'missing', 'products': []}]}
+    with pytest.raises(HTTPException): save(c, broken, None)
+    assert not c.db.new and not c.db.dirty
+    result = save(c, body, None)
+    assert len(result['contractBindings']) == 2
+    assert result['id'] not in ('g1', 'g2')
+
+
+def test_hidden_contract_cannot_be_cleared_but_face_can_change(case, monkeypatch):
+    from app.services import pl_quota_cert_bindings as svc
+    from app.models.crm_models import CrmContract
+    c = case
+    save(c, {'contractBindings': scopes()})
+    original = c.svc.ensure_crm_owner_row_in_scope
+    def deny_second(db, row, *args):
+        if row.id == 'c2': raise ValueError('hidden')
+        return original(db, row, *args)
+    monkeypatch.setattr(c.svc, 'ensure_crm_owner_row_in_scope', deny_second)
+    monkeypatch.setattr(svc, '_visible_contract_ids', lambda db, user: db.query(CrmContract.id).filter(CrmContract.id == 'c1'))
+    with pytest.raises(HTTPException) as error:
+        save(c, {'contractBindings': []})
+    assert error.value.status_code == 403
+    result = save(c, {'remark': 'allowed'})
+    assert result['remark'] == 'allowed'
+    assert [x['contractId'] for x in result['contractBindings']] == ['c1']
+    assert c.db.query(c.m.PlContractCertLink).filter_by(del_flag=0).count() == 2
+
+
+def test_transport_locks_all_certificates_before_contracts_and_current_reads(case, monkeypatch):
+    from app.services import pl_quota_cert_service as certs
+    from sqlalchemy import event
+    from app.models.crm_models import CrmContract, CrmContractProduct
+    c = case
+    c.svc.save_contract_bindings(c.db, c.user, 'c1', {'bindings': [binding('g2')]})
+    certs.save_cert(c.db, c.user, 'transport', dict(certNo='T1', productNameOnCert='硫酸',
+        totalQty='300', quantityUnit='吨', validUseText='多次有效', shipperName='发货人'), None)
+    target = c.db.query(c.m.PlTransportCert).one().id
+    events = []
+    original = certs._get_row
+    contract = c.svc._contract
+    def cert_lock(db, user, kind, rid, lock=False):
+        if lock: events.append(('cert', kind, rid))
+        return original(db, user, kind, rid, lock)
+    def contract_lock(db, user, cid, lock=False):
+        if lock: events.append(('contract', cid))
+        return contract(db, user, cid, lock)
+    monkeypatch.setattr(certs, '_get_row', cert_lock)
+    monkeypatch.setattr(c.svc, '_contract', contract_lock)
+    reads = []
+    def capture(state):
+        if state.is_select and state.statement._for_update_arg is not None:
+            reads.append((state.bind_mapper.class_, state.load_options._populate_existing))
+    event.listen(c.db, 'do_orm_execute', capture)
+    try:
+        result = save(c, {'contractBindings': scopes(), 'purchaseRefs': [
+            {'referenceType': 'purchase_permit', 'referenceNo': 'g1', 'purchaseCertId': 'g1'}]}, target, 'transport')
+    finally:
+        event.remove(c.db, 'do_orm_execute', capture)
+    assert events == [('cert', 'purchase', 'g1'), ('cert', 'purchase', 'g2'),
+                      ('cert', 'transport', target), ('contract', 'c1'), ('contract', 'c2')]
+    for model in (CrmContract, CrmContractProduct, c.m.PlContractCertLink,
+                  c.m.PlTransportCertPurchaseRef):
+        assert any(m == model and refresh for m, refresh in reads)
+    assert len(result['contractBindings']) == 2 and len(result['purchaseRefs']) == 1
+
+
+def test_new_key_after_lock_retries_without_face_mutation(case, monkeypatch):
+    c = case
+    original = c.svc._contract
+    def changed(db, user, cid, lock=False):
+        row = original(db, user, cid, lock)
+        if lock and cid == 'c1':
+            db.add(c.m.PlContractCertLink(id='concurrent', tenant_no='t', contract_id='c1',
+                cert_type='purchase', cert_id='g2', contract_no_snapshot='c1', scope_revision=1, link_status='active'))
+            db.flush()
+        return row
+    monkeypatch.setattr(c.svc, '_contract', changed)
+    with pytest.raises(HTTPException) as error:
+        save(c, {'remark': 'must not write', 'contractBindings': scopes()})
+    assert error.value.status_code == 409
+    assert c.db.query(c.m.PlPurchaseCert).filter_by(id='g1').one().remark is None
+
+
+def test_referenced_scope_cannot_be_removed_or_disabled(case):
+    c = case
+    result = save(c, {'contractBindings': scopes()})
+    link = next(x for x in result['contractBindings'] if x['contractId'] == 'c1')
+    c.db.add(c.m.PlPlanCertUsageItem(id='item', usage_id='usage', tenant_no='t',
+        contract_cert_link_id=link['id'], contract_cert_product_id=link['products'][0]['id'],
+        contract_scope_revision=1, contract_id='c1', contract_product_id='p1', plan_product_id='pp',
+        source_stop_type='delivery', source_stop_id='stop', source_stop_product_id='sp',
+        allocation_seq=1, source_unit='吨', planned_source_qty=10))
+    c.db.flush()
+    for bindings in ([], [{**scopes()[0], 'linkStatus': 'inactive'}, scopes()[1]]):
+        with pytest.raises(HTTPException) as error:
+            save(c, {'remark': 'do not write', 'contractBindings': bindings})
+        assert error.value.status_code == 409
+        assert c.db.query(c.m.PlPurchaseCert).filter_by(id='g1').one().remark is None
+
+
+def test_void_new_or_existing_certificate_cannot_expand(case):
+    c = case
+    with pytest.raises(HTTPException):
+        save(c, {'status': 'void', 'contractBindings': scopes()})
+    assert c.db.query(c.m.PlPurchaseCert).filter_by(id='g1').one().status == 'draft'
+
+
+def test_get_reports_partial_binding_scope_for_editor(case, monkeypatch):
+    from app.services import pl_quota_cert_bindings as svc
+    from app.models.crm_models import CrmContract
+    c = case
+    result = save(c, {'contractBindings': scopes()})
+    assert result['bindingsEditable'] is True
+    monkeypatch.setattr(svc, '_visible_contract_ids', lambda db, user: db.query(CrmContract.id).filter(CrmContract.id == 'c1'))
+    result = svc.get_cert(c.db, c.user, 'purchase', 'g1')
+    assert result['bindingsEditable'] is False
+    assert len(result['contractBindings']) == 1
+
+
+def test_foreign_tenant_contract_rejected_and_no_changes(case):
+    from app.models.crm_models import CrmContract
+    c = case
+    c.db.query(CrmContract).filter_by(id='c2').one().tenant_no = 'other'
+    c.db.commit()
+    with pytest.raises(HTTPException) as error:
+        save(c, {'remark': 'must not write', 'contractBindings': scopes()})
+    assert error.value.status_code == 404
+    assert c.db.query(c.m.PlContractCertLink).count() == 0
+    assert not c.db.dirty and not c.db.new
+
+
+def test_transport_refs_and_face_not_written_when_second_contract_invalid(case):
+    from app.services import pl_quota_cert_service as certs
+    c = case
+    result = certs.save_cert(c.db, c.user, 'transport', dict(certNo='T1', productNameOnCert='硫酸',
+        totalQty='300', quantityUnit='吨', validUseText='多次有效', shipperName='发货人'))
+    data = scopes()
+    data[1]['products'] = [{'contractProductId': 'p1'}]
+    with pytest.raises(HTTPException):
+        save(c, {'remark': 'must not write', 'contractBindings': data, 'purchaseRefs': [
+            {'referenceType': 'purchase_permit', 'referenceNo': 'g1', 'purchaseCertId': 'g1'}]}, result['id'], 'transport')
+    assert c.db.query(c.m.PlTransportCert).one().remark is None
+    assert c.db.query(c.m.PlTransportCertPurchaseRef).count() == 0
+    assert c.db.query(c.m.PlContractCertLink).count() == 0
+    assert not c.db.dirty and not c.db.new