修复：强化发货权限与幂等一致性

shierfeng-fastapi-backend/module_business/service/shipment_service.py

@@ -476,6 +476,15 @@ class ShipmentService:
                 ordered_ids,
                 page_object.source_warehouse_id,
             )
+            await cls._lock_and_authorize_routes(
+                query_db,
+                current_user,
+                warehouse_ids={page_object.source_warehouse_id},
+                site_ids={page_object.target_site_id},
+                hidden_warehouse_ids=set(),
+                hidden_site_ids=set(),
+                require_active=True,
+            )
 
             current_time = datetime.now()
             shipment = BizShipmentOrder(
@@ -538,6 +547,21 @@ class ShipmentService:
                 ordered_ids,
                 page_object.source_warehouse_id,
             )
+            await cls._lock_and_authorize_routes(
+                query_db,
+                current_user,
+                warehouse_ids={
+                    shipment.source_warehouse_id,
+                    page_object.source_warehouse_id,
+                },
+                site_ids={
+                    shipment.target_site_id,
+                    page_object.target_site_id,
+                },
+                hidden_warehouse_ids={shipment.source_warehouse_id},
+                hidden_site_ids={shipment.target_site_id},
+                require_active=True,
+            )
 
             current_time = datetime.now()
             shipment.source_warehouse_id = page_object.source_warehouse_id
@@ -576,6 +600,7 @@ class ShipmentService:
 
             ordered_ids = sorted(shipment_ids)
             warehouse_ids, site_ids = await cls._scopes(query_db, current_user)
+            locked_shipments = []
             for shipment_id in ordered_ids:
                 shipment = await OperationDao.get_shipment_for_update(
                     query_db,
@@ -587,6 +612,19 @@ class ShipmentService:
                     cls._raise_shipment_not_accessible(current_user)
                 if shipment.status != DocumentStatus.DRAFT.value:
                     raise ServiceException(message='已确认发货单不允许删除')
+                locked_shipments.append(shipment)
+
+            route_warehouse_ids = {shipment.source_warehouse_id for shipment in locked_shipments}
+            route_site_ids = {shipment.target_site_id for shipment in locked_shipments}
+            await cls._lock_and_authorize_routes(
+                query_db,
+                current_user,
+                warehouse_ids=route_warehouse_ids,
+                site_ids=route_site_ids,
+                hidden_warehouse_ids=route_warehouse_ids,
+                hidden_site_ids=route_site_ids,
+                require_active=False,
+            )
 
             for shipment_id in ordered_ids:
                 await OperationDao.delete_shipment(query_db, shipment_id)
@@ -613,50 +651,80 @@ class ShipmentService:
             and shipment.confirm_request_hash == request_hash
         )
 
+    @classmethod
+    async def _lock_and_authorize_routes(
+        cls,
+        query_db: AsyncSession,
+        current_user: CurrentUserModel,
+        *,
+        warehouse_ids: set[int],
+        site_ids: set[int],
+        hidden_warehouse_ids: set[int],
+        hidden_site_ids: set[int],
+        require_active: bool,
+    ) -> None:
+        for warehouse_id in sorted(warehouse_ids):
+            warehouse = await MasterDao.get_warehouse(
+                query_db,
+                warehouse_id,
+                for_update=True,
+            )
+            if warehouse is None:
+                raise ServiceException(message='来源仓库不存在')
+            if require_active and warehouse.status != '0':
+                raise ServiceException(message='来源仓库已停用')
+            try:
+                await AccessService.assert_warehouse_access(
+                    query_db,
+                    current_user,
+                    warehouse_id,
+                    for_update=True,
+                )
+            except PermissionException:
+                if warehouse_id in hidden_warehouse_ids:
+                    cls._raise_shipment_not_accessible(current_user)
+                raise
+
+        for site_id in sorted(site_ids):
+            site = await MasterDao.get_site(
+                query_db,
+                site_id,
+                for_update=True,
+            )
+            if site is None:
+                raise ServiceException(message='目标现场不存在')
+            if require_active and site.status != '0':
+                raise ServiceException(message='目标现场已停用')
+            try:
+                await AccessService.assert_site_access(
+                    query_db,
+                    current_user,
+                    site_id,
+                    for_update=True,
+                )
+            except PermissionException:
+                if site_id in hidden_site_ids:
+                    cls._raise_shipment_not_accessible(current_user)
+                raise
+
     @classmethod
     async def _lock_and_reauthorize_route(
         cls,
         query_db: AsyncSession,
         shipment: BizShipmentOrder,
         current_user: CurrentUserModel,
+        *,
+        require_active: bool = True,
     ) -> None:
