feat(wcs): deliver device commands through outbox

2026-08-11 10:39:15 +08:00
parent f191d49319
commit 0d09eb93da
15 changed files with 941 additions and 10 deletions
--- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/CommandDeliveryRecorder.java
+++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/CommandDeliveryRecorder.java
@@ -0,0 +1,111 @@
+package com.ruoyi.wcs.application.device;
+
+import com.ruoyi.wcs.domain.execution.CommandOutboxEntry;
+import com.ruoyi.wcs.domain.execution.DeviceCommand;
+import com.ruoyi.wcs.domain.execution.DeviceCommandState;
+import com.ruoyi.wcs.port.out.CommandOutboxRepository;
+import com.ruoyi.wcs.port.out.DeviceCommandRepository;
+import com.ruoyi.wcs.port.out.RecoveryCaseOpener;
+import com.ruoyi.wcs.port.out.ResourceReservationRepository;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.Instant;
+import java.util.Collections;
+import java.util.Objects;
+
+public final class CommandDeliveryRecorder {
+    private final DeviceCommandRepository commandRepository;
+    private final CommandOutboxRepository outboxRepository;
+    private final RecoveryCaseOpener recoveryCaseOpener;
+    private final ResourceReservationRepository reservationRepository;
+
+    public CommandDeliveryRecorder(DeviceCommandRepository commandRepository,
+                                   CommandOutboxRepository outboxRepository,
+                                   RecoveryCaseOpener recoveryCaseOpener,
+                                   ResourceReservationRepository reservationRepository) {
+        this.commandRepository = Objects.requireNonNull(commandRepository,
+                "commandRepository");
+        this.outboxRepository = Objects.requireNonNull(outboxRepository,
+                "outboxRepository");
+        this.recoveryCaseOpener = Objects.requireNonNull(recoveryCaseOpener,
+                "recoveryCaseOpener");
+        this.reservationRepository = Objects.requireNonNull(reservationRepository,
+                "reservationRepository");
+    }
+
+    @Transactional
+    public OutboxDispatchResult.Outcome record(CommandOutboxEntry entry,
+                                               SendReceipt receipt,
+                                               Instant now) {
+        Objects.requireNonNull(entry, "entry");
+        Objects.requireNonNull(receipt, "receipt");
+        Objects.requireNonNull(now, "now");
+        DeviceCommand command = commandRepository.findByCommandId(
+                entry.getCommandId());
+        if (command == null || command.getState() != DeviceCommandState.PENDING) {
+            entry.markDead("COMMAND_NOT_PENDING", now);
+            outboxRepository.save(entry);
+            return OutboxDispatchResult.Outcome.DEAD;
+        }
+
+        switch (receipt.getStatus()) {
+            case NOT_SENT:
+                entry.reschedule(detail(receipt, "NOT_SENT"),
+                        now.plusSeconds(backoffSeconds(entry.getAttemptCount())));
+                outboxRepository.save(entry);
+                return OutboxDispatchResult.Outcome.RETRY_SCHEDULED;
+            case ACCEPTED:
+                command.markSent(receipt.getReceiptId(), now);
+                entry.markDelivered(now);
+                commandRepository.save(command);
+                outboxRepository.save(entry);
+                return OutboxDispatchResult.Outcome.SENT;
+            case ACCEPTED_UNKNOWN:
+                String reason = detail(receipt, "WRITE_RESULT_AMBIGUOUS");
+                command.markSent(receipt.getReceiptId(), now);
+                command.markUnknown(reason, now);
+                entry.markDelivered(now);
+                reservationRepository.protectForUnknown(command.getCommandKey(),
+                        receipt.getReceiptId());
+                recoveryCaseOpener.open(command.getDeviceId(),
+                        command.getCommandKey(), reason,
+                        Collections.singletonList(receipt.getReceiptId()));
+                commandRepository.save(command);
+                outboxRepository.save(entry);
+                return OutboxDispatchResult.Outcome.ISOLATED_UNKNOWN;
+            case REJECTED:
+                command.markSent(receipt.getReceiptId(), now);
+                command.reject(detail(receipt, "DEVICE_REJECTED"), now);
+                entry.markDelivered(now);
+                commandRepository.save(command);
+                outboxRepository.save(entry);
+                return OutboxDispatchResult.Outcome.REJECTED;
+            default:
+                throw new IllegalStateException("unsupported send status");
+        }
+    }
+
+    DeviceCommand findPendingCommand(String commandId) {
+        DeviceCommand command = commandRepository.findByCommandId(commandId);
+        return command != null && command.getState() == DeviceCommandState.PENDING
+                ? command : null;
+    }
+
+    @Transactional
+    public OutboxDispatchResult.Outcome discardInvalid(
+            CommandOutboxEntry entry, Instant now) {
+        entry.markDead("COMMAND_NOT_PENDING", now);
+        outboxRepository.save(entry);
+        return OutboxDispatchResult.Outcome.DEAD;
+    }
+
+    private static long backoffSeconds(int attemptCount) {
+        int exponent = Math.min(Math.max(attemptCount, 1), 8);
+        return Math.min(300L, 1L << exponent);
+    }
+
+    private static String detail(SendReceipt receipt, String fallback) {
+        String detail = receipt.getDetail();
+        return detail == null || detail.trim().isEmpty() ? fallback : detail.trim();
+    }
+}
--- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/CommandOutboxDispatcher.java
+++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/CommandOutboxDispatcher.java
@@ -0,0 +1,58 @@
+package com.ruoyi.wcs.application.device;
+
+import com.ruoyi.wcs.domain.execution.CommandOutboxEntry;
+import com.ruoyi.wcs.domain.execution.DeviceCommand;
+import com.ruoyi.wcs.port.out.CommandOutboxRepository;
+import com.ruoyi.wcs.port.out.DeviceCommandGateway;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Objects;
+
+public final class CommandOutboxDispatcher {
+    private final CommandOutboxRepository outboxRepository;
+    private final DeviceCommandGateway commandGateway;
+    private final CommandDeliveryRecorder recorder;
+
+    public CommandOutboxDispatcher(CommandOutboxRepository outboxRepository,
+                                   DeviceCommandGateway commandGateway,
+                                   CommandDeliveryRecorder recorder) {
+        this.outboxRepository = Objects.requireNonNull(outboxRepository,
+                "outboxRepository");
+        this.commandGateway = Objects.requireNonNull(commandGateway,
+                "commandGateway");
+        this.recorder = Objects.requireNonNull(recorder, "recorder");
+    }
+
+    public OutboxDispatchResult dispatchDue(int limit, Instant now) {
+        if (limit <= 0) {
+            throw new IllegalArgumentException("limit must be positive");
+        }
+        Objects.requireNonNull(now, "now");
+        List<CommandOutboxEntry> entries = outboxRepository.claimDue(limit, now);
+        OutboxDispatchResult result = new OutboxDispatchResult();
+        for (CommandOutboxEntry entry : entries) {
+            DeviceCommand command = recorder.findPendingCommand(entry.getCommandId());
+            if (command == null) {
+                result.record(recorder.discardInvalid(entry, now));
+                continue;
+            }
+            SendReceipt receipt;
+            try {
+                receipt = commandGateway.send(DeviceCommandEnvelope.from(command,
+                        entry.getAttemptCount(), entry.getTopologyVersion(),
+                        entry.getFencingToken()));
+                if (receipt == null) {
+                    throw new IllegalStateException("gateway returned null receipt");
+                }
+            } catch (RuntimeException ambiguous) {
+                receipt = SendReceipt.acceptedUnknown(
+                        "AMBIGUOUS:" + entry.getCommandId() + ":"
+                                + entry.getAttemptCount(),
+                        "GATEWAY_EXCEPTION_AFTER_SEND_ATTEMPT");
+            }
+            result.record(recorder.record(entry, receipt, now));
+        }
+        return result;
+    }
+}
--- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/DeviceCommandEnvelope.java
+++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/DeviceCommandEnvelope.java
@@ -33,9 +33,16 @@ public final class DeviceCommandEnvelope {
    public static DeviceCommandEnvelope from(DeviceCommand command,
                                             long topologyVersion,
                                             long fencingToken) {
+        return from(command, command.getAttempt(), topologyVersion, fencingToken);
+    }
+
+    public static DeviceCommandEnvelope from(DeviceCommand command,
+                                             int deliveryAttempt,
+                                             long topologyVersion,
+                                             long fencingToken) {
        return new DeviceCommandEnvelope(command.getCommandId(),
                command.getCommandKey(), command.getDeviceId(), command.getType(),
-                command.getPayload(), command.getAttempt(), topologyVersion,
+                command.getPayload(), deliveryAttempt, topologyVersion,
                fencingToken);
    }

--- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/DeviceExecutionService.java
+++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/DeviceExecutionService.java
@@ -0,0 +1,93 @@
+package com.ruoyi.wcs.application.device;
+
+import com.ruoyi.wcs.application.dispatch.GrantedDispatch;
+import com.ruoyi.wcs.domain.execution.CommandOutboxEntry;
+import com.ruoyi.wcs.domain.execution.DeviceCommand;
+import com.ruoyi.wcs.domain.execution.DeviceCommandType;
+import com.ruoyi.wcs.domain.shared.DomainRuleViolation;
+import com.ruoyi.wcs.domain.shared.IdGenerator;
+import com.ruoyi.wcs.port.out.CommandOutboxRepository;
+import com.ruoyi.wcs.port.out.DeviceCommandRepository;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.Instant;
+import java.util.Objects;
+import java.util.function.Supplier;
+
+public final class DeviceExecutionService {
+    private final DeviceCommandRepository commandRepository;
+    private final CommandOutboxRepository outboxRepository;
+    private final IdGenerator idGenerator;
+    private final Supplier<Instant> clock;
+
+    public DeviceExecutionService(DeviceCommandRepository commandRepository,
+                                  CommandOutboxRepository outboxRepository,
+                                  IdGenerator idGenerator,
+                                  Supplier<Instant> clock) {
+        this.commandRepository = Objects.requireNonNull(commandRepository,
+                "commandRepository");
+        this.outboxRepository = Objects.requireNonNull(outboxRepository,
+                "outboxRepository");
+        this.idGenerator = Objects.requireNonNull(idGenerator, "idGenerator");
+        this.clock = Objects.requireNonNull(clock, "clock");
+    }
+
+    @Transactional
+    public DeviceCommand createCommand(GrantedDispatch grant) {
+        Objects.requireNonNull(grant, "grant");
+        String commandKey = "DISPATCH:" + requireText(grant.getOrderId(), "orderId")
+                + ":" + requireText(grant.getStepId(), "stepId");
+        String payload = encode(grant);
+        DeviceCommand existing = commandRepository.findByCommandKey(commandKey);
+        if (existing != null) {
+            if (!existing.getDeviceId().equals(grant.getVehicleId())
+                    || !existing.getPayload().equals(payload)) {
+                throw new DomainRuleViolation(
+                        "command key reused with a different dispatch grant");
+            }
+            return existing;
+        }
+
+        Instant now = Objects.requireNonNull(clock.get(), "clock returned null");
+        DeviceCommand command = DeviceCommand.pending(
+                requireText(idGenerator.nextId(), "generated commandId"),
+                commandKey, grant.getVehicleId(), DeviceCommandType.MOVE_ROUTE,
+                payload, 1, now);
+        CommandOutboxEntry entry = CommandOutboxEntry.pending(
+                command.getCommandId(), command.getCommandKey(), payload, now,
+                grant.getTopologyVersion(), grant.getFencingToken());
+        commandRepository.save(command);
+        outboxRepository.saveIfAbsent(entry);
+        return command;
+    }
+
+    private static String encode(GrantedDispatch grant) {
+        StringBuilder route = new StringBuilder();
+        for (String node : grant.getProtectedRoute().getNodeIds()) {
+            if (route.length() > 0) {
+                route.append('>');
+            }
+            route.append(requireSafe(node, "route node"));
+        }
+        return "orderId=" + requireSafe(grant.getOrderId(), "orderId")
+                + "|stepId=" + requireSafe(grant.getStepId(), "stepId")
+                + "|vehicleId=" + requireSafe(grant.getVehicleId(), "vehicleId")
+                + "|route=" + route;
+    }
+
+    private static String requireSafe(String value, String field) {
+        String text = requireText(value, field);
+        if (text.indexOf('|') >= 0 || text.indexOf('>') >= 0
+                || text.indexOf('=') >= 0) {
+            throw new IllegalArgumentException(field + " contains reserved delimiter");
+        }
+        return text;
+    }
+
+    private static String requireText(String value, String field) {
+        if (value == null || value.trim().isEmpty()) {
+            throw new IllegalArgumentException(field + " must not be blank");
+        }
+        return value.trim();
+    }
+}
--- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/OutboxDispatchResult.java
+++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/OutboxDispatchResult.java
@@ -0,0 +1,35 @@
+package com.ruoyi.wcs.application.device;
+
+import java.util.EnumMap;
+import java.util.Map;
+
+public final class OutboxDispatchResult {
+    public enum Outcome {
+        SENT,
+        RETRY_SCHEDULED,
+        ISOLATED_UNKNOWN,
+        REJECTED,
+        DEAD
+    }
+
+    private final Map<Outcome, Integer> counts =
+            new EnumMap<Outcome, Integer>(Outcome.class);
+
+    void record(Outcome outcome) {
+        Integer count = counts.get(outcome);
+        counts.put(outcome, count == null ? 1 : count + 1);
+    }
+
+    public int count(Outcome outcome) {
+        Integer count = counts.get(outcome);
+        return count == null ? 0 : count;
+    }
+
+    public int getTotal() {
+        int total = 0;
+        for (Integer count : counts.values()) {
+            total += count;
+        }
+        return total;
+    }
+}
--- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/SendReceipt.java
+++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/SendReceipt.java
@@ -16,7 +16,8 @@ public final class SendReceipt {
        this.status = java.util.Objects.requireNonNull(status, "status");
        this.receiptId = receiptId;
        this.detail = detail;
-        if ((status == Status.ACCEPTED || status == Status.ACCEPTED_UNKNOWN)
+        if ((status == Status.ACCEPTED || status == Status.ACCEPTED_UNKNOWN
+                || status == Status.REJECTED)
                && isBlank(receiptId)) {
            throw new IllegalArgumentException(
                    "accepted send receipt must have a receiptId");
@@ -35,8 +36,8 @@ public final class SendReceipt {
        return new SendReceipt(Status.NOT_SENT, null, detail);
    }

-    public static SendReceipt rejected(String detail) {
-        return new SendReceipt(Status.REJECTED, null, detail);
+    public static SendReceipt rejected(String receiptId, String detail) {
+        return new SendReceipt(Status.REJECTED, receiptId, detail);
    }

    public Status getStatus() { return status; }
--- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/WmsNotificationOutboxDispatcher.java
+++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/WmsNotificationOutboxDispatcher.java
@@ -0,0 +1,77 @@
+package com.ruoyi.wcs.application.device;
+
+import com.ruoyi.wcs.domain.execution.WmsNotificationOutboxEntry;
+import com.ruoyi.wcs.port.out.WmsNotificationGateway;
+import com.ruoyi.wcs.port.out.WmsNotificationOutboxRepository;
+import com.ruoyi.wcs.port.out.WmsNotificationReceipt;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Objects;
+
+public final class WmsNotificationOutboxDispatcher {
+    private final WmsNotificationOutboxRepository repository;
+    private final WmsNotificationGateway gateway;
+
+    public WmsNotificationOutboxDispatcher(
+            WmsNotificationOutboxRepository repository,
+            WmsNotificationGateway gateway) {
+        this.repository = Objects.requireNonNull(repository, "repository");
+        this.gateway = Objects.requireNonNull(gateway, "gateway");
+    }
+
+    public OutboxDispatchResult dispatchDue(int limit, Instant now) {
+        if (limit <= 0) {
+            throw new IllegalArgumentException("limit must be positive");
+        }
+        Objects.requireNonNull(now, "now");
+        List<WmsNotificationOutboxEntry> entries = repository.claimDue(limit, now);
+        OutboxDispatchResult result = new OutboxDispatchResult();
+        for (WmsNotificationOutboxEntry entry : entries) {
+            WmsNotificationReceipt receipt;
+            try {
+                receipt = gateway.notify(entry.getNotification());
+                if (receipt == null) {
+                    throw new IllegalStateException("WMS gateway returned null receipt");
+                }
+            } catch (RuntimeException transportFailure) {
+                receipt = new WmsNotificationReceipt(
+                        WmsNotificationReceipt.Status.RETRYABLE_FAILURE,
+                        "WMS_TRANSPORT_EXCEPTION");
+            }
+            switch (receipt.getStatus()) {
+                case DELIVERED:
+                    entry.markDelivered(nonBlank(receipt.getDetail(),
+                            "DELIVERED:" + entry.getEventId()), now);
+                    repository.save(entry);
+                    result.record(OutboxDispatchResult.Outcome.SENT);
+                    break;
+                case RETRYABLE_FAILURE:
+                    entry.reschedule(nonBlank(receipt.getDetail(),
+                                    "WMS_RETRYABLE_FAILURE"),
+                            now.plusSeconds(backoffSeconds(entry.getAttemptCount())));
+                    repository.save(entry);
+                    result.record(OutboxDispatchResult.Outcome.RETRY_SCHEDULED);
+                    break;
+                case PERMANENT_FAILURE:
+                    entry.markDead(nonBlank(receipt.getDetail(),
+                            "WMS_PERMANENT_FAILURE"), now);
+                    repository.save(entry);
+                    result.record(OutboxDispatchResult.Outcome.DEAD);
+                    break;
+                default:
+                    throw new IllegalStateException("unsupported WMS receipt");
+            }
+        }
+        return result;
+    }
+
+    private static long backoffSeconds(int attemptCount) {
+        int exponent = Math.min(Math.max(attemptCount, 1), 8);
+        return Math.min(300L, 1L << exponent);
+    }
+
+    private static String nonBlank(String value, String fallback) {
+        return value == null || value.trim().isEmpty() ? fallback : value.trim();
+    }
+}
--- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/WmsNotificationOutboxProjector.java
+++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/WmsNotificationOutboxProjector.java
@@ -0,0 +1,25 @@
+package com.ruoyi.wcs.application.device;
+
+import com.ruoyi.wcs.domain.execution.WmsNotificationOutboxEntry;
+import com.ruoyi.wcs.port.out.WmsNotification;
+import com.ruoyi.wcs.port.out.WmsNotificationOutboxRepository;
+
+import java.time.Instant;
+import java.util.Objects;
+import org.springframework.transaction.annotation.Transactional;
+
+public final class WmsNotificationOutboxProjector {
+    private final WmsNotificationOutboxRepository repository;
+
+    public WmsNotificationOutboxProjector(
+            WmsNotificationOutboxRepository repository) {
+        this.repository = Objects.requireNonNull(repository, "repository");
+    }
+
+    @Transactional
+    public void project(WmsNotification notification, Instant nextAttemptAt) {
+        Objects.requireNonNull(notification, "notification");
+        repository.saveIfAbsent(WmsNotificationOutboxEntry.pending(
+                notification.getEventId(), notification, nextAttemptAt));
+    }
+}
--- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/CommandOutboxEntry.java
+++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/CommandOutboxEntry.java
@@ -7,23 +7,73 @@ public final class CommandOutboxEntry {
    private final String commandId;
    private final String commandKey;
    private final String payload;
+    private final long topologyVersion;
+    private final long fencingToken;
    private OutboxDeliveryState state;
    private int attemptCount;
    private Instant nextAttemptAt;
    private String lastError;
+    private Instant deliveredAt;

    private CommandOutboxEntry(String commandId, String commandKey,
-                               String payload, Instant nextAttemptAt) {
+                               String payload, Instant nextAttemptAt,
+                               long topologyVersion, long fencingToken) {
        this.commandId = requireText(commandId, "commandId");
        this.commandKey = requireText(commandKey, "commandKey");
        this.payload = requireText(payload, "payload");
+        if (topologyVersion <= 0L || fencingToken <= 0L) {
+            throw new IllegalArgumentException("outbox versions are invalid");
+        }
+        this.topologyVersion = topologyVersion;
+        this.fencingToken = fencingToken;
        this.nextAttemptAt = Objects.requireNonNull(nextAttemptAt, "nextAttemptAt");
        this.state = OutboxDeliveryState.PENDING;
    }

    public static CommandOutboxEntry pending(String commandId, String commandKey,
                                             String payload, Instant nextAttemptAt) {
-        return new CommandOutboxEntry(commandId, commandKey, payload, nextAttemptAt);
+        return pending(commandId, commandKey, payload, nextAttemptAt, 1L, 1L);
+    }
+
+    public static CommandOutboxEntry pending(String commandId, String commandKey,
+                                             String payload, Instant nextAttemptAt,
+                                             long topologyVersion,
+                                             long fencingToken) {
+        return new CommandOutboxEntry(commandId, commandKey, payload,
+                nextAttemptAt, topologyVersion, fencingToken);
+    }
+
+    public boolean claim(Instant now) {
+        Instant claimTime = Objects.requireNonNull(now, "now");
+        if (state != OutboxDeliveryState.PENDING
+                || nextAttemptAt.isAfter(claimTime)) {
+            return false;
+        }
+        state = OutboxDeliveryState.IN_FLIGHT;
+        attemptCount++;
+        return true;
+    }
+
+    public void markDelivered(Instant at) {
+        requireInFlight();
+        deliveredAt = Objects.requireNonNull(at, "at");
+        lastError = null;
+        state = OutboxDeliveryState.DELIVERED;
+    }
+
+    public void reschedule(String error, Instant nextAttemptAt) {
+        requireInFlight();
+        lastError = requireText(error, "error");
+        this.nextAttemptAt = Objects.requireNonNull(nextAttemptAt,
+                "nextAttemptAt");
+        state = OutboxDeliveryState.PENDING;
+    }
+
+    public void markDead(String error, Instant at) {
+        requireInFlight();
+        lastError = requireText(error, "error");
+        deliveredAt = Objects.requireNonNull(at, "at");
+        state = OutboxDeliveryState.DEAD;
    }

    public String getCommandId() {
@@ -54,6 +104,17 @@ public final class CommandOutboxEntry {
        return lastError;
    }

+    public long getTopologyVersion() { return topologyVersion; }
+    public long getFencingToken() { return fencingToken; }
+    public Instant getDeliveredAt() { return deliveredAt; }
+
+    private void requireInFlight() {
+        if (state != OutboxDeliveryState.IN_FLIGHT) {
+            throw new com.ruoyi.wcs.domain.shared.DomainRuleViolation(
+                    "outbox entry must be IN_FLIGHT");
+        }
+    }
+
    private static String requireText(String value, String field) {
        if (value == null || value.trim().isEmpty()) {
            throw new IllegalArgumentException(field + " must not be blank");
--- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/OutboxDeliveryState.java
+++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/OutboxDeliveryState.java
@@ -2,6 +2,7 @@ package com.ruoyi.wcs.domain.execution;

 public enum OutboxDeliveryState {
    PENDING,
+    IN_FLIGHT,
    DELIVERED,
    DEAD
 }
--- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/WmsNotificationOutboxEntry.java
+++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/WmsNotificationOutboxEntry.java
@@ -8,8 +8,12 @@ import java.util.Objects;
 public final class WmsNotificationOutboxEntry {
    private final String eventId;
    private final WmsNotification notification;
-    private final Instant nextAttemptAt;
-    private final OutboxDeliveryState state;
+    private Instant nextAttemptAt;
+    private OutboxDeliveryState state;
+    private int attemptCount;
+    private String lastError;
+    private String externalReceipt;
+    private Instant deliveredAt;

    private WmsNotificationOutboxEntry(String eventId,
                                       WmsNotification notification,
@@ -29,6 +33,40 @@ public final class WmsNotificationOutboxEntry {
        return new WmsNotificationOutboxEntry(eventId, notification, nextAttemptAt);
    }

+    public boolean claim(Instant now) {
+        Instant claimTime = Objects.requireNonNull(now, "now");
+        if (state != OutboxDeliveryState.PENDING
+                || nextAttemptAt.isAfter(claimTime)) {
+            return false;
+        }
+        state = OutboxDeliveryState.IN_FLIGHT;
+        attemptCount++;
+        return true;
+    }
+
+    public void markDelivered(String receipt, Instant at) {
+        requireInFlight();
+        externalReceipt = requireText(receipt, "externalReceipt");
+        deliveredAt = Objects.requireNonNull(at, "at");
+        lastError = null;
+        state = OutboxDeliveryState.DELIVERED;
+    }
+
+    public void reschedule(String error, Instant nextAttemptAt) {
+        requireInFlight();
+        lastError = requireText(error, "error");
+        this.nextAttemptAt = Objects.requireNonNull(nextAttemptAt,
+                "nextAttemptAt");
+        state = OutboxDeliveryState.PENDING;
+    }
+
+    public void markDead(String error, Instant at) {
+        requireInFlight();
+        lastError = requireText(error, "error");
+        deliveredAt = Objects.requireNonNull(at, "at");
+        state = OutboxDeliveryState.DEAD;
+    }
+
    public String getEventId() {
        return eventId;
    }
@@ -45,6 +83,18 @@ public final class WmsNotificationOutboxEntry {
        return state;
    }

+    public int getAttemptCount() { return attemptCount; }
+    public String getLastError() { return lastError; }
+    public String getExternalReceipt() { return externalReceipt; }
+    public Instant getDeliveredAt() { return deliveredAt; }
+
+    private void requireInFlight() {
+        if (state != OutboxDeliveryState.IN_FLIGHT) {
+            throw new com.ruoyi.wcs.domain.shared.DomainRuleViolation(
+                    "WMS outbox entry must be IN_FLIGHT");
+        }
+    }
+
    private static String requireText(String value, String field) {
        if (value == null || value.trim().isEmpty()) {
            throw new IllegalArgumentException(field + " must not be blank");
--- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/CommandOutboxRepository.java
+++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/CommandOutboxRepository.java
@@ -8,5 +8,7 @@ import java.util.List;
 public interface CommandOutboxRepository {
    void saveIfAbsent(CommandOutboxEntry entry);

-    List<CommandOutboxEntry> findDue(int limit, Instant now);
+    void save(CommandOutboxEntry entry);
+
+    List<CommandOutboxEntry> claimDue(int limit, Instant now);
 }
--- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/WmsNotificationOutboxRepository.java
+++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/WmsNotificationOutboxRepository.java
@@ -8,5 +8,7 @@ import java.util.List;
 public interface WmsNotificationOutboxRepository {
    void saveIfAbsent(WmsNotificationOutboxEntry entry);

-    List<WmsNotificationOutboxEntry> findDue(int limit, Instant now);
+    void save(WmsNotificationOutboxEntry entry);
+
+    List<WmsNotificationOutboxEntry> claimDue(int limit, Instant now);
 }
--- a/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/device/CommandOutboxDispatcherTest.java
+++ b/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/device/CommandOutboxDispatcherTest.java
@@ -0,0 +1,278 @@
+package com.ruoyi.wcs.application.device;
+
+import com.ruoyi.wcs.application.dispatch.GrantedDispatch;
+import com.ruoyi.wcs.domain.execution.CommandOutboxEntry;
+import com.ruoyi.wcs.domain.execution.DeviceCommand;
+import com.ruoyi.wcs.domain.execution.DeviceCommandState;
+import com.ruoyi.wcs.domain.execution.OutboxDeliveryState;
+import com.ruoyi.wcs.domain.shared.IdGenerator;
+import com.ruoyi.wcs.domain.traffic.RoutePlan;
+import com.ruoyi.wcs.port.out.CommandOutboxRepository;
+import com.ruoyi.wcs.port.out.DeviceCommandGateway;
+import com.ruoyi.wcs.port.out.DeviceCommandRepository;
+import com.ruoyi.wcs.port.out.RecoveryCaseOpener;
+import com.ruoyi.wcs.port.out.ResourceReservationRepository;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class CommandOutboxDispatcherTest {
+    private FakeCommandRepository commands;
+    private FakeCommandOutboxRepository outbox;
+    private FakeGateway gateway;
+    private FakeIsolation isolation;
+    private DeviceExecutionService service;
+    private CommandOutboxDispatcher dispatcher;
+
+    @BeforeEach
+    void setUp() {
+        commands = new FakeCommandRepository();
+        outbox = new FakeCommandOutboxRepository();
+        gateway = new FakeGateway();
+        isolation = new FakeIsolation();
+        IdGenerator ids = new IdGenerator() {
+            @Override
+            public String nextId() {
+                return "CMD-1";
+            }
+        };
+        service = new DeviceExecutionService(commands, outbox, ids,
+                new java.util.function.Supplier<Instant>() {
+                    @Override
+                    public Instant get() {
+                        return at(0);
+                    }
+                });
+        CommandDeliveryRecorder recorder = new CommandDeliveryRecorder(
+                commands, outbox, isolation, isolation);
+        dispatcher = new CommandOutboxDispatcher(outbox, gateway, recorder);
+    }
+
+    @Test
+    void persistsCommandAndOutboxBeforeCallingTheGateway() {
+        DeviceCommand created = service.createCommand(grantedDispatch());
+
+        assertNotNull(commands.findByCommandId(created.getCommandId()));
+        assertNotNull(outbox.byCommandId(created.getCommandId()));
+        assertEquals(0, gateway.sendCount);
+    }
+
+    @Test
+    void acceptedSendIsNotMarkedCompleted() {
+        service.createCommand(grantedDispatch());
+        gateway.receipt = SendReceipt.accepted("PLC-WRITE-1");
+
+        dispatcher.dispatchDue(10, at(0));
+
+        assertEquals(DeviceCommandState.SENT,
+                commands.findByCommandId("CMD-1").getState());
+        assertEquals(OutboxDeliveryState.DELIVERED,
+                outbox.byCommandId("CMD-1").getState());
+    }
+
+    @Test
+    void acceptedUnknownIsIsolatedAndNeverAutomaticallyRetried() {
+        service.createCommand(grantedDispatch());
+        gateway.receipt = SendReceipt.acceptedUnknown(
+                "PLC-WRITE-UNKNOWN-1", "WRITE_RESULT_AMBIGUOUS");
+
+        dispatcher.dispatchDue(10, at(0));
+        dispatcher.dispatchDue(10, at(100));
+
+        assertEquals(DeviceCommandState.UNKNOWN,
+                commands.findByCommandId("CMD-1").getState());
+        assertEquals(1, gateway.sendCount);
+        assertEquals(Collections.singletonList("WRITE_RESULT_AMBIGUOUS"),
+                isolation.reasons);
+        assertEquals(1, isolation.protectCount);
+    }
+
+    @Test
+    void onlyNotSentOutcomeIsAutomaticallyRetried() {
+        service.createCommand(grantedDispatch());
+        gateway.receipt = SendReceipt.notSent("SOCKET_NOT_OPEN");
+
+        dispatcher.dispatchDue(10, at(0));
+        dispatcher.dispatchDue(10, at(1));
+
+        assertEquals(DeviceCommandState.PENDING,
+                commands.findByCommandId("CMD-1").getState());
+        assertEquals(1, gateway.sendCount);
+        assertEquals(OutboxDeliveryState.PENDING,
+                outbox.byCommandId("CMD-1").getState());
+        assertTrue(outbox.byCommandId("CMD-1").getNextAttemptAt()
+                .isAfter(at(1)));
+    }
+
+    @Test
+    void retryKeepsCommandKeyButIncrementsTheDeliveryAttempt() {
+        service.createCommand(grantedDispatch());
+        gateway.receipts.add(SendReceipt.notSent("SOCKET_NOT_OPEN"));
+        gateway.receipts.add(SendReceipt.accepted("PLC-WRITE-2"));
+
+        dispatcher.dispatchDue(10, at(0));
+        dispatcher.dispatchDue(10, at(2));
+
+        assertEquals(Arrays.asList(1, 2), gateway.deliveryAttempts);
+        assertEquals(DeviceCommandState.SENT,
+                commands.findByCommandId("CMD-1").getState());
+        assertEquals("DISPATCH:ORDER-1:STEP-1",
+                commands.findByCommandId("CMD-1").getCommandKey());
+    }
+
+    @Test
+    void unexpectedGatewayExceptionIsAmbiguousAndNotRetried() {
+        service.createCommand(grantedDispatch());
+        gateway.failure = new IllegalStateException("connection reset after write");
+
+        dispatcher.dispatchDue(10, at(0));
+        dispatcher.dispatchDue(10, at(100));
+
+        assertEquals(DeviceCommandState.UNKNOWN,
+                commands.findByCommandId("CMD-1").getState());
+        assertEquals(1, gateway.sendCount);
+        assertEquals(OutboxDeliveryState.DELIVERED,
+                outbox.byCommandId("CMD-1").getState());
+    }
+
+    @Test
+    void staleOutboxCannotSendAfterTheCommandAlreadyLeftPending() {
+        DeviceCommand command = service.createCommand(grantedDispatch());
+        command.markSent("ALREADY-SENT", at(0));
+        commands.save(command);
+
+        dispatcher.dispatchDue(10, at(0));
+
+        assertEquals(0, gateway.sendCount);
+        assertEquals(OutboxDeliveryState.DEAD,
+                outbox.byCommandId("CMD-1").getState());
+    }
+
+    private static GrantedDispatch grantedDispatch() {
+        RoutePlan route = new RoutePlan(Arrays.asList("A", "B"),
+                Arrays.asList("NODE:A", "EDGE:A:B", "NODE:B"),
+                1, 0, 7L);
+        return new GrantedDispatch("ORDER-1", "STEP-1", "VEH-1", route,
+                Collections.singletonList("RES-1"), 7L, 11L);
+    }
+
+    private static Instant at(long seconds) {
+        return Instant.EPOCH.plusSeconds(seconds);
+    }
+
+    private static final class FakeCommandRepository
+            implements DeviceCommandRepository {
+        private final Map<String, DeviceCommand> values =
+                new LinkedHashMap<String, DeviceCommand>();
+
+        @Override
+        public DeviceCommand findByCommandKey(String commandKey) {
+            for (DeviceCommand command : values.values()) {
+                if (command.getCommandKey().equals(commandKey)) {
+                    return command;
+                }
+            }
+            return null;
+        }
+
+        @Override
+        public DeviceCommand findByCommandId(String commandId) {
+            return values.get(commandId);
+        }
+
+        @Override
+        public void save(DeviceCommand command) {
+            values.put(command.getCommandId(), command);
+        }
+    }
+
+    private static final class FakeCommandOutboxRepository
+            implements CommandOutboxRepository {
+        private final Map<String, CommandOutboxEntry> values =
+                new LinkedHashMap<String, CommandOutboxEntry>();
+
+        @Override
+        public void saveIfAbsent(CommandOutboxEntry entry) {
+            if (!values.containsKey(entry.getCommandId())) {
+                values.put(entry.getCommandId(), entry);
+            }
+        }
+
+        @Override
+        public void save(CommandOutboxEntry entry) {
+            values.put(entry.getCommandId(), entry);
+        }
+
+        @Override
+        public List<CommandOutboxEntry> claimDue(int limit, Instant now) {
+            List<CommandOutboxEntry> result =
+                    new ArrayList<CommandOutboxEntry>();
+            for (CommandOutboxEntry entry : values.values()) {
+                if (result.size() >= limit) {
+                    break;
+                }
+                if (entry.claim(now)) {
+                    result.add(entry);
+                }
+            }
+            return result;
+        }
+
+        CommandOutboxEntry byCommandId(String commandId) {
+            return values.get(commandId);
+        }
+    }
+
+    private static final class FakeGateway implements DeviceCommandGateway {
+        private SendReceipt receipt = SendReceipt.accepted("PLC-WRITE-1");
+        private RuntimeException failure;
+        private int sendCount;
+        private final List<SendReceipt> receipts = new ArrayList<SendReceipt>();
+        private final List<Integer> deliveryAttempts = new ArrayList<Integer>();
+
+        @Override
+        public SendReceipt send(DeviceCommandEnvelope envelope) {
+            sendCount++;
+            deliveryAttempts.add(envelope.getAttempt());
+            if (failure != null) {
+                throw failure;
+            }
+            return receipts.isEmpty() ? receipt : receipts.remove(0);
+        }
+    }
+
+    private static final class FakeIsolation implements RecoveryCaseOpener,
+            ResourceReservationRepository {
+        private final List<String> reasons = new ArrayList<String>();
+        private int protectCount;
+
+        @Override
+        public void open(String deviceId, String commandKey, String reason,
+                         List<String> evidenceIds) {
+            reasons.add(reason);
+        }
+
+        @Override
+        public void releasePassedResources(String commandKey,
+                                           List<String> resourceKeys,
+                                           String releaseEvidenceId,
+                                           long fencingToken) {
+        }
+
+        @Override
+        public void protectForUnknown(String commandKey, String evidenceId) {
+            protectCount++;
+        }
+    }
+}
--- a/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/device/WmsNotificationOutboxDispatcherTest.java
+++ b/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/device/WmsNotificationOutboxDispatcherTest.java
@@ -0,0 +1,130 @@
+package com.ruoyi.wcs.application.device;
+
+import com.ruoyi.wcs.domain.execution.OutboxDeliveryState;
+import com.ruoyi.wcs.domain.execution.WmsNotificationOutboxEntry;
+import com.ruoyi.wcs.port.out.WmsNotification;
+import com.ruoyi.wcs.port.out.WmsNotificationGateway;
+import com.ruoyi.wcs.port.out.WmsNotificationOutboxRepository;
+import com.ruoyi.wcs.port.out.WmsNotificationReceipt;
+import org.junit.jupiter.api.Test;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class WmsNotificationOutboxDispatcherTest {
+
+    @Test
+    void duplicateBusinessEventCreatesOneWmsOutboxEntry() {
+        FakeWmsOutbox repository = new FakeWmsOutbox();
+        WmsNotificationOutboxProjector projector =
+                new WmsNotificationOutboxProjector(repository);
+        WmsNotification notification = notification("EVENT-1");
+
+        projector.project(notification, at(0));
+        projector.project(notification, at(0));
+
+        assertEquals(1, repository.values.size());
+    }
+
+    @Test
+    void transportFailureRetriesWithoutChangingTheBusinessEventKey() {
+        FakeWmsOutbox repository = new FakeWmsOutbox();
+        repository.saveIfAbsent(WmsNotificationOutboxEntry.pending(
+                "EVENT-1", notification("EVENT-1"), at(0)));
+        FakeWmsGateway gateway = new FakeWmsGateway();
+        gateway.receipt = new WmsNotificationReceipt(
+                WmsNotificationReceipt.Status.RETRYABLE_FAILURE, "HTTP_503");
+        WmsNotificationOutboxDispatcher dispatcher =
+                new WmsNotificationOutboxDispatcher(repository, gateway);
+
+        dispatcher.dispatchDue(10, at(0));
+        dispatcher.dispatchDue(10, at(1));
+
+        WmsNotificationOutboxEntry entry = repository.values.get("EVENT-1");
+        assertEquals("EVENT-1", entry.getEventId());
+        assertEquals(1, gateway.calls);
+        assertEquals(OutboxDeliveryState.PENDING, entry.getState());
+    }
+
+    @Test
+    void deliveredCallbackIsNeverSentAgain() {
+        FakeWmsOutbox repository = new FakeWmsOutbox();
+        repository.saveIfAbsent(WmsNotificationOutboxEntry.pending(
+                "EVENT-1", notification("EVENT-1"), at(0)));
+        FakeWmsGateway gateway = new FakeWmsGateway();
+        gateway.receipt = new WmsNotificationReceipt(
+                WmsNotificationReceipt.Status.DELIVERED, "WMS-RECEIPT-1");
+        WmsNotificationOutboxDispatcher dispatcher =
+                new WmsNotificationOutboxDispatcher(repository, gateway);
+
+        dispatcher.dispatchDue(10, at(0));
+        dispatcher.dispatchDue(10, at(100));
+
+        assertEquals(1, gateway.calls);
+        assertEquals(OutboxDeliveryState.DELIVERED,
+                repository.values.get("EVENT-1").getState());
+        assertEquals("WMS-RECEIPT-1",
+                repository.values.get("EVENT-1").getExternalReceipt());
+    }
+
+    private static WmsNotification notification(String eventId) {
+        return new WmsNotification(eventId, "REQ-1", "ORDER-1", at(0),
+                "OrderCompleted", "OK", null,
+                Collections.<String, String>emptyMap());
+    }
+
+    private static Instant at(long seconds) {
+        return Instant.EPOCH.plusSeconds(seconds);
+    }
+
+    private static final class FakeWmsOutbox
+            implements WmsNotificationOutboxRepository {
+        private final Map<String, WmsNotificationOutboxEntry> values =
+                new LinkedHashMap<String, WmsNotificationOutboxEntry>();
+
+        @Override
+        public void saveIfAbsent(WmsNotificationOutboxEntry entry) {
+            if (!values.containsKey(entry.getEventId())) {
+                values.put(entry.getEventId(), entry);
+            }
+        }
+
+        @Override
+        public void save(WmsNotificationOutboxEntry entry) {
+            values.put(entry.getEventId(), entry);
+        }
+
+        @Override
+        public List<WmsNotificationOutboxEntry> claimDue(int limit, Instant now) {
+            List<WmsNotificationOutboxEntry> result =
+                    new ArrayList<WmsNotificationOutboxEntry>();
+            for (WmsNotificationOutboxEntry entry : values.values()) {
+                if (result.size() >= limit) {
+                    break;
+                }
+                if (entry.claim(now)) {
+                    result.add(entry);
+                }
+            }
+            return result;
+        }
+    }
+
+    private static final class FakeWmsGateway
+            implements WmsNotificationGateway {
+        private WmsNotificationReceipt receipt;
+        private int calls;
+
+        @Override
+        public WmsNotificationReceipt notify(WmsNotification notification) {
+            calls++;
+            return receipt;
+        }
+    }
+}