feat(lastleg): reserve pallets for container plans

deploy/sql/海外区域统一增量-MySQL8.sql

@@ -1990,6 +1990,19 @@ DELIMITER ;
 CALL assert_tail_order_commands_ready();
 DROP PROCEDURE IF EXISTS assert_tail_order_commands_ready;
 
+CREATE TABLE IF NOT EXISTS ll_container_plan_conversion_command (
+  id BIGINT NOT NULL COMMENT '柜级转换命令ID', source_container_booking_id VARCHAR(128) NOT NULL COMMENT '稳定来源柜预订ID',
+  idempotency_key VARCHAR(128) NOT NULL COMMENT '首次调用幂等键', request_hash CHAR(64) NOT NULL COMMENT '确认请求SHA-256摘要',
+  snapshot_version BIGINT NOT NULL COMMENT '确认的柜级快照版本', snapshot_hash CHAR(64) NOT NULL COMMENT '确认的柜级快照SHA-256',
+  warehouse_id BIGINT NOT NULL COMMENT '海外接柜仓ID', batch_key VARCHAR(160) NOT NULL COMMENT '海外仓托盘批次幂等键',
+  status VARCHAR(32) NOT NULL COMMENT 'PREPARED/APPLYING/LOCAL_COMMITTED/COMPLETED', preparation_json JSON NOT NULL COMMENT '稳定计划号与托盘业务键快照',
+  result_json JSON NULL COMMENT '本地计划提交结果', last_error VARCHAR(1000) NULL COMMENT '最近一次跨服务错误',
+  created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
+  PRIMARY KEY(id), UNIQUE KEY uk_ll_container_conversion_command(source_container_booking_id),
+  UNIQUE KEY uk_ll_container_conversion_batch(warehouse_id,batch_key), KEY idx_ll_container_conversion_status(status,updated_at)
+) ENGINE=InnoDB COMMENT='柜级入库计划转换Saga与幂等命令';
+ALTER TABLE ll_inbound_plan_pallet MODIFY COLUMN pallet_no VARCHAR(128) NOT NULL COMMENT 'Warehouse权威实物托盘码';
+
 SELECT 'PASS' AS migration_status, '海外尾程人工计划、暂扣与人工节点结构已就绪' AS message;
 
 -- 柜级入库计划跨客户批量托盘预留、确认与失败补偿。

lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/controller/InboundPlanController.java

@@ -16,6 +16,7 @@ import com.baiyue.lastleg.inboundplan.dto.UpdateFbaAppointmentRequest;
 import com.baiyue.lastleg.inboundplan.dto.UpdateInboundPlanRequest;
 import com.baiyue.lastleg.inboundplan.service.InboundFbaAppointmentService;
 import com.baiyue.lastleg.inboundplan.service.InboundPlanService;
+import com.baiyue.lastleg.inboundplan.service.ContainerInboundPlanConversionCoordinator;
 import com.baiyue.lastleg.security.OverseasPermissions;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.tags.Tag;
