fix: make mobile presign transitions concurrency-safe.

2026-10-11 11:17:06 +08:00
1 parent 8453c9c805
commit c5c7e42feb
@@ -1,10 +1,18 @@
 """移动市场接口的白名单请求 DTO。"""
 from __future__ import annotations

+from datetime import datetime
 from decimal import Decimal
 from typing import List, Literal, Optional

-from pydantic import BaseModel, ConfigDict, Field, condecimal
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    condecimal,
+    constr,
+    field_validator,
+)


 MarketPaymentAmount = condecimal(
@@ -13,6 +21,13 @@ MarketPaymentAmount = condecimal(
    decimal_places=2,
    allow_inf_nan=False,
 )
+MarketContractAmount = condecimal(
+    ge=Decimal('0'),
+    max_digits=18,
+    decimal_places=2,
+    allow_inf_nan=False,
+)
+MarketSampleTypeId = constr(min_length=1, max_length=64)


 class MarketPresignTransferBody(BaseModel):
@@ -37,28 +52,72 @@ class MarketPresignOrderBody(BaseModel):
    model_config = ConfigDict(extra='forbid', populate_by_name=True)

    order_date: Optional[str] = Field(None, alias="orderDate")
-    protocol_no: Optional[str] = Field(None, alias="protocolNo")
-    mother_name: Optional[str] = Field(None, alias="motherName")
+    protocol_no: Optional[str] = Field(
+        None, max_length=64, alias="protocolNo"
+    )
+    mother_name: Optional[str] = Field(
+        None, max_length=64, alias="motherName"
+    )
    expected_delivery_date: Optional[str] = Field(
        None, alias="expectedDeliveryDate"
    )
-    region_id: Optional[str] = Field(None, alias="regionId")
-    hospital_id: Optional[str] = Field(None, alias="hospitalId")
+    region_id: Optional[str] = Field(
+        None, max_length=64, alias="regionId"
+    )
+    hospital_id: Optional[str] = Field(
+        None, max_length=64, alias="hospitalId"
+    )
    sample_collect_time: Optional[str] = Field(None, alias="sampleCollectTime")
-    pay_mode: Optional[str] = Field(None, alias="payMode")
-    contract_amount: Optional[float] = Field(None, alias="contractAmount")
-    collector_user_id: Optional[str] = Field(None, alias="collectorUserId")
-    sales_user_id: Optional[str] = Field(None, alias="salesUserId")
+    pay_mode: Optional[str] = Field(None, max_length=8, alias="payMode")
+    contract_amount: Optional[MarketContractAmount] = Field(
+        None, alias="contractAmount"
+    )
+    collector_user_id: Optional[str] = Field(
+        None, max_length=64, alias="collectorUserId"
+    )
+    sales_user_id: Optional[str] = Field(
+        None, max_length=64, alias="salesUserId"
+    )
    region_manager_user_id: Optional[str] = Field(
-        None, alias="regionManagerUserId"
+        None, max_length=64, alias="regionManagerUserId"
    )
    order_status: Optional[
        Literal['draft', 'active', 'converted', 'void']
    ] = Field(None, alias='orderStatus')
-    remark: Optional[str] = None
-    sample_type_ids: Optional[List[str]] = Field(None, alias="sampleTypeIds")
+    remark: Optional[str] = Field(None, max_length=512)
+    sample_type_ids: Optional[List[MarketSampleTypeId]] = Field(
+        None, max_length=100, alias="sampleTypeIds"
+    )
    transfers: Optional[List[MarketPresignTransferBody]] = None

+    @field_validator('order_date', 'expected_delivery_date')
+    @classmethod
+    def validate_ymd(cls, value: Optional[str]) -> Optional[str]:
+        if value in (None, ''):
+            return value
+        text = str(value).strip()
+        if len(text) != 8 or not text.isdigit():
+            raise ValueError('日期必须使用 YYYYMMDD 格式')
+        try:
+            datetime.strptime(text, '%Y%m%d')
+        except ValueError as exc:
+            raise ValueError('日期必须是有效的 YYYYMMDD') from exc
+        return text
+
+    @field_validator('sample_collect_time')
+    @classmethod
+    def validate_sample_collect_time(
+        cls, value: Optional[str]
+    ) -> Optional[str]:
+        if value in (None, ''):
+            return value
+        text = str(value).strip()
+        try:
+            datetime.fromisoformat(text.replace('Z', '+00:00'))
+        except ValueError as exc:
+            raise ValueError('采集时间格式无效') from exc
+        return text
+

 class MarketProtocolCustomerBody(BaseModel):
    model_config = ConfigDict(extra='forbid', populate_by_name=True)
@@ -8,6 +8,7 @@ from fastapi import HTTPException
 from sqlalchemy import func, or_
 from sqlalchemy.orm import Session

+from app.core.utils import parse_datetime
 from app.crud import presign_order as presign_crud
 from app.models.models import (
    BizCellSampleType,
@@ -42,6 +43,26 @@ def _assert_transfer_order_writable(order: BizPresignOrder) -> None:
        raise ValueError('当前状态的预签约订单不可操作转账记录')


+def _scoped_order_for_update(
+    db: Session,
+    order_id: str,
+    ctx: MarketFieldContext,
+    user_info: dict,
+) -> BizPresignOrder:
+    row = (
+        apply_presign_scope(
+            db.query(BizPresignOrder), db, ctx, user_info=user_info
+        )
+        .filter(BizPresignOrder.id == order_id)
+        .populate_existing()
+        .with_for_update()
+        .first()
+    )
+    if row is None:
+        raise HTTPException(status_code=404, detail='数据不存在或无权访问')
+    return row
+
+
 def _current_dept_id(user_info: dict) -> str:
    """兼容 API 用户上下文与测试/内部调用的部门字段命名。"""
    if not isinstance(user_info, dict):
@@ -334,60 +355,64 @@ def update_presign_order(
    ctx: MarketFieldContext,
    user_info: dict,
 ) -> dict:
-    current = assert_presign_access(
-        db, order_id, ctx, user_info=user_info
-    )
-    if (current.order_status or '').strip() == 'converted':
-        raise ValueError('当前状态的预签约订单不可修改')
-    fields = _dump_body(body)
-    sample_type_ids = fields.pop("sample_type_ids", None)
-    fields.pop("transfers", None)
-    resulting_region_id = fields.get("region_id", current.region_id)
-    resulting_hospital_id = fields.get("hospital_id", current.hospital_id)
-    _validate_master_data(
-        db,
-        ctx,
-        region_id=resulting_region_id,
-        hospital_id=resulting_hospital_id,
-        sample_type_ids=sample_type_ids,
-    )
-    if ctx.scope_kind == "specialist":
-        fields["sales_user_id"] = ctx.user_id
-    else:
-        _ensure_manager_assignment(
+    try:
+        current = _scoped_order_for_update(
+            db, order_id, ctx, user_info
+        )
+        if (current.order_status or '').strip() == 'converted':
+            raise ValueError('当前状态的预签约订单不可修改')
+        fields = _dump_body(body)
+        sample_type_ids = fields.pop("sample_type_ids", None)
+        fields.pop("transfers", None)
+        resulting_region_id = fields.get("region_id", current.region_id)
+        resulting_hospital_id = fields.get("hospital_id", current.hospital_id)
+        _validate_master_data(
            db,
            ctx,
            region_id=resulting_region_id,
+            hospital_id=resulting_hospital_id,
+            sample_type_ids=sample_type_ids,
+        )
+        if ctx.scope_kind == "specialist":
+            fields["sales_user_id"] = ctx.user_id
+        else:
+            _ensure_manager_assignment(
+                db,
+                ctx,
+                region_id=resulting_region_id,
+                region_manager_user_id=fields.get(
+                    "region_manager_user_id", current.region_manager_user_id
+                ),
+            )
+        presign_crud.validate_presign_person_ids(
+            db,
+            tenant_no=ctx.tenant_no,
+            collector_user_id=fields.get(
+                "collector_user_id", current.collector_user_id
+            ),
+            sales_user_id=fields.get("sales_user_id", current.sales_user_id),
            region_manager_user_id=fields.get(
                "region_manager_user_id", current.region_manager_user_id
            ),
        )
-    presign_crud.validate_presign_person_ids(
-        db,
-        tenant_no=ctx.tenant_no,
-        collector_user_id=fields.get(
-            "collector_user_id", current.collector_user_id
-        ),
-        sales_user_id=fields.get("sales_user_id", current.sales_user_id),
-        region_manager_user_id=fields.get(
-            "region_manager_user_id", current.region_manager_user_id
-        ),
-    )
-    row = presign_crud.update_presign_order(
-        db,
-        order_id=order_id,
-        tenant_no=ctx.tenant_no,
-        user_id=ctx.user_id,
-        fields=fields,
-        sample_type_ids=sample_type_ids,
-        create_dept_id=_current_dept_id(user_info),
-    )
-    row = assert_presign_access(
-        db, row.id, ctx, user_info=user_info
-    )
-    return presign_crud.order_to_dict(
-        db, row, ctx.tenant_no, with_children=True
-    )
+        row = presign_crud.update_presign_order(
+            db,
+            order_id=order_id,
+            tenant_no=ctx.tenant_no,
+            user_id=ctx.user_id,
+            fields=fields,
+            sample_type_ids=sample_type_ids,
+            create_dept_id=_current_dept_id(user_info),
+        )
+        row = assert_presign_access(
+            db, row.id, ctx, user_info=user_info
+        )
+        return presign_crud.order_to_dict(
+            db, row, ctx.tenant_no, with_children=True
+        )
+    except Exception:
+        db.rollback()
+        raise


 def delete_presign_order(
@@ -396,15 +421,19 @@ def delete_presign_order(
    ctx: MarketFieldContext,
    user_info: dict,
 ) -> None:
-    row = assert_presign_access(db, order_id, ctx, user_info=user_info)
-    if (row.order_status or '').strip() == 'converted':
-        raise ValueError('已转正式的预签约订单不可删除')
-    presign_crud.soft_delete_presign_order(
-        db,
-        order_id=order_id,
-        tenant_no=ctx.tenant_no,
-        user_id=ctx.user_id,
-    )
+    try:
+        row = _scoped_order_for_update(db, order_id, ctx, user_info)
+        if (row.order_status or '').strip() == 'converted':
+            raise ValueError('已转正式的预签约订单不可删除')
+        presign_crud.soft_delete_presign_order(
+            db,
+            order_id=order_id,
+            tenant_no=ctx.tenant_no,
+            user_id=ctx.user_id,
+        )
+    except Exception:
+        db.rollback()
+        raise


 def _scoped_transfer(
@@ -413,7 +442,7 @@ def _scoped_transfer(
    ctx: MarketFieldContext,
    user_info: dict,
 ) -> BizPresignTransfer:
-    row = (
+    probe = (
        db.query(BizPresignTransfer)
        .filter(
            BizPresignTransfer.id == transfer_id,
@@ -422,15 +451,68 @@ def _scoped_transfer(
        )
        .first()
    )
-    if row is None:
+    if probe is None:
        raise HTTPException(status_code=404, detail='转账记录不存在或无权访问')
-    order = assert_presign_access(
-        db,
-        row.presign_order_id,
-        ctx,
-        user_info=user_info,
+    order = _scoped_order_for_update(
+        db, probe.presign_order_id, ctx, user_info
    )
    _assert_transfer_order_writable(order)
+    row = (
+        db.query(BizPresignTransfer)
+        .filter(
+            BizPresignTransfer.id == transfer_id,
+            BizPresignTransfer.tenant_no == ctx.tenant_no,
+            BizPresignTransfer.del_flag == 0,
+        )
+        .populate_existing()
+        .with_for_update()
+        .first()
+    )
+    if row is None:
+        raise HTTPException(status_code=404, detail='转账记录不存在或无权访问')
+    return row
+
+
+def _update_transfer_if_status(
+    db: Session,
+    *,
+    transfer_id: str,
+    tenant_no: str,
+    expected_status: str,
+    values: dict,
+    extra_conditions: tuple = (),
+) -> None:
+    updates = dict(values)
+    changed = (
+        db.query(BizPresignTransfer)
+        .filter(
+            BizPresignTransfer.id == transfer_id,
+            BizPresignTransfer.tenant_no == tenant_no,
+            BizPresignTransfer.del_flag == 0,
+            BizPresignTransfer.pay_status == expected_status,
+            *extra_conditions,
+        )
+        .update(updates, synchronize_session=False)
+    )
+    if changed != 1:
+        raise ValueError('转账状态已变化，请刷新后重试')
+
+
+def _reload_transfer(
+    db: Session, transfer_id: str, tenant_no: str
+) -> BizPresignTransfer:
+    row = (
+        db.query(BizPresignTransfer)
+        .filter(
+            BizPresignTransfer.id == transfer_id,
+            BizPresignTransfer.tenant_no == tenant_no,
+            BizPresignTransfer.del_flag == 0,
+        )
+        .populate_existing()
+        .first()
+    )
+    if row is None:
+        raise HTTPException(status_code=404, detail='转账记录不存在')
    return row


@@ -442,8 +524,8 @@ def create_presign_transfer(
    user_info: dict,
 ) -> dict:
    try:
-        order = assert_presign_access(
-            db, order_id, ctx, user_info=user_info
+        order = _scoped_order_for_update(
+            db, order_id, ctx, user_info
        )
        _assert_transfer_order_writable(order)
        fields = _dump_transfer_body(body)
@@ -495,13 +577,42 @@ def update_presign_transfer(
                existing_refs=current.transfer_attachments,
                business_id=current.presign_order_id,
            )
-        row = presign_crud.update_transfer_draft(
+        values = {
+            BizPresignTransfer.update_by: ctx.user_id,
+            BizPresignTransfer.update_time: datetime.now(),
+        }
+        if 'transfer_amount' in fields:
+            values[BizPresignTransfer.transfer_amount] = fields[
+                'transfer_amount'
+            ]
+        if 'transfer_time' in fields:
+            raw_time = fields['transfer_time']
+            values[BizPresignTransfer.transfer_time] = (
+                None
+                if raw_time in (None, '')
+                else parse_datetime(str(raw_time))
+            )
+        if 'transfer_type' in fields:
+            values[BizPresignTransfer.transfer_type] = (
+                str(fields['transfer_type'] or '').strip() or None
+            )
+        if 'transfer_attachments' in fields:
+            values[BizPresignTransfer.transfer_attachments] = (
+                str(fields['transfer_attachments'] or '').strip() or None
+            )
+        if 'remark' in fields:
+            values[BizPresignTransfer.remark] = (
+                str(fields['remark'] or '').strip() or None
+            )
+        _update_transfer_if_status(
            db,
            transfer_id=transfer_id,
            tenant_no=ctx.tenant_no,
-            user_id=ctx.user_id,
-            fields=fields,
+            expected_status='dfj',
+            values=values,
        )
+        db.commit()
+        row = _reload_transfer(db, transfer_id, ctx.tenant_no)
        return presign_crud._transfer_to_dict(db, row, ctx.tenant_no)
    except Exception:
        db.rollback()
@@ -518,10 +629,17 @@ def delete_presign_transfer(
        row = _scoped_transfer(db, transfer_id, ctx, user_info)
        if (row.pay_status or '').strip() != 'dfj':
            raise ValueError('仅待缴费状态可删除意向金转账笔')
-        row.del_flag = 1
-        row.update_by = ctx.user_id
-        row.update_time = datetime.now()
-        db.add(row)
+        _update_transfer_if_status(
+            db,
+            transfer_id=transfer_id,
+            tenant_no=ctx.tenant_no,
+            expected_status='dfj',
+            values={
+                BizPresignTransfer.del_flag: 1,
+                BizPresignTransfer.update_by: ctx.user_id,
+                BizPresignTransfer.update_time: datetime.now(),
+            },
+        )
        db.commit()
    except Exception:
        db.rollback()
@@ -535,13 +653,32 @@ def submit_presign_transfer(
    user_info: dict,
 ) -> dict:
    try:
-        _scoped_transfer(db, transfer_id, ctx, user_info)
-        row = presign_crud.submit_receipt_finance(
+        row = _scoped_transfer(db, transfer_id, ctx, user_info)
+        if (row.pay_status or '').strip() != 'dfj':
+            raise ValueError('仅待缴费状态可提交财务确认')
+        if row.transfer_amount is None or row.transfer_amount <= 0:
+            raise ValueError('请填写转账金额')
+        if not (row.transfer_type or '').strip():
+            raise ValueError('请选择转账方式')
+        _update_transfer_if_status(
            db,
            transfer_id=transfer_id,
            tenant_no=ctx.tenant_no,
-            user_id=ctx.user_id,
+            expected_status='dfj',
+            values={
+                BizPresignTransfer.pay_status: 'jfqrz',
+                BizPresignTransfer.update_by: ctx.user_id,
+                BizPresignTransfer.update_time: datetime.now(),
+            },
+            extra_conditions=(
+                BizPresignTransfer.transfer_amount.isnot(None),
+                BizPresignTransfer.transfer_amount > 0,
+                BizPresignTransfer.transfer_type.isnot(None),
+                func.length(func.trim(BizPresignTransfer.transfer_type)) > 0,
+            ),
        )
+        db.commit()
+        row = _reload_transfer(db, transfer_id, ctx.tenant_no)
        return presign_crud._transfer_to_dict(db, row, ctx.tenant_no)
    except Exception:
        db.rollback()
@@ -555,13 +692,22 @@ def withdraw_presign_transfer(
    user_info: dict,
 ) -> dict:
    try:
-        _scoped_transfer(db, transfer_id, ctx, user_info)
-        row = presign_crud.withdraw_receipt(
+        row = _scoped_transfer(db, transfer_id, ctx, user_info)
+        if (row.pay_status or '').strip() != 'jfqrz':
+            raise ValueError('仅缴费确认中可撤回')
+        _update_transfer_if_status(
            db,
            transfer_id=transfer_id,
            tenant_no=ctx.tenant_no,
-            user_id=ctx.user_id,
+            expected_status='jfqrz',
+            values={
+                BizPresignTransfer.pay_status: 'dfj',
+                BizPresignTransfer.update_by: ctx.user_id,
+                BizPresignTransfer.update_time: datetime.now(),
+            },
        )
+        db.commit()
+        row = _reload_transfer(db, transfer_id, ctx.tenant_no)
        return presign_crud._transfer_to_dict(db, row, ctx.tenant_no)
    except Exception:
        db.rollback()
@@ -575,20 +721,24 @@ def convert_presign_order(
    ctx: MarketFieldContext,
    user_info: dict,
 ) -> dict:
-    row = assert_presign_access(db, order_id, ctx, user_info=user_info)
-    if (row.order_status or '').strip() not in ('', 'draft', 'active'):
-        raise ValueError('当前状态的预签约订单不可转正式协议')
-    presign_crud.validate_presign_person_ids(
-        db,
-        tenant_no=ctx.tenant_no,
-        collector_user_id=row.collector_user_id or ctx.user_id,
-        sales_user_id=row.sales_user_id,
-        region_manager_user_id=row.region_manager_user_id,
-    )
-    return presign_crud.convert_to_protocol(
-        db,
-        order_id=order_id,
-        tenant_no=ctx.tenant_no,
-        user_id=ctx.user_id,
-        create_dept_id=_current_dept_id(user_info),
-    )
+    try:
+        row = _scoped_order_for_update(db, order_id, ctx, user_info)
+        if (row.order_status or '').strip() not in ('', 'draft', 'active'):
+            raise ValueError('当前状态的预签约订单不可转正式协议')
+        presign_crud.validate_presign_person_ids(
+            db,
+            tenant_no=ctx.tenant_no,
+            collector_user_id=row.collector_user_id or ctx.user_id,
+            sales_user_id=row.sales_user_id,
+            region_manager_user_id=row.region_manager_user_id,
+        )
+        return presign_crud.convert_to_protocol(
+            db,
+            order_id=order_id,
+            tenant_no=ctx.tenant_no,
+            user_id=ctx.user_id,
+            create_dept_id=_current_dept_id(user_info),
+        )
+    except Exception:
+        db.rollback()
+        raise
@@ -32,35 +32,42 @@ from app.models.research_donation import (
 )


-def _session():
-    engine = create_engine("sqlite:///:memory:")
+_TEST_TABLES = [
+    BizCollectionRegionStaff.__table__,
+    BizCollectionRegion.__table__,
+    BizCollectionHospital.__table__,
+    BizPresignOrder.__table__,
+    BizPresignOrderSampleType.__table__,
+    BizPresignTransfer.__table__,
+    BizCellSampleType.__table__,
+    BizCollectionProtocol.__table__,
+    BizProtocolPartyCustomer.__table__,
+    BizProtocolCollectionInfo.__table__,
+    BizProtocolSample.__table__,
+    BizResearchDonation.__table__,
+    BizResearchDonationSample.__table__,
+    BizAttachment.__table__,
+    SysUser.__table__,
+    SysRole.__table__,
+    SysRoleDataScopeDetail.__table__,
+    SysUserRole.__table__,
+    SysPost.__table__,
+    SysPostRole.__table__,
+    SysUserPost.__table__,
+]
+
+
+def _session_factory(database_url="sqlite:///:memory:"):
+    engine = create_engine(database_url)
    Base.metadata.create_all(
        engine,
-        tables=[
-            BizCollectionRegionStaff.__table__,
-            BizCollectionRegion.__table__,
-            BizCollectionHospital.__table__,
-            BizPresignOrder.__table__,
-            BizPresignOrderSampleType.__table__,
-            BizPresignTransfer.__table__,
-            BizCellSampleType.__table__,
-            BizCollectionProtocol.__table__,
-            BizProtocolPartyCustomer.__table__,
-            BizProtocolCollectionInfo.__table__,
-            BizProtocolSample.__table__,
-            BizResearchDonation.__table__,
-            BizResearchDonationSample.__table__,
-            BizAttachment.__table__,
-            SysUser.__table__,
-            SysRole.__table__,
-            SysRoleDataScopeDetail.__table__,
-            SysUserRole.__table__,
-            SysPost.__table__,
-            SysPostRole.__table__,
-            SysUserPost.__table__,
-        ],
+        tables=_TEST_TABLES,
    )
-    return sessionmaker(bind=engine)()
+    return sessionmaker(bind=engine)
+
+
+def _session():
+    return _session_factory()()


 def _order(
@@ -1033,6 +1040,72 @@ def test_market_presign_bodies_forbid_unknown_fields_and_validate_transfer_limit
            MarketPresignTransferBody.model_validate(payload)


+def test_market_presign_order_body_accepts_database_sized_values():
+    from decimal import Decimal
+
+    from app.schemas.market_mobile import MarketPresignOrderBody
+
+    body = MarketPresignOrderBody.model_validate(
+        {
+            'orderDate': '20261011',
+            'protocolNo': 'P' * 64,
+            'motherName': '母' * 64,
+            'expectedDeliveryDate': '20261231',
+            'regionId': 'r' * 64,
+            'hospitalId': 'h' * 64,
+            'sampleCollectTime': '2026-10-11 12:34:56',
+            'payMode': 'x' * 8,
+            'contractAmount': '9999999999999999.99',
+            'collectorUserId': 'c' * 64,
+            'salesUserId': 's' * 64,
+            'regionManagerUserId': 'm' * 64,
+            'remark': '备' * 512,
+            'sampleTypeIds': [f'sample-{index}' for index in range(100)],
+        }
+    )
+
+    assert body.order_date == '20261011'
+    assert body.expected_delivery_date == '20261231'
+    assert body.contract_amount == Decimal('9999999999999999.99')
+    assert len(body.sample_type_ids or []) == 100
+
+
+@pytest.mark.parametrize(
+    'payload',
+    [
+        {'orderDate': '2026-10-11'},
+        {'orderDate': '20260230'},
+        {'expectedDeliveryDate': '20261301'},
+        {'sampleCollectTime': 'not-a-datetime'},
+        {'protocolNo': 'p' * 65},
+        {'motherName': 'm' * 65},
+        {'regionId': 'r' * 65},
+        {'hospitalId': 'h' * 65},
+        {'collectorUserId': 'c' * 65},
+        {'salesUserId': 's' * 65},
+        {'regionManagerUserId': 'm' * 65},
+        {'payMode': 'p' * 9},
+        {'remark': 'r' * 513},
+        {'sampleTypeIds': ['s' * 65]},
+        {'sampleTypeIds': [f'sample-{index}' for index in range(101)]},
+        {'contractAmount': 'NaN'},
+        {'contractAmount': 'Infinity'},
+        {'contractAmount': '-0.01'},
+        {'contractAmount': '10000000000000000.00'},
+        {'contractAmount': '1.001'},
+    ],
+)
+def test_market_presign_order_body_rejects_values_outside_database_bounds(
+    payload,
+):
+    from pydantic import ValidationError
+
+    from app.schemas.market_mobile import MarketPresignOrderBody
+
+    with pytest.raises(ValidationError):
+        MarketPresignOrderBody.model_validate(payload)
+
+
@pytest.mark.parametrize('status', ['active', 'void'])
 def test_specialist_can_soft_delete_own_non_converted_presign(status):
    from app.services.market_presign import delete_presign_order
@@ -1562,6 +1635,174 @@ def test_presign_transfer_attachments_bind_to_parent_order():
    }


+def test_stale_withdraw_does_not_overwrite_later_finance_confirmation(tmp_path):
+    from app.crud import presign_order as presign_crud
+    from app.services.market_presign import withdraw_presign_transfer
+
+    Session = _session_factory(f"sqlite:///{tmp_path / 'withdraw-race.db'}")
+    market_db = Session()
+    finance_db = Session()
+    try:
+        market_db.add(_order('own', sales_user_id='u1'))
+        market_db.add(_transfer('transfer', pay_status='jfqrz'))
+        market_db.commit()
+        stale = market_db.query(BizPresignTransfer).filter_by(id='transfer').one()
+        assert stale.pay_status == 'jfqrz'
+
+        presign_crud.confirm_receipt(
+            finance_db,
+            transfer_id='transfer',
+            tenant_no='tenant-a',
+            user_id='finance-user',
+        )
+
+        with pytest.raises(ValueError, match='状态|撤回'):
+            withdraw_presign_transfer(
+                market_db,
+                transfer_id='transfer',
+                ctx=_ctx('specialist'),
+                user_info=_user(),
+            )
+
+        finance_db.expire_all()
+        current = finance_db.query(BizPresignTransfer).filter_by(id='transfer').one()
+        assert current.pay_status == 'yjf'
+        assert current.update_by == 'finance-user'
+    finally:
+        market_db.close()
+        finance_db.close()
+
+
+@pytest.mark.parametrize('action', ['update', 'delete', 'submit'])
+def test_stale_dfj_action_does_not_mutate_later_paid_transfer(tmp_path, action):
+    from app.crud import presign_order as presign_crud
+    from app.schemas.market_mobile import MarketPresignTransferBody
+    from app.services.market_presign import (
+        delete_presign_transfer,
+        submit_presign_transfer,
+        update_presign_transfer,
+    )
+
+    Session = _session_factory(
+        f"sqlite:///{tmp_path / f'transfer-{action}-race.db'}"
+    )
+    market_db = Session()
+    finance_db = Session()
+    try:
+        market_db.add(_order('own', sales_user_id='u1'))
+        transfer = _transfer('transfer', pay_status='dfj')
+        transfer.remark = 'original'
+        market_db.add(transfer)
+        market_db.commit()
+        stale = market_db.query(BizPresignTransfer).filter_by(id='transfer').one()
+        assert stale.pay_status == 'dfj'
+
+        presign_crud.submit_receipt_finance(
+            finance_db,
+            transfer_id='transfer',
+            tenant_no='tenant-a',
+            user_id='finance-user',
+        )
+        presign_crud.confirm_receipt(
+            finance_db,
+            transfer_id='transfer',
+            tenant_no='tenant-a',
+            user_id='finance-user',
+        )
+        calls = {
+            'update': lambda: update_presign_transfer(
+                market_db,
+                transfer_id='transfer',
+                body=MarketPresignTransferBody(remark='stale-write'),
+                ctx=_ctx('specialist'),
+                user_info=_user(),
+            ),
+            'delete': lambda: delete_presign_transfer(
+                market_db,
+                transfer_id='transfer',
+                ctx=_ctx('specialist'),
+                user_info=_user(),
+            ),
+            'submit': lambda: submit_presign_transfer(
+                market_db,
+                transfer_id='transfer',
+                ctx=_ctx('specialist'),
+                user_info=_user(),
+            ),
+        }
+
+        with pytest.raises(ValueError, match='状态|待缴费|提交'):
+            calls[action]()
+
+        finance_db.expire_all()
+        current = finance_db.query(BizPresignTransfer).filter_by(id='transfer').one()
+        assert current.pay_status == 'yjf'
+        assert current.del_flag == 0
+        assert current.remark == 'original'
+    finally:
+        market_db.close()
+        finance_db.close()
+
+
+@pytest.mark.parametrize('action', ['update', 'delete'])
+def test_stale_order_write_does_not_mutate_later_conversion(tmp_path, action):
+    from app.schemas.market_mobile import MarketPresignOrderBody
+    from app.services.market_presign import (
+        convert_presign_order,
+        delete_presign_order,
+        update_presign_order,
+    )
+
+    Session = _session_factory(
+        f"sqlite:///{tmp_path / f'order-{action}-race.db'}"
+    )
+    market_db = Session()
+    convert_db = Session()
+    try:
+        _add_master_data(market_db)
+        order = _order('own', sales_user_id='u1')
+        order.protocol_no = f'P-RACE-{action}'
+        market_db.add(order)
+        market_db.commit()
+        stale = market_db.query(BizPresignOrder).filter_by(id='own').one()
+        assert stale.order_status == 'active'
+
+        convert_presign_order(
+            convert_db,
+            order_id='own',
+            ctx=_ctx('specialist'),
+            user_info=_user(),
+        )
+        calls = {
+            'update': lambda: update_presign_order(
+                market_db,
+                order_id='own',
+                body=MarketPresignOrderBody(remark='stale-write'),
+                ctx=_ctx('specialist'),
+                user_info=_user(),
+            ),
+            'delete': lambda: delete_presign_order(
+                market_db,
+                order_id='own',
+                ctx=_ctx('specialist'),
+                user_info=_user(),
+            ),
+        }
+
+        with pytest.raises(ValueError, match='不可|已转'):
+            calls[action]()
+
+        convert_db.expire_all()
+        current = convert_db.query(BizPresignOrder).filter_by(id='own').one()
+        assert current.order_status == 'converted'
+        assert current.converted_protocol_id
+        assert current.del_flag == 0
+        assert current.remark is None
+    finally:
+        market_db.close()
+        convert_db.close()
+
+
 def test_conversion_unique_index_migration_is_mysql57_idempotent():
    from pathlib import Path