修复：完善退回并发幂等与信息隐藏

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

@@ -377,7 +377,7 @@ class ReturnService:
             cls._raise_pallet_not_accessible(current_user)
         matching_items = [item for item in return_order.items if item.pallet_id == pallet.pallet_id]
         if len(matching_items) != 1:
-            raise ServiceException(message='扫码托盘不属于该退回单')
+            cls._raise_pallet_not_accessible(current_user, '扫码托盘不属于该退回单')
         inventory = await MasterDao.get_pallet_inventory(query_db, pallet.pallet_id)
         if inventory is None:
             cls._raise_pallet_not_accessible(current_user, '托盘库存不存在')
@@ -445,21 +445,37 @@ class ReturnService:
         query_db: AsyncSession,
         pallet_ids: list[int],
         source_site_id: int,
-    ) -> tuple[list[BizPalletInventory], list[BizPallet], dict[int, BizFillingRecord]]:
+        current_user: CurrentUserModel,
+    ) -> tuple[
+        list[BizPalletInventory],
+        list[BizPallet],
+        dict[int, BizFillingRecord | None],
+        list[int] | None,
+    ]:
         ordered_ids = cls._validate_pallet_ids(pallet_ids)
         inventories = await InventoryDao.get_inventories_for_update(query_db, ordered_ids)
-        pallets = await InventoryDao.get_pallets_by_ids(query_db, ordered_ids)
         inventory_by_id = {inventory.pallet_id: inventory for inventory in inventories}
-        pallet_by_id = {pallet.pallet_id: pallet for pallet in pallets}
-        if set(inventory_by_id) != set(ordered_ids) or set(pallet_by_id) != set(ordered_ids):
-            raise ServiceException(message='存在不存在或库存缺失的托盘')
+        if set(inventory_by_id) != set(ordered_ids):
+            cls._raise_pallet_not_accessible(current_user, '存在不存在或库存缺失的托盘')
 
         ordered_inventories = [inventory_by_id[pallet_id] for pallet_id in ordered_ids]
-        ordered_pallets = [pallet_by_id[pallet_id] for pallet_id in ordered_ids]
-        for pallet, inventory in zip(ordered_pallets, ordered_inventories, strict=True):
-            cls._validate_request_inventory(pallet, inventory, source_site_id)
+        if any(inventory.site_id != source_site_id for inventory in ordered_inventories):
+            cls._raise_pallet_not_accessible(current_user, '同一退回单托盘必须属于同一现场')
+        try:
+            await AccessService.assert_site_access(query_db, current_user, source_site_id)
+        except PermissionException:
+            cls._raise_pallet_not_accessible(current_user)
+        fresh_site_ids = await cls._site_scope(query_db, current_user)
+        if fresh_site_ids is not None and source_site_id not in fresh_site_ids:
+            cls._raise_pallet_not_accessible(current_user)
 
-        filling_by_id = {}
+        pallets = await InventoryDao.get_pallets_by_ids(query_db, ordered_ids)
+        pallet_by_id = {pallet.pallet_id: pallet for pallet in pallets}
+        if set(pallet_by_id) != set(ordered_ids):
+            cls._raise_pallet_not_accessible(current_user, '存在不存在或库存缺失的托盘')
+        ordered_pallets = [pallet_by_id[pallet_id] for pallet_id in ordered_ids]
+
+        filling_by_id: dict[int, BizFillingRecord | None] = {}
         filling_ids = sorted(
             {
                 inventory.active_filling_id
@@ -468,14 +484,31 @@ class ReturnService:
             }
         )
         for filling_id in filling_ids:
-            filling = await OperationDao.get_active_filling_for_update(query_db, filling_id)
-            if filling is None:
-                raise ServiceException(message='托盘活动装填周期不存在或已关闭')
-            filling_by_id[filling_id] = filling
-        for inventory in ordered_inventories:
+            filling_by_id[filling_id] = await OperationDao.get_active_filling_for_update(query_db, filling_id)
+        return ordered_inventories, ordered_pallets, filling_by_id, fresh_site_ids
+
+    @classmethod
+    def _validate_locked_request_pallets(
+        cls,
+        inventories: list[BizPalletInventory],
+        pallets: list[BizPallet],
+        filling_by_id: dict[int, BizFillingRecord | None],
+        source_site_id: int,
+        current_user: CurrentUserModel,
+    ) -> None:
+        for pallet, inventory in zip(pallets, inventories, strict=True):
+            cls._validate_request_inventory(pallet, inventory, source_site_id)
+        for inventory in inventories:
             if inventory.active_filling_id is not None:
