Merge branch 'test' of http://8.163.30.128:10082/java-project/baiyue-firstleg into test

firstleg/src/main/java/com/baiyue/firstleg/dto/LoadingListDtos.java

@@ -200,6 +200,7 @@ public final class LoadingListDtos {
     public record History(
         Long historyId,
         Long itemId,
+        String operationId,
         String actionType,
         String fromStatus,
         String toStatus,

firstleg/src/main/java/com/baiyue/firstleg/dto/LoadingPickDtos.java

@@ -125,12 +125,18 @@ public final class LoadingPickDtos {
                             Long operatorId, String operatorName, LocalDateTime createdAt) {
     }
 
+    public record CompletionCommand(Long id, String operationId, String status, Integer expectedTaskVersion, String remark,
+                                    Integer attemptCount, LocalDateTime nextRetryAt, String failureStage,
+                                    String failureCode, String requestId, String traceId,
+                                    LocalDateTime completedAt) {
+    }
+
     public record Task(Long id, Long loadingPlanId, Long containerBookingId, Long inboundId,
                        Integer inboundVersion, Integer loadingVersion, String cabinetNo,
                        String currentContainerNo, String sourceType, String recordStatus,
                        Long mergedIntoContainerBookingId, String taskStatus,
                        String workflowStatus, Integer version, Assignment assignment,
-                       Review review, List<PickItem> items) {
+                       Review review, CompletionCommand completionCommand, List<PickItem> items) {
     }
 
     /** PDA 工作台只展示当前作业员可领取或已领取的进行中任务。 */

firstleg/src/main/java/com/baiyue/firstleg/entity/LoadingCompletionCommandEntity.java

@@ -0,0 +1,36 @@
+package com.baiyue.firstleg.entity;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.Getter;
+import lombok.Setter;
+
+import java.time.LocalDateTime;
+
+/** Durable PDA completion command. Cross-service work is never executed in the acceptance transaction. */
+@Getter
+@Setter
+@TableName("fl_loading_completion_command")
+public class LoadingCompletionCommandEntity {
+    @TableId(type = IdType.INPUT) private Long id;
+    private Long pickTaskId;
+    private String operationId;
+    private String requestHash;
+    private String commandStatus;
+    private Integer expectedTaskVersion;
+    private String remark;
+    private Long operatorId;
+    private String operatorName;
+    private Integer attemptCount;
+    private LocalDateTime nextRetryAt;
+    private String leaseOwner;
+    private LocalDateTime leaseUntil;
+    private String failureStage;
+    private String failureCode;
+    private String failureRequestId;
+    private String failureTraceId;
+    private LocalDateTime createdAt;
+    private LocalDateTime updatedAt;
+    private LocalDateTime completedAt;
+}

firstleg/src/main/java/com/baiyue/firstleg/loading/repository/LoadingRepository.java

@@ -61,6 +61,17 @@ public class LoadingRepository {
             .eq(LoadingEvidenceEntity::getSealNo, sealNo));
     }
 
+    public LoadingEvidenceEntity findActiveEvidenceByFile(long tenantId, long inboundId,
+                                                           String evidenceType, long fileId) {
+        return evidences.selectOne(new LambdaQueryWrapper<LoadingEvidenceEntity>()
+            .eq(LoadingEvidenceEntity::getTenantId, tenantId)
+            .eq(LoadingEvidenceEntity::getInboundId, inboundId)
+            .eq(LoadingEvidenceEntity::getEvidenceType, evidenceType)
+            .eq(LoadingEvidenceEntity::getFileId, fileId)
+            .eq(LoadingEvidenceEntity::getStatus, "ACTIVE")
+            .last("LIMIT 1"));
+    }
+
     public void createEvidence(LoadingEvidenceEntity entity) { evidences.insert(entity); }
 
     public int voidEvidence(long tenantId, long inboundId, long evidenceId, String reason,

firstleg/src/main/java/com/baiyue/firstleg/loading/service/LoadingDeliveryService.java

@@ -34,6 +34,7 @@ import java.security.MessageDigest;
 import java.util.HexFormat;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -74,6 +75,18 @@ public class LoadingDeliveryService {
                 throw new BizException(ResultCode.CONFLICT, "锁码已用于其他装柜任务");
             }
         }
+        LoadingEvidenceEntity existingFile = repository.findActiveEvidenceByFile(
+            tenantId, inboundId, request.evidenceType(), request.fileId());
+        if (existingFile != null) {
+            String requestedSealNo = LoadingEvidenceType.SEAL.equals(request.evidenceType())
+                ? request.sealNo().trim() : null;
+            if (!Objects.equals(existingFile.getSealNo(), requestedSealNo)) {
+                throw new BizException(ResultCode.CONFLICT, "该带锁码照已绑定其他锁码，请先作废后重新拍摄");
+            }
+            LoadingActionResponse result = new LoadingActionResponse(
+                requireInbound(tenantId, inboundId).processStatus(), evidenceResponse(tenantId, inboundId));
+            return complete(tenantId, inboundId, operatorId, UPSERT_EVIDENCE, key, result);
+        }
         var file = validateFile(tenantId, inboundId, request.fileId(), request.evidenceType());
         String fileName = file.fileName();
         String sealNo = LoadingEvidenceType.SEAL.equals(request.evidenceType()) ? request.sealNo().trim() : null;
@@ -103,23 +116,33 @@ public class LoadingDeliveryService {
     private LoadingActionResponse completeLoading(long tenantId, long inboundId, long evidenceInboundId,
                                                    long operatorId, String operatorName, String key,
                                                    CompleteLoadingRequest request) {
+        return completeLoading(tenantId, inboundId, evidenceInboundId, operatorId, operatorName,
+            key, request, false);
+    }
+
+    private LoadingActionResponse completeLoading(long tenantId, long inboundId, long evidenceInboundId,
+                                                   long operatorId, String operatorName, String key,
+                                                   CompleteLoadingRequest request,
+                                                   boolean acceptedPdaCompletion) {
         String hash = hash(request);
-        LoadingActionResponse replay = claimOrReplay(tenantId, inboundId, operatorId, COMPLETE_LOADING, key, hash);
+        LoadingActionResponse replay = claimOrReplay(tenantId, inboundId, operatorId, COMPLETE_LOADING,
+            key, hash, acceptedPdaCompletion);
         if (replay != null) return replay;
         Snapshot inbound = requireInbound(tenantId, inboundId);
         CompletionTarget target = repository.completionTarget(tenantId, inboundId);
         if (target != null) {
-            return completeFulfillmentSlice(tenantId, evidenceInboundId, operatorId, operatorName, key, request, target);
+            return completeFulfillmentSlice(tenantId, evidenceInboundId, operatorId, operatorName,
+                key, request, target, acceptedPdaCompletion);
         }
         if ("LOADED".equals(inbound.processStatus()) || "DELIVERED".equals(inbound.processStatus())) {
-            loadingModelBridge.requireCompleted(tenantId, inboundId);
+            requireLoadingPlanGate(tenantId, inboundId, acceptedPdaCompletion);
             return complete(tenantId, inboundId, operatorId, COMPLETE_LOADING, key,
                 new LoadingActionResponse(inbound.processStatus(), evidenceResponse(tenantId, inboundId)));
         }
         if (!"LOADING".equals(inbound.processStatus())) throw new BizException(ResultCode.CONFLICT, "当前订单不在装柜中");
-        loadingModelBridge.requireCompleted(tenantId, inboundId);
+        requireLoadingPlanGate(tenantId, inboundId, acceptedPdaCompletion);
         completionGate.requireReleasedBooking(tenantId, inboundId);
-        requireCompletedPicking(tenantId, inboundId);
+        requirePickingGate(tenantId, inboundId, acceptedPdaCompletion);
         requireCompletionEvidence(tenantId, evidenceInboundId);
         if (repository.completeLoading(tenantId, inboundId, operatorId, request.expectedVersion()) != 1) throw concurrent();
         addHistory(tenantId, inboundId, "LOADING", "LOADED", COMPLETE_LOADING, request.remark(), operatorId, operatorName);
@@ -132,12 +155,14 @@ public class LoadingDeliveryService {
      * Resolve and lock the canonical completion row after settlement, then retain optimistic locking
      * with that current server version instead of reusing the pre-settlement task snapshot.
      */
+    @Transactional
     public LoadingActionResponse completeAfterPdaSettlement(long tenantId, long inboundId,
                                                              long operatorId, String operatorName,
                                                              String key, String remark) {
         return completeAfterPdaSettlement(tenantId, inboundId, inboundId, operatorId, operatorName, key, remark);
     }
 
+    @Transactional
     public LoadingActionResponse completeAfterPdaSettlement(long tenantId, long inboundId,
                                                              long evidenceInboundId, long operatorId,
                                                              String operatorName, String key, String remark) {
@@ -148,7 +173,7 @@ public class LoadingDeliveryService {
             throw new BizException(ResultCode.CONFLICT, "PDA结算后的实装履约分片不存在，请刷新后重试");
         }
         return completeLoading(tenantId, inboundId, evidenceInboundId, operatorId, operatorName, key,
-            new CompleteLoadingRequest(current.version(), remark));
+            new CompleteLoadingRequest(current.version(), remark), true);
     }
 
     /**
@@ -172,7 +197,8 @@ public class LoadingDeliveryService {
     private LoadingActionResponse completeFulfillmentSlice(long tenantId, long evidenceInboundId,
                                                             long operatorId, String operatorName, String key,
                                                             CompleteLoadingRequest request,
-                                                            CompletionTarget target) {
+                                                            CompletionTarget target,
+                                                            boolean acceptedPdaCompletion) {
         if ("LOADED".equals(target.loadedChildStatus())) {
             return complete(tenantId, evidenceInboundId, operatorId, COMPLETE_LOADING, key,
                 new LoadingActionResponse("LOADED", evidenceResponse(tenantId, evidenceInboundId)));
@@ -180,9 +206,9 @@ public class LoadingDeliveryService {
         if (!"LOADING".equals(target.loadedChildStatus())) {
             throw new BizException(ResultCode.CONFLICT, "实装履约分片不在装柜中");
         }
-        loadingModelBridge.requireCompleted(tenantId, target.sourceInboundId());
+        requireLoadingPlanGate(tenantId, target.sourceInboundId(), acceptedPdaCompletion);
         completionGate.requireReleasedBooking(tenantId, target.sourceInboundId());
-        requireCompletedPicking(tenantId, target.sourceInboundId());
+        requirePickingGate(tenantId, target.sourceInboundId(), acceptedPdaCompletion);
         requireCompletionEvidence(tenantId, evidenceInboundId);
         if (repository.completeLoadedChild(tenantId, target.sourceInboundId(), operatorId,
             request.expectedVersion()) != 1) throw concurrent();
@@ -295,9 +321,21 @@ public class LoadingDeliveryService {
             task.getExpectedCount(), task.getPickedCount(), task.getRemark());
     }
 
-    private void requireCompletedPicking(long tenantId, long inboundId) {
-        if (!canonicalPicking(tenantId, inboundId).completed()) {
-            throw new BizException(ResultCode.CONFLICT, "PDA拣货复核尚未完成");
+    /** worker 先完成各入库单，之后才完成柜次计划；这里不能循环要求计划已经 COMPLETED。 */
+    private void requireLoadingPlanGate(long tenantId, long inboundId, boolean acceptedPdaCompletion) {
+        if (acceptedPdaCompletion) {
+            loadingModelBridge.requireStarted(tenantId, inboundId);
+            return;
+        }
+        loadingModelBridge.requireCompleted(tenantId, inboundId);
+    }
+
+    /** completion worker 已持有 durable lease；此时 task 尚为 COMPLETING，不能循环要求最终 COMPLETED。 */
+    private void requirePickingGate(long tenantId, long inboundId, boolean acceptedPdaCompletion) {
+        CanonicalPicking picking = canonicalPicking(tenantId, inboundId);
+        if (acceptedPdaCompletion ? !picking.resolved() : !picking.completed()) {
+            throw new BizException(ResultCode.CONFLICT, acceptedPdaCompletion
+                ? "PDA实装明细尚未全部解决" : "PDA拣货复核尚未完成");
         }
     }
 
@@ -324,9 +362,12 @@ public class LoadingDeliveryService {
 
     private record CanonicalPicking(int expectedCount, int pickedCount, int taskCount,
                                     int completedTaskCount, int unresolvedCount) {
-        boolean completed() {
+        boolean resolved() {
             return expectedCount > 0 && expectedCount == pickedCount && taskCount > 0
-                && completedTaskCount == taskCount && unresolvedCount == 0;
+                && unresolvedCount == 0;
+        }
+        boolean completed() {
+            return resolved() && completedTaskCount == taskCount;
         }
     }
 
@@ -418,10 +459,17 @@ public class LoadingDeliveryService {
 
     private LoadingActionResponse claimOrReplay(long tenantId, long inboundId, long operatorId,
                                                 String operation, String key, String hash) {
+        return claimOrReplay(tenantId, inboundId, operatorId, operation, key, hash, false);
+    }
+
+    private LoadingActionResponse claimOrReplay(long tenantId, long inboundId, long operatorId,
+                                                String operation, String key, String hash,
+                                                boolean resumeProcessing) {
         validateKey(key);
         if (repository.claimOperation(tenantId, inboundId, operatorId, operation, key, hash)) return null;
         LoadingOperationEntity existing = repository.findOperation(tenantId, inboundId, operatorId, operation, key);
         if (existing == null || !hash.equals(existing.getRequestHash())) throw new BizException(ResultCode.CONFLICT, "幂等键已用于其他请求");
+        if (resumeProcessing && "PROCESSING".equals(existing.getStatus())) return null;
         if (!"SUCCESS".equals(existing.getStatus()) || !StringUtils.hasText(existing.getResultJson())) throw new BizException(ResultCode.CONFLICT, "相同请求正在处理中，请稍后重试");
         try { return objectMapper.readValue(existing.getResultJson(), LoadingActionResponse.class); }
         catch (JsonProcessingException e) { throw new IllegalStateException("无法读取装柜幂等结果", e); }

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

@@ -7,6 +7,7 @@ import org.apache.ibatis.annotations.Param;
 import org.apache.ibatis.annotations.Select;
 
 import java.math.BigDecimal;
+import java.time.LocalDateTime;
 import java.util.List;
 
 /** 读取 PDA 结算后的最终实装入仓单和冻结货物快照。 */
@@ -15,6 +16,9 @@ public interface LoadingCompletedEventMapper {
     @Select("""
         SELECT p.id AS loading_plan_id, p.plan_no, p.container_booking_id, p.version AS aggregate_version,
                b.container_no, b.cabinet_no, b.warehouse_id, b.warehouse_name,
+               b.customer_id AS booking_customer_id, b.customer_name,
+               b.estimated_port_arrival_at, b.actual_port_arrival_at,
+               b.estimated_warehouse_arrival_at,
                i.account_id, i.inbound_id AS source_inbound_id,
                COALESCE(o.root_inbound_id, o.id) AS root_inbound_id,
                o.order_id AS source_order_id, o.order_no AS source_order_no,
@@ -49,7 +53,8 @@ public interface LoadingCompletedEventMapper {
     /** 读取一键代发托盘及其箱唛快照；普通订单不会使用该结果。 */
     @Select("""
         SELECT i.account_id, i.inbound_id AS source_inbound_id,
-               a.pallet_code, a.allocation_no, a.picked_package_count AS allocated_package_count,
+               a.pallet_id, a.reservation_id, a.pallet_code, a.allocation_no,
+               a.picked_package_count AS allocated_package_count,
                i.inbound_detail_id AS source_inbound_detail_id, i.sku_code,
                i.product_name_zh, i.package_count, i.product_quantity,
                label.box_code
@@ -79,7 +84,10 @@ public interface LoadingCompletedEventMapper {
     class Row {
         private Long loadingPlanId; private String planNo; private Long containerBookingId;
         private Integer aggregateVersion; private String containerNo; private String cabinetNo;
-        private Long warehouseId; private String warehouseName; private Long accountId;
+        private Long warehouseId; private String warehouseName; private Long bookingCustomerId;
+        private String customerName; private Long accountId;
+        private LocalDateTime estimatedPortArrivalAt; private LocalDateTime actualPortArrivalAt;
+        private LocalDateTime estimatedWarehouseArrivalAt;
         private Long sourceInboundId; private Long rootInboundId; private Long sourceOrderId;
         private Long loadingPlanItemId;
         private String sourceOrderNo; private Long rootOrderId; private String inboundCode;
@@ -95,7 +103,8 @@ public interface LoadingCompletedEventMapper {
 
     @Getter @Setter
     class PalletRow {
-        private Long accountId; private Long sourceInboundId; private String palletCode;
+        private Long accountId; private Long sourceInboundId; private Long palletId;
+        private Long reservationId; private String palletCode;
         private Integer allocationNo; private Integer allocatedPackageCount;
         private Long sourceInboundDetailId; private String skuCode; private String productNameZh;
         private Integer packageCount; private Integer productQuantity; private String boxCode;

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

@@ -0,0 +1,109 @@
+package com.baiyue.firstleg.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.baiyue.firstleg.entity.LoadingCompletionCommandEntity;
+import org.apache.ibatis.annotations.Insert;
+import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Select;
+import org.apache.ibatis.annotations.Update;
+
+import java.time.LocalDateTime;
+import java.util.List;
+
+@Mapper
+public interface LoadingCompletionCommandMapper extends BaseMapper<LoadingCompletionCommandEntity> {
+    @Insert("""
+        INSERT IGNORE INTO fl_loading_completion_command
+          (id,pick_task_id,operation_id,request_hash,command_status,expected_task_version,remark,
+           operator_id,operator_name,attempt_count,next_retry_at)
+        VALUES
+          (#{id},#{pickTaskId},#{operationId},#{requestHash},'PROCESSING',#{expectedTaskVersion},#{remark},
+           #{operatorId},#{operatorName},0,CURRENT_TIMESTAMP(3))
+        """)
+    int insertCommand(LoadingCompletionCommandEntity command);
+
+    @Select("SELECT * FROM fl_loading_completion_command WHERE operation_id=#{operationId}")
+    LoadingCompletionCommandEntity findByOperation(@Param("operationId") String operationId);
+
+    @Select("""
+        SELECT * FROM fl_loading_completion_command
+         WHERE pick_task_id=#{taskId} AND command_status IN ('PROCESSING','RETRYABLE','FAILED')
+         ORDER BY created_at DESC LIMIT 1
+        """)
+    LoadingCompletionCommandEntity findActiveByTask(@Param("taskId") long taskId);
+
+    @Select("SELECT * FROM fl_loading_completion_command WHERE pick_task_id=#{taskId} ORDER BY created_at DESC LIMIT 1")
+    LoadingCompletionCommandEntity findLatestByTask(@Param("taskId") long taskId);
+
+    @Select("""
+        SELECT * FROM fl_loading_completion_command
+         WHERE command_status IN ('PROCESSING','RETRYABLE') AND next_retry_at<=CURRENT_TIMESTAMP(3)
+           AND (lease_until IS NULL OR lease_until<CURRENT_TIMESTAMP(3))
+         ORDER BY next_retry_at,id LIMIT #{limit}
+        """)
+    List<LoadingCompletionCommandEntity> findDue(@Param("limit") int limit);
+
+    @Update("""
+        UPDATE fl_loading_completion_command
+           SET command_status='PROCESSING',lease_owner=#{worker},lease_until=#{leaseUntil},
+               attempt_count=attempt_count+1,updated_at=CURRENT_TIMESTAMP(3)
+         WHERE id=#{id} AND command_status IN ('PROCESSING','RETRYABLE')
+           AND next_retry_at<=CURRENT_TIMESTAMP(3)
+           AND (lease_until IS NULL OR lease_until<CURRENT_TIMESTAMP(3))
+        """)
+    int claim(@Param("id") long id, @Param("worker") String worker,
+              @Param("leaseUntil") LocalDateTime leaseUntil);
+
+    @Update("""
+        UPDATE fl_loading_completion_command SET lease_until=#{leaseUntil},updated_at=CURRENT_TIMESTAMP(3)
+         WHERE id=#{id} AND command_status='PROCESSING' AND lease_owner=#{worker}
+        """)
+    int renew(@Param("id") long id, @Param("worker") String worker,
+              @Param("leaseUntil") LocalDateTime leaseUntil);
+
+    @Update("""
+        UPDATE fl_loading_completion_command
+           SET command_status='COMPLETED',completed_at=CURRENT_TIMESTAMP(3),updated_at=CURRENT_TIMESTAMP(3),
+               lease_owner=NULL,lease_until=NULL,failure_stage=NULL,failure_code=NULL,
+               failure_request_id=NULL,failure_trace_id=NULL
+         WHERE id=#{id} AND command_status='PROCESSING' AND lease_owner=#{worker}
+        """)
+    int complete(@Param("id") long id, @Param("worker") String worker);
+
+    @Update("""
+        UPDATE fl_loading_completion_command
+           SET command_status=#{status},next_retry_at=#{nextRetryAt},updated_at=CURRENT_TIMESTAMP(3),
+               lease_owner=NULL,lease_until=NULL,failure_stage=#{stage},failure_code=#{code},
+               failure_request_id=#{requestId},failure_trace_id=#{traceId}
+         WHERE id=#{id} AND command_status='PROCESSING' AND lease_owner=#{worker}
+        """)
+    int fail(@Param("id") long id, @Param("worker") String worker, @Param("status") String status,
+             @Param("nextRetryAt") LocalDateTime nextRetryAt, @Param("stage") String stage,
+             @Param("code") String code, @Param("requestId") String requestId, @Param("traceId") String traceId);
+
+    @Update("""
+        UPDATE fl_loading_completion_command
+           SET command_status='RETRYABLE',next_retry_at=CURRENT_TIMESTAMP(3),updated_at=CURRENT_TIMESTAMP(3),
+               lease_owner=NULL,lease_until=NULL
+         WHERE id=#{id} AND command_status='FAILED'
+        """)
+    int resume(@Param("id") long id);
+
+    @Select("""
+        SELECT checkpoint_stage FROM fl_loading_completion_checkpoint
+         WHERE command_id=#{commandId} AND tenant_id=#{tenantId} AND inbound_id=#{inboundId}
+        """)
+    String checkpoint(@Param("commandId") long commandId, @Param("tenantId") long tenantId,
+                      @Param("inboundId") long inboundId);
+
+    @Insert("""
+        INSERT INTO fl_loading_completion_checkpoint
+          (command_id,tenant_id,inbound_id,owner_sequence,checkpoint_stage)
+        VALUES (#{commandId},#{tenantId},#{inboundId},#{sequence},#{stage})
+        ON DUPLICATE KEY UPDATE checkpoint_stage=VALUES(checkpoint_stage),updated_at=CURRENT_TIMESTAMP(3)
+        """)
+    int checkpointCompleted(@Param("commandId") long commandId, @Param("tenantId") long tenantId,
+                            @Param("inboundId") long inboundId, @Param("sequence") int sequence,
+                            @Param("stage") String stage);
+}

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

@@ -232,6 +232,7 @@ public interface LoadingPickItemMapper extends BaseMapper<LoadingPickItemEntity>
     /**
      * 以结算后的装柜计划事实为完成口径，不能只检查当前 PDA 任务中的明细。
      * PARTIAL 在本查询前已把计划箱数收敛为实际拣货箱数；收敛前的严格小于关系由结算计划校验。
+     * 异步完成期间任务会在 COMPLETING/COMPLETION_FAILED 间切换，不能把任务状态当成明细未完成事实。
      */
     @Select("""
         SELECT COUNT(*)
@@ -244,7 +245,6 @@ public interface LoadingPickItemMapper extends BaseMapper<LoadingPickItemEntity>
                  JOIN fl_loading_pick_task pick_task ON pick_task.id = pick_item.pick_task_id
                 WHERE pick_item.loading_plan_item_id = plan_item.id
                   AND pick_task.loading_plan_id = #{planId}
-                  AND pick_task.task_status = 'LOADING'
                   AND (
                       (pick_item.item_status = 'PICKED'
                        AND pick_item.picked_package_count = plan_item.package_count)

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

@@ -19,6 +19,29 @@ public interface LoadingPickTaskMapper extends BaseMapper<LoadingPickTaskEntity>
         + "AND task_status='LOADING' AND version=#{expectedVersion}")
     int touch(@Param("taskId") long taskId, @Param("expectedVersion") int expectedVersion);
 
+    @Update("""
+        UPDATE fl_loading_pick_task SET task_status='COMPLETING',version=version+1
+         WHERE id=#{taskId} AND task_status='LOADING' AND version=#{expectedVersion}
+        """)
+    int acceptCompletion(@Param("taskId") long taskId, @Param("expectedVersion") int expectedVersion);
+
+    @Update("""
+        UPDATE fl_loading_pick_task
+           SET task_status='COMPLETED',completed_by=#{operatorId},completed_by_name=#{operatorName},
+               completed_at=CURRENT_TIMESTAMP(3),version=version+1
+         WHERE id=#{taskId} AND task_status='COMPLETING'
+        """)
+    int completeAccepted(@Param("taskId") long taskId, @Param("operatorId") long operatorId,
+                         @Param("operatorName") String operatorName);
+
+    @Update("UPDATE fl_loading_pick_task SET task_status='COMPLETION_FAILED',version=version+1 "
+        + "WHERE id=#{taskId} AND task_status='COMPLETING'")
+    int failCompletion(@Param("taskId") long taskId);
+
+    @Update("UPDATE fl_loading_pick_task SET task_status='COMPLETING',version=version+1 "
+        + "WHERE id=#{taskId} AND task_status='COMPLETION_FAILED'")
+    int resumeCompletion(@Param("taskId") long taskId);
+
     @Select("""
         SELECT DISTINCT task.*
           FROM fl_loading_pick_task task
@@ -60,6 +83,8 @@ public interface LoadingPickTaskMapper extends BaseMapper<LoadingPickTaskEntity>
                booking.merged_into_container_booking_id AS mergedIntoContainerBookingId,
                booking.warehouse_name AS warehouseName, task.task_status AS taskStatus,
                CASE WHEN task.task_status = 'COMPLETED' THEN 'COMPLETED'
+                    WHEN task.task_status = 'COMPLETING' THEN 'PROCESSING'
+                    WHEN task.task_status = 'COMPLETION_FAILED' THEN 'COMPLETION_FAILED'
                     WHEN assignment.pick_task_id IS NULL THEN 'UNASSIGNED'
                     WHEN SUM(CASE WHEN pick_item.item_status IN ('PENDING', 'PICKING') THEN 1 ELSE 0 END) = 0
                       THEN 'AWAITING_REVIEW'
@@ -72,7 +97,7 @@ public interface LoadingPickTaskMapper extends BaseMapper<LoadingPickTaskEntity>
           JOIN fl_container_booking booking ON booking.id = plan.container_booking_id
           JOIN fl_loading_pick_item pick_item ON pick_item.pick_task_id = task.id
           LEFT JOIN fl_loading_pick_assignment assignment ON assignment.pick_task_id = task.id
-         WHERE task.task_status = 'LOADING' AND booking.record_status = 'ACTIVE'
+         WHERE task.task_status IN ('LOADING','COMPLETING','COMPLETION_FAILED') AND booking.record_status = 'ACTIVE'
            AND (assignment.pick_task_id IS NULL OR assignment.assignee_id = #{operatorId})
            <if test="keyword != null and keyword != ''">
              AND (booking.container_no LIKE CONCAT('%', #{keyword}, '%')

firstleg/src/main/java/com/baiyue/firstleg/repository/LoadingCompletionCommandRepository.java

@@ -0,0 +1,34 @@
+package com.baiyue.firstleg.repository;
+
+import com.baiyue.firstleg.entity.LoadingCompletionCommandEntity;
+import com.baiyue.firstleg.mapper.LoadingCompletionCommandMapper;
+import jakarta.annotation.Resource;
+import org.springframework.stereotype.Repository;
+
+import java.time.LocalDateTime;
+import java.util.List;
+
+@Repository
+public class LoadingCompletionCommandRepository {
+    @Resource private LoadingCompletionCommandMapper mapper;
+
+    public int insert(LoadingCompletionCommandEntity command) { return mapper.insertCommand(command); }
+    public LoadingCompletionCommandEntity operation(String operationId) { return mapper.findByOperation(operationId); }
+    public LoadingCompletionCommandEntity active(long taskId) { return mapper.findActiveByTask(taskId); }
+    public LoadingCompletionCommandEntity latest(long taskId) { return mapper.findLatestByTask(taskId); }
+    public List<LoadingCompletionCommandEntity> due(int limit) { return mapper.findDue(limit); }
+    public boolean claim(long id, String worker, LocalDateTime leaseUntil) { return mapper.claim(id, worker, leaseUntil) == 1; }
+    public boolean renew(long id, String worker, LocalDateTime leaseUntil) { return mapper.renew(id, worker, leaseUntil) == 1; }
+    public boolean complete(long id, String worker) { return mapper.complete(id, worker) == 1; }
+    public boolean fail(long id, String worker, String status, LocalDateTime nextRetryAt, String stage,
+                        String code, String requestId, String traceId) {
+        return mapper.fail(id, worker, status, nextRetryAt, stage, code, requestId, traceId) == 1;
+    }
+    public boolean resume(long id) { return mapper.resume(id) == 1; }
+    public String checkpoint(long commandId, long tenantId, long inboundId) {
+        return mapper.checkpoint(commandId, tenantId, inboundId);
+    }
+    public void checkpoint(long commandId, long tenantId, long inboundId, int sequence, String stage) {
+        mapper.checkpointCompleted(commandId, tenantId, inboundId, sequence, stage);
+    }
+}

firstleg/src/main/java/com/baiyue/firstleg/repository/LoadingListRepository.java

@@ -188,7 +188,7 @@ public class LoadingListRepository {
 
     private List<History> histories(long planId) {
         return mapper.selectHistories(planId).stream().map(row -> new History(row.getHistoryId(), row.getItemId(),
-            row.getActionType(), row.getFromStatus(), row.getToStatus(), row.getBeforeSnapshot(),
+            row.getOperationId(), row.getActionType(), row.getFromStatus(), row.getToStatus(), row.getBeforeSnapshot(),
             row.getAfterSnapshot(), row.getOperatorId(), row.getOperatorName(), row.getOperationReason(),
             row.getOccurredAt())).toList();
     }

firstleg/src/main/java/com/baiyue/firstleg/repository/LoadingPickRepository.java

@@ -57,6 +57,18 @@ public class LoadingPickRepository {
         return tasks.touch(taskId, expectedVersion);
     }
 
+    public int acceptCompletion(long taskId, int expectedVersion) {
+        return tasks.acceptCompletion(taskId, expectedVersion);
+    }
+
+    public int completeAccepted(long taskId, long operatorId, String operatorName) {
+        return tasks.completeAccepted(taskId, operatorId, operatorName);
+    }
+
+    public int failCompletion(long taskId) { return tasks.failCompletion(taskId); }
+
+    public int resumeCompletion(long taskId) { return tasks.resumeCompletion(taskId); }
+
     public LoadingPickTaskEntity taskByPlan(long planId) {
         return tasks.selectOne(new LambdaQueryWrapper<LoadingPickTaskEntity>()
             .eq(LoadingPickTaskEntity::getLoadingPlanId, planId));

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

@@ -37,7 +37,7 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 
-/** 在装柜完成事务内按客户写跨区业务 Outbox。 */
+/** 在装柜完成事务内按客户及可信客户名称快照写跨区业务 Outbox。 */
 @Service
 public class LoadingCompletedOutboxService {
     private static final long HANDOVER_VERSION = 2L;
@@ -49,6 +49,7 @@ public class LoadingCompletedOutboxService {
     public void publish(long planId) {
         List<Row> rows = snapshotMapper.rows(planId);
         if (rows.isEmpty()) throw new BizException(ResultCode.CONFLICT, "装柜计划没有最终实装入仓单，不能生成尾程交接事件");
+        validateCustomerSnapshots(rows);
         Map<Long, List<String>> cartonMarks = cartonLabelPools(rows);
         List<PalletRow> palletRows = snapshotMapper.palletRows(planId);
         rows.stream().collect(java.util.stream.Collectors.groupingBy(Row::getAccountId,
@@ -67,7 +68,8 @@ public class LoadingCompletedOutboxService {
                 .filter(item -> Objects.equals(item.getSourceInboundId(), value.get(0).getSourceInboundId())).toList(), cartonMarks)).toList();
         var payload = new FirstLegLoadingPlanCompletedV2(String.valueOf(first.getLoadingPlanId()),
             first.getPlanNo(), string(first.getContainerBookingId()), first.getContainerNo(), first.getCabinetNo(),
-            string(first.getWarehouseId()), first.getWarehouseName(), orders);
+            string(first.getWarehouseId()), first.getWarehouseName(), iso(first.getEstimatedPortArrivalAt()),
+            iso(first.getActualPortArrivalAt()), iso(first.getEstimatedWarehouseArrivalAt()), orders);
         JsonNode payloadNode = objectMapper.valueToTree(payload);
         String hash = EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalize(payloadNode));
         String eventKey = "FIRSTLEG_LOADING_COMPLETED:" + first.getLoadingPlanId() + ':' + accountId + ':' + HANDOVER_VERSION;
@@ -102,12 +104,38 @@ public class LoadingCompletedOutboxService {
             string(first.getRootInboundId()), string(first.getSourceOrderId()), first.getSourceOrderNo(),
             string(first.getRootOrderId()), first.getInboundCode(), string(first.getDestinationWarehouseId()),
             first.getDestinationWarehouseCode(), first.getDestinationWarehouseName(), sumInt(sorted, Row::getPackageCount),
-            sumNullableInt(sorted, Row::getProductQuantity), "ONE_CLICK_DROPSHIP".equals(orderType)
-                ? pallets.size() : 0, sum(sorted, Row::getGrossWeightKg), sum(sorted, Row::getNetWeightKg),
-            sum(sorted, Row::getVolumeCbm), items, string(first.getAccountId()), orderType, pallets,
+            sumNullableInt(sorted, Row::getProductQuantity), actualPalletCount(palletRows),
+            sum(sorted, Row::getGrossWeightKg), sum(sorted, Row::getNetWeightKg),
+            sum(sorted, Row::getVolumeCbm), sum(sorted, Row::getDimensionalWeightKg), items,
+            string(first.getAccountId()), first.getCustomerName().trim(), orderType, pallets,
             destinationType(sorted));
     }
 
+    /** 只按 Warehouse 已分配的物理 palletId 计数；无分配事实时保持 null。 */
+    private static Integer actualPalletCount(List<PalletRow> palletRows) {
+        Set<Long> palletIds = palletRows.stream().map(PalletRow::getPalletId)
+            .filter(Objects::nonNull).filter(value -> value > 0).collect(java.util.stream.Collectors.toSet());
+        return palletIds.isEmpty() ? null : palletIds.size();
+    }
+
+    private void validateCustomerSnapshots(List<Row> rows) {
+        for (Row row : rows) {
+            if (row.getAccountId() == null || !Objects.equals(row.getAccountId(), row.getBookingCustomerId())) {
+                throw new BizException(ResultCode.CONFLICT, "装柜计划客户与柜次客户快照不一致，不能生成尾程交接事件");
+            }
+            if (row.getCustomerName() == null || row.getCustomerName().isBlank()) {
+                throw new BizException(ResultCode.CONFLICT, "柜次缺少客户名称快照，不能生成尾程交接事件");
+            }
+        }
+        rows.stream().collect(java.util.stream.Collectors.groupingBy(Row::getAccountId)).forEach((accountId, values) -> {
+            Set<String> names = values.stream().map(Row::getCustomerName).map(String::trim)
+                .collect(java.util.stream.Collectors.toSet());
+            if (names.size() != 1) {
+                throw new BizException(ResultCode.CONFLICT, "同一客户事件存在多个客户名称快照");
+            }
+        });
+    }
+
     /** 国内按实装箱数从可用标签池稳定随机抽取箱唛，重复发布同一计划时结果不变。 */
     private Map<Long, List<String>> cartonLabelPools(List<Row> rows) {
         Map<Long, List<String>> result = new java.util.LinkedHashMap<>();
@@ -256,6 +284,7 @@ public class LoadingCompletedOutboxService {
     private String json(Object value) { try { return objectMapper.writeValueAsString(value); }
         catch (JsonProcessingException e) { throw new BizException(ResultCode.INTERNAL_ERROR, "装柜完成事件序列化失败"); } }
     private static String string(Long value) { return value == null ? null : String.valueOf(value); }
+    private static String iso(java.time.LocalDateTime value) { return value == null ? null : value.toString(); }
     private static int sumInt(List<Row> rows, java.util.function.Function<Row,Integer> getter) { return rows.stream().map(getter).filter(java.util.Objects::nonNull).mapToInt(Integer::intValue).sum(); }
     private static Integer sumNullableInt(List<Row> rows, java.util.function.Function<Row,Integer> getter) { List<Integer> values=rows.stream().map(getter).filter(java.util.Objects::nonNull).toList(); return values.isEmpty()?null:values.stream().mapToInt(Integer::intValue).sum(); }
     private static BigDecimal sum(List<Row> rows, java.util.function.Function<Row,BigDecimal> getter) { List<BigDecimal> values=rows.stream().map(getter).filter(java.util.Objects::nonNull).toList(); return values.isEmpty()?null:values.stream().reduce(BigDecimal.ZERO, BigDecimal::add); }

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

@@ -0,0 +1,64 @@
+package com.baiyue.firstleg.service;
+
+import com.baiyue.common.core.exception.BizException;
+import com.baiyue.firstleg.entity.LoadingCompletionCommandEntity;
+import com.baiyue.firstleg.repository.LoadingCompletionCommandRepository;
+import com.baiyue.firstleg.repository.LoadingPickRepository;
+import jakarta.annotation.Resource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+
+import java.time.LocalDateTime;
+import java.util.UUID;
+
+/** Lease-based recovery worker for PDA completion commands. */
+@Service
+public class LoadingCompletionWorker {
+    private static final Logger log = LoggerFactory.getLogger(LoadingCompletionWorker.class);
+    private static final int MAX_ATTEMPTS = 5;
+    private final String workerId = "pda-completion-" + UUID.randomUUID();
+
+    @Resource private LoadingCompletionCommandRepository commands;
+    @Resource private LoadingPickRepository picks;
+    @Resource private LoadingPickService loadingPicks;
+
+    @Scheduled(fixedDelayString = "${baiyue.firstleg.pda-completion.fixed-delay-ms:2000}")
+    public void dispatch() {
+        for (LoadingCompletionCommandEntity candidate : commands.due(10)) {
+            if (!commands.claim(candidate.getId(), workerId, LocalDateTime.now().plusMinutes(10))) {
+                continue;
+            }
+            try {
+                loadingPicks.processCompletion(candidate, workerId);
+            } catch (RuntimeException exception) {
+                handleFailure(candidate, exception);
+            }
+        }
+    }
+
+    private void handleFailure(LoadingCompletionCommandEntity command, RuntimeException exception) {
+        Throwable cause = exception instanceof LoadingPickService.CompletionProcessingException
+            && exception.getCause() != null ? exception.getCause() : exception;
+        String stage = exception instanceof LoadingPickService.CompletionProcessingException staged
+            ? staged.stage() : "UNEXPECTED";
+        String code = cause instanceof BizException business
+            ? business.getResultCode().code() : "S0500";
+        String requestId = cause instanceof PdaPickSettlementService.SettlementFailure remote
+            ? remote.requestId() : null;
+        String traceId = cause instanceof PdaPickSettlementService.SettlementFailure remote
+            ? remote.traceId() : null;
+        int attempt = command.getAttemptCount() == null ? 1 : command.getAttemptCount() + 1;
+        boolean terminal = attempt >= MAX_ATTEMPTS;
+        LocalDateTime retryAt = LocalDateTime.now().plusSeconds(Math.min(300L, 15L << Math.min(attempt - 1, 4)));
+        boolean recorded = commands.fail(command.getId(), workerId, terminal ? "FAILED" : "RETRYABLE", retryAt,
+            stage, code, requestId, traceId);
+        if (terminal && recorded) {
+            picks.failCompletion(command.getPickTaskId());
+        }
+        log.warn("pda_completion_worker_failure commandId={} taskId={} operationId={} stage={} code={} "
+                + "requestId={} traceId={} attempt={} terminal={}", command.getId(), command.getPickTaskId(),
+            command.getOperationId(), stage, code, requestId, traceId, attempt, terminal, cause);
+    }
+}

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

@@ -16,6 +16,7 @@ import com.baiyue.firstleg.dto.LoadingPickDtos.AssignRequest;
 import com.baiyue.firstleg.dto.LoadingPickDtos.ReviewRequest;
 import com.baiyue.firstleg.dto.LoadingPickDtos.CompleteLoadingRequest;
 import com.baiyue.firstleg.dto.LoadingPickDtos.Operation;
+import com.baiyue.firstleg.dto.LoadingPickDtos.CompletionCommand;
 import com.baiyue.firstleg.dto.LoadingPickDtos.BatchRequest;
 import com.baiyue.firstleg.dto.LoadingPickDtos.BatchItemRequest;
 import com.baiyue.firstleg.dto.LoadingPickDtos.InventoryResource;
@@ -26,6 +27,7 @@ import com.baiyue.firstleg.entity.LoadingPickAssignmentEntity;
 import com.baiyue.firstleg.entity.LoadingPickHistoryEntity;
 import com.baiyue.firstleg.entity.LoadingPickItemEntity;
 import com.baiyue.firstleg.entity.LoadingPickOperationEntity;
+import com.baiyue.firstleg.entity.LoadingCompletionCommandEntity;
 import com.baiyue.firstleg.entity.LoadingPickReviewEntity;
 import com.baiyue.firstleg.entity.LoadingPickTaskEntity;
 import com.baiyue.firstleg.entity.LoadingPlanItemEntity;
@@ -39,6 +41,7 @@ import com.baiyue.firstleg.mapper.InboundStatusHistoryMapper;
 import com.baiyue.firstleg.mapper.LoadingPickItemMapper.CartonLabelRow;
 import com.baiyue.firstleg.mapper.LoadingPickItemMapper;
 import com.baiyue.firstleg.repository.LoadingPickRepository;
+import com.baiyue.firstleg.repository.LoadingCompletionCommandRepository;
 import com.baiyue.firstleg.repository.SmartLoadingRepository;
 import com.baiyue.firstleg.oceanbooking.repository.OceanBookingRepository;
 import jakarta.annotation.Resource;
@@ -66,6 +69,7 @@ import java.util.Objects;
 import java.util.Set;
 import java.util.stream.Collectors;
 import java.time.LocalDateTime;
+import com.baomidou.mybatisplus.core.toolkit.IdWorker;
 
 /** 装柜中 PDA 拣货任务及明细同步。 */
 @Service
@@ -76,6 +80,8 @@ public class LoadingPickService {
     @Resource
     private LoadingPickRepository repository;
     @Resource
+    private LoadingCompletionCommandRepository completionCommands;
+    @Resource
     private SmartLoadingRepository smartLoadingRepository;
     @Resource
     private OceanBookingRepository oceanBookingRepository;
@@ -440,9 +446,31 @@ public class LoadingPickService {
         String operationId = operation(request.operationId());
         String remark = trim(request.remark());
         String requestHash = hash("COMPLETE|" + taskId + '|' + request.expectedTaskVersion() + '|' + remark);
-        Task replay = replayTaskOperation(taskId, operationId, requestHash, operatorId);
-        if (replay != null) return replay;
-        LoadingPickTaskEntity task = requiredLoadingTaskForUpdate(taskId);
+        LoadingCompletionCommandEntity replay = completionCommands.operation(operationId);
+        if (replay != null) {
+            requireSameCompletionCommand(replay, taskId, requestHash, operatorId);
+            if ("FAILED".equals(replay.getCommandStatus())) {
+                LoadingPickTaskEntity failed = requiredTaskForUpdate(taskId);
+                if (!"COMPLETION_FAILED".equals(failed.getTaskStatus())
+                    || !completionCommands.resume(replay.getId()) || repository.resumeCompletion(taskId) != 1) {
+                    throw conflict("装柜完成命令状态已变化，请刷新后重试");
+                }
+            }
+            return view(requiredTask(taskId), null);
+        }
+        LoadingPickTaskEntity task = requiredTaskForUpdate(taskId);
+        LoadingCompletionCommandEntity concurrentReplay = completionCommands.operation(operationId);
+        if (concurrentReplay != null) {
+            requireSameCompletionCommand(concurrentReplay, taskId, requestHash, operatorId);
+            return view(task, null);
+        }
+        LoadingCompletionCommandEntity active = completionCommands.active(taskId);
+        if (active != null) {
+            throw conflict("已有其他装柜完成命令正在处理");
+        }
+        if (!"LOADING".equals(task.getTaskStatus())) {
+            throw conflict("装柜任务正在完成或已经完成");
+        }
         if (!request.expectedTaskVersion().equals(task.getVersion())) {
             throw conflict("PDA任务版本已变化，请刷新后复核");
         }
@@ -454,40 +482,122 @@ public class LoadingPickService {
             .filter(owner -> owner.tenantId() != null && owner.inboundId() != null)
             .toList();
         if (discoveredOwners.isEmpty()) throw conflict("PDA任务未关联有效入库单");
-        LoadingPickItemMapper.InboundOwner evidenceOwner = discoveredOwners.get(0);
-        List<LoadingPickItemMapper.InboundOwner> owners = discoveredOwners.stream()
+        LoadingCompletionCommandEntity command = new LoadingCompletionCommandEntity();
+        command.setId(IdWorker.getId());
+        command.setPickTaskId(taskId);
+        command.setOperationId(operationId);
+        command.setRequestHash(requestHash);
+        command.setExpectedTaskVersion(request.expectedTaskVersion());
+        command.setRemark(remark);
+        command.setOperatorId(operatorId);
+        command.setOperatorName(trim(operatorName));
+        if (completionCommands.insert(command) != 1) {
+            LoadingCompletionCommandEntity concurrent = completionCommands.operation(operationId);
+            requireSameCompletionCommand(concurrent, taskId, requestHash, operatorId);
+            return view(requiredTask(taskId), null);
+        }
+        if (repository.acceptCompletion(taskId, task.getVersion()) != 1) {
+            throw conflict("PDA任务版本已变化，请刷新后复核");
+        }
+        return view(requiredTask(taskId), null);
+    }
+
+    /** Called only by the completion worker, after its durable lease has committed. */
+    public void processCompletion(LoadingCompletionCommandEntity command, String worker) {
+        long taskId = command.getPickTaskId();
+        LoadingPickTaskEntity task = requiredTask(taskId);
+        List<LoadingPickItemMapper.InboundOwner> owners = repository.inboundOwners(taskId).stream()
+            .filter(owner -> owner.tenantId() != null && owner.inboundId() != null)
             .sorted(Comparator.comparing(LoadingPickItemMapper.InboundOwner::tenantId)
                 .thenComparing(LoadingPickItemMapper.InboundOwner::inboundId))
             .toList();
-        for (LoadingPickItemMapper.InboundOwner owner : owners) {
-            LoadingInboundMapper.Snapshot snapshot = repository.inboundForUpdate(owner.tenantId(), owner.inboundId());
-            if (snapshot == null) throw conflict("PDA任务关联入库单不存在");
-            completionStage(taskId, "完成前业务门禁", () ->
-                loadingDeliveryService.requirePdaCompletionReady(owner.tenantId(), owner.inboundId(),
-                    evidenceOwner.inboundId()));
+        if (owners.isEmpty()) throw new CompletionProcessingException("OWNER_DISCOVERY", conflict("PDA任务未关联有效入库单"));
+        long evidenceInboundId = owners.get(0).inboundId();
+        for (int index = 0; index < owners.size(); index++) {
+            requireCompletionLease(command.getId(), worker);
+            LoadingPickItemMapper.InboundOwner owner = owners.get(index);
+            String checkpoint = completionCommands.checkpoint(command.getId(), owner.tenantId(), owner.inboundId());
+            if (!checkpointAtLeast(checkpoint, "READY")) {
+                runCompletionStage("OWNER_READY", () -> loadingDeliveryService.requirePdaCompletionReady(
+                    owner.tenantId(), owner.inboundId(), evidenceInboundId));
+                requireCompletionLease(command.getId(), worker);
+                completionCommands.checkpoint(command.getId(), owner.tenantId(), owner.inboundId(), index + 1, "READY");
+            }
+            if (!checkpointAtLeast(checkpoint, "SETTLED")) {
+                runCompletionStage("WAREHOUSE_SETTLEMENT", () -> settlementService.settleOwner(
+                    taskId, owner.inboundId(), command.getOperatorId(), command.getOperatorName()));
+                requireCompletionLease(command.getId(), worker);
+                completionCommands.checkpoint(command.getId(), owner.tenantId(), owner.inboundId(), index + 1, "SETTLED");
+            }
+            if (!checkpointAtLeast(checkpoint, "INBOUND_COMPLETED")) {
+                runCompletionStage("INBOUND_COMPLETE", () -> loadingDeliveryService.completeAfterPdaSettlement(
+                    owner.tenantId(), owner.inboundId(), evidenceInboundId, command.getOperatorId(),
+                    command.getOperatorName(), "PDA-" + command.getOperationId() + '-' + owner.inboundId(),
+                    command.getRemark()));
+                requireCompletionLease(command.getId(), worker);
+                completionCommands.checkpoint(command.getId(), owner.tenantId(), owner.inboundId(), index + 1,
+                    "INBOUND_COMPLETED");
+            }
         }
-        completionStage(taskId, "拆单与库位结算", () ->
-            settlementService.settle(taskId, operatorId, operatorName));
         if (!repository.allPlanItemsPicked(task.getLoadingPlanId())) {
-            throw conflict("柜次仍有未完成PDA拣货的箱数");
+            throw new CompletionProcessingException("PLAN_GATE", conflict("柜次仍有未完成PDA拣货的箱数"));
         }
-        if (repository.approve(taskId, operatorId, operatorName, remark) != 1) {
-            throw conflict("PDA任务已经复核");
+        if (repository.review(taskId) == null && repository.approve(taskId, command.getOperatorId(),
+            command.getOperatorName(), command.getRemark()) != 1) {
+            throw new CompletionProcessingException("TASK_REVIEW", conflict("PDA任务已经复核"));
         }
-        insertOperation(taskId, operationId, requestHash, "COMPLETE_LOADING", operatorId, operatorName);
-        completionStage(taskId, "装柜计划完成", () ->
-            loadingListService.finishFromPda(task.getLoadingPlanId(), taskId, operatorId, operatorName));
-        repository.syncInboundProjections(taskId, operatorId, operatorName);
-        for (LoadingPickItemMapper.InboundOwner owner : owners) {
-            completionStage(taskId, "入库单状态完成", () ->
-                loadingDeliveryService.completeAfterPdaSettlement(owner.tenantId(), owner.inboundId(),
-                    evidenceOwner.inboundId(), operatorId, operatorName,
-                    "PDA-" + operationId + '-' + owner.inboundId(), remark));
+        if (repository.operation(command.getOperationId()) == null) {
+            insertOperation(taskId, command.getOperationId(), command.getRequestHash(), "COMPLETE_LOADING",
+                command.getOperatorId(), command.getOperatorName());
         }
-        completionStage(taskId, "完成事件发布", () ->
-            loadingListService.publishCompletedEvent(task.getLoadingPlanId()));
-        completeTruckAfterCommit(task.getContainerBookingId(), taskId, operatorId, operatorName);
-        return view(requiredTask(taskId), null);
+        runCompletionStage("PLAN_COMPLETE", () -> loadingListService.finishFromPda(task.getLoadingPlanId(), taskId,
+            command.getOperatorId(), command.getOperatorName()));
+        repository.syncInboundProjections(taskId, command.getOperatorId(), command.getOperatorName());
+        runCompletionStage("COMPLETED_EVENT", () -> loadingListService.publishCompletedEvent(task.getLoadingPlanId()));
+        requireCompletionLease(command.getId(), worker);
+        LoadingPickTaskEntity current = requiredTask(taskId);
+        if (!"COMPLETED".equals(current.getTaskStatus())
+            && repository.completeAccepted(taskId, command.getOperatorId(), command.getOperatorName()) != 1) {
+            throw new CompletionProcessingException("TASK_FINALIZE", conflict("PDA任务完成状态已变化"));
+        }
+        if (!completionCommands.complete(command.getId(), worker)) {
+            throw new CompletionProcessingException("COMMAND_FINALIZE", conflict("装柜完成命令租约已变化"));
+        }
+        completeTruckAfterCommit(task.getContainerBookingId(), taskId, command.getOperatorId(), command.getOperatorName());
+    }
+
+    private void requireCompletionLease(long commandId, String worker) {
+        if (!completionCommands.renew(commandId, worker, LocalDateTime.now().plusMinutes(10))) {
+            throw new CompletionProcessingException("LEASE_LOST", conflict("装柜完成命令租约已变化"));
+        }
+    }
+
+    private static void requireSameCompletionCommand(LoadingCompletionCommandEntity command, long taskId,
+                                                     String requestHash, long operatorId) {
+        if (command == null || !Long.valueOf(taskId).equals(command.getPickTaskId())
+            || !Long.valueOf(operatorId).equals(command.getOperatorId())
+            || !requestHash.equals(command.getRequestHash())) {
+            throw conflict("operationId已用于不同装柜完成命令");
+        }
+    }
+
+    private static boolean checkpointAtLeast(String checkpoint, String target) {
+        List<String> stages = List.of("READY", "SETTLED", "INBOUND_COMPLETED");
+        return checkpoint != null && stages.indexOf(checkpoint) >= stages.indexOf(target);
+    }
+
+    private static void runCompletionStage(String stage, Runnable action) {
+        try {
+            action.run();
+        } catch (RuntimeException exception) {
+            throw new CompletionProcessingException(stage, exception);
+        }
+    }
+
+    public static final class CompletionProcessingException extends RuntimeException {
+        private final String stage;
+        CompletionProcessingException(String stage, RuntimeException cause) { super(cause); this.stage = stage; }
+        public String stage() { return stage; }
     }
 
     private void completionStage(long taskId, String stage, Runnable action) {
@@ -692,6 +802,13 @@ public class LoadingPickService {
     }
 
     private Task view(LoadingPickTaskEntity task, Long warehouseId) {
+        var booking = repository.activeBookingByTask(task.getId());
+        if (booking == null || !Objects.equals(task.getContainerBookingId(), booking.getId())) {
+            throw new BizException(ResultCode.NOT_FOUND, "PDA拣货任务不存在或柜次已非有效存续记录");
+        }
+        if (warehouseId != null && !Objects.equals(warehouseId, booking.getWarehouseId())) {
+            throw new BizException(ResultCode.NOT_FOUND, "PDA拣货任务不存在或不属于当前仓库");
+        }
         List<LoadingPickItemEntity> source = warehouseId == null
             ? repository.items(task.getId()) : repository.items(task.getId(), warehouseId);
         Map<Long, List<CartonLabelRow>> cartonsByItem = repository.cartonLabels(task.getId()).stream()
@@ -714,24 +831,28 @@ public class LoadingPickService {
         Review review = reviewed == null ? null : new Review(reviewed.getReviewStatus(),
             reviewed.getReviewedBy(), reviewed.getReviewedByName(), reviewed.getRemark(), reviewed.getVersion());
         String workflowStatus = "COMPLETED".equals(task.getTaskStatus()) ? "COMPLETED"
+            : "COMPLETING".equals(task.getTaskStatus()) ? "PROCESSING"
+            : "COMPLETION_FAILED".equals(task.getTaskStatus()) ? "COMPLETION_FAILED"
             : assigned == null ? "UNASSIGNED"
             : repository.allTaskItemsResolved(task.getId()) ? "AWAITING_REVIEW" : "IN_PROGRESS";
-        var booking = repository.activeBookingByTask(task.getId());
-        if (booking == null || !Objects.equals(task.getContainerBookingId(), booking.getId())) {
-            throw new BizException(ResultCode.NOT_FOUND, "PDA拣货任务不存在或柜次已非有效存续记录");
-        }
         var owners = repository.inboundOwners(task.getId());
         Long detailInboundId = source.isEmpty() ? null : source.get(0).getInboundId();
         var owner = detailInboundId == null ? null : owners.stream()
             .filter(value -> detailInboundId.equals(value.inboundId())).findFirst().orElse(null);
         var inbound = owner == null || owner.tenantId() == null || owner.inboundId() == null
             ? null : repository.inbound(owner.tenantId(), owner.inboundId());
+        LoadingCompletionCommandEntity command = "LOADING".equals(task.getTaskStatus())
+            ? null : completionCommands.latest(task.getId());
+        CompletionCommand completionCommand = command == null ? null : new CompletionCommand(command.getId(),
+            command.getOperationId(), command.getCommandStatus(), command.getExpectedTaskVersion(), command.getRemark(),
+            command.getAttemptCount(), command.getNextRetryAt(), command.getFailureStage(), command.getFailureCode(),
+            command.getFailureRequestId(), command.getFailureTraceId(), command.getCompletedAt());
         return new Task(task.getId(), task.getLoadingPlanId(), task.getContainerBookingId(),
             owner == null ? null : owner.inboundId(), inbound == null ? null : inbound.version(),
             inbound == null ? null : inbound.version(), task.getCabinetNo(),
             booking.getContainerNo(), booking.getSourceType(), booking.getRecordStatus(),
             booking.getMergedIntoContainerBookingId(), task.getTaskStatus(),
-            workflowStatus, task.getVersion(), assignment, review, items);
+            workflowStatus, task.getVersion(), assignment, review, completionCommand, items);
     }
 
     private List<BatchLine> validateBatchLines(LoadingPickTaskEntity task, BatchRequest request) {

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

@@ -18,6 +18,8 @@ import jakarta.annotation.Resource;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.Comparator;
@@ -32,6 +34,7 @@ import java.util.stream.Collectors;
 /** PDA 任务结算：按柜次生成实装/余量履约分片，并结算测量与库位资源。 */
 @Service
 public class PdaPickSettlementService {
+    private static final Logger log = LoggerFactory.getLogger(PdaPickSettlementService.class);
     private static final long PLATFORM_WAREHOUSE_RESOURCE_TENANT_ID = 0L;
     private static final String SOURCE_SERVICE = "firstleg-service";
 
@@ -82,6 +85,35 @@ public class PdaPickSettlementService {
         }
     }
 
+    /** Worker boundary: one inbound owner per transaction, so a cabinet never holds 15 owner locks at once. */
+    @Transactional
+    public void settleOwner(long taskId, long inboundId, long operatorId, String operatorName) {
+        if (!picks.allTaskItemsResolved(taskId)) {
+            throw conflict("PDA任务仍有未处理明细");
+        }
+        List<LoadingPickItemEntity> taskItems = picks.items(taskId);
+        Map<Long, Integer> pickedByInbound = pickedByInbound(taskItems);
+        Integer pickedPackageCount = pickedByInbound.get(inboundId);
+        if (pickedPackageCount == null) {
+            throw conflict("PDA任务不包含指定入库单");
+        }
+        PdaPickSettlementPlan.Resolution resolution = PdaPickSettlementPlan.resolve(taskItems);
+        for (Long planItemId : resolution.skippedPlanItemIds()) {
+            int changed = splits.removeSkippedPlanItem(planItemId, operatorId, operatorName);
+            if (changed != 0 && changed != 1) {
+                throw conflict("跳单明细已被并发修改，请刷新后重试");
+            }
+        }
+        boolean newlySplit = settleInbound(inboundId, pickedPackageCount, taskItems, taskId, operatorId, operatorName);
+        if (newlySplit) {
+            for (PdaPickSettlementPlan.SplitLine line : resolution.partialByInbound()
+                .getOrDefault(inboundId, List.of())) {
+                requireAffected(splits.convergePartialPlanItem(line.loadingPlanItemId(), operatorId, operatorName),
+                    "部分拣货装柜明细");
+            }
+        }
+    }
+
     private boolean settleInbound(long parentInboundId, int pickedPackageCount,
                                   List<LoadingPickItemEntity> taskItems, long taskId,
                                   long operatorId, String operatorName) {
@@ -379,7 +411,7 @@ public class PdaPickSettlementService {
 
     private static List<Long> releasedReservationIds(List<InboundPalletAllocationEntity> allocations) {
         return allocations.stream().filter(item -> remaining(item) == 0).map(PdaPickSettlementService::reservationId)
-            .sorted().toList();
+            .distinct().sorted().toList();
     }
 
     private static List<WarehousePalletLocationClient.SettlementTransferItem> remainingTransferItems(
@@ -422,18 +454,51 @@ public class PdaPickSettlementService {
                 operatorId, operatorName, "PDA-SETTLE-" + taskId + '-' + parentInboundId,
                 releasedReservationIds, remainingItems));
         } catch (FeignException exception) {
+            log.warn("PDA warehouse settlement transport failure taskId={} sourceInboundId={} operationId={} "
+                    + "httpStatus={} releasedReservationIds={} remainingReservations={}",
+                taskId, parentInboundId, "PDA-SETTLE-" + taskId + '-' + parentInboundId,
+                exception.status(), releasedReservationIds, settlementTransferFacts(remainingItems));
             throw new BizException(ResultCode.SERVICE_UNAVAILABLE, "仓储服务不可用，无法结算PDA库位资源");
         }
         if (response == null || !response.successful() || response.data() == null
             || response.data().released() == null || response.data().transferred() == null
             || !sameReleasedReservations(releasedReservationIds, response.data().released())
             || !sameTransferredResources(remainingItems, response.data().transferred())) {
+            log.warn("PDA warehouse settlement rejected taskId={} sourceInboundId={} operationId={} "
+                    + "responseCode={} requestId={} traceId={} expectedAction=RELEASE_OR_TRANSFER "
+                    + "releasedReservationIds={} remainingReservations={}",
+                taskId, parentInboundId, "PDA-SETTLE-" + taskId + '-' + parentInboundId,
+                response == null ? null : response.code(), response == null ? null : response.requestId(),
+                response == null ? null : response.traceId(), releasedReservationIds,
+                settlementTransferFacts(remainingItems));
             ResultCode code = response == null ? ResultCode.SERVICE_UNAVAILABLE : ResultCode.of(response.code());
-            throw new BizException(code, response == null || response.message() == null
-                ? "仓储服务原子结算PDA库位资源失败" : response.message());
+            throw new SettlementFailure(code, response == null || response.message() == null
+                ? "仓储服务原子结算PDA库位资源失败" : response.message(),
+                response == null ? null : response.requestId(), response == null ? null : response.traceId());
         }
     }
 
+    /** Carries only correlation facts needed by the durable command diagnosis record. */
+    public static final class SettlementFailure extends BizException {
+        private final String requestId;
+        private final String traceId;
+
+        SettlementFailure(ResultCode code, String message, String requestId, String traceId) {
+            super(code, message);
+            this.requestId = requestId;
+            this.traceId = traceId;
+        }
+
+        public String requestId() { return requestId; }
+        public String traceId() { return traceId; }
+    }
+
+    private static List<String> settlementTransferFacts(
+        List<WarehousePalletLocationClient.SettlementTransferItem> remainingItems) {
+        return remainingItems.stream().map(item -> item.reservationId() + ":" + item.remainingPackageCount())
+            .sorted().toList();
+    }
+
     private long existingSettlementTarget(long sourceInboundId, String idempotencyKey) {
         try {
             ApiResponse<Long> response = warehouse.settlementTarget(internalToken, sourceInboundId, idempotencyKey);