fix: 恢复服务重启后的自动导入任务

easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/document/exception/DocumentParseBridgeException.java

@@ -75,6 +75,10 @@ public class DocumentParseBridgeException extends RuntimeException {
         return new DocumentParseBridgeException("task_failed", message);
     }
 
+    public static DocumentParseBridgeException taskNotFound(String message, Throwable cause) {
+        return new DocumentParseBridgeException("task_not_found", message, cause);
+    }
+
     public static DocumentParseBridgeException resultFetchFailed(String message, Throwable cause) {
         return new DocumentParseBridgeException("result_fetch_failed", message, cause);
     }

easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/document/service/impl/DocumentParseBridgeServiceImpl.java

@@ -5,6 +5,7 @@ import com.easyagents.document.core.entity.ParseResponse;
 import com.easyagents.document.core.entity.ParseResult;
 import com.easyagents.document.core.entity.ParseTaskInfo;
 import com.easyagents.document.core.entity.ParseTaskStatus;
+import com.easyagents.document.core.exception.DocumentAsyncTaskNotFoundException;
 import com.easyagents.document.pdf.PdfDocumentParseService;
 import com.easyagents.document.pptx.PptxDocumentParseService;
 import com.easyagents.document.xlsx.XlsxDocumentParseService;
@@ -140,6 +141,8 @@ public class DocumentParseBridgeServiceImpl implements DocumentParseBridgeServic
             return parseResultMapper.map(taskStatus);
         } catch (DocumentParseBridgeException e) {
             throw e;
+        } catch (DocumentAsyncTaskNotFoundException e) {
+            throw DocumentParseBridgeException.taskNotFound("异步文档解析任务不存在", e);
         } catch (Exception e) {
             throw DocumentParseBridgeException.taskFailed("查询异步文档解析任务状态失败", e);
         }