-                cls._validate_active_filling(filling_by_id[inventory.active_filling_id], inventory)
-        return ordered_inventories, ordered_pallets, filling_by_id
+                filling = filling_by_id[inventory.active_filling_id]
+                if filling is None:
+                    cls._raise_pallet_not_accessible(current_user, '托盘活动装填周期不存在或已关闭')
+                try:
+                    cls._validate_active_filling(filling, inventory)
+                except ServiceException:
+                    if not cls._is_admin(current_user):
+                        cls._raise_pallet_not_accessible(current_user)
+                    raise
 
     @staticmethod
     def _build_return_items(
@@ -585,7 +618,7 @@ class ReturnService:
             await query_db.rollback()
 
     @classmethod
-    async def request_return_services(
+    async def request_return_services(  # noqa: PLR0915
         cls,
         query_db: AsyncSession,
         page_object: ReturnRequestModel,
@@ -595,30 +628,34 @@ class ReturnService:
             user_id, user_name = cls._current_identity(current_user)
             ordered_ids = cls._validate_pallet_ids(page_object.pallet_ids)
             request_hash = cls._build_request_hash(page_object)
-            site_ids = await cls._site_scope(query_db, current_user)
-            if site_ids is not None and page_object.source_site_id not in site_ids:
-                raise PermissionException(data='', message='无权访问该客户现场')
+            preflight_site_ids = await cls._site_scope(query_db, current_user)
+            if preflight_site_ids is not None and page_object.source_site_id not in preflight_site_ids:
+                cls._raise_pallet_not_accessible(current_user)
+            try:
                 await AccessService.assert_site_access(
                     query_db,
                     current_user,
                     page_object.source_site_id,
                 )
-            request_warehouse_ids = None if site_ids is None else []
+            except PermissionException:
+                cls._raise_pallet_not_accessible(current_user)
+            request_warehouse_ids = None if preflight_site_ids is None else []
             existing = await OperationDao.get_return_by_request_idempotency_key(
                 query_db,
                 page_object.idempotency_key,
-                site_ids=site_ids,
+                site_ids=preflight_site_ids,
                 warehouse_ids=request_warehouse_ids,
             )
             if existing is not None and not cls._is_exact_request_result(existing, page_object, request_hash):
                 raise ServiceException(message='申请退回失败，幂等键已被其他请求使用')
+            preflight_existing_id = int(existing.return_id) if existing is not None else None
 
             await query_db.rollback()
-            if existing is not None:
+            if preflight_existing_id is not None:
                 locked_existing = await OperationDao.get_return_for_update(
                     query_db,
-                    existing.return_id,
-                    site_ids=site_ids,
+                    preflight_existing_id,
+                    site_ids=preflight_site_ids,
                     warehouse_ids=request_warehouse_ids,
                 )
                 if locked_existing is None:
@@ -635,12 +672,35 @@ class ReturnService:
                 await query_db.rollback()
                 return cls._success('申请成功', return_id)
 
-            inventories, _, _ = await cls._lock_request_pallets(
+            inventories, pallets, filling_by_id, fresh_site_ids = await cls._lock_request_pallets(
                 query_db,
                 ordered_ids,
                 page_object.source_site_id,
+                current_user,
             )
+            fresh_warehouse_ids = None if fresh_site_ids is None else []
+            fresh_existing = await OperationDao.get_return_by_request_idempotency_key(
+                query_db,
+                page_object.idempotency_key,
+                site_ids=fresh_site_ids,
+                warehouse_ids=fresh_warehouse_ids,
+            )
+            if fresh_existing is not None:
+                if not cls._is_exact_request_result(fresh_existing, page_object, request_hash):
+                    raise ServiceException(message='申请退回失败，幂等键已被其他请求使用')
+                fresh_existing_id = int(fresh_existing.return_id)
+                await cls._lock_and_reauthorize_request_site(query_db, fresh_existing, current_user)
+                await query_db.rollback()
+                return cls._success('申请成功', fresh_existing_id)
+
             await cls._lock_new_request_route(query_db, page_object, current_user)
+            cls._validate_locked_request_pallets(
+                inventories,
+                pallets,
+                filling_by_id,
+                page_object.source_site_id,
+                current_user,
+            )
             for inventory in inventories:
                 PalletStateMachine.ensure_transition(
                     PalletFlowState(inventory.flow_state),

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

@@ -278,6 +278,26 @@ class ReturnDaoScopeRedTest(unittest.IsolatedAsyncioTestCase):
         self.assertIn('biz_return_order.target_warehouse_id IN (7)', sql)
         self.assertIn(' OR ', sql)
 
+    async def test_request_key_read_is_scoped_and_never_locks_missing_unique_key(self) -> None:
+        result = MagicMock()
+        result.scalar_one_or_none.return_value = None
+        self.db.execute.return_value = result
+
+        await OperationDao.get_return_by_request_idempotency_key(
+            self.db,
+            'request-key',
+            site_ids=[9],
+            warehouse_ids=[],
+        )
+
+        statement = self.db.execute.await_args.args[0]
+        self.assertTrue(statement.get_final_froms()[0].compare(BizReturnOrder.__table__))
+        for dialect in (mysql.dialect(), postgresql.dialect()):
+            sql = compile_sql(statement, dialect)
+            self.assertIn("biz_return_order.request_idempotency_key = 'request-key'", sql)
+            self.assertIn('biz_return_order.source_site_id IN (9)', sql)
+            self.assertNotIn('FOR UPDATE', sql)
+
 
 @unittest.skipUnless(ReturnService is not None, 'ReturnService 尚未实现')
 class ReturnServiceContractRedTest(unittest.TestCase):
@@ -561,6 +581,51 @@ class ReturnReadAndScanRedTest(unittest.IsolatedAsyncioTestCase):
             await ReturnService.scan_return_pallet_services(self.db, 'qr-2', query, self.user)
         self.assertEqual('无权访问该退回单', hidden_error.exception.message)
 
+    async def test_confirm_scan_hides_missing_and_foreign_pallet_equally_for_normal_user(self) -> None:
+        order = make_return_order(snapshots=((2, 102, 8),))
+        query = ReturnScanQueryModel(mode='confirm', returnId=13)
+        messages = []
+        for pallet in (None, make_pallet(5)):
+            with self.subTest(pallet_exists=pallet is not None):
+                inventory_read = AsyncMock()
+                with (
+                    patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])),
+                    patch.object(OperationDao, 'get_return_by_id', new=AsyncMock(return_value=order)),
+                    patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=pallet)),
+                    patch.object(MasterDao, 'get_pallet_inventory', new=inventory_read),
+                ):
+                    try:
+                        await ReturnService.scan_return_pallet_services(self.db, 'qr-5', query, self.user)
+                    except Exception as exc:
+                        raised = exc
+                    else:
+                        raised = None
+                self.assertIsInstance(raised, PermissionException)
+                messages.append(getattr(raised, 'message', '未抛出异常'))
+                inventory_read.assert_not_awaited()
+
+        self.assertEqual(['无权访问该托盘', '无权访问该托盘'], messages)
+
+    async def test_admin_confirm_scan_keeps_precise_missing_and_foreign_pallet_errors(self) -> None:
+        admin = make_user(1)
+        order = make_return_order(snapshots=((2, 102, 8),))
+        query = ReturnScanQueryModel(mode='confirm', returnId=13)
+        cases = (
+            (None, '托盘不存在'),
+            (make_pallet(5), '扫码托盘不属于该退回单'),
+        )
+        for pallet, expected_message in cases:
+            with self.subTest(expected_message=expected_message):
+                with (
+                    patch.object(OperationDao, 'get_return_by_id', new=AsyncMock(return_value=order)),
+                    patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=pallet)),
+                    patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock()) as inventory_read,
+                    self.assertRaises(ServiceException) as raised,
+                ):
+                    await ReturnService.scan_return_pallet_services(self.db, 'qr-5', query, admin)
+                self.assertEqual(expected_message, raised.exception.message)
+                inventory_read.assert_not_awaited()
+
 
 @unittest.skipUnless(ReturnService is not None, 'ReturnService 尚未实现')
 class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase):
