fix: validate overseas handover before settlement

firstleg/src/main/java/com/baiyue/firstleg/mapper/LoadingCompletedEventMapper.java

@@ -43,9 +43,7 @@ public interface LoadingCompletedEventMapper {
            AND actual_task.loading_plan_id = p.id
           JOIN fl_inbound_order o ON o.id = i.inbound_id AND o.tenant_id = i.account_id AND o.deleted = 0
           LEFT JOIN fl_inbound_order root_order ON root_order.id = COALESCE(o.root_inbound_id, o.id)
-         WHERE p.id = #{planId} AND p.plan_status = 'COMPLETED'
+         WHERE p.id = #{planId} AND i.package_count > 0
-           AND o.process_status = 'LOADED'
-           AND i.package_count > 0
          ORDER BY i.account_id, i.inbound_id, i.inbound_detail_id, i.id
         """)
     List<Row> rows(@Param("planId") long planId);

firstleg/src/main/java/com/baiyue/firstleg/service/LoadingCompletedOutboxService.java

@@ -48,16 +48,26 @@ public class LoadingCompletedOutboxService {
     @Resource private ObjectMapper objectMapper;
 
     public void publish(long planId) {
+        preparedEnvelopes(planId).forEach(this::insert);
+    }
+
+    /** 在任何入库结算前验证跨区交接快照，避免整托/箱唛冲突留下半完成状态。 */
+    public void requirePublishable(long planId) {
+        preparedEnvelopes(planId);
+    }
+
+    private List<EventEnvelope> preparedEnvelopes(long planId) {
         List<Row> rows = snapshotMapper.rows(planId);
         if (rows.isEmpty()) throw new BizException(ResultCode.CONFLICT, "装柜计划没有最终实装入仓单，不能生成尾程交接事件");
         validateContainerIdentity(rows);
         validateCustomerSnapshots(rows);
         Map<Long, List<String>> cartonMarks = cartonLabelPools(rows);
         List<PalletRow> palletRows = palletRows(planId);
-        rows.stream().collect(java.util.stream.Collectors.groupingBy(Row::getAccountId,
+        return rows.stream().collect(java.util.stream.Collectors.groupingBy(Row::getAccountId,
             java.util.TreeMap::new, java.util.stream.Collectors.toList()))
-            .forEach((accountId, accountRows) -> insert(envelope(accountId, accountRows,
+            .entrySet().stream().map(entry -> envelope(entry.getKey(), entry.getValue(),
-                palletRows.stream().filter(item -> accountId.equals(item.getAccountId())).toList(), cartonMarks)));
+                palletRows.stream().filter(item -> entry.getKey().equals(item.getAccountId())).toList(), cartonMarks))
+            .toList();
     }
 
     /** 已开展逐箱打托的入仓单使用现场事实；未开展的存量入仓单继续兼容原托盘分配快照。 */

firstleg/src/main/java/com/baiyue/firstleg/service/LoadingListService.java

@@ -571,6 +571,12 @@ public class LoadingListService {
         loadingCompletedOutboxService.publish(planId);
     }
 
+    /** 完成 worker 在远程结算前验证最终跨区快照能够完整生成。 */
+    @Transactional(readOnly = true)
+    public void requireCompletedEventReady(long planId) {
+        loadingCompletedOutboxService.requirePublishable(planId);
+    }
+
     private AggregateRow requirePlan(long planId) {
         AggregateRow row = repository.aggregate(planId);
         if (row == null) {

firstleg/src/main/java/com/baiyue/firstleg/service/LoadingPickService.java

@@ -535,6 +535,8 @@ public class LoadingPickService {
         List<GroupSnapshot> taskGroups = coloadGroups(owners);
         runCompletionStage("CABINET_EVIDENCE", () ->
             cabinetEvidenceService.requireComplete(task.getContainerBookingId()));
+        runCompletionStage("COMPLETED_EVENT_READY", () ->
+            loadingListService.requireCompletedEventReady(task.getLoadingPlanId()));
 
         // Phase 1 is read-only: cabinet evidence is checked once, then every owner must pass its business gates
         // before any warehouse resource or inbound status can be changed. This prevents a later customer gate from

firstleg/src/test/java/com/baiyue/firstleg/service/LoadingCompletedOutboxServiceTest.java

@@ -234,7 +234,7 @@ class LoadingCompletedOutboxServiceTest {
         assertThat(firstMarks).doesNotContainAnyElementsOf(secondMarks);
     }
 
-    @Test void snapshotOnlyPublishesFinalizedLoadedInbounds() throws Exception {
+    @Test void snapshotSupportsReadOnlyPreflightBeforeFinalStatusMutation() throws Exception {
         String rowsSql = String.join(" ", LoadingCompletedEventMapper.class.getMethod("rows", long.class)
             .getAnnotation(Select.class).value()).replaceAll("\\s+", " ");
         String palletsSql = String.join(" ", LoadingCompletedEventMapper.class.getMethod("palletRows", long.class)
@@ -243,8 +243,8 @@ class LoadingCompletedOutboxServiceTest {
             .getMethod("actualPalletRows", long.class).getAnnotation(Select.class).value())
             .replaceAll("\\s+", " ");
 
-        assertThat(rowsSql).contains("p.plan_status = 'COMPLETED'")
+        assertThat(rowsSql).doesNotContain("p.plan_status = 'COMPLETED'")
-            .contains("o.process_status = 'LOADED'")
+            .doesNotContain("o.process_status = 'LOADED'")
             .contains("i.customer_name")
             .doesNotContain("b.customer_id AS booking_customer_id")
             .doesNotContain("b.customer_name")

firstleg/src/test/java/com/baiyue/firstleg/service/LoadingPickServiceTest.java

@@ -966,6 +966,33 @@ class LoadingPickServiceTest {
         verify(loadingListService, never()).publishCompletedEvent(anyLong());
     }
 
+    @Test
+    void unpublishableCrossRegionSnapshotFailsBeforeAnyOwnerSettlement() {
+        LoadingPickTaskEntity completing = task();
+        completing.setTaskStatus("COMPLETING");
+        when(repository.task(20L)).thenReturn(completing);
+        when(repository.inboundOwners(20L)).thenReturn(List.of(
+            new LoadingPickItemMapper.InboundOwner(101L, 50L),
+            new LoadingPickItemMapper.InboundOwner(101L, 60L)));
+        doThrow(new BizException(com.baiyue.common.core.api.ResultCode.CONFLICT,
+            "一件代发现场托盘箱唛数与实装箱数不一致"))
+            .when(loadingListService).requireCompletedEventReady(10L);
+
+        assertThatThrownBy(() -> service.processCompletion(
+            completionCommand(905L, "async-unpublishable"), "worker-unpublishable"))
+            .isInstanceOf(LoadingPickService.CompletionProcessingException.class)
+            .satisfies(error -> assertThat(((LoadingPickService.CompletionProcessingException) error).stage())
+                .isEqualTo("COMPLETED_EVENT_READY"));
+
+        verify(settlementService, never()).settleOwner(anyLong(), anyLong(), anyLong(), anyString());
+        verify(loadingDeliveryService, never()).completeAfterPdaSettlement(anyLong(), anyLong(), anyLong(),
+            anyString(), anyString(), anyString());
+        verify(completionCommands, never()).checkpoint(anyLong(), anyLong(), anyLong(), anyInt(), anyString());
+        verify(repository, never()).insert(any(LoadingPickOperationEntity.class));
+        verify(loadingListService, never()).finishFromPda(anyLong(), anyLong(), anyLong(), anyString());
+        verify(loadingListService, never()).publishCompletedEvent(anyLong());
+    }
+
     @Test
     void workerRestartResumesFromOwnerCheckpointsWithoutRepeatingSettlements() {
         assertNoTruckCoupling();

scripts/admin-test-env-acceptance.mjs

@@ -1299,10 +1299,12 @@ export class AcceptanceRun {
     this.context.phase3LoadingRequireSharedOverflow = false
     this.context.phase3LoadingIncrementalPackageCount = 2
     this.context.phase3LoadingIncrementalDetailCount = 4
-    if (taxModes?.incremental) {
+    this.context.orderProfile = {
-      this.context.orderProfile = {
+      ...(this.context.orderProfile || {}),
-        ...(this.context.orderProfile || {}), taxMode: taxModes.incremental,
+      // The incremental order exercises partial/SKIP/PARTIAL mutations. Keep it STANDARD so a
-      }
+      // ONE_CLICK_DROPSHIP primary order remains a physically indivisible whole-pallet handover.
+      fulfillmentType: 'STANDARD',
+      ...(taxModes?.incremental ? { taxMode: taxModes.incremental } : {}),
     }
     await this.executeModule('orders-phase3-loading-incremental-fixture', () => orders.run(this))
     const measurement = await import(new URL('./admin-test-env-modules/inbound-measurement.mjs', import.meta.url))

scripts/admin-test-env-modules/phase3-loading.mjs

@@ -226,15 +226,18 @@ export async function run(context) {
   const loadingPlan = context.apiSuccess(MODULE, '读取柜次创建时原子生成的智能装柜计划', loadingPlanResponse, '', OPTIONS)
   const loadingPlanId = id(loadingPlan?.id)
   if (!loadingPlanId) return
+  const initialAssignmentCount = source?.fulfillmentType === 'ONE_CLICK_DROPSHIP'
+    ? Number(candidate.availablePackageCount) : 1
   const assignmentResponse = await request(context, 'POST', `${smartLoadingBase}/plans/${loadingPlanId}/assignments`, {
-    inboundId, inboundDetailId: id(candidate.inboundDetailId), packageCount: 1,
+    inboundId, inboundDetailId: id(candidate.inboundDetailId), packageCount: initialAssignmentCount,
     expectedVersion: Number(loadingPlan.version), operationId: `${context.marker}-bridge-assign`.slice(0, 128),
   })
   const assignedLoadingPlan = context.apiSuccess(MODULE, '将本轮入库货物分配到柜次计划', assignmentResponse, '', OPTIONS)
   context.assert(MODULE, '柜次计划明细和版本真实持久化', assignmentResponse,
-    assignedLoadingPlan?.items?.some?.((item) => id(item.inboundId) === inboundId && Number(item.packageCount) === 1)
+    assignedLoadingPlan?.items?.some?.((item) => id(item.inboundId) === inboundId
+      && Number(item.packageCount) === initialAssignmentCount)
       && Number(assignedLoadingPlan?.version) === 1,
-    `loadingPlanId=${loadingPlanId}, version=${assignedLoadingPlan?.version ?? 'missing'}`, OPTIONS)
+    `loadingPlanId=${loadingPlanId}, assignedPackages=${initialAssignmentCount}, version=${assignedLoadingPlan?.version ?? 'missing'}`, OPTIONS)
 
   const assignedItem = assignedLoadingPlan?.items?.find?.((item) => id(item.inboundId) === inboundId)
   const orderRemarks = source || {}

scripts/admin-test-env-modules/phase4-overseas-pda.mjs

@@ -279,11 +279,19 @@ async function convertReceivingContainers(context, producers) {
     const summaries = success(summaryResponse) && Array.isArray(summaryResponse.body?.data?.records)
       ? summaryResponse.body.data.records : []
     const summary = summaries.find((item) => id(item?.containerNo) === containerNo)
+    const totalOrders = Number(summary?.totalOrderCount)
+    const inboundCodes = Number(summary?.inboundCodeCount)
+    const existingPlans = Number(summary?.planCount)
+    const pendingPlans = Number(summary?.pendingPlanCount)
+    const convertible = nonBlank(summary?.sourceContainerBookingId)
+      && totalOrders > 0 && inboundCodes === totalOrders
+      && Number(summary?.missingInboundCodeCount) === 0
+      && ((existingPlans === 0 && pendingPlans === totalOrders)
+        || (summary?.allPlansReady === true && existingPlans === totalOrders && pendingPlans === 0))
     context.addStep(MODULE, `读取待分配柜级摘要 ${containerNo}`, summaryResponse,
-      Boolean(summary?.eligible && nonBlank(summary?.sourceContainerBookingId)
+      convertible,
-        && Number(summary?.inboundCodeCount) > 0),
+      `sourceContainerBookingId=${id(summary?.sourceContainerBookingId) || 'missing'}, allPlansReady=${summary?.allPlansReady === true}, orders=${totalOrders}, inboundCodes=${inboundCodes}, plans=${existingPlans}, pending=${pendingPlans}; ${trace(summaryResponse)}`, OPTIONS)
-      `sourceContainerBookingId=${id(summary?.sourceContainerBookingId) || 'missing'}, eligible=${summary?.eligible === true}, inboundCodes=${summary?.inboundCodeCount ?? 'missing'}; ${trace(summaryResponse)}`, OPTIONS)
+    if (!convertible) return false
-    if (!summary?.eligible || !nonBlank(summary?.sourceContainerBookingId)) return false
     const body = { sourceContainerBookingId: id(summary.sourceContainerBookingId) }
     const key = `${context.marker}-pda-convert-${containerNo}`.slice(0, 128)
     const convertedResponse = await context.requestOverseas('POST',
@@ -295,11 +303,11 @@ async function convertReceivingContainers(context, producers) {
       .flatMap((item) => Array.isArray(item?.lines) ? item.lines : [])
       .map((line) => id(line?.inboundCode)).filter(nonBlank))
     context.addStep(MODULE, `整柜转换为逐入仓码独立计划 ${containerNo}`, convertedResponse,
-      plans.length === Number(summary.inboundCodeCount)
+      plans.length === inboundCodes
         && plans.every((item) => Array.isArray(item?.lines) && item.lines.length > 0)
-        && convertedInboundCodes.size === Number(summary.inboundCodeCount),
+        && convertedInboundCodes.size === inboundCodes,
-      `plans=${plans.length}, inboundCodes=${convertedInboundCodes.size}, expected=${summary.inboundCodeCount}; ${trace(convertedResponse)}`, OPTIONS)
+      `plans=${plans.length}, inboundCodes=${convertedInboundCodes.size}, expected=${inboundCodes}; ${trace(convertedResponse)}`, OPTIONS)
-    if (plans.length !== Number(summary.inboundCodeCount)) return false
+    if (plans.length !== inboundCodes) return false
     const replayResponse = await context.requestOverseas('POST',
       '/api/platform/v1/last-leg/inbound-plans/convert-container', body,
       { 'Idempotency-Key': key }, context.overseasToken)
@@ -307,6 +315,15 @@ async function convertReceivingContainers(context, producers) {
       success(replayResponse) && JSON.stringify(replayResponse.body?.data) === JSON.stringify(plans),
       trace(replayResponse), OPTIONS)
     if (!success(replayResponse)) return false
+    const progressResponse = await context.requestOverseas('GET',
+      `/api/platform/v1/last-leg/orders/container-summaries/${encodeURIComponent(summary.sourceContainerBookingId)}`,
+      undefined, {}, context.overseasToken)
+    const progress = success(progressResponse) ? progressResponse.body?.data?.summary : null
+    context.addStep(MODULE, `整柜转换后计划完整度闭环 ${containerNo}`, progressResponse,
+      progress?.allPlansReady === true && Number(progress?.planCount) === totalOrders
+        && Number(progress?.pendingPlanCount) === 0,
+      `allPlansReady=${progress?.allPlansReady === true}, plans=${progress?.planCount ?? 'missing'}, pending=${progress?.pendingPlanCount ?? 'missing'}, plannedArrivalAt=${progress?.plannedArrivalAt || 'missing'}; ${trace(progressResponse)}`, OPTIONS)
+    if (progress?.allPlansReady !== true) return false
   }
   return true
 }