-        warehouse = await MasterDao.get_warehouse(
+        await cls._lock_and_authorize_routes(
             query_db,
-            shipment.source_warehouse_id,
+            current_user,
-            for_update=True,
+            warehouse_ids={shipment.source_warehouse_id},
+            site_ids={shipment.target_site_id},
+            hidden_warehouse_ids={shipment.source_warehouse_id},
+            hidden_site_ids={shipment.target_site_id},
+            require_active=require_active,
         )
-        if warehouse is None:
-            raise ServiceException(message='来源仓库不存在')
-        if warehouse.status != '0':
-            raise ServiceException(message='来源仓库已停用')
-        try:
-            await AccessService.assert_warehouse_access(
-                query_db,
-                current_user,
-                shipment.source_warehouse_id,
-                for_update=True,
-            )
-        except PermissionException:
-            cls._raise_shipment_not_accessible(current_user)
-
-        site = await MasterDao.get_site(
-            query_db,
-            shipment.target_site_id,
-            for_update=True,
-        )
-        if site is None:
-            raise ServiceException(message='目标现场不存在')
-        if site.status != '0':
-            raise ServiceException(message='目标现场已停用')
-        try:
-            await AccessService.assert_site_access(
-                query_db,
-                current_user,
-                shipment.target_site_id,
-                for_update=True,
-            )
-        except PermissionException:
-            cls._raise_shipment_not_accessible(current_user)
 
     @classmethod
     async def _recover_confirm_integrity_error(
@@ -684,11 +752,26 @@ class ShipmentService:
                 request_hash,
             ):
                 raise ServiceException(message='确认发货失败，幂等键已被其他请求使用')
-            await cls._validate_active_route(
+            locked_recovered = await OperationDao.get_shipment_for_update(
                 query_db,
-                recovered.source_warehouse_id,
+                shipment_id,
-                recovered.target_site_id,
+                warehouse_ids=warehouse_ids,
+                site_ids=site_ids,
+            )
+            if locked_recovered is None:
+                cls._raise_shipment_not_accessible(current_user)
+            if not cls._is_exact_confirmed(
+                locked_recovered,
+                shipment_id,
+                idempotency_key,
+                request_hash,
+            ):
+                raise ServiceException(message='确认发货失败，幂等键已被其他请求使用')
+            await cls._lock_and_reauthorize_route(
+                query_db,
+                locked_recovered,
                 current_user,
+                require_active=False,
             )
             return cls._success('确认成功')
         finally:
@@ -745,7 +828,12 @@ class ShipmentService:
                     idempotency_key,
                     request_hash,
                 ):
-                    await cls._lock_and_reauthorize_route(query_db, locked_shipment, current_user)
+                    await cls._lock_and_reauthorize_route(
+                        query_db,
+                        locked_shipment,
+                        current_user,
+                        require_active=False,
+                    )
                     await query_db.rollback()
                     return cls._success('确认成功')
                 raise ServiceException(message='发货单已确认，幂等信息不匹配')

shierfeng-fastapi-backend/tests/module_business/test_shipment_service.py

@@ -745,6 +745,10 @@ class ShipmentDraftServiceTest(unittest.IsolatedAsyncioTestCase):
                 'pallet-lock',
                 'filling-lock-102',
                 'filling-lock-105',
+                'warehouse-read',
+                'warehouse-access',
+                'site-read',
+                'site-access',
                 'add-order',
                 'replace-items',
                 'commit',
@@ -756,6 +760,114 @@ class ShipmentDraftServiceTest(unittest.IsolatedAsyncioTestCase):
         self.db.commit.assert_awaited_once()
         self.db.rollback.assert_not_awaited()
 
+    async def test_add_rechecks_revoked_mapping_after_pallet_locks_before_writing(self) -> None:
+        page = ShipmentCreateModel(sourceWarehouseId=7, targetSiteId=9, palletIds=[2])
+
+        async def warehouse_access(*_args: Any, **kwargs: Any) -> None:
+            if kwargs.get('for_update'):
+                raise PermissionException(data='', message='无权访问该仓库')
+
+        add_order = AsyncMock(return_value=make_shipment(pallet_ids=(2,)))
+        replace_items = AsyncMock()
+        warehouse_mapping = AsyncMock(side_effect=warehouse_access)
+        with (
+            patch.object(AccessService, 'assert_warehouse_access', new=warehouse_mapping),
+            patch.object(AccessService, 'assert_site_access', new=AsyncMock()),
+            patch.object(
+                MasterDao,
+                'get_warehouse',
+                new=AsyncMock(
+                    return_value=BizWarehouse(
+                        warehouse_id=7,
+                        warehouse_code='W-7',
+                        warehouse_name='七号仓',
+                        status='0',
+                    )
+                ),
+            ) as warehouse_read,
+            patch.object(
+                MasterDao,
+                'get_site',
+                new=AsyncMock(
+                    return_value=BizCustomerSite(
+                        site_id=9,
+                        customer_id=1,
+                        site_code='S-9',
+                        site_name='九号现场',
+                        status='0',
+                    )
+                ),
+            ),
+            patch.object(
+                ShipmentService,
+                '_lock_and_validate_pallets',
+                new=AsyncMock(return_value=([], [], [])),
+            ),
+            patch.object(OperationDao, 'add_shipment', new=add_order),
+            patch.object(OperationDao, 'replace_shipment_items', new=replace_items),
+            self.assertRaises(PermissionException) as raised,
+        ):
+            await ShipmentService.add_shipment_services(self.db, page, self.user)
+
+        self.assertEqual('无权访问该仓库', raised.exception.message)
+        self.assertEqual(2, warehouse_read.await_count)
+        warehouse_read.assert_awaited_with(self.db, 7, for_update=True)
+        self.assertEqual(2, warehouse_mapping.await_count)
+        warehouse_mapping.assert_awaited_with(self.db, self.user, 7, for_update=True)
+        add_order.assert_not_awaited()
+        replace_items.assert_not_awaited()
+        self.db.commit.assert_not_awaited()
+        self.db.rollback.assert_awaited_once()
+
+    async def test_add_rechecks_disabled_master_after_pallet_locks_before_writing(self) -> None:
+        page = ShipmentCreateModel(sourceWarehouseId=7, targetSiteId=9, palletIds=[2])
+
+        async def get_warehouse(*_args: Any, **kwargs: Any) -> BizWarehouse:
+            return BizWarehouse(
+                warehouse_id=7,
+                warehouse_code='W-7',
+                warehouse_name='七号仓',
+                status='1' if kwargs.get('for_update') else '0',
+            )
+
+        add_order = AsyncMock(return_value=make_shipment(pallet_ids=(2,)))
+        replace_items = AsyncMock()
+        warehouse_read = AsyncMock(side_effect=get_warehouse)
+        with (
+            patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()),
+            patch.object(AccessService, 'assert_site_access', new=AsyncMock()),
+            patch.object(MasterDao, 'get_warehouse', new=warehouse_read),
+            patch.object(
+                MasterDao,
+                'get_site',
+                new=AsyncMock(
+                    return_value=BizCustomerSite(
+                        site_id=9,
+                        customer_id=1,
+                        site_code='S-9',
+                        site_name='九号现场',
+                        status='0',
+                    )
+                ),
+            ),
+            patch.object(
+                ShipmentService,
+                '_lock_and_validate_pallets',
+                new=AsyncMock(return_value=([], [], [])),
+            ),
+            patch.object(OperationDao, 'add_shipment', new=add_order),
+            patch.object(OperationDao, 'replace_shipment_items', new=replace_items),
+            self.assertRaises(ServiceException) as raised,
+        ):
+            await ShipmentService.add_shipment_services(self.db, page, self.user)
+
+        self.assertEqual('来源仓库已停用', raised.exception.message)
+        warehouse_read.assert_awaited_with(self.db, 7, for_update=True)
+        add_order.assert_not_awaited()
+        replace_items.assert_not_awaited()
+        self.db.commit.assert_not_awaited()
+        self.db.rollback.assert_awaited_once()
+
     async def test_edit_revalidates_current_draft_and_keeps_snapshots_null(self) -> None:
         page = ShipmentUpdateModel(
             shipmentId=12,
@@ -848,6 +960,229 @@ class ShipmentDraftServiceTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual('FILLED_IN_WAREHOUSE', inventory.flow_state)
         self.db.commit.assert_awaited_once()
 
+    async def test_edit_rechecks_old_and_new_routes_before_writing(self) -> None:
+        scenarios = (
+            ('current-warehouse-revoked', PermissionException, '无权访问该发货单'),
+            ('new-site-revoked', PermissionException, '无权访问该客户现场'),
+            ('new-site-disabled', ServiceException, '目标现场已停用'),
+        )
+        for scenario, exception_type, expected_message in scenarios:
+            with self.subTest(scenario=scenario):
+                db = MagicMock(spec=AsyncSession)
+                shipment = make_shipment(pallet_ids=(2,))
+                page = ShipmentUpdateModel(
+                    shipmentId=12,
+                    sourceWarehouseId=8,
+                    targetSiteId=10,
+                    palletIds=[2],
+                )
+                current_warehouse_id = shipment.source_warehouse_id
+                new_site_id = page.target_site_id
+
+                async def warehouse_access(
+                    _db: AsyncSession,
+                    _user: CurrentUserModel,
+                    warehouse_id: int,
+                    _scenario: str = scenario,
+                    _current_warehouse_id: int = current_warehouse_id,
+                    **kwargs: Any,
+                ) -> None:
+                    if (
+                        _scenario == 'current-warehouse-revoked'
+                        and warehouse_id == _current_warehouse_id
+                        and kwargs.get('for_update')
+                    ):
+                        raise PermissionException(data='', message='无权访问该仓库')
+
+                async def site_access(
+                    _db: AsyncSession,
+                    _user: CurrentUserModel,
+                    site_id: int,
+                    _scenario: str = scenario,
+                    _new_site_id: int = new_site_id,
+                    **kwargs: Any,
+                ) -> None:
+                    if _scenario == 'new-site-revoked' and site_id == _new_site_id and kwargs.get('for_update'):
+                        raise PermissionException(data='', message='无权访问该客户现场')
+
+                async def get_warehouse(
+                    _db: AsyncSession,
+                    warehouse_id: int,
+                    **_kwargs: Any,
+                ) -> BizWarehouse:
+                    return BizWarehouse(
+                        warehouse_id=warehouse_id,
+                        warehouse_code=f'W-{warehouse_id}',
+                        warehouse_name=f'{warehouse_id}号仓',
+                        status='0',
+                    )
+
+                async def get_site(
+                    _db: AsyncSession,
+                    site_id: int,
+                    _scenario: str = scenario,
+                    _new_site_id: int = new_site_id,
+                    **kwargs: Any,
+                ) -> BizCustomerSite:
+                    return BizCustomerSite(
+                        site_id=site_id,
+                        customer_id=1,
+                        site_code=f'S-{site_id}',
+                        site_name=f'{site_id}号现场',
+                        status=(
+                            '1'
+                            if _scenario == 'new-site-disabled' and site_id == _new_site_id and kwargs.get('for_update')
+                            else '0'
+                        ),
+                    )
+
+                update = AsyncMock()
+                replace_items = AsyncMock()
+                warehouse_mapping = AsyncMock(side_effect=warehouse_access)
+                site_mapping = AsyncMock(side_effect=site_access)
+                site_read = AsyncMock(side_effect=get_site)
+                with (
+                    patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7, 8])),
+                    patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9, 10])),
+                    patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(return_value=shipment)),
+                    patch.object(AccessService, 'assert_warehouse_access', new=warehouse_mapping),
+                    patch.object(AccessService, 'assert_site_access', new=site_mapping),
+                    patch.object(MasterDao, 'get_warehouse', new=AsyncMock(side_effect=get_warehouse)),
+                    patch.object(MasterDao, 'get_site', new=site_read),
+                    patch.object(
+                        ShipmentService,
+                        '_lock_and_validate_pallets',
+                        new=AsyncMock(return_value=([], [], [])),
+                    ),
+                    patch.object(OperationDao, 'update_shipment', new=update),
+                    patch.object(OperationDao, 'replace_shipment_items', new=replace_items),
+                    self.assertRaises(exception_type) as raised,
+                ):
+                    await ShipmentService.edit_shipment_services(db, page, self.user)
+
+                self.assertEqual(expected_message, raised.exception.message)
+                if scenario == 'current-warehouse-revoked':
+                    warehouse_mapping.assert_any_await(db, self.user, 7, for_update=True)
+                else:
+                    site_read.assert_any_await(db, 10, for_update=True)
+                if scenario == 'new-site-revoked':
+                    site_mapping.assert_any_await(db, self.user, 10, for_update=True)
+                update.assert_not_awaited()
+                replace_items.assert_not_awaited()
+                db.commit.assert_not_awaited()
+                db.rollback.assert_awaited_once()
+
+    async def test_edit_locks_old_and_new_routes_in_stable_grouped_order(self) -> None:
+        db = MagicMock(spec=AsyncSession)
+        shipment = make_shipment(
+            source_warehouse_id=8,
+            target_site_id=10,
+            pallet_ids=(2,),
+        )
+        page = ShipmentUpdateModel(
+            shipmentId=12,
+            sourceWarehouseId=7,
+            targetSiteId=9,
+            palletIds=[2],
+        )
+        actions: list[str] = []
+
+        async def warehouse_access(
+            _db: AsyncSession,
+            _user: CurrentUserModel,
+            warehouse_id: int,
+            **kwargs: Any,
+        ) -> None:
+            if kwargs.get('for_update'):
+                actions.append(f'warehouse-mapping-{warehouse_id}')
+
+        async def site_access(
+            _db: AsyncSession,
+            _user: CurrentUserModel,
+            site_id: int,
+            **kwargs: Any,
+        ) -> None:
+            if kwargs.get('for_update'):
+                actions.append(f'site-mapping-{site_id}')
+
+        async def get_warehouse(
+            _db: AsyncSession,
+            warehouse_id: int,
+            **kwargs: Any,
+        ) -> BizWarehouse:
+            if kwargs.get('for_update'):
+                actions.append(f'warehouse-master-{warehouse_id}')
+            return BizWarehouse(
+                warehouse_id=warehouse_id,
+                warehouse_code=f'W-{warehouse_id}',
+                warehouse_name=f'{warehouse_id}号仓',
+                status='0',
+            )
+
+        async def get_site(
+            _db: AsyncSession,
+            site_id: int,
+            **kwargs: Any,
+        ) -> BizCustomerSite:
+            if kwargs.get('for_update'):
+                actions.append(f'site-master-{site_id}')
+            return BizCustomerSite(
+                site_id=site_id,
+                customer_id=1,
+                site_code=f'S-{site_id}',
+                site_name=f'{site_id}号现场',
+                status='0',
+            )
+
+        async def lock_pallets(*_args: Any, **_kwargs: Any) -> tuple[list[Any], list[Any], list[Any]]:
+            actions.append('pallet-locks')
+            return [], [], []
+
+        async def update(*_args: Any, **_kwargs: Any) -> BizShipmentOrder:
+            actions.append('update')
+            return shipment
+
+        async def replace(*_args: Any, **_kwargs: Any) -> list[BizShipmentItem]:
+            actions.append('replace')
+            return []
+
+        async def commit() -> None:
+            actions.append('commit')
+
+        db.commit.side_effect = commit
+        with (
+            patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7, 8])),
+            patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9, 10])),
+            patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(return_value=shipment)),
+            patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock(side_effect=warehouse_access)),
+            patch.object(AccessService, 'assert_site_access', new=AsyncMock(side_effect=site_access)),
+            patch.object(MasterDao, 'get_warehouse', new=AsyncMock(side_effect=get_warehouse)),
+            patch.object(MasterDao, 'get_site', new=AsyncMock(side_effect=get_site)),
+            patch.object(ShipmentService, '_lock_and_validate_pallets', new=AsyncMock(side_effect=lock_pallets)),
+            patch.object(OperationDao, 'update_shipment', new=AsyncMock(side_effect=update)),
+            patch.object(OperationDao, 'replace_shipment_items', new=AsyncMock(side_effect=replace)),
+        ):
+            result = await ShipmentService.edit_shipment_services(db, page, self.user)
+
+        self.assertTrue(result.is_success)
+        self.assertEqual(
+            [
+                'pallet-locks',
+                'warehouse-master-7',
+                'warehouse-mapping-7',
+                'warehouse-master-8',
+                'warehouse-mapping-8',
+                'site-master-9',
+                'site-mapping-9',
+                'site-master-10',
+                'site-mapping-10',
+                'update',
+                'replace',
+                'commit',
+            ],
+            actions,
+        )
+
     async def test_batch_delete_preflights_all_drafts_before_any_delete(self) -> None:
         documents = {
             11: make_shipment(11, status='DRAFT'),
@@ -879,6 +1214,173 @@ class ShipmentDraftServiceTest(unittest.IsolatedAsyncioTestCase):
         self.db.commit.assert_not_awaited()
         self.db.rollback.assert_awaited_once()
 
+    async def test_batch_delete_rechecks_all_current_routes_before_first_delete(self) -> None:
+        documents = {
+            11: make_shipment(11, source_warehouse_id=8, target_site_id=10, status='DRAFT'),
+            12: make_shipment(12, source_warehouse_id=7, target_site_id=9, status='DRAFT'),
+        }
+        revoked_warehouse_id = documents[11].source_warehouse_id
+        parent_lock_order: list[int] = []
+
+        async def lock_order(
+            _db: AsyncSession,
+            shipment_id: int,
+            **_kwargs: Any,
+        ) -> BizShipmentOrder:
+            parent_lock_order.append(shipment_id)
+            return documents[shipment_id]
+
+        async def get_warehouse(
+            _db: AsyncSession,
+            warehouse_id: int,
+            **_kwargs: Any,
+        ) -> BizWarehouse:
+            return BizWarehouse(
+                warehouse_id=warehouse_id,
+                warehouse_code=f'W-{warehouse_id}',
+                warehouse_name=f'{warehouse_id}号仓',
+                status='0',
+            )
+
+        async def warehouse_access(
+            _db: AsyncSession,
+            _user: CurrentUserModel,
+            warehouse_id: int,
+            **kwargs: Any,
+        ) -> None:
+            if warehouse_id == revoked_warehouse_id and kwargs.get('for_update'):
+                raise PermissionException(data='', message='无权访问该仓库')
+
+        delete = AsyncMock()
+        warehouse_mapping = AsyncMock(side_effect=warehouse_access)
+        with (
+            patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7, 8])),
+            patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9, 10])),
+            patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(side_effect=lock_order)),
+            patch.object(MasterDao, 'get_warehouse', new=AsyncMock(side_effect=get_warehouse)),
+            patch.object(AccessService, 'assert_warehouse_access', new=warehouse_mapping),
+            patch.object(OperationDao, 'delete_shipment', new=delete),
+            self.assertRaises(PermissionException) as raised,
+        ):
+            await ShipmentService.delete_shipment_services(self.db, [12, 11], self.user)
+
+        self.assertEqual('无权访问该发货单', raised.exception.message)
+        self.assertEqual([11, 12], parent_lock_order)
+        self.assertEqual([7, 8], [call.args[2] for call in warehouse_mapping.await_args_list])
+        self.assertTrue(all(call.kwargs.get('for_update') for call in warehouse_mapping.await_args_list))
+        delete.assert_not_awaited()
+        self.db.commit.assert_not_awaited()
+        self.db.rollback.assert_awaited_once()
+
+    async def test_batch_delete_locks_deduplicated_routes_stably_without_requiring_active(self) -> None:
+        db = MagicMock(spec=AsyncSession)
+        documents = {
+            11: make_shipment(11, source_warehouse_id=8, target_site_id=10, status='DRAFT'),
+            12: make_shipment(12, source_warehouse_id=7, target_site_id=10, status='DRAFT'),
+            13: make_shipment(13, source_warehouse_id=8, target_site_id=9, status='DRAFT'),
+        }
+        actions: list[str] = []
+
+        async def lock_order(
+            _db: AsyncSession,
+            shipment_id: int,
+            **_kwargs: Any,
+        ) -> BizShipmentOrder:
+            actions.append(f'parent-{shipment_id}')
+            return documents[shipment_id]
+
+        async def get_warehouse(
+            _db: AsyncSession,
+            warehouse_id: int,
+            **kwargs: Any,
+        ) -> BizWarehouse:
+            self.assertTrue(kwargs.get('for_update'))
+            actions.append(f'warehouse-master-{warehouse_id}')
+            return BizWarehouse(
+                warehouse_id=warehouse_id,
+                warehouse_code=f'W-{warehouse_id}',
+                warehouse_name=f'{warehouse_id}号仓',
+                status='1',
+            )
+
+        async def warehouse_access(
+            _db: AsyncSession,
+            _user: CurrentUserModel,
+            warehouse_id: int,
+            **kwargs: Any,
+        ) -> None:
+            self.assertTrue(kwargs.get('for_update'))
+            actions.append(f'warehouse-mapping-{warehouse_id}')
+
+        async def get_site(
+            _db: AsyncSession,
+            site_id: int,
+            **kwargs: Any,
+        ) -> BizCustomerSite:
+            self.assertTrue(kwargs.get('for_update'))
+            actions.append(f'site-master-{site_id}')
+            return BizCustomerSite(
+                site_id=site_id,
+                customer_id=1,
+                site_code=f'S-{site_id}',
+                site_name=f'{site_id}号现场',
+                status='1',
+            )
+
+        async def site_access(
+            _db: AsyncSession,
+            _user: CurrentUserModel,
+            site_id: int,
+            **kwargs: Any,
+        ) -> None:
+            self.assertTrue(kwargs.get('for_update'))
+            actions.append(f'site-mapping-{site_id}')
+
+        async def delete(_db: AsyncSession, shipment_id: int) -> None:
+            actions.append(f'delete-{shipment_id}')
+
+        async def commit() -> None:
+            actions.append('commit')
+
+        db.commit.side_effect = commit
+        with (
+            patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7, 8])),
+            patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9, 10])),
+            patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(side_effect=lock_order)),
+            patch.object(MasterDao, 'get_warehouse', new=AsyncMock(side_effect=get_warehouse)),
+            patch.object(
+                AccessService,
+                'assert_warehouse_access',
+                new=AsyncMock(side_effect=warehouse_access),
+            ),
+            patch.object(MasterDao, 'get_site', new=AsyncMock(side_effect=get_site)),
+            patch.object(AccessService, 'assert_site_access', new=AsyncMock(side_effect=site_access)),
+            patch.object(OperationDao, 'delete_shipment', new=AsyncMock(side_effect=delete)),
+        ):
+            result = await ShipmentService.delete_shipment_services(db, [13, 11, 12], self.user)
+
+        self.assertTrue(result.is_success)
+        self.assertEqual(
+            [
+                'parent-11',
+                'parent-12',
+                'parent-13',
+                'warehouse-master-7',
+                'warehouse-mapping-7',
+                'warehouse-master-8',
+                'warehouse-mapping-8',
+                'site-master-9',
+                'site-mapping-9',
+                'site-master-10',
+                'site-mapping-10',
+                'delete-11',
+                'delete-12',
+                'delete-13',
+                'commit',
+            ],
+            actions,
+        )
+
 
 class ShipmentConfirmServiceTest(unittest.IsolatedAsyncioTestCase):
     def setUp(self) -> None:
@@ -1161,6 +1663,153 @@ class ShipmentConfirmServiceTest(unittest.IsolatedAsyncioTestCase):
         self.db.commit.assert_not_awaited()
         self.db.rollback.assert_awaited_once()
 
+    async def test_exact_confirm_retry_ignores_later_route_disable_but_rechecks_current_access(
+        self,
+    ) -> None:
+        for warehouse_status, site_status in (('1', '0'), ('0', '1')):
+            with self.subTest(warehouse_status=warehouse_status, site_status=site_status):
+                db = MagicMock(spec=AsyncSession)
+                confirmed = make_shipment(
+                    status='CONFIRMED',
+                    idempotency_key='confirm-key',
+                    request_hash=self.request_hash,
+                )
+                for item in confirmed.items:
+                    item.active_filling_id = 100 + item.pallet_id
+                    item.package_qty = 8
+                original_snapshots = [
+                    (item.pallet_id, item.active_filling_id, item.package_qty) for item in confirmed.items
+                ]
+                inventory_lock = AsyncMock()
+                update = AsyncMock()
+                add_event = AsyncMock()
+                warehouse_access = AsyncMock()
+                site_access = AsyncMock()
+
+                with (
+                    patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])),
+                    patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])),
+                    patch.object(OperationDao, 'get_shipment_by_id', new=AsyncMock(return_value=confirmed)),
+                    patch.object(
+                        OperationDao,
+                        'get_shipment_by_confirm_idempotency_key',
+                        new=AsyncMock(return_value=confirmed),
+                    ),
+                    patch.object(
+                        OperationDao,
+                        'get_shipment_for_update',
+                        new=AsyncMock(return_value=confirmed),
+                    ),
+                    patch.object(InventoryDao, 'get_inventories_for_update', new=inventory_lock),
+                    patch.object(
+                        MasterDao,
+                        'get_warehouse',
+                        new=AsyncMock(
+                            return_value=BizWarehouse(
+                                warehouse_id=7,
+                                warehouse_code='W-7',
+                                warehouse_name='七号仓',
+                                status=warehouse_status,
+                            )
+                        ),
+                    ) as warehouse_lock,
+                    patch.object(AccessService, 'assert_warehouse_access', new=warehouse_access),
+                    patch.object(
+                        MasterDao,
+                        'get_site',
+                        new=AsyncMock(
+                            return_value=BizCustomerSite(
+                                site_id=9,
+                                customer_id=1,
+                                site_code='S-9',
+                                site_name='九号现场',
+                                status=site_status,
+                            )
+                        ),
+                    ) as site_lock,
+                    patch.object(AccessService, 'assert_site_access', new=site_access),
+                    patch.object(OperationDao, 'update_shipment', new=update),
+                    patch.object(InventoryDao, 'add_event', new=add_event),
+                ):
+                    result = await ShipmentService.confirm_shipment_services(
+                        db,
+                        12,
+                        self.command,
+                        self.user,
+                    )
+
+                self.assertTrue(result.is_success)
+                warehouse_lock.assert_awaited_once_with(db, 7, for_update=True)
+                warehouse_access.assert_awaited_once_with(db, self.user, 7, for_update=True)
+                site_lock.assert_awaited_once_with(db, 9, for_update=True)
+                site_access.assert_awaited_once_with(db, self.user, 9, for_update=True)
+                self.assertEqual(
+                    original_snapshots,
+                    [(item.pallet_id, item.active_filling_id, item.package_qty) for item in confirmed.items],
+                )
+                inventory_lock.assert_not_awaited()
+                update.assert_not_awaited()
+                add_event.assert_not_awaited()
+                db.commit.assert_not_awaited()
+                db.rollback.assert_awaited_once()
+
+    async def test_exact_confirm_retry_hides_document_when_current_mapping_is_revoked(self) -> None:
+        confirmed = make_shipment(
+            status='CONFIRMED',
+            idempotency_key='confirm-key',
+            request_hash=self.request_hash,
+        )
+        inventory_lock = AsyncMock()
+        update = AsyncMock()
+        add_event = AsyncMock()
+        site_lock = AsyncMock()
+        warehouse_mapping = AsyncMock(side_effect=PermissionException(data='', message='无权访问该仓库'))
+        with (
+            patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])),
+            patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])),
+            patch.object(OperationDao, 'get_shipment_by_id', new=AsyncMock(return_value=confirmed)),
+            patch.object(
+                OperationDao,
+                'get_shipment_by_confirm_idempotency_key',
+                new=AsyncMock(return_value=confirmed),
+            ),
+            patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(return_value=confirmed)),
+            patch.object(InventoryDao, 'get_inventories_for_update', new=inventory_lock),
+            patch.object(
+                MasterDao,
+                'get_warehouse',
+                new=AsyncMock(
+                    return_value=BizWarehouse(
+                        warehouse_id=7,
+                        warehouse_code='W-7',
+                        warehouse_name='七号仓',
+                        status='1',
+                    )
+                ),
+            ) as warehouse_lock,
+            patch.object(AccessService, 'assert_warehouse_access', new=warehouse_mapping),
+            patch.object(MasterDao, 'get_site', new=site_lock),
+            patch.object(OperationDao, 'update_shipment', new=update),
+            patch.object(InventoryDao, 'add_event', new=add_event),
+            self.assertRaises(PermissionException) as raised,
+        ):
+            await ShipmentService.confirm_shipment_services(
+                self.db,
+                12,
+                self.command,
+                self.user,
+            )
+
+        self.assertEqual('无权访问该发货单', raised.exception.message)
+        warehouse_lock.assert_awaited_once_with(self.db, 7, for_update=True)
+        warehouse_mapping.assert_awaited_once_with(self.db, self.user, 7, for_update=True)
+        site_lock.assert_not_awaited()
+        inventory_lock.assert_not_awaited()
+        update.assert_not_awaited()
+        add_event.assert_not_awaited()
+        self.db.commit.assert_not_awaited()
+        self.db.rollback.assert_awaited_once()
+
     async def test_confirm_uses_locked_current_items_not_initial_rr_snapshot(self) -> None:
         initial = make_shipment(pallet_ids=(99,))
         locked = make_shipment(pallet_ids=(2,))