@@ -575,6 +640,337 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase):
             idempotencyKey='request-key',
         )
 
+    async def test_concurrent_same_request_returns_winner_after_inventory_wait(self) -> None:  # noqa: PLR0915
+        request_hash = ReturnService._build_request_hash(self.page)
+        winner = make_return_order(request_hash=request_hash)
+        inventories = [
+            make_inventory(2, state=PalletFlowState.RETURNING.value),
+            make_inventory(5, remaining=0, state=PalletFlowState.RETURNING.value),
+        ]
+        actions: list[str] = []
+        key_count = 0
+        scope_count = 0
+        access_count = 0
+        rollback_count = 0
+
+        async def read_key(*_args: Any, **_kwargs: Any) -> BizReturnOrder | None:
+            nonlocal key_count
+            key_count += 1
+            actions.append('key-preflight' if key_count == 1 else 'key-fresh')
+            return None if key_count == 1 else winner
+
+        async def read_scope(*_args: Any) -> list[int]:
+            nonlocal scope_count
+            scope_count += 1
+            actions.append('scope-preflight' if scope_count == 1 else 'scope-fresh')
+            return [9]
+
+        async def check_access(*_args: Any, **kwargs: Any) -> None:
+            nonlocal access_count
+            access_count += 1
+            if kwargs.get('for_update', False):
+                actions.append('mapping-lock')
+            else:
+                actions.append('access-preflight' if access_count == 1 else 'access-fresh')
+
+        async def rollback() -> None:
+            nonlocal rollback_count
+            rollback_count += 1
+            actions.append('rollback-preflight' if rollback_count == 1 else 'rollback-result')
+
+        async def lock_inventories(*_args: Any) -> list[BizPalletInventory]:
+            actions.append('inventory-lock')
+            return inventories
+
+        async def lock_pallets(*_args: Any) -> list[BizPallet]:
+            actions.append('pallet-lock')
+            return [make_pallet(2), make_pallet(5)]
+
+        async def lock_filling(*_args: Any) -> BizFillingRecord:
+            actions.append('filling-lock')
+            return make_filling(2)
+
+        async def lock_site(*_args: Any, **_kwargs: Any) -> SimpleNamespace:
+            actions.append('site-lock')
+            return SimpleNamespace(status='0')
+
+        key_read = AsyncMock(side_effect=read_key)
+        scope_read = AsyncMock(side_effect=read_scope)
+        self.db.rollback.side_effect = rollback
+        with (
+            patch.object(AccessDao, 'list_user_site_ids', new=scope_read),
+            patch.object(OperationDao, 'get_return_by_request_idempotency_key', new=key_read),
+            patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(side_effect=lock_inventories)),
+            patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(side_effect=lock_pallets)),
+            patch.object(
+                OperationDao,
+                'get_active_filling_for_update',
+                new=AsyncMock(side_effect=lock_filling),
+            ),
+            patch.object(AccessService, 'assert_site_access', new=AsyncMock(side_effect=check_access)),
+            patch.object(MasterDao, 'get_site', new=AsyncMock(side_effect=lock_site)),
+            patch.object(OperationDao, 'add_return_order', new=AsyncMock()) as add_order,
+            patch.object(OperationDao, 'add_return_items', new=AsyncMock()) as add_items,
+            patch.object(InventoryDao, 'add_event', new=AsyncMock()) as add_event,
+        ):
+            try:
+                result = await ReturnService.request_return_services(self.db, self.page, self.user)
+            except ServiceException as exc:
+                self.fail(f'锁等待后的精确重试被库存状态提前拒绝：{exc.message}')
+
+        self.assertEqual(13, result.result)
+        self.assertEqual(
+            [
+                'scope-preflight',
+                'access-preflight',
+                'key-preflight',
+                'rollback-preflight',
+                'inventory-lock',
+                'access-fresh',
+                'scope-fresh',
+                'pallet-lock',
+                'filling-lock',
+                'key-fresh',
+                'site-lock',
+                'mapping-lock',
+                'rollback-result',
+            ],
+            actions,
+        )
+        self.assertEqual(2, key_read.await_count)
+        self.assertEqual(2, scope_read.await_count)
+        add_order.assert_not_awaited()
+        add_items.assert_not_awaited()
+        add_event.assert_not_awaited()
+        self.db.commit.assert_not_awaited()
+        self.assertEqual([PalletFlowState.RETURNING.value] * 2, [item.flow_state for item in inventories])
+
+    async def test_concurrent_same_key_different_payload_is_stable_idempotency_conflict(self) -> None:
+        winner = make_return_order(request_hash=ReturnService._build_request_hash(self.page))
+        conflict_page = self.page.model_copy(update={'target_warehouse_id': 8})
+        inventories = [
+            make_inventory(2, state=PalletFlowState.RETURNING.value),
+            make_inventory(5, remaining=0, state=PalletFlowState.RETURNING.value),
+        ]
+        key_read = AsyncMock(side_effect=[None, winner])
+        with (
+            patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])),
+            patch.object(OperationDao, 'get_return_by_request_idempotency_key', new=key_read),
+            patch.object(AccessService, 'assert_site_access', new=AsyncMock()),
+            patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(return_value=inventories)),
+            patch.object(
+                InventoryDao,
+                'get_pallets_by_ids',
+                new=AsyncMock(return_value=[make_pallet(2), make_pallet(5)]),
+            ),
+            patch.object(
+                OperationDao,
+                'get_active_filling_for_update',
+                new=AsyncMock(return_value=make_filling(2)),
+            ),
+            patch.object(OperationDao, 'add_return_order', new=AsyncMock()) as add_order,
+            self.assertRaises(ServiceException) as raised,
+        ):
+            await ReturnService.request_return_services(self.db, conflict_page, self.user)
+
+        self.assertEqual('申请退回失败，幂等键已被其他请求使用', raised.exception.message)
+        self.assertEqual(2, key_read.await_count)
+        add_order.assert_not_awaited()
+        self.db.commit.assert_not_awaited()
+        self.assertEqual(2, self.db.rollback.await_count)
+
+    async def test_concurrent_winner_uses_fresh_expanded_site_scope(self) -> None:
+        winner = make_return_order(request_hash=ReturnService._build_request_hash(self.page))
+        inventories = [
+            make_inventory(2, state=PalletFlowState.RETURNING.value),
+            make_inventory(5, remaining=0, state=PalletFlowState.RETURNING.value),
+        ]
+        scope_read = AsyncMock(side_effect=[[9], [9, 10]])
+        key_read = AsyncMock(side_effect=[None, winner])
+        with (
+            patch.object(AccessDao, 'list_user_site_ids', new=scope_read),
+            patch.object(OperationDao, 'get_return_by_request_idempotency_key', new=key_read),
+            patch.object(AccessService, 'assert_site_access', new=AsyncMock()),
+            patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(return_value=inventories)),
+            patch.object(
+                InventoryDao,
+                'get_pallets_by_ids',
+                new=AsyncMock(return_value=[make_pallet(2), make_pallet(5)]),
+            ),
+            patch.object(
+                OperationDao,
+                'get_active_filling_for_update',
+                new=AsyncMock(return_value=make_filling(2)),
+            ),
+            patch.object(MasterDao, 'get_site', new=AsyncMock(return_value=SimpleNamespace(status='0'))),
+        ):
+            result = await ReturnService.request_return_services(self.db, self.page, self.user)
+
+        self.assertEqual(13, result.result)
+        self.assertEqual(
+            [
+                {'site_ids': [9], 'warehouse_ids': []},
+                {'site_ids': [9, 10], 'warehouse_ids': []},
+            ],
+            [call.kwargs for call in key_read.await_args_list],
+        )
+        self.db.commit.assert_not_awaited()
+        self.assertEqual(2, self.db.rollback.await_count)
+
+    async def test_request_hides_site_revoked_during_inventory_wait(self) -> None:
+        inventories = [
+            make_inventory(2),
+            make_inventory(5, remaining=0, state=PalletFlowState.AT_SITE_EMPTY.value),
+        ]
+        scope_read = AsyncMock(side_effect=[[9], []])
+        key_read = AsyncMock(return_value=None)
+        inventory_locks = AsyncMock(return_value=inventories)
+        pallet_locks = AsyncMock(return_value=[make_pallet(2), make_pallet(5)])
+        with (
+            patch.object(AccessDao, 'list_user_site_ids', new=scope_read),
+            patch.object(OperationDao, 'get_return_by_request_idempotency_key', new=key_read),
+            patch.object(AccessService, 'assert_site_access', new=AsyncMock()),
+            patch.object(InventoryDao, 'get_inventories_for_update', new=inventory_locks),
+            patch.object(InventoryDao, 'get_pallets_by_ids', new=pallet_locks),
+            self.assertRaises(PermissionException) as raised,
+        ):
+            await ReturnService.request_return_services(self.db, self.page, self.user)
+
+        self.assertEqual('无权访问该托盘', raised.exception.message)
+        inventory_locks.assert_awaited_once_with(self.db, [2, 5])
+        pallet_locks.assert_not_awaited()
+        self.assertEqual(2, scope_read.await_count)
+        self.assertEqual(1, key_read.await_count)
+        self.db.commit.assert_not_awaited()
+        self.assertEqual(2, self.db.rollback.await_count)
+
+    async def test_preflight_existing_uses_only_copied_id_after_rollback(self) -> None:
+        request_hash = ReturnService._build_request_hash(self.page)
+
+        class ExpiringReturnProbe:
+            def __init__(self) -> None:
+                self.expired = False
+                self.return_id_reads = 0
+                self.request_idempotency_key = 'request-key'
+                self.request_hash = request_hash
+                self.source_site_id = 9
+                self.target_warehouse_id = 7
+                self.status = ReturnStatus.REQUESTED.value
+
+            @property
+            def return_id(self) -> int:
+                if self.expired:
+                    raise AssertionError('rollback 后禁止访问预检 return ORM')
+                self.return_id_reads += 1
+                return 13
+
+        probe = ExpiringReturnProbe()
+        fresh = make_return_order(request_hash=request_hash)
+
+        async def rollback() -> None:
+            probe.expired = True
+
+        self.db.rollback.side_effect = rollback
+        order_lock = AsyncMock(return_value=fresh)
+        with (
+            patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])),
+            patch.object(AccessService, 'assert_site_access', new=AsyncMock()),
+            patch.object(
+                OperationDao,
+                'get_return_by_request_idempotency_key',
+                new=AsyncMock(return_value=probe),
+            ),
+            patch.object(OperationDao, 'get_return_for_update', new=order_lock),
+            patch.object(MasterDao, 'get_site', new=AsyncMock(return_value=SimpleNamespace(status='0'))),
+            patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock()) as inventory_locks,
+        ):
+            result = await ReturnService.request_return_services(self.db, self.page, self.user)
+
+        self.assertEqual(13, result.result)
+        self.assertTrue(probe.expired)
+        self.assertEqual(1, probe.return_id_reads)
+        order_lock.assert_awaited_once_with(self.db, 13, site_ids=[9], warehouse_ids=[])
+        inventory_locks.assert_not_awaited()
+        self.assertEqual(2, self.db.rollback.await_count)
+
+    async def test_request_hides_missing_and_foreign_site_pallets_equally_for_normal_user(self) -> None:
+        cases = (
+            ('missing', []),
+            ('foreign-site', [make_inventory(2, site_id=10), make_inventory(5, site_id=10)]),
+        )
+        messages = []
+        for name, inventories in cases:
+            with self.subTest(case=name):
+                db = MagicMock(spec=AsyncSession)
+                pallet_locks = AsyncMock(return_value=[make_pallet(2), make_pallet(5)])
+                with (
+                    patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])),
+                    patch.object(
+                        OperationDao,
+                        'get_return_by_request_idempotency_key',
+                        new=AsyncMock(return_value=None),
+                    ),
+                    patch.object(AccessService, 'assert_site_access', new=AsyncMock()),
+                    patch.object(
+                        InventoryDao,
+                        'get_inventories_for_update',
+                        new=AsyncMock(return_value=inventories),
+                    ),
+                    patch.object(InventoryDao, 'get_pallets_by_ids', new=pallet_locks),
+                ):
+                    try:
+                        await ReturnService.request_return_services(db, self.page, self.user)
+                    except Exception as exc:
+                        raised = exc
+                    else:
+                        raised = None
+                self.assertIsInstance(raised, PermissionException)
+                messages.append(getattr(raised, 'message', '未抛出异常'))
+                pallet_locks.assert_not_awaited()
+                db.commit.assert_not_awaited()
+                db.rollback.assert_awaited()
+
+        self.assertEqual(['无权访问该托盘', '无权访问该托盘'], messages)
+
+    async def test_request_hides_invalid_active_filling_association_for_normal_user(self) -> None:
+        inventories = [
+            make_inventory(2),
+            make_inventory(5, remaining=0, state=PalletFlowState.AT_SITE_EMPTY.value),
+        ]
+        with (
+            patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])),
+            patch.object(
+                OperationDao,
+                'get_return_by_request_idempotency_key',
+                new=AsyncMock(return_value=None),
+            ),
+            patch.object(AccessService, 'assert_site_access', new=AsyncMock()),
+            patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(return_value=inventories)),
+            patch.object(
+                InventoryDao,
+                'get_pallets_by_ids',
+                new=AsyncMock(return_value=[make_pallet(2), make_pallet(5)]),
+            ),
+            patch.object(
+                OperationDao,
+                'get_active_filling_for_update',
+                new=AsyncMock(return_value=make_filling(5)),
+            ),
+            patch.object(MasterDao, 'get_site', new=AsyncMock(return_value=SimpleNamespace(status='0'))),
+            patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=SimpleNamespace(status='0'))),
+        ):
+            try:
+                await ReturnService.request_return_services(self.db, self.page, self.user)
+            except Exception as exc:
+                raised = exc
+            else:
+                raised = None
+
+        self.assertIsInstance(raised, PermissionException)
+        self.assertEqual('无权访问该托盘', getattr(raised, 'message', None))
+        self.db.commit.assert_not_awaited()
+        self.db.rollback.assert_awaited()
+
     async def test_request_freezes_all_pallets_saves_snapshots_and_appends_events(self) -> None:
         inventories = [
             make_inventory(2),
@@ -603,12 +999,16 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase):
             return event
 
         with (
-            patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9, 9])),
+            patch.object(
+                AccessDao,
+                'list_user_site_ids',
+                new=AsyncMock(return_value=[9, 9]),
+            ) as scope_read,
             patch.object(
                 OperationDao,
                 'get_return_by_request_idempotency_key',
                 new=AsyncMock(return_value=None),
-            ),
+            ) as key_read,
             patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(return_value=inventories)) as locks,
             patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=pallets)),
             patch.object(
@@ -657,6 +1057,12 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase):
 
         self.assertTrue(result.is_success)
         self.assertEqual(13, result.result)