@@ -175,6 +178,9 @@ public class DocumentParseBridgeServiceImpl implements DocumentParseBridgeServic
         } catch (DocumentParseBridgeException e) {
             LOG.error("桥接服务获取异步解析结果失败: providerTaskId={}", taskId, e);
             throw e;
+        } catch (DocumentAsyncTaskNotFoundException e) {
+            LOG.warn("桥接服务异步解析执行已丢失: providerTaskId={}", taskId);
+            throw DocumentParseBridgeException.taskNotFound("异步文档解析任务不存在", e);
         } catch (Exception e) {
             LOG.error("桥接服务获取异步解析结果异常: providerTaskId={}", taskId, e);
             throw DocumentParseBridgeException.resultFetchFailed("获取异步文档解析结果失败", e);
@@ -212,6 +218,9 @@ public class DocumentParseBridgeServiceImpl implements DocumentParseBridgeServic
         } catch (DocumentParseBridgeException e) {
             LOG.error("桥接服务查询异步解析任务状态失败: providerTaskId={}", taskId, e);
             throw e;
+        } catch (DocumentAsyncTaskNotFoundException e) {
+            LOG.warn("桥接服务异步解析执行已丢失: providerTaskId={}", taskId);
+            throw DocumentParseBridgeException.taskNotFound("异步文档解析任务不存在", e);
         } catch (Exception e) {
             LOG.error("桥接服务查询异步解析任务状态异常: providerTaskId={}", taskId, e);
             throw DocumentParseBridgeException.taskFailed("聚合查询异步文档解析任务信息失败", e);

easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchAppService.java

@@ -80,6 +80,7 @@ public class DocumentImportBatchAppService {
     private final DocumentMapper documentMapper;
     private final RedisLockExecutor redisLockExecutor;
     private final DocumentImportBatchCircuitBreaker circuitBreaker;
+    private final DocumentImportRecoveryCoordinator recoveryCoordinator;
 
     @Resource(name = "default")
     private FileStorageService storageService;
@@ -97,6 +98,7 @@ public class DocumentImportBatchAppService {
      * @param documentMapper 文档 Mapper
      * @param redisLockExecutor 分布式锁执行器
      * @param circuitBreaker 自动导入批次熔断器
+     * @param recoveryCoordinator 恢复令牌事务协调器
      */
     public DocumentImportBatchAppService(DocumentImportBatchService batchService,
                                          DocumentImportBatchItemService itemService,
@@ -107,7 +109,8 @@ public class DocumentImportBatchAppService {
                                          DocumentImportBatchItemMapper itemMapper,
                                          DocumentMapper documentMapper,
                                          RedisLockExecutor redisLockExecutor,
-                                         DocumentImportBatchCircuitBreaker circuitBreaker) {
+                                         DocumentImportBatchCircuitBreaker circuitBreaker,
+                                         DocumentImportRecoveryCoordinator recoveryCoordinator) {
         this.batchService = batchService;
         this.itemService = itemService;
         this.batchTracker = batchTracker;
@@ -118,6 +121,7 @@ public class DocumentImportBatchAppService {
         this.documentMapper = documentMapper;
         this.redisLockExecutor = redisLockExecutor;
         this.circuitBreaker = circuitBreaker;
+        this.recoveryCoordinator = recoveryCoordinator;
     }
 
     /**
@@ -1246,13 +1250,10 @@ public class DocumentImportBatchAppService {
         Date claimedAt = new Date();
         Date leaseUntil = new Date(
             claimedAt.getTime() + RECOVERY_DISPATCH_LEASE.toMillis());
-        if (batchMapper.claimRecoveryPending(
-            batchId, recoveryToken, leaseUntil, claimedAt) <= 0) {
-            return;
-        }
         try {
             DocumentImportBatch claimedBatch =
-                batchMapper.selectClaimedRecovery(batchId, recoveryToken);
+                recoveryCoordinator.claim(
+                    batchId, recoveryToken, leaseUntil, claimedAt);
             if (claimedBatch == null) {
                 LOG.info(
                     "批次恢复调度令牌已失效，旧持有者停止恢复: "
@@ -1284,9 +1285,8 @@ public class DocumentImportBatchAppService {
                 );
                 return;
             }
-            int finalized = batchMapper.finalizeRecoveryPending(
+            if (!recoveryCoordinator.finalizeRecovery(
-                batchId, recoveryToken, new Date());
+                batchId, recoveryToken, new Date())) {
-            if (finalized <= 0) {
                 LOG.info(
                     "批次恢复待办已变更，当前实例跳过收尾: "
                         + "batchId={}, recoveryToken={}",
@@ -1299,7 +1299,8 @@ public class DocumentImportBatchAppService {
                 if (!circuitBreaker.interruptRecoveryBatch(
                     batchId, recoveryToken, error)) {
                     LOG.info(
-                        "批次恢复异常发生时令牌已失效，跳过旧持有者熔断: "
+                        "批次恢复异常发生时令牌已失效或已有并发推进，"
+                            + "跳过当前持有者熔断: "
                             + "batchId={}, recoveryToken={}",
                         batchId,
                         recoveryToken
@@ -1350,13 +1351,13 @@ public class DocumentImportBatchAppService {
             Date renewedLeaseUntil = new Date(
                 nowMillis + RECOVERY_DISPATCH_LEASE.toMillis()
             );
-            int renewed = batchMapper.renewRecoveryPendingLease(
+            boolean renewed = recoveryCoordinator.renew(
                 batchId,
                 recoveryToken,
                 renewedLeaseUntil,
                 now
             );
-            if (renewed <= 0) {
+            if (!renewed) {
                 return false;
             }
             renewAfter.set(nowMillis + renewIntervalMillis);

easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchCircuitBreaker.java

@@ -9,7 +9,9 @@ import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
 import tech.easyflow.ai.entity.DocumentImportBatch;
+import tech.easyflow.ai.entity.DocumentImportBatchItem;
 import tech.easyflow.ai.entity.DocumentImportTask;
+import tech.easyflow.ai.enums.DocumentImportBatchItemStatus;
 import tech.easyflow.ai.enums.DocumentImportBatchStatus;
 import tech.easyflow.ai.enums.DocumentImportMode;
 import tech.easyflow.ai.enums.DocumentImportTaskStatus;
@@ -23,6 +25,8 @@ import java.sql.SQLRecoverableException;
 import java.sql.SQLTimeoutException;
 import java.sql.SQLTransientConnectionException;
 import java.util.Date;
+import java.util.List;
+import java.util.Objects;
 import java.util.concurrent.RejectedExecutionException;
 
 /**
@@ -135,13 +139,18 @@ public class DocumentImportBatchCircuitBreaker {
         if (recoveryToken == null || recoveryToken.isBlank()) {
             throw new IllegalArgumentException("恢复调度令牌不能为空");
         }
+        DocumentImportRecoveryException recoveryError =
+            findCause(error, DocumentImportRecoveryException.class);
         return interruptBatch(
             batchId,
             resolveReason(error),
             null,
             null,
             error,
-            recoveryToken
+            recoveryToken,
+            recoveryError == null
+                ? List.of()
+                : recoveryError.getFailureFenceItemIds()
         );
     }
 
@@ -211,7 +220,7 @@ public class DocumentImportBatchCircuitBreaker {
                                    String phase,
                                    Throwable error) {
         return interruptBatch(
-            batchId, reason, taskId, phase, error, null);
+            batchId, reason, taskId, phase, error, null, List.of());
     }
 
     /**
@@ -223,6 +232,7 @@ public class DocumentImportBatchCircuitBreaker {
      * @param phase 触发任务阶段，可为空
      * @param error 原始异常
      * @param recoveryToken 恢复调度令牌，可为空
+     * @param recoveryFailureItemIds 零重排时的初始失败项围栏
      * @return 批次是否已经停止运行；围栏失效时返回 {@code false}
      */
     private boolean interruptBatch(BigInteger batchId,
@@ -230,11 +240,17 @@ public class DocumentImportBatchCircuitBreaker {
                                    BigInteger taskId,
                                    String phase,
                                    Throwable error,
-                                   String recoveryToken) {
+                                   String recoveryToken,
+                                   List<BigInteger> recoveryFailureItemIds) {
         if (batchId == null) {
             return false;
         }
-        DocumentImportBatch batch = batchMapper.selectOneById(batchId);
+        boolean fencedRecoveryFailure = recoveryToken != null
+            && recoveryFailureItemIds != null
+            && !recoveryFailureItemIds.isEmpty();
+        // 批次行是后续 task/item 状态收口的根锁；所有熔断路径先锁定
+        // batch，避免多表围栏更新与任务完成事务形成 task -> batch 逆序。
+        DocumentImportBatch batch = batchMapper.selectForUpdate(batchId);
         if (batch == null) {
             LOG.warn(
                 "忽略无法关联批次的自动导入熔断请求: taskId={}, batchId={}",
@@ -249,6 +265,39 @@ public class DocumentImportBatchCircuitBreaker {
         }
 
         Date now = new Date();
+        if (fencedRecoveryFailure) {
+            if (!Boolean.TRUE.equals(batch.getRecoveryPending())
+                || !Objects.equals(recoveryToken, batch.getRecoveryToken())
+                || batch.getRecoveryLeaseUntil() == null
+                || !batch.getRecoveryLeaseUntil().after(now)) {
+                return false;
+            }
+            if (itemMapper.countActiveItems(batchId) > 0) {
+                LOG.info(
+                    "恢复失败熔断发现批次已有活跃文件项，保留运行批次: "
+                        + "batchId={}",
+                    batchId
+                );
+                return false;
+            }
+            for (BigInteger itemId : recoveryFailureItemIds) {
+                DocumentImportBatchItem item =
+                    itemMapper.selectForUpdate(itemId);
+                if (item == null
+                    || !batchId.equals(item.getBatchId())
+                    || !DocumentImportBatchItemStatus.FAILED
+                        .name().equals(item.getStatus())) {
+                    LOG.info(
+                        "恢复失败熔断发现等价并发推进，保留运行批次: "
+                            + "batchId={}, itemId={}, status={}",
+                        batchId,
+                        itemId,
+                        item == null ? null : item.getStatus()
+                    );
+                    return false;
+                }
+            }
+        }
         int interrupted;
         if (recoveryToken != null) {
             interrupted = batchMapper.interruptOwnedRecoveryBatch(
@@ -305,6 +354,14 @@ public class DocumentImportBatchCircuitBreaker {
      * @return 稳定错误码与用户可见摘要
      */
     private InterruptionReason resolveReason(Throwable error) {
+        DocumentImportRecoveryException recoveryError =
+            findCause(error, DocumentImportRecoveryException.class);
+        if (recoveryError != null) {
+            return new InterruptionReason(
+                recoveryError.getCode(),
+                recoveryError.getUserMessage()
+            );
+        }
         if (containsRedisFailure(error)) {
             return new InterruptionReason(
                 REDIS_UNAVAILABLE,
@@ -329,6 +386,25 @@ public class DocumentImportBatchCircuitBreaker {
         );
     }
 
+    /**
+     * 从异常链中查找指定类型。
+     *
+     * @param error 原始异常
+     * @param type 目标异常类型
+     * @param <T> 异常类型
+     * @return 匹配异常；不存在时返回 {@code null}
+     */
+    private <T extends Throwable> T findCause(Throwable error, Class<T> type) {
+        Throwable current = error;
+        while (current != null) {
+            if (type.isInstance(current)) {
+                return type.cast(current);
+            }
+            current = current.getCause();
+        }
+        return null;
+    }
+
     /**
      * 判断异常链是否来自 Redis/Lettuce。
      *

easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchTracker.java

@@ -27,6 +27,18 @@ import java.util.Date;
 @Service
 public class DocumentImportBatchTracker {
 
+    /**
+     * 单文件恢复失败原因的事务内处理结果。
+     */
+    public enum FailedRetryErrorOutcome {
+        /** 失败原因已写回，文件项仍等待后续人工恢复。 */
+        ERROR_RECORDED,
+        /** 等价并发操作已把文件项推进出失败状态。 */
+        ALREADY_ADVANCED,
+        /** 批次或文件项已不再允许当前恢复请求写回。 */
+        REJECTED
+    }
+
     private final DocumentImportBatchService batchService;
     private final DocumentImportBatchItemService itemService;
     private final DocumentImportBatchMapper batchMapper;
@@ -64,6 +76,31 @@ public class DocumentImportBatchTracker {
         return batch;
     }
 
+    /**
+     * 在任务行变更前锁定所属批次，统一 batch -> task 锁序。
+     *
+     * <p>自动导入的任务生命周期由运行批次驱动，批次结束后不允许
+     * 旧任务继续推进。手动导入的上传批次可以先结束，用户之后再手动
+     * 执行分块和索引，因此只参与锁序，不以批次终态阻断后续任务。</p>
+     *
+     * @param batchId 批次 ID；无批次任务传入 {@code null}
+     * @return 批次存在且允许当前任务推进时为 {@code true}
+     */
+    public boolean lockBatchForTaskMutation(BigInteger batchId) {
+        if (batchId == null) {
+            return true;
+        }
+        DocumentImportBatch batch = batchMapper.selectForUpdate(batchId);
+        if (batch == null) {
+            return false;
+        }
+        if (DocumentImportMode.AUTO.name().equals(batch.getImportMode())) {
+            return DocumentImportBatchStatus.RUNNING.name().equals(
+                batch.getStatus());
+        }
+        return DocumentImportMode.MANUAL.name().equals(batch.getImportMode());
+    }
+
     /**
      * 查询批次文件项。
      *
@@ -107,10 +144,8 @@ public class DocumentImportBatchTracker {
         if (itemId == null) {
             return false;
         }
-        DocumentImportBatchItem current = requireItem(itemId);
         int attemptDelta = (status == DocumentImportBatchItemStatus.PENDING
             || status == DocumentImportBatchItemStatus.RUNNING)
-            && DocumentImportBatchItemStatus.FAILED.name().equals(current.getStatus())
             ? 1
             : 0;
         return transitionItem(itemId, stage, status, errorSummary,
@@ -162,28 +197,34 @@ public class DocumentImportBatchTracker {
         if (itemId == null) {
             return false;
         }
-        for (int attempt = 0; attempt < 3; attempt++) {
+        LockedBatchItem locked = lockBatchThenItem(itemId);
-            DocumentImportBatchItem current = requireItem(itemId);
+        if (DocumentImportBatchStatus.INTERRUPTED.name()
+            .equals(locked.batch.getStatus())) {
+            return false;
+        }
+        DocumentImportBatchItem current = locked.item;
         DocumentImportBatchItemStatus currentStatus =
             DocumentImportBatchItemStatus.valueOf(current.getStatus());
         if (!isAllowedTransition(currentStatus, status)) {
             return false;
         }
-            String expectedStatus = current.getStatus();
         Date now = new Date();
+        int effectiveAttemptDelta = currentStatus == DocumentImportBatchItemStatus.FAILED
+            ? Math.max(0, attemptDelta)
+            : 0;
         int updated = itemMapper.transitionStatus(
             itemId,
-                expectedStatus,
+            current.getStatus(),
             stage.name(),
             status.name(),
             errorSummary,
             failureCode,
             retryable,
-                Math.max(0, attemptDelta),
+            effectiveAttemptDelta,
             now
         );
         if (updated <= 0) {
-                continue;
+            return false;
         }
         CounterDelta delta = CounterDelta.between(current, status, retryable);
         if (!delta.isZero()) {
@@ -203,8 +244,6 @@ public class DocumentImportBatchTracker {
         refreshBatch(current.getBatchId());
         return true;
     }
-        return false;
-    }
 
     /**
      * 校验文件项状态机，拒绝迟到任务覆盖终态。
@@ -243,7 +282,12 @@ public class DocumentImportBatchTracker {
      */
     @org.springframework.transaction.annotation.Transactional
     public void bindDocument(BigInteger itemId, BigInteger documentId) {
-        DocumentImportBatchItem item = requireItem(itemId);
+        LockedBatchItem locked = lockBatchThenItem(itemId);
+        if (DocumentImportBatchStatus.INTERRUPTED.name()
+            .equals(locked.batch.getStatus())) {
+            throw new BusinessException("导入批次已中断，请继续批次后重试");
+        }
+        DocumentImportBatchItem item = locked.item;
         if (documentId.equals(item.getDocumentId())
             && DocumentImportBatchItemStatus.PENDING.name().equals(item.getStatus())) {
             return;
@@ -252,6 +296,11 @@ public class DocumentImportBatchTracker {
         boolean recoveringFailedItem =
             DocumentImportBatchItemStatus.FAILED.name().equals(item.getStatus())
                 && item.getDocumentId() == null;
+        if (recoveringFailedItem
+            && !DocumentImportBatchStatus.RUNNING.name()
+                .equals(locked.batch.getStatus())) {
+            throw new BusinessException("导入批次状态已变化，请刷新后重试");
+        }
         int updated = recoveringFailedItem
             ? itemMapper.bindFailedDocument(itemId, documentId, now)
             : itemMapper.bindDocument(itemId, documentId, now);
@@ -268,6 +317,91 @@ public class DocumentImportBatchTracker {
         refreshBatch(item.getBatchId());
     }
 
+    /**
+     * 将引用已丢失文档的失败项重新绑定到恢复创建的新文档。
+     *
+     * @param itemId 文件项 ID
+     * @param missingDocumentId 已不存在的旧文档 ID
+     * @param documentId 恢复创建的新文档 ID
+     */
+    @org.springframework.transaction.annotation.Transactional
+    public void replaceMissingDocument(BigInteger itemId,
+                                       BigInteger missingDocumentId,
+                                       BigInteger documentId) {
+        LockedBatchItem locked = lockBatchThenItem(itemId);
+        if (!DocumentImportBatchStatus.RUNNING.name()
+            .equals(locked.batch.getStatus())) {
+            throw new BusinessException("导入批次状态已变化，请刷新后重试");
+        }
+        DocumentImportBatchItem item = locked.item;
+        if (!DocumentImportBatchItemStatus.FAILED.name().equals(item.getStatus())
+            || !java.util.Objects.equals(missingDocumentId, item.getDocumentId())) {
+            throw new BusinessException("导入文件状态已变化，请刷新后重试");
+        }
+        Date now = new Date();
+        if (itemMapper.replaceMissingDocument(
+            itemId, missingDocumentId, documentId, now) <= 0) {
+            throw new BusinessException("原文档状态已变化，请刷新后重试");
+        }
+        batchMapper.refreshCountersFromItems(item.getBatchId(), now);
+        refreshBatch(item.getBatchId());
+    }
+
+    /**
+     * 按批次到文件项的固定顺序取得失败项恢复权。
+     *
+     * @param itemId 文件项 ID
+     * @return 可恢复文件项；状态已变化时返回 null
+     */
+    @org.springframework.transaction.annotation.Transactional
+    public DocumentImportBatchItem lockFailedItemForRetry(BigInteger itemId) {
+        if (itemId == null) {
+            return null;
+        }
+        LockedBatchItem locked = lockBatchThenItem(itemId);
+        if (!DocumentImportBatchStatus.RUNNING.name()
+            .equals(locked.batch.getStatus())
+            || !DocumentImportBatchItemStatus.FAILED.name()
+                .equals(locked.item.getStatus())) {
+            return null;
+        }
+        return locked.item;
+    }
+
+    /**
+     * 在运行批次内保留单文件恢复失败原因。
+     *
+     * @param itemId 文件项 ID
+     * @param batchId 批次 ID
+     * @param errorSummary 错误摘要
+     * @return 失败原因写回、并发推进或拒绝结果
+     */
+    @org.springframework.transaction.annotation.Transactional(
+        propagation = org.springframework.transaction.annotation.Propagation.REQUIRES_NEW)
+    public FailedRetryErrorOutcome updateFailedRetryError(
+        BigInteger itemId,
+        BigInteger batchId,
+        String errorSummary
+    ) {
+        if (itemId == null || batchId == null) {
+            return FailedRetryErrorOutcome.REJECTED;
+        }
+        LockedBatchItem locked = lockBatchThenItem(itemId);
+        if (!batchId.equals(locked.batch.getId())
+            || !DocumentImportBatchStatus.RUNNING.name()
+                .equals(locked.batch.getStatus())) {
+            return FailedRetryErrorOutcome.REJECTED;
+        }
+        if (!DocumentImportBatchItemStatus.FAILED.name()
+            .equals(locked.item.getStatus())) {
+            return FailedRetryErrorOutcome.ALREADY_ADVANCED;
+        }
+        return itemMapper.updateFailedRetryError(
+            itemId, batchId, errorSummary, new Date()) > 0
+                ? FailedRetryErrorOutcome.ERROR_RECORDED
+                : FailedRetryErrorOutcome.REJECTED;
+    }
+
     /**
      * 原子完成文件上传并增量更新批次上传数。
      *
@@ -280,7 +414,12 @@ public class DocumentImportBatchTracker {
     public boolean completeUpload(BigInteger itemId,
                                   String filePath,
                                   String storageLocator) {
-        DocumentImportBatchItem item = requireItem(itemId);
+        LockedBatchItem locked = lockBatchThenItem(itemId);
+        if (DocumentImportBatchStatus.INTERRUPTED.name()
+            .equals(locked.batch.getStatus())) {
+            return false;
+        }
+        DocumentImportBatchItem item = locked.item;
         if (DocumentImportBatchItemStatus.UPLOADED.name().equals(item.getStatus())) {
             return filePath.equals(item.getFilePath())
                 && storageLocator.equals(item.getStorageLocator());
@@ -304,6 +443,34 @@ public class DocumentImportBatchTracker {
         return true;
     }
 
+    /**
+     * 先锁批次再锁文件项，统一所有计数迁移路径的加锁顺序。
+     *
+     * @param itemId 文件项 ID
+     * @return 已锁定的批次与文件项
+     */
+    private LockedBatchItem lockBatchThenItem(BigInteger itemId) {
+        DocumentImportBatchItem candidate = requireItem(itemId);
+        DocumentImportBatch batch = batchMapper.selectForUpdate(candidate.getBatchId());
+        if (batch == null) {
+            throw new BusinessException("导入批次不存在");
+        }
+        DocumentImportBatchItem item = itemMapper.selectForUpdate(itemId);
+        if (item == null) {
+            throw new BusinessException("导入文件不存在");
+        }
+        if (!java.util.Objects.equals(batch.getId(), item.getBatchId())) {
+            throw new IllegalStateException("导入文件批次归属已变化");
+        }
+        return new LockedBatchItem(batch, item);
+    }
+
+    private record LockedBatchItem(
+        DocumentImportBatch batch,
+        DocumentImportBatchItem item
+    ) {
+    }
+
     /**
      * 记录文件项成功后需要清理的历史文档。
      *

easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportRecoveryCoordinator.java

@@ -0,0 +1,75 @@
+package tech.easyflow.ai.documentimport.task;
+
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+import tech.easyflow.ai.entity.DocumentImportBatch;
+import tech.easyflow.ai.mapper.DocumentImportBatchMapper;
+
+import java.math.BigInteger;
+import java.util.Date;
+
+/**
+ * 在独立短事务中维护批次恢复令牌。
+ *
+ * <p>继续操作通常从原事务的 {@code afterCommit} 回调启动。将领取、续租和
+ * 收尾放入独立事务，可保证恢复令牌在单文件重试事务开始前已经提交，并避免
+ * 复用刚完成提交但尚未解绑的事务资源。</p>
+ */
+@Service
+public class DocumentImportRecoveryCoordinator {
+
+    private final DocumentImportBatchMapper batchMapper;
+
+    public DocumentImportRecoveryCoordinator(DocumentImportBatchMapper batchMapper) {
+        this.batchMapper = batchMapper;
+    }
+
+    /**
+     * 领取恢复待办并读取本次令牌对应的恢复参数。
+     *
+     * @param batchId 批次 ID
+     * @param recoveryToken 恢复令牌
+     * @param leaseUntil 租约到期时间
+     * @param claimedAt 领取时间
+     * @return 已领取批次；领取失败时返回 null
+     */
+    @Transactional(propagation = Propagation.REQUIRES_NEW)
+    public DocumentImportBatch claim(BigInteger batchId,
+                                     String recoveryToken,
+                                     Date leaseUntil,
+                                     Date claimedAt) {
+        if (batchMapper.claimRecoveryPending(
+            batchId, recoveryToken, leaseUntil, claimedAt) <= 0) {
+            return null;
+        }
+        return batchMapper.selectClaimedRecovery(batchId, recoveryToken);
+    }
+
+    /**
+     * 续期当前恢复令牌。
+     *
+     * @return 当前令牌仍有效且续期成功时返回 true
+     */
+    @Transactional(propagation = Propagation.REQUIRES_NEW)
+    public boolean renew(BigInteger batchId,
+                         String recoveryToken,
+                         Date leaseUntil,
+                         Date renewedAt) {
+        return batchMapper.renewRecoveryPendingLease(
+            batchId, recoveryToken, leaseUntil, renewedAt) > 0;
+    }
+
+    /**
+     * 按真实计数收尾并清除恢复待办。
+     *
+     * @return 当前令牌仍有效且收尾成功时返回 true
+     */
+    @Transactional(propagation = Propagation.REQUIRES_NEW)
+    public boolean finalizeRecovery(BigInteger batchId,
+                                    String recoveryToken,
+                                    Date finalizedAt) {
+        return batchMapper.finalizeRecoveryPending(
+            batchId, recoveryToken, finalizedAt) > 0;
+    }
+}

easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportRecoveryException.java

@@ -0,0 +1,71 @@
+package tech.easyflow.ai.documentimport.task;
+
+import java.math.BigInteger;
+import java.util.Collection;
+import java.util.List;
+
+/**
+ * 批次恢复未能重新排队任何失败项。
+ *
+ * <p>该异常只携带稳定错误码和用户可见摘要，供恢复令牌持有者
+ * 中断批次，避免“继续”请求在没有启动任何新任务时表现为成功。</p>
+ */
+public class DocumentImportRecoveryException extends RuntimeException {
+
+    private static final String ERROR_CODE = "document_import_recovery_failed";
+    private static final String USER_MESSAGE =
+        "未能重新排队任何失败文件，请查看文件错误后继续";
+    private final List<BigInteger> failureFenceItemIds;
+
+    /**
+     * 创建批次恢复失败异常。
+     */
+    public DocumentImportRecoveryException() {
+        this(List.of());
+    }
+
+    /**
+     * 创建带初始失败项快照的批次恢复失败异常。
+     *
+     * @param failureFenceItemIds 领取恢复时仍为失败的文件项 ID
+     */
+    public DocumentImportRecoveryException(
+        Collection<BigInteger> failureFenceItemIds
+    ) {
+        super(USER_MESSAGE);
+        this.failureFenceItemIds = failureFenceItemIds == null
+            ? List.of()
+            : failureFenceItemIds.stream()
+                .filter(java.util.Objects::nonNull)
+                .distinct()
+                .sorted()
+                .toList();
+    }
+
+    /**
+     * 获取稳定错误码。
+     *
+     * @return 稳定错误码
+     */
+    public String getCode() {
+        return ERROR_CODE;
+    }
+
+    /**
+     * 获取用户可见摘要。
+     *
+     * @return 用户可见摘要
+     */
+    public String getUserMessage() {
+        return USER_MESSAGE;
+    }
+
+    /**
+     * 获取用于原子熔断围栏的初始失败项。
+     *
+     * @return 按 ID 排序的不可变列表
+     */
+    public List<BigInteger> getFailureFenceItemIds() {
+        return failureFenceItemIds;
+    }
+}

easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/mapper/DocumentImportBatchItemMapper.java

@@ -17,22 +17,49 @@ import java.util.Date;
  */
 public interface DocumentImportBatchItemMapper extends BaseMapper<DocumentImportBatchItem> {
 
+    String SELECT_COLUMNS = "id, batch_id AS batchId, knowledge_id AS knowledgeId, "
+        + "document_id AS documentId, replaced_document_id AS replacedDocumentId, "
+        + "client_file_key AS clientFileKey, file_name AS fileName, "
+        + "relative_path AS relativePath, file_size AS fileSize, "
+        + "file_path AS filePath, storage_locator AS storageLocator, "
+        + "cleanup_pending AS cleanupPending, content_sha256 AS contentSha256, "
+        + "stage, status, error_summary AS errorSummary, "
+        + "failure_code AS failureCode, applied_strategy_code AS appliedStrategyCode, "
+        + "strategy_snapshot_json AS strategySnapshotJson, retryable, "
+        + "attempt_count AS attemptCount, created, created_by AS createdBy, "
+        + "modified, modified_by AS modifiedBy";
+
     /**
-     * 锁定运行批次中的失败项并取得本轮恢复权。
+     * 锁定并读取批次文件项。
      *
-     * <p>同时锁定批次和文件项，使恢复任务创建与批次熔断串行化。</p>
+     * <p>调用方必须先锁定所属批次，再调用本方法，避免批次计数更新时
+     * 出现共享锁升级死锁。</p>
      *
      * @param itemId 文件项 ID
-     * @return 可恢复文件项；状态已变化时返回 null
+     * @return 文件项；不存在时返回 null
      */
-    @Select("SELECT item.* FROM tb_document_import_batch_item item "
+    @Select("SELECT " + SELECT_COLUMNS + " FROM tb_document_import_batch_item "
-        + "INNER JOIN tb_document_import_batch batch ON batch.id=item.batch_id "
+        + "WHERE id=#{itemId} FOR UPDATE")
-        + "WHERE item.id=#{itemId} AND item.status='FAILED' "
+    DocumentImportBatchItem selectForUpdate(
-        + "AND batch.status='RUNNING' FOR UPDATE")
-    DocumentImportBatchItem selectFailedForRetry(
         @Param("itemId") BigInteger itemId
     );
 
+    /**
+     * 统计运行批次内已经由等价请求推进的活跃文件项。
+     *
+     * <p>调用方必须先锁定批次；所有文件项状态迁移遵循批次到文件项
+     * 的锁顺序，因此该当前状态检查到后续熔断之间不会新增活跃项。</p>
+     *
+     * @param batchId 批次 ID
+     * @return 活跃文件项数量
+     */
+    @Select("SELECT COUNT(*) FROM tb_document_import_batch_item "
+        + "WHERE batch_id=#{batchId} "
+        + "AND status IN ('PENDING','UPLOADING','RUNNING','UPLOADED')")
+    int countActiveItems(
+        @Param("batchId") BigInteger batchId
+    );
+
     /**
      * 将中断批次中尚未结束的文件项统一收口为可恢复失败。
      *
@@ -100,14 +127,13 @@ public interface DocumentImportBatchItemMapper extends BaseMapper<DocumentImport
      * @param modified 修改时间
      * @return 更新行数
      */
-    @Update("UPDATE tb_document_import_batch_item item "
+    @Update("UPDATE tb_document_import_batch_item SET "
-        + "INNER JOIN tb_document_import_batch batch ON batch.id=item.batch_id SET "
+        + "stage=#{stage}, status=#{status}, "
-        + "item.stage=#{stage}, item.status=#{status}, "
+        + "error_summary=#{errorSummary}, failure_code=#{failureCode}, "
-        + "item.error_summary=#{errorSummary}, item.failure_code=#{failureCode}, "
+        + "retryable=#{retryable}, "
-        + "item.retryable=#{retryable}, "
+        + "attempt_count=attempt_count + #{attemptDelta}, "
-        + "item.attempt_count=item.attempt_count + #{attemptDelta}, "
+        + "modified=#{modified} WHERE id=#{id} "
-        + "item.modified=#{modified} WHERE item.id=#{id} "
+        + "AND status=#{expectedStatus}")
-        + "AND item.status=#{expectedStatus} AND batch.status<>'INTERRUPTED'")
     int transitionStatus(
         @Param("id") BigInteger id,
         @Param("expectedStatus") String expectedStatus,
@@ -146,21 +172,46 @@ public interface DocumentImportBatchItemMapper extends BaseMapper<DocumentImport
      * @param modified 修改时间
      * @return 文件项仍属于运行批次且绑定成功时返回 1
      */
-    @Update("UPDATE tb_document_import_batch_item item "
+    @Update("UPDATE tb_document_import_batch_item SET "
-        + "INNER JOIN tb_document_import_batch batch ON batch.id=item.batch_id SET "
+        + "document_id=#{documentId}, stage='PARSE', "
-        + "item.document_id=#{documentId}, item.stage='PARSE', "
+        + "status='PENDING', error_summary=NULL, "
-        + "item.status='PENDING', item.error_summary=NULL, "
+        + "failure_code=NULL, retryable=0, "
-        + "item.failure_code=NULL, item.retryable=0, "
+        + "attempt_count=COALESCE(attempt_count, 0) + 1, "
-        + "item.attempt_count=COALESCE(item.attempt_count, 0) + 1, "
+        + "modified=#{modified} "
-        + "item.modified=#{modified} "
+        + "WHERE id=#{id} AND status='FAILED' "
-        + "WHERE item.id=#{id} AND item.status='FAILED' "
+        + "AND document_id IS NULL")
-        + "AND item.document_id IS NULL AND batch.status='RUNNING'")
     int bindFailedDocument(
         @Param("id") BigInteger id,
         @Param("documentId") BigInteger documentId,
         @Param("modified") Date modified
     );
 
+    /**
+     * 将引用已丢失文档的失败项绑定到恢复创建的新文档。
+     *
+     * @param id 文件项 ID
+     * @param missingDocumentId 已不存在的旧文档 ID
+     * @param documentId 恢复创建的新文档 ID
+     * @param modified 修改时间
+     * @return 文件项仍处于预期失败状态且旧文档确实不存在时返回 1
+     */
+    @Update("UPDATE tb_document_import_batch_item SET "
+        + "document_id=#{documentId}, stage='PARSE', "
+        + "status='PENDING', error_summary=NULL, "
+        + "failure_code=NULL, retryable=0, "
+        + "attempt_count=COALESCE(attempt_count, 0) + 1, "
+        + "modified=#{modified} "
+        + "WHERE id=#{id} AND status='FAILED' "
+        + "AND document_id=#{missingDocumentId} "
+        + "AND NOT EXISTS (SELECT 1 FROM tb_document "
+        + "WHERE id=#{missingDocumentId})")
+    int replaceMissingDocument(
+        @Param("id") BigInteger id,
+        @Param("missingDocumentId") BigInteger missingDocumentId,
+        @Param("documentId") BigInteger documentId,
+        @Param("modified") Date modified
+    );
+
     /**
      * 更新恢复失败项的业务错误，但不提前汇总批次终态。
      *
@@ -170,12 +221,11 @@ public interface DocumentImportBatchItemMapper extends BaseMapper<DocumentImport
      * @param modified 修改时间
      * @return 文件项仍属于运行批次且更新成功时返回 1
      */
-    @Update("UPDATE tb_document_import_batch_item item "
+    @Update("UPDATE tb_document_import_batch_item SET "
-        + "INNER JOIN tb_document_import_batch batch ON batch.id=item.batch_id SET "
+        + "error_summary=#{errorSummary}, failure_code=NULL, "
-        + "item.error_summary=#{errorSummary}, item.failure_code=NULL, "
+        + "retryable=1, modified=#{modified} "
-        + "item.retryable=1, item.modified=#{modified} "
+        + "WHERE id=#{id} AND batch_id=#{batchId} "
-        + "WHERE item.id=#{id} AND item.batch_id=#{batchId} "
+        + "AND status='FAILED'")
-        + "AND item.status='FAILED' AND batch.status='RUNNING'")
     int updateFailedRetryError(
         @Param("id") BigInteger id,
         @Param("batchId") BigInteger batchId,

easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/mapper/DocumentImportBatchMapper.java

@@ -18,6 +18,38 @@ import java.util.List;
  */
 public interface DocumentImportBatchMapper extends BaseMapper<DocumentImportBatch> {
 
+    String SELECT_COLUMNS = "id, knowledge_id AS knowledgeId, "
+        + "caller_type AS callerType, caller_id AS callerId, "
+        + "idempotency_key_hash AS idempotencyKeyHash, "
+        + "request_digest AS requestDigest, duplicate_policy AS duplicatePolicy, "
+        + "requested_strategy_json AS requestedStrategyJson, "
+        + "retry_generation AS retryGeneration, version, "
+        + "import_mode AS importMode, status, total_count AS totalCount, "
+        + "total_bytes AS totalBytes, completed_count AS completedCount, "
+        + "processing_count AS processingCount, failed_count AS failedCount, "
+        + "pending_count AS pendingCount, uploaded_count AS uploadedCount, "
+        + "skipped_count AS skippedCount, cancelled_count AS cancelledCount, "
+        + "retryable_failed_count AS retryableFailedCount, "
+        + "interrupt_code AS interruptCode, interrupt_message AS interruptMessage, "
+        + "interrupted_at AS interruptedAt, recovery_pending AS recoveryPending, "
+        + "recovery_file_keys_json AS recoveryFileKeysJson, "
+        + "recovery_token AS recoveryToken, "
+        + "recovery_lease_until AS recoveryLeaseUntil, "
+        + "started_at AS startedAt, finished_at AS finishedAt, created, "
+        + "created_by AS createdBy, modified, modified_by AS modifiedBy";
+
+    /**
+     * 锁定并读取批次，统一批次计数与文件项状态更新的加锁顺序。
+     *
+     * @param batchId 批次 ID
+     * @return 批次；不存在时返回 null
+     */
+    @Select("SELECT " + SELECT_COLUMNS + " FROM tb_document_import_batch "
+        + "WHERE id=#{batchId} FOR UPDATE")
+    DocumentImportBatch selectForUpdate(
+        @Param("batchId") BigInteger batchId
+    );
+
     /**
      * 锁定并读取调用者拥有的批次，保证后续重试基于最新状态。
      *
@@ -26,7 +58,7 @@ public interface DocumentImportBatchMapper extends BaseMapper<DocumentImportBatc
      * @param callerId 调用者 ID
      * @return 批次；不存在时返回 null
      */
-    @Select("SELECT * FROM tb_document_import_batch "
+    @Select("SELECT " + SELECT_COLUMNS + " FROM tb_document_import_batch "
         + "WHERE id=#{batchId} AND caller_type=#{callerType} "
         + "AND caller_id=#{callerId} FOR UPDATE")
     DocumentImportBatch selectOwnedForUpdate(
@@ -195,7 +227,7 @@ public interface DocumentImportBatchMapper extends BaseMapper<DocumentImportBatc
      * @param limit 最大批次数
      * @return 待恢复批次
      */
-    @Select("SELECT * FROM tb_document_import_batch "
+    @Select("SELECT " + SELECT_COLUMNS + " FROM tb_document_import_batch "
         + "WHERE status='RUNNING' AND recovery_pending=1 "
         + "AND (recovery_token IS NULL OR recovery_lease_until <= #{now}) "
         + "ORDER BY modified, id LIMIT #{limit}")
@@ -256,7 +288,7 @@ public interface DocumentImportBatchMapper extends BaseMapper<DocumentImportBatc
      * @param recoveryToken 恢复调度令牌
      * @return 当前令牌持有的批次；令牌失效时返回空
      */
-    @Select("SELECT * FROM tb_document_import_batch "
+    @Select("SELECT " + SELECT_COLUMNS + " FROM tb_document_import_batch "
         + "WHERE id=#{batchId} AND status='RUNNING' "
         + "AND recovery_pending=1 AND recovery_token=#{recoveryToken}")
     DocumentImportBatch selectClaimedRecovery(
@@ -383,7 +415,7 @@ public interface DocumentImportBatchMapper extends BaseMapper<DocumentImportBatc
      * @param limit 最大返回数量
      * @return 待回收批次
      */
-    @Select("SELECT * FROM tb_document_import_batch "
+    @Select("SELECT " + SELECT_COLUMNS + " FROM tb_document_import_batch "
         + "WHERE status IN ('UPLOADING','READY') "
         + "AND ((modified IS NOT NULL AND modified < #{incompleteCutoff}) "
         + "OR (modified IS NULL AND created < #{incompleteCutoff})) "

easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/mapper/DocumentImportTaskMapper.java

@@ -18,6 +18,26 @@ import java.util.List;
  */
 public interface DocumentImportTaskMapper extends BaseMapper<DocumentImportTask> {
 
+    String SELECT_COLUMNS = "id, document_id AS documentId, "
+        + "knowledge_id AS knowledgeId, batch_id AS batchId, "
+        + "batch_item_id AS batchItemId, phase, status, "
+        + "provider_task_id AS providerTaskId, payload_json AS payloadJson, "
+        + "error_summary AS errorSummary, failure_code AS failureCode, "
+        + "attempt_no AS attemptNo, execution_token AS executionToken, "
+        + "lease_until AS leaseUntil, version, started_at AS startedAt, "
+        + "finished_at AS finishedAt, created, created_by AS createdBy, "
+        + "modified, modified_by AS modifiedBy";
+
+    String QUALIFIED_SELECT_COLUMNS = "task.id, task.document_id AS documentId, "
+        + "task.knowledge_id AS knowledgeId, task.batch_id AS batchId, "
+        + "task.batch_item_id AS batchItemId, task.phase, task.status, "
+        + "task.provider_task_id AS providerTaskId, task.payload_json AS payloadJson, "
+        + "task.error_summary AS errorSummary, task.failure_code AS failureCode, "
+        + "task.attempt_no AS attemptNo, task.execution_token AS executionToken, "
+        + "task.lease_until AS leaseUntil, task.version, task.started_at AS startedAt, "
+        + "task.finished_at AS finishedAt, task.created, task.created_by AS createdBy, "
+        + "task.modified, task.modified_by AS modifiedBy";
+
     /**
      * 按任务阶段和批次轮转查询待投递任务，避免大批次独占投递窗口。
      *
@@ -25,14 +45,16 @@ public interface DocumentImportTaskMapper extends BaseMapper<DocumentImportTask>
      * @param limit 最大任务数
      * @return 公平排序后的待投递任务
      */
-    @Select("SELECT task.* FROM tb_document_import_task task JOIN ("
+    @Select("SELECT " + QUALIFIED_SELECT_COLUMNS
+        + " FROM tb_document_import_task task JOIN ("
         + "SELECT task.id, ROW_NUMBER() OVER ("
         + "PARTITION BY task.phase, COALESCE(task.batch_id, task.id) "
         + "ORDER BY task.created, task.id) AS lane_row "
         + "FROM tb_document_import_task task "
         + "LEFT JOIN tb_document_import_batch batch ON batch.id=task.batch_id "
         + "WHERE task.status='PENDING' AND task.modified <= #{redispatchBefore} "
-        + "AND (task.batch_id IS NULL OR batch.status='RUNNING')"
+        + "AND (task.batch_id IS NULL OR batch.import_mode='MANUAL' "
+        + "OR batch.status='RUNNING')"
         + ") ranked ON ranked.id=task.id "
         + "ORDER BY ranked.lane_row, task.created, task.id LIMIT #{limit}")
     List<DocumentImportTask> selectPendingFairly(
@@ -52,11 +74,7 @@ public interface DocumentImportTaskMapper extends BaseMapper<DocumentImportTask>
     @Update("UPDATE tb_document_import_task SET modified=#{now}, "
         + "modified_by=#{operatorId}, version=COALESCE(version, 0) + 1 "
         + "WHERE id=#{id} AND status='PENDING' "
-        + "AND modified <= #{redispatchBefore} "
+        + "AND modified <= #{redispatchBefore}")
-        + "AND (batch_id IS NULL OR EXISTS ("
-        + "SELECT 1 FROM tb_document_import_batch batch "
-        + "WHERE batch.id=tb_document_import_task.batch_id "
-        + "AND batch.status='RUNNING'))")
     int touchPendingForDispatch(
         @Param("id") BigInteger id,
         @Param("redispatchBefore") Date redispatchBefore,
@@ -72,7 +90,7 @@ public interface DocumentImportTaskMapper extends BaseMapper<DocumentImportTask>
      * @param limit 最大任务数
      * @return 已失去执行租约的运行任务
      */
-    @Select("SELECT * FROM tb_document_import_task "
+    @Select("SELECT " + SELECT_COLUMNS + " FROM tb_document_import_task "
         + "WHERE status='RUNNING' AND ("
         + "(lease_until IS NOT NULL AND lease_until <= #{now}) OR "
         + "(lease_until IS NULL AND modified <= #{legacyCutoff})) "
@@ -99,11 +117,7 @@ public interface DocumentImportTaskMapper extends BaseMapper<DocumentImportTask>
         + "started_at=COALESCE(started_at, #{now}), finished_at=NULL, "
         + "error_summary=NULL, failure_code=NULL, modified=#{now}, "
         + "modified_by=#{operatorId}, version=COALESCE(version, 0) + 1 "
-        + "WHERE id=#{id} AND status='PENDING' "
+        + "WHERE id=#{id} AND status='PENDING'")
-        + "AND (batch_id IS NULL OR EXISTS ("
-        + "SELECT 1 FROM tb_document_import_batch batch "
-        + "WHERE batch.id=tb_document_import_task.batch_id "
-        + "AND batch.status='RUNNING'))")
     int claimPending(
         @Param("id") BigInteger id,
         @Param("executionToken") String executionToken,
@@ -126,11 +140,7 @@ public interface DocumentImportTaskMapper extends BaseMapper<DocumentImportTask>
         + "modified=#{now}, modified_by=#{operatorId}, "
         + "version=COALESCE(version, 0) + 1 "
         + "WHERE id=#{id} AND status='RUNNING' "
-        + "AND execution_token=#{executionToken} "
+        + "AND execution_token=#{executionToken}")
-        + "AND (batch_id IS NULL OR EXISTS ("
-        + "SELECT 1 FROM tb_document_import_batch batch "
-        + "WHERE batch.id=tb_document_import_task.batch_id "
-        + "AND batch.status='RUNNING'))")
     int renewLease(
         @Param("id") BigInteger id,
         @Param("executionToken") String executionToken,
@@ -156,11 +166,7 @@ public interface DocumentImportTaskMapper extends BaseMapper<DocumentImportTask>
         + "lease_until=NULL, finished_at=#{now}, modified=#{now}, "
         + "modified_by=#{operatorId}, version=COALESCE(version, 0) + 1 "
         + "WHERE id=#{id} AND status='RUNNING' "
-        + "AND execution_token=#{executionToken} "
+        + "AND execution_token=#{executionToken}")
-        + "AND (batch_id IS NULL OR EXISTS ("
-        + "SELECT 1 FROM tb_document_import_batch batch "
-        + "WHERE batch.id=tb_document_import_task.batch_id "
-        + "AND batch.status='RUNNING'))")
     int finishOwned(
         @Param("id") BigInteger id,
         @Param("executionToken") String executionToken,

easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/document/service/impl/DocumentParseBridgeServiceImplTest.java

@@ -6,6 +6,7 @@ import com.easyagents.document.core.entity.ParseResponse;
 import com.easyagents.document.core.entity.ParseResult;
 import com.easyagents.document.core.entity.ParseTaskInfo;
 import com.easyagents.document.core.entity.ParseTaskStatus;
+import com.easyagents.document.core.exception.DocumentAsyncTaskNotFoundException;
 import com.easyagents.document.pdf.PdfDocumentParseService;
 import com.easyagents.document.pptx.PptxDocumentParseService;
 import com.easyagents.document.xlsx.XlsxDocumentParseService;
@@ -128,6 +129,33 @@ public class DocumentParseBridgeServiceImplTest {
         Assert.assertEquals(0, pptxService.queryResultCallCount);
     }
 
+    /**
+     * 验证本地 Office 内存任务丢失会转换为稳定桥接错误码。
+     */
+    @Test
+    public void shouldExposeMissingOfficeTaskAsStableBridgeError() {
+        FakePptxDocumentParseService pptxService =
+            new FakePptxDocumentParseService();
+        pptxService.queryTaskInfoError =
+            new DocumentAsyncTaskNotFoundException("lost-task");
+        DocumentParseBridgeServiceImpl bridgeService =
+            buildBridgeService(null, pptxService, null, null);
+
+        try {
+            bridgeService.queryTaskInfo(
+                "lost-task",
+                buildSource(
+                    "slides.pptx",
+                    "application/vnd.openxmlformats-officedocument.presentationml.presentation"
+                )
+            );
+            Assert.fail("expected DocumentParseBridgeException");
+        } catch (DocumentParseBridgeException error) {
+            Assert.assertEquals("task_not_found", error.getCode());
+            Assert.assertSame(pptxService.queryTaskInfoError, error.getCause());
+        }
+    }
+
     /**
      * 验证缺少底层服务时抛出稳定错误码。
      */
@@ -418,6 +446,7 @@ public class DocumentParseBridgeServiceImplTest {
         private int parseCallCount;
         private int queryTaskInfoCallCount;
         private int queryResultCallCount;
+        private RuntimeException queryTaskInfoError;
 
         @Override
         public ParseResponse parse(ParseRequest request) {
@@ -450,6 +479,9 @@ public class DocumentParseBridgeServiceImplTest {
         @Override
         public ParseTaskInfo queryTaskInfo(String taskId) {
             queryTaskInfoCallCount++;
+            if (queryTaskInfoError != null) {
+                throw queryTaskInfoError;
+            }
             throw new UnsupportedOperationException();
         }
     }

easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchAppServiceTest.java

@@ -873,8 +873,8 @@ public class DocumentImportBatchAppServiceTest {
                 Mockito.anyString(),
                 Mockito.same(recoveryError)
             );
-        Mockito.verify(context.batchMapper, Mockito.never())
+        Mockito.verify(context.recoveryCoordinator, Mockito.never())
-            .finalizeRecoveryPending(
+            .finalizeRecovery(
                 Mockito.any(), Mockito.anyString(), Mockito.any(Date.class));
     }
 
@@ -892,8 +892,9 @@ public class DocumentImportBatchAppServiceTest {
         Mockito.when(context.batchMapper.selectRecoveryPendingBatches(
             Mockito.any(Date.class), Mockito.anyInt()
         )).thenReturn(List.of(batch));
-        Mockito.when(context.batchMapper.selectClaimedRecovery(
+        Mockito.when(context.recoveryCoordinator.claim(
-            Mockito.eq(batch.getId()), Mockito.anyString()
+            Mockito.eq(batch.getId()), Mockito.anyString(),
+            Mockito.any(Date.class), Mockito.any(Date.class)
         )).thenReturn(null);
 
         int recovered = context.service.recoverPendingBatchRetries();
@@ -924,8 +925,9 @@ public class DocumentImportBatchAppServiceTest {
         Mockito.when(context.batchMapper.selectRecoveryPendingBatches(
             Mockito.any(Date.class), Mockito.anyInt()
         )).thenReturn(List.of(batch));
-        Mockito.when(context.batchMapper.selectClaimedRecovery(
+        Mockito.when(context.recoveryCoordinator.claim(
-            Mockito.eq(batch.getId()), Mockito.anyString()
+            Mockito.eq(batch.getId()), Mockito.anyString(),
+            Mockito.any(Date.class), Mockito.any(Date.class)
         )).thenReturn(claimedBatch);
 
         int recovered = context.service.recoverPendingBatchRetries();
@@ -939,16 +941,14 @@ public class DocumentImportBatchAppServiceTest {
             );
         ArgumentCaptor<String> recoveryToken =
             ArgumentCaptor.forClass(String.class);
-        Mockito.verify(context.batchMapper).claimRecoveryPending(
+        Mockito.verify(context.recoveryCoordinator).claim(
             Mockito.eq(batch.getId()),
             recoveryToken.capture(),
             Mockito.any(Date.class),
             Mockito.any(Date.class)
         );
-        Mockito.verify(context.batchMapper).selectClaimedRecovery(
+        Mockito.verify(context.recoveryCoordinator)
-            batch.getId(), recoveryToken.getValue());
+            .finalizeRecovery(
-        Mockito.verify(context.batchMapper)
-            .finalizeRecoveryPending(
                 Mockito.eq(batch.getId()),
                 Mockito.eq(recoveryToken.getValue()),
                 Mockito.any(Date.class));
@@ -975,8 +975,8 @@ public class DocumentImportBatchAppServiceTest {
 
         Assert.assertEquals(1, context.service.recoverPendingBatchRetries());
 
-        Mockito.verify(context.batchMapper, Mockito.never())
+        Mockito.verify(context.recoveryCoordinator, Mockito.never())
-            .finalizeRecoveryPending(
+            .finalizeRecovery(
                 Mockito.any(), Mockito.anyString(), Mockito.any(Date.class));
         Mockito.verifyNoInteractions(context.circuitBreaker);
     }
@@ -990,12 +990,12 @@ public class DocumentImportBatchAppServiceTest {
         BigInteger batchId = BigInteger.valueOf(84);
         String recoveryToken = "recovery-token";
         AtomicLong now = new AtomicLong(1_000L);
-        Mockito.when(context.batchMapper.renewRecoveryPendingLease(
+        Mockito.when(context.recoveryCoordinator.renew(
             Mockito.eq(batchId),
             Mockito.eq(recoveryToken),
             Mockito.any(Date.class),
             Mockito.any(Date.class)
-        )).thenReturn(1, 0);
+        )).thenReturn(true, false);
         BooleanSupplier guard = context.service.createRecoveryLeaseGuard(
             batchId,
             recoveryToken,
@@ -1004,8 +1004,8 @@ public class DocumentImportBatchAppServiceTest {
         );
 
         Assert.assertTrue(guard.getAsBoolean());
-        Mockito.verify(context.batchMapper, Mockito.never())
+        Mockito.verify(context.recoveryCoordinator, Mockito.never())
-            .renewRecoveryPendingLease(
+            .renew(
                 Mockito.any(),
                 Mockito.anyString(),
                 Mockito.any(Date.class),
@@ -1017,8 +1017,8 @@ public class DocumentImportBatchAppServiceTest {
         now.set(121_000L);
         Assert.assertFalse(guard.getAsBoolean());
 
-        Mockito.verify(context.batchMapper, Mockito.times(2))
+        Mockito.verify(context.recoveryCoordinator, Mockito.times(2))
-            .renewRecoveryPendingLease(
+            .renew(
                 Mockito.eq(batchId),
                 Mockito.eq(recoveryToken),
                 Mockito.any(Date.class),
@@ -1198,6 +1198,8 @@ public class DocumentImportBatchAppServiceTest {
         RedisLockExecutor redisLockExecutor = Mockito.mock(RedisLockExecutor.class);
         DocumentImportBatchCircuitBreaker circuitBreaker =
             Mockito.mock(DocumentImportBatchCircuitBreaker.class);
+        DocumentImportRecoveryCoordinator recoveryCoordinator =
+            Mockito.mock(DocumentImportRecoveryCoordinator.class);
         RedisLockExecutor.LockHandle lockHandle = Mockito.mock(RedisLockExecutor.LockHandle.class);
         Mockito.when(redisLockExecutor.tryAcquire(
             Mockito.anyString(), Mockito.any(), Mockito.any()
@@ -1213,21 +1215,16 @@ public class DocumentImportBatchAppServiceTest {
         Mockito.when(batchMapper.selectOwnedForUpdate(
             Mockito.any(), Mockito.anyString(), Mockito.any()
         )).thenReturn(batch);
-        Mockito.when(batchMapper.claimRecoveryPending(
+        Mockito.when(recoveryCoordinator.claim(
-            Mockito.any(),
+            Mockito.any(), Mockito.anyString(),
-            Mockito.anyString(),
+            Mockito.any(Date.class), Mockito.any(Date.class)
-            Mockito.any(Date.class),
-            Mockito.any(Date.class)
-        )).thenReturn(1);
-        Mockito.when(batchMapper.selectClaimedRecovery(
-            Mockito.any(), Mockito.anyString()
         )).thenReturn(batch);
         Mockito.when(taskAppService.retryBatchFailures(
             Mockito.any(), Mockito.anySet(), Mockito.any(BooleanSupplier.class)
         )).thenReturn(true);
-        Mockito.when(batchMapper.finalizeRecoveryPending(
+        Mockito.when(recoveryCoordinator.finalizeRecovery(
             Mockito.any(), Mockito.anyString(), Mockito.any(Date.class)
-        )).thenReturn(1);
+        )).thenReturn(true);
 
         DocumentImportBatchAppService service = new DocumentImportBatchAppService(
             batchService,
@@ -1239,12 +1236,13 @@ public class DocumentImportBatchAppServiceTest {
             itemMapper,
             documentMapper,
             redisLockExecutor,
-            circuitBreaker
+            circuitBreaker,
+            recoveryCoordinator
         );
         return new TestContext(
             service, batchService, itemService, batchTracker,
             taskAppService, batchMapper, itemMapper, documentMapper,
-            circuitBreaker, lockHandle
+            circuitBreaker, recoveryCoordinator, lockHandle
         );
     }
 
@@ -1356,6 +1354,7 @@ public class DocumentImportBatchAppServiceTest {
         DocumentImportBatchItemMapper itemMapper,
         DocumentMapper documentMapper,
         DocumentImportBatchCircuitBreaker circuitBreaker,
+        DocumentImportRecoveryCoordinator recoveryCoordinator,
         RedisLockExecutor.LockHandle lockHandle
     ) {
     }

easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchCircuitBreakerTest.java

@@ -8,7 +8,9 @@ import org.springframework.dao.DataAccessResourceFailureException;
 import org.springframework.jdbc.BadSqlGrammarException;
 import org.springframework.data.redis.RedisSystemException;
 import tech.easyflow.ai.entity.DocumentImportBatch;
+import tech.easyflow.ai.entity.DocumentImportBatchItem;
 import tech.easyflow.ai.entity.DocumentImportTask;
+import tech.easyflow.ai.enums.DocumentImportBatchItemStatus;
 import tech.easyflow.ai.enums.DocumentImportBatchStatus;
 import tech.easyflow.ai.enums.DocumentImportMode;
 import tech.easyflow.ai.enums.DocumentImportTaskPhase;
@@ -50,7 +52,7 @@ public class DocumentImportBatchCircuitBreakerTest {
         DocumentImportTaskMapper taskMapper =
             Mockito.mock(DocumentImportTaskMapper.class);
         Mockito.when(taskMapper.selectOneById(taskId)).thenReturn(task);
-        Mockito.when(batchMapper.selectOneById(batchId)).thenReturn(batch);
+        Mockito.when(batchMapper.selectForUpdate(batchId)).thenReturn(batch);
         Mockito.when(batchMapper.interruptRunningBatchForActiveTask(
             Mockito.eq(batchId),
             Mockito.eq(taskId),
@@ -161,6 +163,7 @@ public class DocumentImportBatchCircuitBreakerTest {
         DocumentImportTaskMapper taskMapper =
             Mockito.mock(DocumentImportTaskMapper.class);
         Mockito.when(batchMapper.selectOneById(batchId)).thenReturn(batch);
+        Mockito.when(batchMapper.selectForUpdate(batchId)).thenReturn(batch);
         Mockito.when(batchMapper.interruptOwnedRecoveryBatch(
             Mockito.eq(batchId),
             Mockito.eq("stale-token"),
@@ -183,6 +186,136 @@ public class DocumentImportBatchCircuitBreakerTest {
             Mockito.any(Date.class), Mockito.any());
     }
 
+    /**
+     * 验证零重排熔断在批次与失败项锁内完成，且未发生并发推进时生效。
+     */
+    @Test
+    public void shouldInterruptRecoveryFailureWhenItemsRemainFailed() {
+        BigInteger batchId = BigInteger.valueOf(42);
+        BigInteger itemId = BigInteger.valueOf(43);
+        String recoveryToken = "current-token";
+        DocumentImportBatch batch = runningAutoBatch(batchId);
+        batch.setRecoveryPending(true);
+        batch.setRecoveryToken(recoveryToken);
+        batch.setRecoveryLeaseUntil(
+            new Date(System.currentTimeMillis() + 60_000));
+        DocumentImportBatchItem item = new DocumentImportBatchItem();
+        item.setId(itemId);
+        item.setBatchId(batchId);
+        item.setStatus(DocumentImportBatchItemStatus.FAILED.name());
+        DocumentImportBatchMapper batchMapper =
+            Mockito.mock(DocumentImportBatchMapper.class);
+        DocumentImportBatchItemMapper itemMapper =
+            Mockito.mock(DocumentImportBatchItemMapper.class);
+        DocumentImportTaskMapper taskMapper =
+            Mockito.mock(DocumentImportTaskMapper.class);
+        Mockito.when(batchMapper.selectForUpdate(batchId)).thenReturn(batch);
+        Mockito.when(itemMapper.selectForUpdate(itemId)).thenReturn(item);
+        Mockito.when(batchMapper.interruptOwnedRecoveryBatch(
+            Mockito.eq(batchId), Mockito.eq(recoveryToken),
+            Mockito.eq("document_import_recovery_failed"),
+            Mockito.anyString(), Mockito.any(Date.class)
+        )).thenReturn(1);
+        DocumentImportBatchCircuitBreaker circuitBreaker =
+            new DocumentImportBatchCircuitBreaker(
+                batchMapper, itemMapper, taskMapper);
+
+        Assert.assertTrue(circuitBreaker.interruptRecoveryBatch(
+            batchId,
+            recoveryToken,
+            new DocumentImportRecoveryException(java.util.List.of(itemId))
+        ));
+
+        org.mockito.InOrder order = Mockito.inOrder(batchMapper, itemMapper);
+        order.verify(batchMapper).selectForUpdate(batchId);
+        order.verify(itemMapper).countActiveItems(batchId);
+        order.verify(itemMapper).selectForUpdate(itemId);
+        order.verify(batchMapper).interruptOwnedRecoveryBatch(
+            Mockito.eq(batchId), Mockito.eq(recoveryToken),
+            Mockito.eq("document_import_recovery_failed"),
+            Mockito.anyString(), Mockito.any(Date.class));
+    }
+
+    /**
+     * 验证错误写回提交后出现的等价并发推进会阻止旧请求熔断整批。
+     */
+    @Test
+    public void shouldKeepRunningBatchWhenRecoveryItemAdvanced() {
+        BigInteger batchId = BigInteger.valueOf(44);
+        BigInteger itemId = BigInteger.valueOf(45);
+        String recoveryToken = "current-token";
+        DocumentImportBatch batch = runningAutoBatch(batchId);
+        batch.setRecoveryPending(true);
+        batch.setRecoveryToken(recoveryToken);
+        batch.setRecoveryLeaseUntil(
+            new Date(System.currentTimeMillis() + 60_000));
+        DocumentImportBatchItem item = new DocumentImportBatchItem();
+        item.setId(itemId);
+        item.setBatchId(batchId);
+        item.setStatus(DocumentImportBatchItemStatus.PENDING.name());
+        DocumentImportBatchMapper batchMapper =
+            Mockito.mock(DocumentImportBatchMapper.class);
+        DocumentImportBatchItemMapper itemMapper =
+            Mockito.mock(DocumentImportBatchItemMapper.class);
+        Mockito.when(batchMapper.selectForUpdate(batchId)).thenReturn(batch);
+        Mockito.when(itemMapper.countActiveItems(batchId)).thenReturn(1);
+        DocumentImportBatchCircuitBreaker circuitBreaker =
+            new DocumentImportBatchCircuitBreaker(
+                batchMapper,
+                itemMapper,
+                Mockito.mock(DocumentImportTaskMapper.class));
+
+        Assert.assertFalse(circuitBreaker.interruptRecoveryBatch(
+            batchId,
+            recoveryToken,
+            new DocumentImportRecoveryException(java.util.List.of(itemId))
+        ));
+        Mockito.verify(batchMapper, Mockito.never())
+            .interruptOwnedRecoveryBatch(
+                Mockito.any(), Mockito.anyString(), Mockito.anyString(),
+                Mockito.anyString(), Mockito.any(Date.class));
+        Mockito.verify(itemMapper, Mockito.never()).selectForUpdate(itemId);
+    }
+
+    /**
+     * 验证失败快照外已有活跃项时，选择性恢复也不能熔断并发任务。
+     */
+    @Test
+    public void shouldKeepRunningBatchWhenAnotherItemAdvancedBeforeSnapshot() {
+        BigInteger batchId = BigInteger.valueOf(46);
+        BigInteger failedItemId = BigInteger.valueOf(47);
+        String recoveryToken = "current-token";
+        DocumentImportBatch batch = runningAutoBatch(batchId);
+        batch.setRecoveryPending(true);
+        batch.setRecoveryToken(recoveryToken);
+        batch.setRecoveryLeaseUntil(
+            new Date(System.currentTimeMillis() + 60_000));
+        DocumentImportBatchMapper batchMapper =
+            Mockito.mock(DocumentImportBatchMapper.class);
+        DocumentImportBatchItemMapper itemMapper =
+            Mockito.mock(DocumentImportBatchItemMapper.class);
+        Mockito.when(batchMapper.selectForUpdate(batchId)).thenReturn(batch);
+        Mockito.when(itemMapper.countActiveItems(batchId)).thenReturn(1);
+        DocumentImportBatchCircuitBreaker circuitBreaker =
+            new DocumentImportBatchCircuitBreaker(
+                batchMapper,
+                itemMapper,
+                Mockito.mock(DocumentImportTaskMapper.class));
+
+        Assert.assertFalse(circuitBreaker.interruptRecoveryBatch(
+            batchId,
+            recoveryToken,
+            new DocumentImportRecoveryException(
+                java.util.List.of(failedItemId))
+        ));
+        Mockito.verify(itemMapper, Mockito.never())
+            .selectForUpdate(failedItemId);
+        Mockito.verify(batchMapper, Mockito.never())
+            .interruptOwnedRecoveryBatch(
+                Mockito.any(), Mockito.anyString(), Mockito.anyString(),
+                Mockito.anyString(), Mockito.any(Date.class));
+    }
+
     /**
      * 验证 SQL 语法错误保留为通用系统异常，避免伪装成数据库不可用。
      */
@@ -192,7 +325,7 @@ public class DocumentImportBatchCircuitBreakerTest {
         DocumentImportBatch batch = runningAutoBatch(batchId);
         DocumentImportBatchMapper batchMapper =
             Mockito.mock(DocumentImportBatchMapper.class);
-        Mockito.when(batchMapper.selectOneById(batchId)).thenReturn(batch);
+        Mockito.when(batchMapper.selectForUpdate(batchId)).thenReturn(batch);
         Mockito.when(batchMapper.interruptRunningBatch(
             Mockito.eq(batchId),
             Mockito.eq("document_import_infrastructure_failure"),
@@ -222,7 +355,7 @@ public class DocumentImportBatchCircuitBreakerTest {
         DocumentImportBatch batch = runningAutoBatch(batchId);
         DocumentImportBatchMapper batchMapper =
             Mockito.mock(DocumentImportBatchMapper.class);
-        Mockito.when(batchMapper.selectOneById(batchId)).thenReturn(batch);
+        Mockito.when(batchMapper.selectForUpdate(batchId)).thenReturn(batch);
         Mockito.when(batchMapper.interruptRunningBatch(
             Mockito.eq(batchId),
             Mockito.eq("database_unavailable"),
@@ -242,6 +375,36 @@ public class DocumentImportBatchCircuitBreakerTest {
         ));
     }
 
+    /**
+     * 验证没有文件重新排队时使用稳定恢复失败语义中断批次。
+     */
+    @Test
+    public void shouldExposeRecoveryFailureInsteadOfGenericInfrastructureError() {
+        BigInteger batchId = BigInteger.valueOf(71);
+        DocumentImportBatch batch = runningAutoBatch(batchId);
+        DocumentImportBatchMapper batchMapper =
+            Mockito.mock(DocumentImportBatchMapper.class);
+        Mockito.when(batchMapper.selectForUpdate(batchId)).thenReturn(batch);
+        Mockito.when(batchMapper.interruptRunningBatch(
+            Mockito.eq(batchId),
+            Mockito.eq("document_import_recovery_failed"),
+            Mockito.eq("未能重新排队任何失败文件，请查看文件错误后继续"),
+            Mockito.any(Date.class)
+        )).thenReturn(1);
+        DocumentImportBatchCircuitBreaker circuitBreaker =
+            new DocumentImportBatchCircuitBreaker(
+                batchMapper,
+                Mockito.mock(DocumentImportBatchItemMapper.class),
+                Mockito.mock(DocumentImportTaskMapper.class)
+            );
+
+        Assert.assertTrue(circuitBreaker.interruptBatch(
+            batchId,
+            new IllegalStateException(
+                "recovery failed", new DocumentImportRecoveryException())
+        ));
+    }
+
     /**
      * 验证恢复与任务熔断 SQL 都包含对应所有权围栏。
      *

easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchTrackerTest.java

@@ -1,6 +1,8 @@
 package tech.easyflow.ai.documentimport.task;
 
 import org.junit.Test;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
 import tech.easyflow.ai.documentimport.DocumentImportBatchDtos;
 import tech.easyflow.ai.entity.DocumentImportBatch;
 import tech.easyflow.ai.entity.DocumentImportBatchItem;
@@ -13,6 +15,7 @@ import tech.easyflow.ai.mapper.DocumentImportBatchMapper;
 import tech.easyflow.ai.service.DocumentImportBatchItemService;
 import tech.easyflow.ai.service.DocumentImportBatchService;
 
+import java.lang.reflect.Method;
 import java.math.BigInteger;
 import java.util.Date;
 
@@ -25,6 +28,7 @@ import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -62,6 +66,39 @@ public class DocumentImportBatchTrackerTest {
         verify(batchService, never()).updateById(batch, false);
     }
 
+    /**
+     * 验证任务变更始终先锁批次，但只有自动批次受运行态门禁约束。
+     */
+    @Test
+    public void shouldAllowManualTaskAfterUploadBatchCompleted() {
+        BigInteger automaticBatchId = BigInteger.valueOf(101);
+        BigInteger manualBatchId = BigInteger.valueOf(102);
+        DocumentImportBatch automatic = new DocumentImportBatch();
+        automatic.setId(automaticBatchId);
+        automatic.setImportMode(DocumentImportMode.AUTO.name());
+        automatic.setStatus(DocumentImportBatchStatus.COMPLETED.name());
+        DocumentImportBatch manual = new DocumentImportBatch();
+        manual.setId(manualBatchId);
+        manual.setImportMode(DocumentImportMode.MANUAL.name());
+        manual.setStatus(DocumentImportBatchStatus.COMPLETED.name());
+        DocumentImportBatchMapper batchMapper =
+            mock(DocumentImportBatchMapper.class);
+        when(batchMapper.selectForUpdate(automaticBatchId))
+            .thenReturn(automatic);
+        when(batchMapper.selectForUpdate(manualBatchId)).thenReturn(manual);
+        DocumentImportBatchTracker tracker = new DocumentImportBatchTracker(
+            mock(DocumentImportBatchService.class),
+            mock(DocumentImportBatchItemService.class),
+            batchMapper,
+            mock(DocumentImportBatchItemMapper.class)
+        );
+
+        assertFalse(tracker.lockBatchForTaskMutation(automaticBatchId));
+        assertTrue(tracker.lockBatchForTaskMutation(manualBatchId));
+        verify(batchMapper).selectForUpdate(automaticBatchId);
+        verify(batchMapper).selectForUpdate(manualBatchId);
+    }
+
     /**
      * 验证兼容可重试计数与人工可继续的失败总数保持一致。
      */
@@ -111,6 +148,8 @@ public class DocumentImportBatchTrackerTest {
         item.setStatus(DocumentImportBatchItemStatus.FAILED.name());
         item.setRetryable(true);
         when(itemService.getById(item.getId())).thenReturn(item);
+        when(batchMapper.selectForUpdate(batch.getId())).thenReturn(batch);
+        when(itemMapper.selectForUpdate(item.getId())).thenReturn(item);
         when(itemMapper.transitionStatus(
             eq(item.getId()),
             eq(DocumentImportBatchItemStatus.FAILED.name()),
@@ -143,6 +182,14 @@ public class DocumentImportBatchTrackerTest {
             eq(-1),
             any(Date.class)
         );
+        org.mockito.InOrder lockOrder = inOrder(batchMapper, itemMapper);
+        lockOrder.verify(batchMapper).selectForUpdate(batch.getId());
+        lockOrder.verify(itemMapper).selectForUpdate(item.getId());
+        lockOrder.verify(itemMapper).transitionStatus(
+            eq(item.getId()), eq(DocumentImportBatchItemStatus.FAILED.name()),
+            eq(DocumentImportBatchItemStage.INDEX.name()),
+            eq(DocumentImportBatchItemStatus.PENDING.name()), eq(null), eq(null),
+            eq(false), eq(1), any(Date.class));
     }
 
     /**
@@ -167,6 +214,8 @@ public class DocumentImportBatchTrackerTest {
         item.setRetryable(true);
         BigInteger documentId = BigInteger.valueOf(99);
         when(itemService.getById(item.getId())).thenReturn(item);
+        when(batchMapper.selectForUpdate(batch.getId())).thenReturn(batch);
+        when(itemMapper.selectForUpdate(item.getId())).thenReturn(item);
         when(itemMapper.bindFailedDocument(
             eq(item.getId()), eq(documentId), any(Date.class)
         )).thenReturn(1);
@@ -184,6 +233,35 @@ public class DocumentImportBatchTrackerTest {
             anyInt(), anyInt(), anyInt(), any(Date.class));
     }
 
+    /**
+     * 验证失败项只能在运行批次内绑定恢复创建的文档。
+     */
+    @Test(expected = tech.easyflow.common.web.exceptions.BusinessException.class)
+    public void shouldRejectRecoveredDocumentOutsideRunningBatch() {
+        DocumentImportBatchService batchService =
+            mock(DocumentImportBatchService.class);
+        DocumentImportBatchItemService itemService =
+            mock(DocumentImportBatchItemService.class);
+        DocumentImportBatchMapper batchMapper =
+            mock(DocumentImportBatchMapper.class);
+        DocumentImportBatchItemMapper itemMapper =
+            mock(DocumentImportBatchItemMapper.class);
+        DocumentImportBatch batch = batch(
+            DocumentImportBatchStatus.PARTIAL_SUCCEEDED, 1);
+        DocumentImportBatchItem item = new DocumentImportBatchItem();
+        item.setId(BigInteger.TEN);
+        item.setBatchId(batch.getId());
+        item.setStatus(DocumentImportBatchItemStatus.FAILED.name());
+        when(itemService.getById(item.getId())).thenReturn(item);
+        when(batchMapper.selectForUpdate(batch.getId())).thenReturn(batch);
+        when(itemMapper.selectForUpdate(item.getId())).thenReturn(item);
+        DocumentImportBatchTracker tracker =
+            new DocumentImportBatchTracker(
+                batchService, itemService, batchMapper, itemMapper);
+
+        tracker.bindDocument(item.getId(), BigInteger.valueOf(99));
+    }
+
     /**
      * 验证迟到任务不能把已完成文件重新改为处理中。
      */
@@ -198,6 +276,9 @@ public class DocumentImportBatchTrackerTest {
         item.setBatchId(BigInteger.ONE);
         item.setStatus(DocumentImportBatchItemStatus.COMPLETED.name());
         when(itemService.getById(item.getId())).thenReturn(item);
+        DocumentImportBatch batch = batch(DocumentImportBatchStatus.RUNNING, 1);
+        when(batchMapper.selectForUpdate(batch.getId())).thenReturn(batch);
+        when(itemMapper.selectForUpdate(item.getId())).thenReturn(item);
 
         DocumentImportBatchTracker tracker =
             new DocumentImportBatchTracker(batchService, itemService, batchMapper, itemMapper);
@@ -210,6 +291,87 @@ public class DocumentImportBatchTrackerTest {
             any(), any(), any(), any(), any(), any(), anyBoolean(), anyInt(), any(Date.class));
     }
 
+    /**
+     * 验证批次中断后，已到达的工作线程不能再迁移文件项。
+     */
+    @Test
+    public void shouldFenceLateTransitionAfterBatchInterrupted() {
+        DocumentImportBatchService batchService = mock(DocumentImportBatchService.class);
+        DocumentImportBatchItemService itemService = mock(DocumentImportBatchItemService.class);
+        DocumentImportBatchMapper batchMapper = mock(DocumentImportBatchMapper.class);
+        DocumentImportBatchItemMapper itemMapper = mock(DocumentImportBatchItemMapper.class);
+        DocumentImportBatch batch = batch(DocumentImportBatchStatus.INTERRUPTED, 1);
+        DocumentImportBatchItem item = new DocumentImportBatchItem();
+        item.setId(BigInteger.TEN);
+        item.setBatchId(batch.getId());
+        item.setStatus(DocumentImportBatchItemStatus.PENDING.name());
+        when(itemService.getById(item.getId())).thenReturn(item);
+        when(batchMapper.selectForUpdate(batch.getId())).thenReturn(batch);
+        when(itemMapper.selectForUpdate(item.getId())).thenReturn(item);
+
+        DocumentImportBatchTracker tracker =
+            new DocumentImportBatchTracker(batchService, itemService, batchMapper, itemMapper);
+
+        assertFalse(tracker.updateItem(item.getId(),
+            DocumentImportBatchItemStage.PARSE,
+            DocumentImportBatchItemStatus.RUNNING,
+            null));
+        verify(itemMapper, never()).transitionStatus(
+            any(), any(), any(), any(), any(), any(), anyBoolean(), anyInt(), any(Date.class));
+    }
+
+    /**
+     * 验证恢复失败原因在提交后回调内仍使用独立事务持久化。
+     *
+     * @throws Exception 方法不存在时抛出
+     */
+    @Test
+    public void failedRetryErrorShouldUseIndependentTransaction()
+        throws Exception {
+        Method method = DocumentImportBatchTracker.class.getMethod(
+            "updateFailedRetryError",
+            BigInteger.class,
+            BigInteger.class,
+            String.class
+        );
+        Transactional transactional = method.getAnnotation(Transactional.class);
+
+        assertEquals(Propagation.REQUIRES_NEW, transactional.propagation());
+    }
+
+    /**
+     * 验证失败原因写回与等价并发推进在同一锁内完成判定。
+     */
+    @Test
+    public void failedRetryErrorShouldRecognizeConcurrentAdvance() {
+        DocumentImportBatchService batchService =
+            mock(DocumentImportBatchService.class);
+        DocumentImportBatchItemService itemService =
+            mock(DocumentImportBatchItemService.class);
+        DocumentImportBatchMapper batchMapper =
+            mock(DocumentImportBatchMapper.class);
+        DocumentImportBatchItemMapper itemMapper =
+            mock(DocumentImportBatchItemMapper.class);
+        DocumentImportBatch batch = batch(DocumentImportBatchStatus.RUNNING, 1);
+        DocumentImportBatchItem item = new DocumentImportBatchItem();
+        item.setId(BigInteger.TEN);
+        item.setBatchId(batch.getId());
+        item.setStatus(DocumentImportBatchItemStatus.PENDING.name());
+        when(itemService.getById(item.getId())).thenReturn(item);
+        when(batchMapper.selectForUpdate(batch.getId())).thenReturn(batch);
+        when(itemMapper.selectForUpdate(item.getId())).thenReturn(item);
+        DocumentImportBatchTracker tracker =
+            new DocumentImportBatchTracker(
+                batchService, itemService, batchMapper, itemMapper);
+
+        assertEquals(
+            DocumentImportBatchTracker.FailedRetryErrorOutcome.ALREADY_ADVANCED,
+            tracker.updateFailedRetryError(
+                item.getId(), batch.getId(), "状态已变化"));
+        verify(itemMapper, never()).updateFailedRetryError(
+            any(), any(), any(), any(Date.class));
+    }
+
     private DocumentImportBatch batch(DocumentImportBatchStatus status, int totalCount) {
         DocumentImportBatch batch = new DocumentImportBatch();
         batch.setId(BigInteger.ONE);

easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/DocumentImportRecoveryCoordinatorTest.java

@@ -0,0 +1,74 @@
+package tech.easyflow.ai.documentimport.task;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+import tech.easyflow.ai.entity.DocumentImportBatch;
+import tech.easyflow.ai.mapper.DocumentImportBatchMapper;
+
+import java.lang.reflect.Method;
+import java.math.BigInteger;
+import java.util.Date;
+
+/**
+ * {@link DocumentImportRecoveryCoordinator} 恢复令牌事务测试。
+ */
+public class DocumentImportRecoveryCoordinatorTest {
+
+    @Test
+    public void claimShouldReturnOnlyCurrentTokenBatch() {
+        DocumentImportBatchMapper mapper =
+            Mockito.mock(DocumentImportBatchMapper.class);
+        DocumentImportRecoveryCoordinator coordinator =
+            new DocumentImportRecoveryCoordinator(mapper);
+        BigInteger batchId = BigInteger.valueOf(81);
+        String token = "token";
+        Date claimedAt = new Date(1_000L);
+        Date leaseUntil = new Date(121_000L);
+        DocumentImportBatch batch = new DocumentImportBatch();
+        batch.setId(batchId);
+        Mockito.when(mapper.claimRecoveryPending(
+            batchId, token, leaseUntil, claimedAt)).thenReturn(1);
+        Mockito.when(mapper.selectClaimedRecovery(batchId, token))
+            .thenReturn(batch);
+
+        Assert.assertSame(batch, coordinator.claim(
+            batchId, token, leaseUntil, claimedAt));
+    }
+
+    @Test
+    public void claimShouldSkipReadWhenTokenWasNotAcquired() {
+        DocumentImportBatchMapper mapper =
+            Mockito.mock(DocumentImportBatchMapper.class);
+        DocumentImportRecoveryCoordinator coordinator =
+            new DocumentImportRecoveryCoordinator(mapper);
+
+        Assert.assertNull(coordinator.claim(
+            BigInteger.ONE, "token", new Date(2_000L), new Date(1_000L)));
+        Mockito.verify(mapper, Mockito.never())
+            .selectClaimedRecovery(Mockito.any(), Mockito.anyString());
+    }
+
+    @Test
+    public void tokenMutationsShouldAlwaysUseIndependentTransactions()
+        throws Exception {
+        assertRequiresNew("claim", BigInteger.class, String.class,
+            Date.class, Date.class);
+        assertRequiresNew("renew", BigInteger.class, String.class,
+            Date.class, Date.class);
+        assertRequiresNew("finalizeRecovery", BigInteger.class,
+            String.class, Date.class);
+    }
+
+    private void assertRequiresNew(String methodName, Class<?>... parameterTypes)
+        throws Exception {
+        Method method = DocumentImportRecoveryCoordinator.class.getMethod(
+            methodName, parameterTypes);
+        Transactional transactional = method.getAnnotation(Transactional.class);
+        Assert.assertNotNull(transactional);
+        Assert.assertEquals(Propagation.REQUIRES_NEW,
+            transactional.propagation());
+    }
+}