@@ -32,6 +33,7 @@ import org.springframework.web.server.ResponseStatusException;
 @Tag(name = "尾程入仓计划")
 public class InboundPlanController {
     @Resource private InboundPlanService service;
+    @Resource private ContainerInboundPlanConversionCoordinator containerConversion;
     @Resource private InboundFbaAppointmentService appointmentService;
 
     /** 按入仓计划明细分页查询可预约的入库订单行。 */
@@ -65,11 +67,12 @@ public class InboundPlanController {
     @Operation(summary = "按整柜批量生成剩余入库计划", description = "平台按 sourceContainerBookingId 锁定整柜订单；保留已有有效计划，只为剩余入仓码分别生成独立计划，整批原子且可幂等重放。Required authority: " + OverseasPermissions.INBOUND_ORDER_CREATE)
     public ApiResponse<java.util.List<InboundPlanResponse>> convertContainer(
         @AuthenticationPrincipal AuthenticatedUser user,
+        @RequestHeader("Idempotency-Key") String idempotencyKey,
         @Valid @RequestBody ConvertContainerInboundPlanRequest request) {
         if (user == null || user.tenantId() == null || user.tenantId() != 0L || !"PLATFORM".equals(user.appCode())) {
             throw new ResponseStatusException(HttpStatus.FORBIDDEN, "仅平台账号可按整柜生成入库计划");
         }
-        return ApiResponse.success(service.convertContainer(request.sourceContainerBookingId()));
+        return ApiResponse.success(containerConversion.confirm(request, idempotencyKey));
     }
 
     @GetMapping("/container-preview")

lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/dto/ConvertContainerInboundPlanRequest.java

@@ -1,10 +1,14 @@
 package com.baiyue.lastleg.inboundplan.dto;
 
 import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.Min;
+import jakarta.validation.constraints.Pattern;
 import jakarta.validation.constraints.Size;
 
 /** 平台按来源头程柜预订身份批量生成入库计划。 */
 public record ConvertContainerInboundPlanRequest(
-    @NotBlank @Size(max = 128) String sourceContainerBookingId
+    @NotBlank @Size(max = 128) String sourceContainerBookingId,
+    @Min(0) long snapshotVersion,
+    @NotBlank @Pattern(regexp = "^[0-9a-f]{64}$") String snapshotHash
 ) {
 }

lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/entity/ContainerPlanConversionCommand.java

@@ -0,0 +1,27 @@
+package com.baiyue.lastleg.inboundplan.entity;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.Data;
+
+import java.time.LocalDateTime;
+
+@Data
+@TableName("ll_container_plan_conversion_command")
+public class ContainerPlanConversionCommand {
+    @TableId(type = IdType.INPUT) private Long id;
+    private String sourceContainerBookingId;
+    private String idempotencyKey;
+    private String requestHash;
+    private Long snapshotVersion;
+    private String snapshotHash;
+    private Long warehouseId;
+    private String batchKey;
+    private String status;
+    private String preparationJson;
+    private String resultJson;
+    private String lastError;
+    private LocalDateTime createdAt;
+    private LocalDateTime updatedAt;
+}

lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/mapper/ContainerPlanConversionCommandMapper.java

@@ -0,0 +1,33 @@
+package com.baiyue.lastleg.inboundplan.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.baiyue.lastleg.inboundplan.entity.ContainerPlanConversionCommand;
+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;
+
+@Mapper
+public interface ContainerPlanConversionCommandMapper extends BaseMapper<ContainerPlanConversionCommand> {
+    @Select("SELECT * FROM ll_container_plan_conversion_command WHERE source_container_booking_id=#{bookingId} AND idempotency_key=#{key} LIMIT 1")
+    ContainerPlanConversionCommand find(@Param("bookingId") String bookingId, @Param("key") String key);
+
+    @Select("SELECT * FROM ll_container_plan_conversion_command WHERE source_container_booking_id=#{bookingId} LIMIT 1")
+    ContainerPlanConversionCommand findByBooking(@Param("bookingId") String bookingId);
+
+    @Update("UPDATE ll_container_plan_conversion_command SET status=#{status},result_json=COALESCE(#{resultJson},result_json),last_error=#{lastError},updated_at=#{updatedAt} WHERE id=#{id} AND status=#{expectedStatus}")
+    int updateState(@Param("id") long id, @Param("expectedStatus") String expectedStatus,
+                    @Param("status") String status,
+                    @Param("resultJson") String resultJson, @Param("lastError") String lastError,
+                    @Param("updatedAt") LocalDateTime updatedAt);
+
+    @Update("UPDATE ll_container_plan_conversion_command SET status=#{nextStatus},last_error=NULL,updated_at=#{updatedAt} WHERE id=#{id} AND status=#{expectedStatus}")
+    int claim(@Param("id") long id, @Param("expectedStatus") String expectedStatus,
+              @Param("nextStatus") String nextStatus, @Param("updatedAt") LocalDateTime updatedAt);
+
+    @Update("UPDATE ll_container_plan_conversion_command SET last_error=NULL,updated_at=#{updatedAt} WHERE id=#{id} AND status='APPLYING' AND updated_at=#{expectedUpdatedAt}")
+    int reclaimStale(@Param("id") long id, @Param("expectedUpdatedAt") LocalDateTime expectedUpdatedAt,
+                     @Param("updatedAt") LocalDateTime updatedAt);
+}

lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/gateway/HttpWarehouseResourceGateway.java

@@ -230,6 +230,39 @@ public class HttpWarehouseResourceGateway implements WarehouseResourceGateway {
         return response.data();
     }
 
+    @Override
+    public PalletBatchResponse ensurePalletBatch(PalletBatchEnsure request) {
+        return palletBatch("/internal/v1/warehouse-pallet-batches/ensure-reserved", request);
+    }
+
+    @Override
+    public PalletBatchResponse confirmPalletBatch(PalletBatchCommand request) {
+        return palletBatch("/internal/v1/warehouse-pallet-batches/confirm", request);
+    }
+
+    @Override
+    public PalletBatchResponse compensatePalletBatch(PalletBatchCompensation request) {
+        return palletBatch("/internal/v1/warehouse-pallet-batches/compensate", request);
+    }
+
+    private PalletBatchResponse palletBatch(String path, Object body) {
+        PalletBatchEnvelope response = client.post().uri(path)
+            .header("X-Internal-Service-Token", internalToken).body(body).retrieve()
+            .onStatus(HttpStatusCode::isError, (request, result) -> {
+                throw new ResponseStatusException(result.getStatusCode().is5xxServerError()
+                    ? HttpStatus.SERVICE_UNAVAILABLE : HttpStatus.BAD_REQUEST,
+                    "Warehouse pallet batch request failed");
+            }).body(PalletBatchEnvelope.class);
+        if (response == null || !"200".equals(response.code()) || response.data() == null) {
+            HttpStatus status = response != null && "A0409".equals(response.code()) ? HttpStatus.CONFLICT
+                : response != null && "A0404".equals(response.code()) ? HttpStatus.NOT_FOUND
+                : HttpStatus.SERVICE_UNAVAILABLE;
+            throw new ResponseStatusException(status, response == null || response.message() == null
+                ? "Warehouse pallet batch returned no usable response" : response.message());
+        }
+        return response.data();
+    }
+
     private record WarehousePalletResponse(String code, String message, WarehousePalletData data) {}
     private record WarehousePalletData(@JsonAlias("palletId") long id, String palletCode, long warehouseId, long locationId,
                                        String status, String fbaWarehouseCode, java.math.BigDecimal unitPrice, int version,
@@ -239,4 +272,5 @@ public class HttpWarehouseResourceGateway implements WarehouseResourceGateway {
     private record ReserveData(long lockId) {}
     private record WarehouseReferenceResponse(String code, String message, WarehouseReference data) {}
     private record WarehouseLocationResponse(String code, String message, WarehouseLocationReference data) {}
+    private record PalletBatchEnvelope(String code, String message, PalletBatchResponse data) {}
 }

lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/gateway/WarehouseResourceGateway.java

@@ -1,5 +1,7 @@
 package com.baiyue.lastleg.inboundplan.pda.gateway;
 
+import java.util.List;
+
 public interface WarehouseResourceGateway {
     WarehousePalletReference ensureDropshipPallet(String customerId, long warehouseId, String businessId);
     long reserve(String customerId, long warehouseId, long palletId, long locationId,
@@ -18,6 +20,15 @@ public interface WarehouseResourceGateway {
     default WarehouseReference requireActiveSharedWarehouse(long warehouseId) {
         throw new UnsupportedOperationException("Shared warehouse resolution is not configured");
     }
+    default PalletBatchResponse ensurePalletBatch(PalletBatchEnsure request) {
+        throw new UnsupportedOperationException("Warehouse pallet batch reservation is not configured");
+    }
+    default PalletBatchResponse confirmPalletBatch(PalletBatchCommand request) {
+        throw new UnsupportedOperationException("Warehouse pallet batch confirmation is not configured");
+    }
+    default PalletBatchResponse compensatePalletBatch(PalletBatchCompensation request) {
+        throw new UnsupportedOperationException("Warehouse pallet batch compensation is not configured");
+    }
 
     record WarehousePalletReference(long palletId, String palletNo, long warehouseId, long locationId,
                                     String status, String fbaWarehouseCode, java.math.BigDecimal unitPrice, int version) {
@@ -35,4 +46,14 @@ public interface WarehouseResourceGateway {
                               String status) { }
     record WarehouseLocationReference(long locationId, String locationCode, String locationType,
                                       java.math.BigDecimal maxLoadKg, String status) { }
+    record PalletBatchEnsure(long warehouseId, String batchKey, List<PalletBatchItemRequest> items) { }
+    record PalletBatchItemRequest(long customerId, String businessKey, String inboundPlanNo,
+                                  int sequenceNo, String palletType, String fbaWarehouseCode) { }
+    record PalletBatchCommand(long warehouseId, String batchKey, String commandKey) { }
+    record PalletBatchCompensation(long warehouseId, String batchKey, String commandKey, String reason) { }
+    record PalletBatchResponse(long batchId, String batchKey, String status, int attemptNo,
+                               boolean replayed, List<PalletBatchItem> items) { }
+    record PalletBatchItem(long customerId, String businessKey, String inboundPlanNo, int sequenceNo,
+                           long palletId, String palletCode, long locationId, long resourceLockId,
+                           String status) { }
 }

lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/service/ContainerInboundPlanConversionCoordinator.java

@@ -0,0 +1,289 @@
+package com.baiyue.lastleg.inboundplan.service;
+
+import com.baomidou.mybatisplus.core.toolkit.IdWorker;
+import com.baiyue.lastleg.inboundplan.dto.ContainerInboundPlanPreviewResponse;
+import com.baiyue.lastleg.inboundplan.dto.ConvertContainerInboundPlanRequest;
+import com.baiyue.lastleg.inboundplan.dto.InboundPlanResponse;
+import com.baiyue.lastleg.inboundplan.entity.ContainerPlanConversionCommand;
+import com.baiyue.lastleg.inboundplan.mapper.ContainerPlanConversionCommandMapper;
+import com.baiyue.lastleg.inboundplan.pda.gateway.WarehouseResourceGateway;
+import com.baiyue.lastleg.service.LastlegOrderNoGenerator;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import jakarta.annotation.Resource;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.http.HttpStatus;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.support.TransactionTemplate;
+import org.springframework.web.server.ResponseStatusException;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HexFormat;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+@Service
+public class ContainerInboundPlanConversionCoordinator {
+    private static final java.time.Duration APPLYING_LEASE = java.time.Duration.ofMinutes(2);
+    @Resource private InboundPlanService plans;
+    @Resource private WarehouseResourceGateway warehouse;
+    @Resource private ContainerPlanConversionCommandMapper commands;
+    @Resource private LastlegOrderNoGenerator numbers;
+    @Resource private ObjectMapper objectMapper;
+    @Resource private PlatformTransactionManager transactionManager;
+
+    public List<InboundPlanResponse> confirm(ConvertContainerInboundPlanRequest request, String idempotencyKey) {
+        String bookingId = required(request.sourceContainerBookingId(), 128, "来源柜预订 ID 不能为空");
+        String key = required(idempotencyKey, 128, "Idempotency-Key 必须为1至128个字符");
+        String requestHash = hash(List.of(bookingId, request.snapshotVersion(), request.snapshotHash()));
+        ContainerPlanConversionCommand resolved = commands.find(bookingId, key);
+        if (resolved == null) resolved = commands.findByBooking(bookingId);
+        if (resolved == null) resolved = createCommand(bookingId, key, request, requestHash);
+        final ContainerPlanConversionCommand command = resolved;
+        requireSameRequest(command, requestHash);
+        Preparation preparation = read(command.getPreparationJson(), Preparation.class);
+
+        if ("COMPLETED".equals(command.getStatus())) return result(command);
+        if ("APPLYING".equals(command.getStatus())) {
+            LocalDateTime now = LocalDateTime.now();
+            if (command.getUpdatedAt() == null || command.getUpdatedAt().plus(APPLYING_LEASE).isAfter(now)
+                || commands.reclaimStale(command.getId(), command.getUpdatedAt(), now) != 1) {
+                throw new ResponseStatusException(HttpStatus.CONFLICT, "该柜正在生成入库计划，请勿重复提交");
+            }
+            command.setUpdatedAt(now);
+        }
+        if (!"LOCAL_COMMITTED".equals(command.getStatus())) {
+            if (commands.claim(command.getId(), "PREPARED", "APPLYING", LocalDateTime.now()) != 1) {
+                ContainerPlanConversionCommand current = commands.findByBooking(bookingId);
+                if (current != null && "COMPLETED".equals(current.getStatus())) return result(current);
+                throw new ResponseStatusException(HttpStatus.CONFLICT, "该柜正在生成入库计划，请勿重复提交");
+            }
+            command.setStatus("APPLYING");
+        }
+        if (preparation.items().isEmpty()) {
+            try {
+                List<InboundPlanResponse> existing = inTransaction(() -> {
+                    List<InboundPlanResponse> converted = plans.convertContainerReserved(bookingId,
+                        request.snapshotVersion(), request.snapshotHash(), List.of());
+                    state(command.getId(), "APPLYING", "COMPLETED", write(converted), null);
+                    return converted;
+                });
+                command.setStatus("COMPLETED");
+                return existing;
+            } catch (RuntimeException failure) {
+                resetForRetry(command, failure);
+                throw failure;
+            }
+        }
+
+        WarehouseResourceGateway.PalletBatchEnsure ensure = new WarehouseResourceGateway.PalletBatchEnsure(
+            preparation.warehouseId(), command.getBatchKey(), preparation.items().stream().map(item ->
+                new WarehouseResourceGateway.PalletBatchItemRequest(Long.parseLong(item.customerId()), item.businessKey(),
+                    item.planNo(), item.globalSequenceNo(), "STANDARD", item.fbaWarehouseCode())).toList());
+        WarehouseResourceGateway.PalletBatchResponse reservation;
+        try {
+            reservation = warehouse.ensurePalletBatch(ensure);
+            requireBatch(reservation, command.getBatchKey(), preparation.items().size(), "RESERVED", "CONFIRMED");
+        } catch (RuntimeException failure) {
+            resetForRetry(command, failure);
+            throw failure;
+        }
+
+        List<InboundPlanResponse> localResult;
+        if ("LOCAL_COMMITTED".equals(command.getStatus())) {
+            localResult = result(command);
+        } else {
+            try {
+                List<InboundPlanService.PreparedPlan> preparedPlans = bind(preparation, reservation);
+                localResult = inTransaction(() -> {
+                    List<InboundPlanResponse> created = plans.convertContainerReserved(bookingId,
+                        request.snapshotVersion(), request.snapshotHash(), preparedPlans);
+                    state(command.getId(), "APPLYING", "LOCAL_COMMITTED", write(created), null);
+                    return created;
+                });
+                command.setStatus("LOCAL_COMMITTED");
+            } catch (RuntimeException failure) {
+                try {
+                    warehouse.compensatePalletBatch(new WarehouseResourceGateway.PalletBatchCompensation(
+                        preparation.warehouseId(), command.getBatchKey(), "COMPENSATE:" + command.getId(),
+                        "Lastleg local transaction did not commit"));
+                } catch (RuntimeException compensationFailure) {
+                    failure.addSuppressed(compensationFailure);
+                }
+                resetForRetry(command, failure);
+                throw failure;
+            }
+        }
+
+        WarehouseResourceGateway.PalletBatchResponse confirmed = warehouse.confirmPalletBatch(
+            new WarehouseResourceGateway.PalletBatchCommand(preparation.warehouseId(), command.getBatchKey(),
+                "CONFIRM:" + command.getId()));
+        requireBatch(confirmed, command.getBatchKey(), preparation.items().size(), "CONFIRMED");
+        complete(command, localResult);
+        return localResult;
+    }
+
+    private ContainerPlanConversionCommand createCommand(String bookingId, String key,
+                                                          ConvertContainerInboundPlanRequest request,
+                                                          String requestHash) {
+        ContainerInboundPlanPreviewResponse preview = plans.previewContainer(bookingId);
+        if (preview.snapshotVersion() != request.snapshotVersion()
+            || !Objects.equals(preview.snapshotHash(), request.snapshotHash()) || !preview.ready()) {
+            throw new ResponseStatusException(HttpStatus.CONFLICT, "柜级预览已变化或存在阻断，请刷新预览后重试");
+        }
+        long warehouseId;
+        try { warehouseId = Long.parseLong(preview.overseasWarehouseId()); }
+        catch (RuntimeException exception) {
+            throw new ResponseStatusException(HttpStatus.CONFLICT, "柜级接柜仓 ID 无效", exception);
+        }
+        long commandId = IdWorker.getId();
+        List<PreparedItem> items = new ArrayList<>();
+        int globalSequence = 0;
+        for (ContainerInboundPlanPreviewResponse.Order order : preview.groups().stream()
+            .flatMap(group -> group.orders().stream()).filter(item -> "READY".equals(item.readiness())).toList()) {
+            long planId = IdWorker.getId();
+            String planNo = numbers.next("INBOUND_PLAN", order.destinationWarehouseCode(), planId);
+            for (int localSequence = 1; localSequence <= order.suggestedPalletCount(); localSequence++) {
+                items.add(new PreparedItem(order.customerId(), Long.parseLong(order.orderId()), planId, planNo,
+                    localSequence, ++globalSequence, "CIP:" + commandId + ":" + globalSequence,
+                    "FBA".equals(order.destinationType()) ? order.destinationWarehouseCode() : null));
+            }
+        }
+        Preparation preparation = new Preparation(warehouseId, List.copyOf(items));
+        ContainerPlanConversionCommand command = new ContainerPlanConversionCommand();
+        command.setId(commandId); command.setSourceContainerBookingId(bookingId); command.setIdempotencyKey(key);
+        command.setRequestHash(requestHash); command.setSnapshotVersion(request.snapshotVersion());
+        command.setSnapshotHash(request.snapshotHash()); command.setWarehouseId(warehouseId);
+        command.setBatchKey("CIP:" + hash(bookingId) + ":" + commandId); command.setStatus("PREPARED");
+        command.setPreparationJson(write(preparation)); command.setCreatedAt(LocalDateTime.now());
+        command.setUpdatedAt(command.getCreatedAt());
+        try { inTransaction(() -> { commands.insert(command); return null; }); }
+        catch (DuplicateKeyException concurrent) {
+            ContainerPlanConversionCommand existing = commands.findByBooking(bookingId);
+            if (existing == null) throw concurrent;
+            return existing;
+        }
+        return command;
+    }
+
+    private List<InboundPlanService.PreparedPlan> bind(Preparation preparation,
+                                                        WarehouseResourceGateway.PalletBatchResponse response) {
+        Map<String, WarehouseResourceGateway.PalletBatchItem> byKey = new LinkedHashMap<>();
+        for (WarehouseResourceGateway.PalletBatchItem item : response.items()) {
+            if (item.businessKey() == null || byKey.putIfAbsent(item.businessKey(), item) != null) {
+                throw new ResponseStatusException(HttpStatus.CONFLICT, "海外仓托盘批次包含重复业务键");
+            }
+        }
+        if (response.items().stream().map(WarehouseResourceGateway.PalletBatchItem::palletId).distinct().count() != response.items().size()
+            || response.items().stream().map(WarehouseResourceGateway.PalletBatchItem::resourceLockId).distinct().count() != response.items().size()) {
+            throw new ResponseStatusException(HttpStatus.CONFLICT, "海外仓托盘批次包含重复托盘或资源锁");
+        }
+        return preparation.items().stream().collect(Collectors.groupingBy(PreparedItem::orderId)).values().stream()
+            .map(items -> {
+                PreparedItem first = items.get(0);
+                return new InboundPlanService.PreparedPlan(first.planId(), first.planNo(), first.customerId(),
+                    first.orderId(), items.stream().sorted(Comparator.comparingInt(PreparedItem::localSequenceNo))
+                        .map(item -> new InboundPlanService.PreparedPallet(item.localSequenceNo(), item.businessKey(),
+                            requiredReservation(byKey.get(item.businessKey()), item))).toList());
+            }).toList();
+    }
+
+    private WarehouseResourceGateway.PalletBatchItem requiredReservation(
+        WarehouseResourceGateway.PalletBatchItem value, PreparedItem expected) {
+        if (value == null || value.customerId() != Long.parseLong(expected.customerId())
+            || !Objects.equals(value.businessKey(), expected.businessKey())
+            || !Objects.equals(value.inboundPlanNo(), expected.planNo())
+            || value.sequenceNo() != expected.globalSequenceNo() || value.palletId() <= 0
+            || value.locationId() <= 0 || value.resourceLockId() <= 0
+            || value.palletCode() == null || value.palletCode().isBlank()) {
+            throw new ResponseStatusException(HttpStatus.CONFLICT, "海外仓托盘预留响应与计划快照不一致");
+        }
+        return value;
+    }
+
+    private void requireBatch(WarehouseResourceGateway.PalletBatchResponse response, String batchKey, int size,
+                              String... acceptedStatuses) {
+        if (response == null || !Objects.equals(batchKey, response.batchKey()) || response.items() == null
+            || response.items().size() != size || java.util.Arrays.stream(acceptedStatuses)
+                .noneMatch(status -> status.equals(response.status()))) {
+            throw new ResponseStatusException(HttpStatus.CONFLICT, "海外仓托盘批次响应不完整或状态不一致");
+        }
+    }
+
+    private void requireSameRequest(ContainerPlanConversionCommand command, String requestHash) {
+        if (!MessageDigest.isEqual(command.getRequestHash().getBytes(StandardCharsets.UTF_8),
+            requestHash.getBytes(StandardCharsets.UTF_8))) {
+            throw new ResponseStatusException(HttpStatus.CONFLICT, "同一幂等键不能用于不同柜级确认请求");
+        }
+    }
+
+    private List<InboundPlanResponse> result(ContainerPlanConversionCommand command) {
+        if (command.getResultJson() == null) {
+            throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE, "柜级转换结果尚未可用");
+        }
+        return read(command.getResultJson(), new TypeReference<List<InboundPlanResponse>>() { });
+    }
+
+    private void complete(ContainerPlanConversionCommand command, List<InboundPlanResponse> result) {
+        inTransaction(() -> { state(command.getId(), "LOCAL_COMMITTED", "COMPLETED", write(result), null); return null; });
+        command.setStatus("COMPLETED");
+    }
+
+    private void resetForRetry(ContainerPlanConversionCommand command, RuntimeException failure) {
+        if ("APPLYING".equals(command.getStatus())) {
+            state(command.getId(), "APPLYING", "PREPARED", null, safeMessage(failure));
+            command.setStatus("PREPARED");
+        }
+    }
+
+    private void state(long id, String expectedStatus, String status, String resultJson, String lastError) {
+        if (commands.updateState(id, expectedStatus, status, resultJson, lastError, LocalDateTime.now()) != 1) {
+            throw new IllegalStateException("无法更新柜级转换命令状态");
+        }
+    }
+
+    private <T> T inTransaction(java.util.function.Supplier<T> action) {
+        return new TransactionTemplate(transactionManager).execute(status -> action.get());
+    }
+
+    private String hash(Object value) {
+        try { return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(write(value).getBytes(StandardCharsets.UTF_8))); }
+        catch (NoSuchAlgorithmException exception) { throw new IllegalStateException(exception); }
+    }
+    private String write(Object value) {
+        try { return objectMapper.writeValueAsString(value); }
+        catch (JsonProcessingException exception) { throw new IllegalStateException("无法序列化柜级转换快照", exception); }
+    }
+    private <T> T read(String value, Class<T> type) {
+        try { return objectMapper.readValue(value, type); }
+        catch (JsonProcessingException exception) { throw new IllegalStateException("无法读取柜级转换快照", exception); }
+    }
+    private <T> T read(String value, TypeReference<T> type) {
+        try { return objectMapper.readValue(value, type); }
+        catch (JsonProcessingException exception) { throw new IllegalStateException("无法读取柜级转换结果", exception); }
+    }
+    private static String required(String value, int max, String message) {
+        String normalized = value == null || value.isBlank() ? null : value.trim();
+        if (normalized == null || normalized.length() > max) throw new ResponseStatusException(HttpStatus.BAD_REQUEST, message);
+        return normalized;
+    }
+    private static String safeMessage(Throwable value) {
+        String message = value.getMessage();
+        return message == null ? value.getClass().getSimpleName() : message.substring(0, Math.min(1000, message.length()));
+    }
+
+    public record Preparation(long warehouseId, List<PreparedItem> items) { }
+    public record PreparedItem(String customerId, long orderId, long planId, String planNo,
+                               int localSequenceNo, int globalSequenceNo, String businessKey,
+                               String fbaWarehouseCode) { }
+}

lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/service/InboundPlanService.java

@@ -231,6 +231,7 @@ public class InboundPlanService {
                 || row.packageCount() > 0 && row.volumeCbm().signum() == 0) blockers.add("CARGO_FACT_INVALID");
         }
         Integer suggested = suggestedPalletCount(row, blockers);
+        if (!planned && (suggested == null || suggested <= 0)) blockers.add("PALLET_COUNT_NOT_READY");
         String destinationType = destinationType(row);
         String destinationCode = "FBA".equals(destinationType) && trimToNull(row.fbaWarehouseCode()) != null
             ? row.fbaWarehouseCode().trim() : row.destinationWarehouseCode();
@@ -335,11 +336,23 @@ public class InboundPlanService {
                                            List<SourceOrderSnapshot> orders, List<String> sourcePalletNos,
                                            List<WholePalletInput> wholePallets, ManualInboundPlanRequest manual,
                                            WarehouseResourceGateway.WarehouseReference receivingWarehouse) {
+        return createPlan(customerId, selectedOrderId, orders, sourcePalletNos, wholePallets, manual,
+            receivingWarehouse, null, Map.of());
+    }
+
+    private InboundPlanResponse createPlan(String customerId, Long selectedOrderId,
+                                           List<SourceOrderSnapshot> orders, List<String> sourcePalletNos,
+                                           List<WholePalletInput> wholePallets, ManualInboundPlanRequest manual,
+                                           WarehouseResourceGateway.WarehouseReference receivingWarehouse,
+                                           PreparedPlan preparedPlan,
+                                           Map<Integer, WarehouseResourceGateway.PalletBatchItem> reservedPallets) {
         String containerNo = requiredContainerNo(orders.get(0).containerNo());
         LocalDateTime now = LocalDateTime.now();
         InboundPlan plan = new InboundPlan();
-        plan.setId(IdWorker.getId());
-        plan.setPlanNo(orderNoGenerator.next("INBOUND_PLAN", orders.get(0).destinationWarehouseCode(), plan.getId()));
+        plan.setId(preparedPlan == null ? IdWorker.getId() : preparedPlan.planId());
+        plan.setPlanNo(preparedPlan == null
+            ? orderNoGenerator.next("INBOUND_PLAN", orders.get(0).destinationWarehouseCode(), plan.getId())
+            : preparedPlan.planNo());
         plan.setCustomerId(customerId);
         plan.setSourceOrderId(selectedOrderId);
         plan.setContainerNo(containerNo);
@@ -384,14 +397,14 @@ public class InboundPlanService {
             palletCounts = Map.of(orders.get(0).orderId(), manual.palletCount());
         } else if (wholePallets == null) {
             Map<PalletAllocator.PalletDraft, InboundPlanPallet> persistedPallets = createPallets(plan, allocation,
-                orderById, sourcePalletNos == null ? List.of() : sourcePalletNos);
+                orderById, sourcePalletNos == null ? List.of() : sourcePalletNos, reservedPallets);
             createAllocations(plan, allocation, lineByOrder, persistedPallets);
             palletCounts = allocation.allocations().stream().collect(Collectors.groupingBy(
                 PalletAllocator.AllocationDraft::orderId, Collectors.collectingAndThen(Collectors.mapping(
                     item -> item.pallet().sequenceNo(), Collectors.toSet()), java.util.Set::size)));
         } else {
             SourceOrderSnapshot order = orders.get(0);
-            createWholePallets(plan, order, lineByOrder.get(order.orderId()), wholePallets);
+            createWholePallets(plan, order, lineByOrder.get(order.orderId()), wholePallets, reservedPallets);
             palletCounts = Map.of(order.orderId(), wholePallets.size());
         }
         for (SourceOrderSnapshot order : orders) {
@@ -417,10 +430,32 @@ public class InboundPlanService {
     /** 同一来源柜的每个尾程订单独立建计划；整个柜级批次共用一个事务。 */
     @Transactional
     public List<InboundPlanResponse> convertContainer(String sourceContainerBookingId) {
+        return convertContainerInternal(sourceContainerBookingId, null, null, Map.of());
+    }
+
+    @Transactional
+    public List<InboundPlanResponse> convertContainerReserved(String sourceContainerBookingId,
+                                                               long snapshotVersion, String snapshotHash,
+                                                               List<PreparedPlan> preparedPlans) {
+        Map<Long, PreparedPlan> prepared = preparedPlans.stream().collect(Collectors.toMap(
+            PreparedPlan::orderId, value -> value));
+        return convertContainerInternal(sourceContainerBookingId, snapshotVersion, snapshotHash, prepared);
+    }
+
+    private List<InboundPlanResponse> convertContainerInternal(String sourceContainerBookingId,
+                                                                Long snapshotVersion, String snapshotHash,
+                                                                Map<Long, PreparedPlan> preparedPlans) {
         String bookingId = trimToNull(sourceContainerBookingId);
         if (bookingId == null) throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "来源柜预订 ID 不能为空");
         List<InboundPlanOrderMapper.ContainerOrderRef> refs = orderMapper.lockContainerOrders(bookingId);
         if (refs.isEmpty()) throw new ResponseStatusException(HttpStatus.NOT_FOUND, "来源柜没有尾程订单");
+        if (snapshotVersion != null) {
+            ContainerInboundPlanPreviewResponse current = previewContainer(bookingId);
+            if (current.snapshotVersion() != snapshotVersion || !Objects.equals(current.snapshotHash(), snapshotHash)
+                || !current.ready()) {
+                throw new ResponseStatusException(HttpStatus.CONFLICT, "柜级预览已变化或存在阻断，请刷新预览后重试");
+            }
+        }
         String containerNo = null;
         for (InboundPlanOrderMapper.ContainerOrderRef ref : refs) {
             String current = requiredContainerNo(ref.containerNo());
@@ -481,8 +516,15 @@ public class InboundPlanService {
         for (int pendingIndex = 0; pendingIndex < pendingIndexes.size(); pendingIndex++) {
             int refIndex = pendingIndexes.get(pendingIndex);
             InboundPlanOrderMapper.ContainerOrderRef ref = refs.get(refIndex);
+            PreparedPlan prepared = preparedPlans.get(ref.orderId());
+            if (snapshotVersion != null && (prepared == null || !Objects.equals(prepared.customerId(), ref.customerId()))) {
+                throw new ResponseStatusException(HttpStatus.CONFLICT, "柜级托盘预留与待生成计划不一致");
+            }
+            Map<Integer, WarehouseResourceGateway.PalletBatchItem> reserved = prepared == null ? Map.of()
+                : prepared.pallets().stream().collect(Collectors.toMap(PreparedPallet::sequenceNo,
+                    PreparedPallet::reservation));
             created.put(refIndex, createPlan(ref.customerId(), ref.orderId(), snapshots.get(pendingIndex), List.of(),
-                wholePallets.get(pendingIndex), null, null));
+                wholePallets.get(pendingIndex), null, null, prepared, reserved));
         }
         List<InboundPlanResponse> result = new ArrayList<>(refs.size());
         for (int index = 0; index < refs.size(); index++) {
@@ -492,6 +534,11 @@ public class InboundPlanService {
         return List.copyOf(result);
     }
 
+    public record PreparedPlan(long planId, String planNo, String customerId, long orderId,
+                               List<PreparedPallet> pallets) { }
+    public record PreparedPallet(int sequenceNo, String businessKey,
+                                 WarehouseResourceGateway.PalletBatchItem reservation) { }
+
     private WholePalletInput wholePallet(DropshipPalletSnapshot pallet) {
         return new WholePalletInput(pallet.sourcePalletNo(), Math.toIntExact(pallet.packageCount()));
     }
@@ -569,7 +616,8 @@ public class InboundPlanService {
     }
 
     private void createWholePallets(InboundPlan plan, SourceOrderSnapshot order, InboundPlanLine line,
-                                    List<WholePalletInput> sourcePallets) {
+                                    List<WholePalletInput> sourcePallets,
+                                    Map<Integer, WarehouseResourceGateway.PalletBatchItem> reservedPallets) {
         BigDecimal remainingVolume = volume(order);
         int remainingPackages = order.packageCount();
         for (int index = 0; index < sourcePallets.size(); index++) {
@@ -586,6 +634,7 @@ public class InboundPlanService {
             pallet.setDestinationWarehouseName(order.destinationWarehouseName());
             pallet.setCapacityCbm(PalletAllocator.PALLET_CAPACITY_CBM); pallet.setAllocatedVolumeCbm(palletVolume);
             pallet.setSequenceNo(index + 1); pallet.setAssignmentStatus("UNASSIGNED");
+            applyReservedPallet(pallet, reservedPallets.get(index + 1));
             pallet.setActualVolumeCbm(BigDecimal.ZERO); pallet.setActualPackageCount(0); palletMapper.insert(pallet);
 
             InboundPlanPalletAllocation allocation = new InboundPlanPalletAllocation();
@@ -707,7 +756,7 @@ public class InboundPlanService {
 
     private Map<PalletAllocator.PalletDraft, InboundPlanPallet> createPallets(
         InboundPlan plan, PalletAllocator.Result allocation, Map<Long, SourceOrderSnapshot> orderById,
-        List<String> sourcePalletNos) {
+        List<String> sourcePalletNos, Map<Integer, WarehouseResourceGateway.PalletBatchItem> reservedPallets) {
         Map<String, SourceOrderSnapshot> warehouseSamples = orderById.values().stream()
             .collect(Collectors.toMap(this::warehouseKey, value -> value, (left, right) -> left));
         Map<PalletAllocator.PalletDraft, InboundPlanPallet> result = new HashMap<>();
@@ -729,6 +778,7 @@ public class InboundPlanService {
             pallet.setAllocatedVolumeCbm(draft.allocatedVolumeCbm());
             pallet.setSequenceNo(result.size() + 1);
             pallet.setAssignmentStatus("UNASSIGNED");
+            applyReservedPallet(pallet, reservedPallets.get(pallet.getSequenceNo()));
             pallet.setActualVolumeCbm(BigDecimal.ZERO);
             pallet.setActualPackageCount(0);
             palletMapper.insert(pallet);
@@ -737,6 +787,20 @@ public class InboundPlanService {
         return result;
     }
 
+    private void applyReservedPallet(InboundPlanPallet pallet,
+                                     WarehouseResourceGateway.PalletBatchItem reserved) {
+        if (reserved == null) return;
+        if (trimToNull(reserved.palletCode()) == null || reserved.palletCode().length() > 128) {
+            throw new ResponseStatusException(HttpStatus.CONFLICT, "海外仓返回的真实托盘码无效");
+        }
+        pallet.setPalletNo(reserved.palletCode());
+        pallet.setWarehousePalletId(reserved.palletId());
+        pallet.setWarehouseLocationId(reserved.locationId());
+        pallet.setWarehouseResourceLockId(reserved.resourceLockId());
+        pallet.setAssignmentStatus("RESERVED");
+        pallet.setLockedAt(LocalDateTime.now());
+    }
+
     private void createAllocations(InboundPlan plan, PalletAllocator.Result allocation,
                                    Map<Long, InboundPlanLine> lineByOrder,
                                    Map<PalletAllocator.PalletDraft, InboundPlanPallet> pallets) {

lastleg-service/src/main/resources/db/lastleg.sql

@@ -578,6 +578,27 @@ CREATE TABLE IF NOT EXISTS ll_inbound_plan (
   KEY idx_ll_inbound_plan_container (customer_id, container_no)
 ) ENGINE=InnoDB COMMENT='尾程入库计划表：每张来源入仓单生成一张入库计划，同柜计划共享集装箱号';
 
+CREATE TABLE IF NOT EXISTS ll_container_plan_conversion_command (
+  id BIGINT NOT NULL COMMENT '柜级转换命令ID',
+  source_container_booking_id VARCHAR(128) NOT NULL COMMENT '稳定来源柜预订ID',
+  idempotency_key VARCHAR(128) NOT NULL COMMENT '调用方幂等键',
+  request_hash CHAR(64) NOT NULL COMMENT '确认请求SHA-256摘要',
+  snapshot_version BIGINT NOT NULL COMMENT '确认的柜级快照版本',
+  snapshot_hash CHAR(64) NOT NULL COMMENT '确认的柜级快照SHA-256',
+  warehouse_id BIGINT NOT NULL COMMENT '海外接柜仓ID',
+  batch_key VARCHAR(160) NOT NULL COMMENT '海外仓托盘批次幂等键',
+  status VARCHAR(32) NOT NULL COMMENT 'PREPARED/APPLYING/LOCAL_COMMITTED/COMPLETED',
+  preparation_json JSON NOT NULL COMMENT '稳定计划号与托盘业务键快照',
+  result_json JSON NULL COMMENT '本地计划提交结果',
+  last_error VARCHAR(1000) NULL COMMENT '最近一次跨服务错误',
+  created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
+  updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
+  PRIMARY KEY(id),
+  UNIQUE KEY uk_ll_container_conversion_command(source_container_booking_id),
+  UNIQUE KEY uk_ll_container_conversion_batch(warehouse_id,batch_key),
+  KEY idx_ll_container_conversion_status(status,updated_at)
+) ENGINE=InnoDB COMMENT='柜级入库计划转换Saga与幂等命令';
+
 CREATE TABLE IF NOT EXISTS ll_inbound_plan_line (
   id BIGINT NOT NULL COMMENT '入库计划明细ID（雪花ID）',
   customer_id VARCHAR(128) NOT NULL COMMENT '客户账号标识',
@@ -616,7 +637,7 @@ CREATE TABLE IF NOT EXISTS ll_inbound_plan_pallet (
   id BIGINT NOT NULL COMMENT '计划托盘ID（雪花ID）',
   customer_id VARCHAR(128) NOT NULL COMMENT '客户账号标识',
   plan_id BIGINT NOT NULL COMMENT '入库计划ID',
-  pallet_no VARCHAR(64) NOT NULL COMMENT '当前托盘编号：分配前为计划占位码，分配后为海外仓实物托盘码',
+  pallet_no VARCHAR(128) NOT NULL COMMENT '海外仓资源服务返回的真实托盘码',
   source_pallet_no VARCHAR(128) NULL COMMENT '一件代发不可变来源托盘号快照',
   pallet_type VARCHAR(32) NULL COMMENT '人工计划托盘类型',
   destination_warehouse_id VARCHAR(128) NULL COMMENT '目的仓ID快照',
@@ -694,7 +715,7 @@ ALTER TABLE ll_inbound_plan_pallet COMMENT='尾程入库计划托盘表：按目
 ALTER TABLE ll_inbound_plan_pallet
   MODIFY COLUMN id BIGINT NOT NULL COMMENT '计划托盘ID（雪花ID）',
   MODIFY COLUMN plan_id BIGINT NOT NULL COMMENT '入库计划ID',
-  MODIFY COLUMN pallet_no VARCHAR(64) NOT NULL COMMENT '当前托盘编号：分配前为计划占位码，分配后为海外仓实物托盘码',
+  MODIFY COLUMN pallet_no VARCHAR(128) NOT NULL COMMENT '海外仓资源服务返回的真实托盘码',
   MODIFY COLUMN destination_warehouse_id VARCHAR(128) NULL COMMENT '目的仓ID快照',
   MODIFY COLUMN destination_warehouse_code VARCHAR(128) NULL COMMENT '目的仓编码快照',
   MODIFY COLUMN destination_warehouse_name VARCHAR(255) NULL COMMENT '目的仓名称快照',

lastleg-service/src/main/resources/openapi/overseas-business-v1.yaml

@@ -433,17 +433,21 @@ paths:
   /api/platform/v1/last-leg/inbound-plans/convert-container:
     post:
       summary: 按整柜批量生成剩余入库计划
-      description: 权限 platform:overseas-warehouse:inbound-order:create；使用稳定 sourceContainerBookingId 锁定整柜订单，并要求全柜尾程订单及既有有效计划的真实 containerNo 与接柜仓快照一致。已有有效计划保持不变，只为剩余尾程订单/入仓码分别生成独立计划；响应按柜内订单稳定顺序返回全部已有及新增计划。整批原子且幂等，剩余为0时直接返回已有结果；任何快照漂移均冲突且不部分写入。
+      description: 权限 platform:overseas-warehouse:inbound-order:create；调用方必须先取柜级预览并原样回传 snapshotVersion/snapshotHash。服务端以 sourceContainerBookingId 建立唯一 Saga 命令，先向海外仓资源服务按票、按托盘批量预留真实 palletId/palletCode/locationId/resourceLockId，再在一个本地事务中锁整柜并为每票生成独立计划，最后确认资源批次。本地任一票失败时整柜回滚并补偿资源；本地已提交但远端确认失败时重放只重试确认。同柜不同 Idempotency-Key 也不会产生第二批资源。
       security: [{ bearerAuth: [] }]
+      parameters:
+        - { name: Idempotency-Key, in: header, required: true, schema: { type: string, minLength: 1, maxLength: 128 } }
       requestBody:
         required: true
         content:
           application/json:
             schema:
               type: object
-              required: [sourceContainerBookingId]
+              required: [sourceContainerBookingId, snapshotVersion, snapshotHash]
               properties:
                 sourceContainerBookingId: { type: string, minLength: 1, maxLength: 128 }
+                snapshotVersion: { type: integer, format: int64, minimum: 0 }
+                snapshotHash: { type: string, pattern: '^[0-9a-f]{64}$' }
       responses: { '200': { description: 统一 ApiResponse 包装的入库计划数组 } }
   /api/platform/v1/last-leg/inbound-plans/container-preview:
     get:

lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/controller/InboundPlanControllerTest.java

@@ -8,6 +8,7 @@ import com.baiyue.lastleg.inboundplan.dto.ManualInboundPlanRequest;
 import com.baiyue.lastleg.inboundplan.dto.UpdateFbaAppointmentRequest;
 import com.baiyue.lastleg.inboundplan.service.InboundFbaAppointmentService;
 import com.baiyue.lastleg.inboundplan.service.InboundPlanService;
+import com.baiyue.lastleg.inboundplan.service.ContainerInboundPlanConversionCoordinator;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -30,6 +31,7 @@ import static org.mockito.Mockito.when;
 class InboundPlanControllerTest {
     @Mock InboundPlanService service;
     @Mock InboundFbaAppointmentService appointmentService;
+    @Mock ContainerInboundPlanConversionCoordinator containerConversion;
     private InboundPlanController controller;
 
     @BeforeEach
@@ -37,6 +39,7 @@ class InboundPlanControllerTest {
         controller = new InboundPlanController();
         ReflectionTestUtils.setField(controller, "service", service);
         ReflectionTestUtils.setField(controller, "appointmentService", appointmentService);
+        ReflectionTestUtils.setField(controller, "containerConversion", containerConversion);
     }
 
     @Test
@@ -79,19 +82,21 @@ class InboundPlanControllerTest {
 
     @Test
     void platformContainerConversionUsesStableBookingIdentity() {
-        controller.convertContainer(platform(), new ConvertContainerInboundPlanRequest(" BOOKING-1 "));
+        var request = new ConvertContainerInboundPlanRequest(" BOOKING-1 ", 3, "a".repeat(64));
+        controller.convertContainer(platform(), "convert-key", request);
 
-        verify(service).convertContainer(" BOOKING-1 ");
+        verify(containerConversion).confirm(request, "convert-key");
     }
 
     @Test
     void merchantCannotConvertAWholeContainer() {
         AuthenticatedUser customer = new AuthenticatedUser(8L, 100L, 100L, 9L, "customer", "MERCHANT", "WEB", Set.of());
 
-        assertThatThrownBy(() -> controller.convertContainer(customer,
-            new ConvertContainerInboundPlanRequest("BOOKING-1")))
+        assertThatThrownBy(() -> controller.convertContainer(customer, "convert-key",
+            new ConvertContainerInboundPlanRequest("BOOKING-1", 3, "a".repeat(64))))
             .isInstanceOf(ResponseStatusException.class);
-        verify(service, never()).convertContainer(org.mockito.ArgumentMatchers.any());
+        verify(containerConversion, never()).confirm(org.mockito.ArgumentMatchers.any(),
+            org.mockito.ArgumentMatchers.any());
     }
 
     private static AuthenticatedUser platform() {

lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/pda/gateway/HttpWarehouseResourceGatewayTest.java

@@ -8,6 +8,8 @@ import org.springframework.test.web.client.MockRestServiceServer;
 import org.springframework.web.client.RestClient;
 import org.springframework.web.server.ResponseStatusException;
 
+import java.util.List;
+
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.springframework.test.web.client.match.MockRestRequestMatchers.requestTo;
@@ -141,4 +143,38 @@ class HttpWarehouseResourceGatewayTest {
         server.verify();
     }
 
+    @Test
+    void ensurePalletBatchAcceptsJacksonStringEncodedLongs() {
+        server.expect(requestTo("http://warehouse/internal/v1/warehouse-pallet-batches/ensure-reserved"))
+            .andRespond(withSuccess("""
+                {"code":"200","data":{"batchId":"41","batchKey":"BATCH-1","status":"RESERVED",
+                "attemptNo":1,"replayed":false,"items":[{"customerId":"100","businessKey":"CIP:1:1",
+                "inboundPlanNo":"IP-21","sequenceNo":1,"palletId":"51","palletCode":"PLT-51",
+                "locationId":"61","resourceLockId":"71","status":"RESERVED"}]}}
+                """, MediaType.APPLICATION_JSON));
+
+        var result = gateway.ensurePalletBatch(new WarehouseResourceGateway.PalletBatchEnsure(31L, "BATCH-1",
+            List.of(new WarehouseResourceGateway.PalletBatchItemRequest(
+                100L, "CIP:1:1", "IP-21", 1, "STANDARD", "ONT8"))));
+
+        assertThat(result.batchId()).isEqualTo(41L);
+        assertThat(result.items().get(0).resourceLockId()).isEqualTo(71L);
+        server.verify();
+    }
+
+    @Test
+    void palletBatchMapsBusinessConflictEnvelopeEvenWithHttp200() {
+        server.expect(requestTo("http://warehouse/internal/v1/warehouse-pallet-batches/confirm"))
+            .andRespond(withSuccess("""
+                {"code":"A0409","message":"Batch request hash conflict","data":null}
+                """, MediaType.APPLICATION_JSON));
+
+        assertThatThrownBy(() -> gateway.confirmPalletBatch(
+            new WarehouseResourceGateway.PalletBatchCommand(31L, "BATCH-1", "CONFIRM:1")))
+            .isInstanceOf(ResponseStatusException.class)
+            .hasMessageContaining("409")
+            .hasMessageContaining("Batch request hash conflict");
+        server.verify();
+    }
+
 }

lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/service/ContainerInboundPlanConversionCoordinatorTest.java

@@ -0,0 +1,126 @@
+package com.baiyue.lastleg.inboundplan.service;
+
+import com.baiyue.lastleg.inboundplan.dto.ConvertContainerInboundPlanRequest;
+import com.baiyue.lastleg.inboundplan.entity.ContainerPlanConversionCommand;
+import com.baiyue.lastleg.inboundplan.mapper.ContainerPlanConversionCommandMapper;
+import com.baiyue.lastleg.inboundplan.pda.gateway.WarehouseResourceGateway;
+import com.fasterxml.jackson.databind.json.JsonMapper;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.web.server.ResponseStatusException;
+
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class ContainerInboundPlanConversionCoordinatorTest {
+    private static final String BOOKING_ID = "BOOKING-1";
+    private static final String SNAPSHOT_HASH = "a".repeat(64);
+
+    @Mock InboundPlanService plans;
+    @Mock WarehouseResourceGateway warehouse;
+    @Mock ContainerPlanConversionCommandMapper commands;
+
+    private ContainerInboundPlanConversionCoordinator coordinator;
+    private JsonMapper objectMapper;
+
+    @BeforeEach
+    void setUp() {
+        coordinator = new ContainerInboundPlanConversionCoordinator();
+        objectMapper = JsonMapper.builder().findAndAddModules().build();
+        ReflectionTestUtils.setField(coordinator, "plans", plans);
+        ReflectionTestUtils.setField(coordinator, "warehouse", warehouse);
+        ReflectionTestUtils.setField(coordinator, "commands", commands);
+        ReflectionTestUtils.setField(coordinator, "objectMapper", objectMapper);
+    }
+
+    @Test
+    void anotherIdempotencyKeyForCompletedCabinetReplaysWithoutCreatingAnotherBatch() throws Exception {
+        ConvertContainerInboundPlanRequest request = request();
+        ContainerPlanConversionCommand existing = command("FIRST-KEY", "COMPLETED", request);
+        existing.setResultJson("[]");
+        when(commands.find(BOOKING_ID, "SECOND-KEY")).thenReturn(null);
+        when(commands.findByBooking(BOOKING_ID)).thenReturn(existing);
+
+        assertEquals(List.of(), coordinator.confirm(request, "SECOND-KEY"));
+
+        verify(warehouse, never()).ensurePalletBatch(org.mockito.ArgumentMatchers.any());
+        verify(commands, never()).insert(org.mockito.ArgumentMatchers.any(ContainerPlanConversionCommand.class));
+    }
+
+    @Test
+    void concurrentCabinetClaimRejectsFollowerBeforeWarehouseReservation() throws Exception {
+        ConvertContainerInboundPlanRequest request = request();
+        ContainerPlanConversionCommand applying = command("FIRST-KEY", "APPLYING", request);
+        when(commands.find(BOOKING_ID, "SECOND-KEY")).thenReturn(null);
+        when(commands.findByBooking(BOOKING_ID)).thenReturn(applying);
+
+        ResponseStatusException error = assertThrows(ResponseStatusException.class,
+            () -> coordinator.confirm(request, "SECOND-KEY"));
+
+        assertEquals(409, error.getStatusCode().value());
+        verify(warehouse, never()).ensurePalletBatch(org.mockito.ArgumentMatchers.any());
+    }
+
+    @Test
+    void failedConditionalClaimStopsBeforeWarehouseReservation() throws Exception {
+        ConvertContainerInboundPlanRequest request = request();
+        ContainerPlanConversionCommand prepared = command("FIRST-KEY", "PREPARED", request);
+        when(commands.find(BOOKING_ID, "SECOND-KEY")).thenReturn(null);
+        when(commands.findByBooking(BOOKING_ID)).thenReturn(prepared);
+        when(commands.claim(org.mockito.ArgumentMatchers.eq(1L), org.mockito.ArgumentMatchers.eq("PREPARED"),
+            org.mockito.ArgumentMatchers.eq("APPLYING"), org.mockito.ArgumentMatchers.any())).thenReturn(0);
+
+        ResponseStatusException error = assertThrows(ResponseStatusException.class,
+            () -> coordinator.confirm(request, "SECOND-KEY"));
+
+        assertEquals(409, error.getStatusCode().value());
+        verify(warehouse, never()).ensurePalletBatch(org.mockito.ArgumentMatchers.any());
+    }
+
+    @Test
+    void reservationMustMatchEveryStableItemIdentity() throws Exception {
+        var expected = new ContainerInboundPlanConversionCoordinator.PreparedItem(
+            "100", 11L, 21L, "IP-21", 1, 1, "CIP:1:1", "ONT8");
+        var preparation = new ContainerInboundPlanConversionCoordinator.Preparation(31L, List.of(expected));
+        var response = new WarehouseResourceGateway.PalletBatchResponse(41L, "BATCH-1", "RESERVED", 1,
+            false, List.of(new WarehouseResourceGateway.PalletBatchItem(
+                100L, "CIP:1:1", "WRONG-PLAN", 1, 51L, "PLT-51", 61L, 71L, "RESERVED")));
+
+        ResponseStatusException error = assertThrows(ResponseStatusException.class,
+            () -> ReflectionTestUtils.invokeMethod(coordinator, "bind", preparation, response));
+
+        assertEquals(409, error.getStatusCode().value());
+    }
+
+    private ConvertContainerInboundPlanRequest request() {
+        return new ConvertContainerInboundPlanRequest(BOOKING_ID, 7L, SNAPSHOT_HASH);
+    }
+
+    private ContainerPlanConversionCommand command(String key, String status,
+                                                     ConvertContainerInboundPlanRequest request) throws Exception {
+        ContainerPlanConversionCommand command = new ContainerPlanConversionCommand();
+        command.setId(1L);
+        command.setSourceContainerBookingId(BOOKING_ID);
+        command.setIdempotencyKey(key);
+        command.setRequestHash(ReflectionTestUtils.invokeMethod(coordinator, "hash",
+            List.of(BOOKING_ID, request.snapshotVersion(), request.snapshotHash())));
+        command.setSnapshotVersion(request.snapshotVersion());
+        command.setSnapshotHash(request.snapshotHash());
+        command.setWarehouseId(31L);
+        command.setBatchKey("BATCH-1");
+        command.setStatus(status);
+        command.setPreparationJson(objectMapper.writeValueAsString(
+            new ContainerInboundPlanConversionCoordinator.Preparation(31L, List.of())));
+        return command;
+    }
+}

lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/service/InboundPlanServiceTest.java

@@ -424,6 +424,8 @@ class InboundPlanServiceTest {
     void containerConversionPrevalidatesEveryOrderBeforeWritingIndependentPlans() throws Exception {
         assertNotNull(InboundPlanService.class.getMethod("convertContainer", String.class)
             .getAnnotation(Transactional.class));
+        assertNotNull(InboundPlanService.class.getMethod("convertContainerReserved", String.class,
+            long.class, String.class, List.class).getAnnotation(Transactional.class));
         when(orderMapper.lockContainerOrders("BOOKING-1")).thenReturn(List.of(
             new InboundPlanOrderMapper.ContainerOrderRef("100", 1L, "PENDING_ASSIGNMENT", "MSCU6639870", "STANDARD"),
             new InboundPlanOrderMapper.ContainerOrderRef("200", 2L, "DESTINATION_ASSIGNED", "MSCU6639870", "STANDARD")));