+        self.assertEqual(2, scope_read.await_count)
+        self.assertEqual(2, key_read.await_count)
+        self.assertEqual(
+            [{'site_ids': [9], 'warehouse_ids': []}, {'site_ids': [9], 'warehouse_ids': []}],
+            [call.kwargs for call in key_read.await_args_list],
+        )
         locks.assert_awaited_once_with(self.db, [2, 5])
         site_access.assert_awaited_with(self.db, self.user, 9, for_update=True)
         warehouse_access.assert_not_awaited()
@@ -676,7 +1082,7 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual([4, 4], [event.after_line_id for event in events])
         self.db.commit.assert_awaited_once()
 
-    async def test_request_rejects_mixed_site_and_rolls_back_whole_order(self) -> None:
+    async def test_request_hides_mixed_site_and_rolls_back_whole_order(self) -> None:
         inventories = [make_inventory(2), make_inventory(5, site_id=10)]
         with (
             patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])),
@@ -696,15 +1102,45 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase):
                 new=AsyncMock(side_effect=[make_filling(2), make_filling(5)]),
             ),
             patch.object(OperationDao, 'add_return_order', new=AsyncMock()) as add_order,
-            self.assertRaises(ServiceException) as raised,
+            self.assertRaises(PermissionException) as raised,
         ):
             await ReturnService.request_return_services(self.db, self.page, self.user)
 