@@ -1265,7 +1914,11 @@ class ShipmentConfirmServiceTest(unittest.IsolatedAsyncioTestCase):
                 'get_shipment_by_confirm_idempotency_key',
                 new=AsyncMock(side_effect=[None, nonexact]),
             ),
-            patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(side_effect=integrity_error())),
+            patch.object(
+                OperationDao,
+                'get_shipment_for_update',
+                new=AsyncMock(side_effect=integrity_error()),
+            ),
             self.assertRaises(ServiceException) as raised,
         ):
             await ShipmentService.confirm_shipment_services(self.db, 12, self.command, self.user)
@@ -1274,6 +1927,80 @@ class ShipmentConfirmServiceTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(2, self.db.rollback.await_count)
         self.db.commit.assert_not_awaited()
 
+    async def test_integrity_recovery_locks_parent_and_rechecks_revoked_current_mapping(self) -> None:
+        initial = make_shipment(status='DRAFT')
+        recovered = make_shipment(
+            status='CONFIRMED',
+            idempotency_key='confirm-key',
+            request_hash=self.request_hash,
+        )
+        parent_lock_count = 0
+
+        async def lock_parent(*_args: Any, **_kwargs: Any) -> BizShipmentOrder:
+            nonlocal parent_lock_count
+            parent_lock_count += 1
+            if parent_lock_count == 1:
+                raise integrity_error()
+            return recovered
+
+        async def warehouse_access(*_args: Any, **kwargs: Any) -> None:
+            if kwargs.get('for_update'):
+                raise PermissionException(data='', message='无权访问该仓库')
+
+        parent_lock = AsyncMock(side_effect=lock_parent)
+        warehouse_mapping = AsyncMock(side_effect=warehouse_access)
+        site_lock = AsyncMock(
+            return_value=BizCustomerSite(
+                site_id=9,
+                customer_id=1,
+                site_code='S-9',
+                site_name='九号现场',
+                status='0',
+            )
+        )
+        with (
+            patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(side_effect=[[7], [7]])),
+            patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(side_effect=[[9], [9]])),
+            patch.object(OperationDao, 'get_shipment_by_id', new=AsyncMock(return_value=initial)),
+            patch.object(
+                OperationDao,
+                'get_shipment_by_confirm_idempotency_key',
+                new=AsyncMock(side_effect=[None, recovered]),
+            ),
+            patch.object(OperationDao, 'get_shipment_for_update', new=parent_lock),
+            patch.object(
+                MasterDao,
+                'get_warehouse',
+                new=AsyncMock(
+                    return_value=BizWarehouse(
+                        warehouse_id=7,
+                        warehouse_code='W-7',
+                        warehouse_name='七号仓',
+                        status='0',
+                    )
+                ),
+            ) as warehouse_lock,
+            patch.object(AccessService, 'assert_warehouse_access', new=warehouse_mapping),
+            patch.object(MasterDao, 'get_site', new=site_lock),
+            patch.object(AccessService, 'assert_site_access', new=AsyncMock()),
+            self.assertRaises(PermissionException) as raised,
+        ):
+            await ShipmentService.confirm_shipment_services(self.db, 12, self.command, self.user)
+
+        self.assertEqual('无权访问该发货单', raised.exception.message)
+        self.assertEqual(2, parent_lock.await_count)
+        parent_lock.assert_awaited_with(
+            self.db,
+            12,
+            warehouse_ids=[7],
+            site_ids=[9],
+        )
+        warehouse_lock.assert_awaited_once_with(self.db, 7, for_update=True)
+        warehouse_mapping.assert_awaited_once_with(self.db, self.user, 7, for_update=True)
+        site_lock.assert_not_awaited()
+        self.db.commit.assert_not_awaited()
+        self.assertEqual(2, self.db.rollback.await_count)
+
     async def test_disabled_source_after_pallet_locks_rolls_back_before_mutation(self) -> None:
         shipment = make_shipment(pallet_ids=(2,))
         inventory = make_inventory(2, version=4)
@@ -1338,7 +2065,11 @@ class ShipmentConfirmServiceTest(unittest.IsolatedAsyncioTestCase):
             patch.object(AccessDao, 'list_user_site_ids', new=site_scope),
             patch.object(OperationDao, 'get_shipment_by_id', new=AsyncMock(return_value=initial)),
             patch.object(OperationDao, 'get_shipment_by_confirm_idempotency_key', new=key_read),
-            patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(side_effect=integrity_error())),
+            patch.object(
+                OperationDao,
+                'get_shipment_for_update',
+                new=AsyncMock(side_effect=[integrity_error(), recovered]),
+            ),
             patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()),
             patch.object(AccessService, 'assert_site_access', new=AsyncMock()),
             patch.object(