-        self.assertIn('同一现场', raised.exception.message)
+        self.assertEqual('无权访问该托盘', raised.exception.message)
         add_order.assert_not_awaited()
         self.db.commit.assert_not_awaited()
         self.db.rollback.assert_awaited()
 
+    async def test_admin_request_keeps_precise_missing_and_foreign_site_errors(self) -> None:
+        admin = make_user(1)
+        cases = (
+            ([], '存在不存在或库存缺失的托盘'),
+            ([make_inventory(2, site_id=10), make_inventory(5, site_id=10)], '同一退回单托盘必须属于同一现场'),
+        )
+        for inventories, expected_message in cases:
+            with self.subTest(expected_message=expected_message):
+                db = MagicMock(spec=AsyncSession)
+                with (
+                    patch.object(
+                        OperationDao,
+                        'get_return_by_request_idempotency_key',
+                        new=AsyncMock(return_value=None),
+                    ),
+                    patch.object(AccessService, 'assert_site_access', new=AsyncMock()),
+                    patch.object(
+                        InventoryDao,
+                        'get_inventories_for_update',
+                        new=AsyncMock(return_value=inventories),
+                    ),
+                    patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock()) as pallet_locks,
+                    self.assertRaises(ServiceException) as raised,
+                ):
+                    await ReturnService.request_return_services(db, self.page, admin)
+                self.assertEqual(expected_message, raised.exception.message)
+                pallet_locks.assert_not_awaited()
+                db.commit.assert_not_awaited()
+                db.rollback.assert_awaited()
+
     async def test_request_same_key_same_hash_returns_original_and_different_hash_conflicts(self) -> None:
         request_hash = ReturnService._build_request_hash(self.page)
         exact = make_return_order(request_hash=request_hash)
@@ -743,12 +1179,13 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase):
         conflict_db.rollback.assert_awaited()
 
     async def test_request_checks_site_mapping_before_locking_pallets(self) -> None:
+        key_read = AsyncMock(return_value=None)
         with (
             patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])),
             patch.object(
                 OperationDao,
                 'get_return_by_request_idempotency_key',
-                new=AsyncMock(return_value=None),
+                new=key_read,
             ),
             patch.object(
                 AccessService,
@@ -770,10 +1207,30 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase):
                 raised = None
 
         self.assertIsInstance(raised, PermissionException)
-        self.assertEqual('无权访问该客户现场', raised.message)
+        self.assertEqual('无权访问该托盘', raised.message)
+        key_read.assert_not_awaited()
         pallet_locks.assert_not_awaited()
         self.db.rollback.assert_awaited()
 
+    async def test_request_rejects_unscoped_requested_site_before_inventory_locks(self) -> None:
+        key_read = AsyncMock(return_value=None)
+        access = AsyncMock()
+        inventory_locks = AsyncMock()
+        with (
+            patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[])),
+            patch.object(OperationDao, 'get_return_by_request_idempotency_key', new=key_read),
+            patch.object(AccessService, 'assert_site_access', new=access),
+            patch.object(InventoryDao, 'get_inventories_for_update', new=inventory_locks),
+            self.assertRaises(PermissionException) as raised,
+        ):
+            await ReturnService.request_return_services(self.db, self.page, self.user)
+
+        self.assertEqual('无权访问该托盘', raised.exception.message)
+        key_read.assert_not_awaited()
+        access.assert_not_awaited()
+        inventory_locks.assert_not_awaited()
+        self.db.rollback.assert_awaited_once()
+
     async def test_admin_request_idempotency_scope_is_unrestricted(self) -> None:
         admin = make_user(1)
         request_hash = ReturnService._build_request_hash(self.page)