From 7a64cfcaa4fcb133a8e4d0584c94ff52ee842a3e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E5=AD=90=E9=BB=98?= <925456043@qq.com> Date: Fri, 7 Aug 2026 12:39:27 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E7=9F=A5=E8=AF=86?= =?UTF-8?q?=E5=BA=93=E5=A4=B1=E8=B4=A5=E6=89=B9=E6=AC=A1=E7=BB=A7=E7=BB=AD?= =?UTF-8?q?=E4=B8=8E=E9=87=8D=E5=88=86=E5=9D=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 统一继续和重试逻辑，恢复全部失败明细 - 向量化失败时废弃旧快照并重新分块 - 增加 BGE-M3 分块硬上限与索引写入结果校验 --- .../task/DocumentImportBatchAppService.java | 97 +--- .../task/DocumentImportBatchTracker.java | 3 +- ...KnowledgeDocumentImportTaskAppService.java | 442 ++++++++++++------ .../task/KnowledgeImportBatchFacade.java | 9 +- .../DocumentImportBatchAppServiceTest.java | 91 +++- .../task/DocumentImportBatchTrackerTest.java | 6 +- .../task/KnowledgeImportBatchFacadeTest.java | 51 ++ .../DocumentImportBatchStatus.test.ts | 74 +++ .../DocumentImportBatchStatus.vue | 4 +- 9 files changed, 539 insertions(+), 238 deletions(-) diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchAppService.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchAppService.java index d38ce287..16aa2980 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchAppService.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchAppService.java @@ -441,9 +441,14 @@ public class DocumentImportBatchAppService { throw new BusinessException("当前批次无需继续"); } assertNoOtherActiveAutoBatch(batch); - if (valueOrZero(batch.getRetryableFailedCount()) <= 0) { - throw new BusinessException("当前批次没有可继续的失败项"); + List failedItems = + listFailedItems(batchId, Set.of()); + if (failedItems.isEmpty()) { + throw new BusinessException("当前批次没有失败项"); } + Set selectedKeys = failedItems.stream() + .map(DocumentImportBatchItem::getClientFileKey) + .collect(java.util.stream.Collectors.toCollection(LinkedHashSet::new)); Date now = new Date(); DocumentImportBatch update = new DocumentImportBatch(); update.setStatus(DocumentImportBatchStatus.RUNNING.name()); @@ -459,8 +464,11 @@ public class DocumentImportBatchAppService { if (claimed <= 0) { throw new BusinessException("批次状态已变化，请刷新后重试"); } - runAfterCommit(() -> resumeBatchFailures(batchId)); - return batchTracker.refreshBatch(batchId); + batch.setStatus(DocumentImportBatchStatus.RUNNING.name()); + batch.setFinishedAt(null); + batch.setModified(now); + runAfterCommit(() -> resumeBatchFailures(batchId, selectedKeys)); + return batchTracker.toStatusResponse(batch); } /** @@ -732,7 +740,7 @@ public class DocumentImportBatchAppService { * * @param taskId 批次任务 ID * @param caller 调用者上下文 - * @param fileKeys 指定文件键；为空时重试全部可恢复失败项 + * @param fileKeys 指定文件键；为空时重试全部失败项 * @return 稳定的重试响应快照 */ @Transactional @@ -759,11 +767,10 @@ public class DocumentImportBatchAppService { } Set requestedFileKeys = fileKeys == null ? Set.of() : fileKeys; List retryItems = - listRetryItems(batch.getId(), requestedFileKeys); + listFailedItems(batch.getId(), requestedFileKeys); if (retryItems.isEmpty()) { - throw new BusinessException(409, 40905, "当前任务没有可重试的失败文件"); + throw new BusinessException(409, 40905, "当前任务没有失败文件"); } - restoreLegacyRetryability(retryItems); Date now = new Date(); int expectedGeneration = valueOrZero(batch.getRetryGeneration()); int claimedGeneration = expectedGeneration + 1; @@ -1140,15 +1147,6 @@ public class DocumentImportBatchAppService { return false; } - /** - * 在批次状态提交后启动失败项重试；启动器异常时恢复为可继续状态。 - * - * @param batchId 批次 ID - */ - private void resumeBatchFailures(BigInteger batchId) { - resumeBatchFailures(batchId, Set.of()); - } - /** * 在批次状态提交后启动选定失败项重试。 * @@ -1165,14 +1163,14 @@ public class DocumentImportBatchAppService { } /** - * 查询并校验本次重试选中的可恢复失败项。 + * 查询并校验本次人工重试选中的失败项。 * * @param batchId 批次 ID * @param fileKeys 指定文件键 - * @return 可恢复失败项 + * @return 失败项 */ - private List listRetryItems(BigInteger batchId, - Set fileKeys) { + private List listFailedItems(BigInteger batchId, + Set fileKeys) { QueryWrapper wrapper = QueryWrapper.create() .eq(DocumentImportBatchItem::getBatchId, batchId) .eq(DocumentImportBatchItem::getStatus, DocumentImportBatchItemStatus.FAILED.name()); @@ -1185,64 +1183,13 @@ public class DocumentImportBatchAppService { throw new BusinessException("fileKeys 包含空值或重复值"); } wrapper.in(DocumentImportBatchItem::getClientFileKey, normalized); - List selected = itemService.list(wrapper).stream() - .filter(this::isRetryCandidate) - .toList(); + List selected = itemService.list(wrapper); if (selected.size() != normalized.size()) { - throw new BusinessException(409, 40906, "部分文件当前不可重试"); + throw new BusinessException(409, 40906, "部分文件当前不是失败状态"); } return selected; } - return itemService.list(wrapper).stream() - .filter(this::isRetryCandidate) - .toList(); - } - - /** - * 判断失败项是否可以进入本次人工重试。 - * - * @param item 批次失败项 - * @return 是否可以重试 - */ - private boolean isRetryCandidate(DocumentImportBatchItem item) { - return Boolean.TRUE.equals(item.getRetryable()) - || taskAppService.isRecoverableBatchFailure(item); - } - - /** - * 在领取重试前恢复旧失败项的持久化重试资格及批次计数。 - * - * @param retryItems 本次重试项 - */ - private void restoreLegacyRetryability( - List retryItems) { - for (DocumentImportBatchItem item : retryItems) { - if (Boolean.TRUE.equals(item.getRetryable())) { - continue; - } - DocumentImportBatchItemStage stage; - try { - stage = DocumentImportBatchItemStage.valueOf(item.getStage()); - } catch (IllegalArgumentException | NullPointerException error) { - throw new BusinessException(409, 40906, "部分文件当前不可重试"); - } - if (!batchTracker.transitionItem( - item.getId(), - stage, - DocumentImportBatchItemStatus.FAILED, - item.getErrorSummary(), - true, - 0, - item.getFailureCode() - )) { - throw new BusinessException( - 409, - 40903, - "文件状态已变化，请刷新任务状态" - ); - } - item.setRetryable(true); - } + return itemService.list(wrapper); } /** diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchTracker.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchTracker.java index 206b5c3e..e2bed47b 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchTracker.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchTracker.java @@ -401,7 +401,8 @@ public class DocumentImportBatchTracker { response.setPendingCount(valueOrZero(batch.getPendingCount())); response.setSkippedCount(valueOrZero(batch.getSkippedCount())); response.setCancelledCount(valueOrZero(batch.getCancelledCount())); - response.setRetryableFailedCount(valueOrZero(batch.getRetryableFailedCount())); + // 人工继续统一覆盖全部失败项，兼容字段返回相同数量，避免旧计数影响前端判断。 + response.setRetryableFailedCount(valueOrZero(batch.getFailedCount())); int total = Math.max(1, valueOrZero(batch.getTotalCount())); int terminalCount = valueOrZero(batch.getCompletedCount()) + valueOrZero(batch.getFailedCount()) diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/KnowledgeDocumentImportTaskAppService.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/KnowledgeDocumentImportTaskAppService.java index 0338131a..5d754305 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/KnowledgeDocumentImportTaskAppService.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/KnowledgeDocumentImportTaskAppService.java @@ -8,6 +8,7 @@ import com.easyagents.core.model.embedding.EmbeddingOptions; import com.easyagents.core.store.DocumentStore; import com.easyagents.core.store.StoreOptions; import com.easyagents.core.store.StoreResult; +import com.easyagents.rag.core.BgeM3ChunkSafety; import com.easyagents.rag.core.RagChunk; import com.easyagents.rag.core.RagDefaults; import com.easyagents.rag.core.RagStrategyCodes; @@ -129,6 +130,8 @@ public class KnowledgeDocumentImportTaskAppService { private static final String TASK_ERROR_EXECUTION_INTERRUPTED = "execution_interrupted"; private static final String TASK_ERROR_SPLIT_FAILED = "split_failed"; private static final String TASK_ERROR_INDEX_FAILED = "index_failed"; + private static final String DEFAULT_INDEX_FAILURE_MESSAGE = "分块或向量化失败，请重试"; + private static final String ROLLBACK_FAILURE_SUFFIX = "；外部索引回滚未完成，请联系管理员处理"; private static final Pattern HTTP_SERVER_ERROR_PATTERN = Pattern.compile("\\bstatus=5\\d{2}\\b"); private static final Pattern MARKDOWN_IMAGE_PATTERN = Pattern.compile("!\\[(?:[^\\]]*)\\]\\(([^)]+)\\)"); private final FlexIDKeyGenerator flexIdKeyGenerator = new FlexIDKeyGenerator(); @@ -316,8 +319,8 @@ public class KnowledgeDocumentImportTaskAppService { buildDocumentPayload(document)); documentImportBatchTracker.bindDocument(item.getId(), document.getId()); taskIds.add(task.getId()); - documentImportTaskStatusStreamService.publishAfterCommit(document.getId()); } + // 启动接口返回后列表会统一刷新；批量建档阶段不逐文件推送，避免提交后形成 SSE 风暴。 int initialDispatchCount = Math.min( taskIds.size(), Math.max(1, bulkProperties.getPerBatchParseMaxRunning()) @@ -339,13 +342,12 @@ public class KnowledgeDocumentImportTaskAppService { * 重试批次中选定的失败文件，单个文件重试失败不会中止其他文件。 * * @param batchId 批次 ID - * @param fileKeys 指定文件键；为空时重试全部可恢复失败项 + * @param fileKeys 指定文件键；为空时重试全部失败项 */ public void retryBatchFailures(BigInteger batchId, Set fileKeys) { QueryWrapper query = QueryWrapper.create() .eq(DocumentImportBatchItem::getBatchId, batchId) - .eq(DocumentImportBatchItem::getStatus, DocumentImportBatchItemStatus.FAILED.name()) - .eq(DocumentImportBatchItem::getRetryable, true); + .eq(DocumentImportBatchItem::getStatus, DocumentImportBatchItemStatus.FAILED.name()); if (fileKeys != null && !fileKeys.isEmpty()) { query.in(DocumentImportBatchItem::getClientFileKey, fileKeys); } @@ -827,6 +829,15 @@ public class KnowledgeDocumentImportTaskAppService { */ @Transactional public Result retryIndexTask(DocumentImportDtos.TaskRetryRequest request) { + DocumentCollection knowledge = assertDocumentCollection(request.getKnowledgeId()); + tech.easyflow.ai.entity.Document document = + requireDocumentForKnowledge(request.getDocumentId(), knowledge.getId()); + BigInteger batchId = optionAsBigInteger( + document.getOptions(), DocumentImportKeys.KEY_DOCUMENT_IMPORT_BATCH_ID); + if (DocumentProcessStatus.INDEX_FAILED.name().equals(document.getProcessStatus()) + && isAutomaticBatch(batchId)) { + return retryFailedTask(request); + } DocumentImportDtos.TaskStartIndexRequest startRequest = new DocumentImportDtos.TaskStartIndexRequest(); startRequest.setKnowledgeId(request.getKnowledgeId()); startRequest.setDocumentId(request.getDocumentId()); @@ -849,21 +860,25 @@ public class KnowledgeDocumentImportTaskAppService { if (DocumentProcessStatus.PARSE_FAILED.name().equals(status)) { return retryParseTask(request); } - if (DocumentProcessStatus.INDEX_FAILED.name().equals(status)) { - return retryIndexTask(request); - } - if (!DocumentProcessStatus.SPLIT_FAILED.name().equals(status)) { - throw new BusinessException("当前文档无需重试"); - } - if (!claimDocumentSplitRetry(document)) { - throw new BusinessException("文档状态已变化，请刷新后重试"); - } BigInteger batchId = optionAsBigInteger( document.getOptions(), DocumentImportKeys.KEY_DOCUMENT_IMPORT_BATCH_ID); BigInteger batchItemId = optionAsBigInteger( document.getOptions(), DocumentImportKeys.KEY_DOCUMENT_IMPORT_BATCH_ITEM_ID); + if (DocumentProcessStatus.INDEX_FAILED.name().equals(status) + && !isAutomaticBatch(batchId)) { + return retryIndexTask(request); + } + if (!DocumentProcessStatus.SPLIT_FAILED.name().equals(status) + && !DocumentProcessStatus.INDEX_FAILED.name().equals(status)) { + throw new BusinessException("当前文档无需重试"); + } + if (!claimDocumentSplitRetry(document, status)) { + throw new BusinessException("文档状态已变化，请刷新后重试"); + } + String staleSnapshotPath = invalidateChunkSnapshot(document); DocumentImportTask splitTask = enqueueAutomaticSplit(batchId, batchItemId, document); + deleteChunkSnapshotAfterCommit(staleSnapshotPath); DocumentImportDtos.TaskStartIndexResponse response = new DocumentImportDtos.TaskStartIndexResponse(); response.setTaskId(splitTask.getId()); response.setProcessStatus(DocumentProcessStatus.SPLITTING.name()); @@ -874,19 +889,6 @@ public class KnowledgeDocumentImportTaskAppService { if (item.getDocumentId() == null) { throw new BusinessException("失败文件尚未生成文档"); } - int completedRetries = item.getAttemptCount() == null ? 0 : item.getAttemptCount(); - if (completedRetries >= Math.max(1, bulkProperties.getMaxTaskAttempts()) - 1) { - documentImportBatchTracker.transitionItem( - item.getId(), - DocumentImportBatchItemStage.valueOf(item.getStage()), - DocumentImportBatchItemStatus.FAILED, - item.getErrorSummary(), - false, - 0, - item.getFailureCode() - ); - throw new BusinessException("该文件已达到最大重试次数"); - } tech.easyflow.ai.entity.Document document = requireDocument(item.getDocumentId()); DocumentImportDtos.TaskRetryRequest request = new DocumentImportDtos.TaskRetryRequest(); request.setKnowledgeId(item.getKnowledgeId()); @@ -1232,10 +1234,14 @@ public class KnowledgeDocumentImportTaskAppService { return; } clearPersistedChunks(document.getId()); + boolean rollbackSucceeded = true; if (storeContext != null && !storedChunks.isEmpty()) { - rollbackStoredChunks(taskId, document.getId(), storeContext, storedChunks); + rollbackSucceeded = rollbackStoredChunks( + taskId, document.getId(), storeContext, storedChunks); } - markIndexFailed(task, document, "分块或向量化失败，请重试"); + markIndexFailed( + task, document, + resolveIndexFailureMessage(e, rollbackSucceeded)); } finally { closeStoreContext(storeContext); } @@ -1441,8 +1447,7 @@ public class KnowledgeDocumentImportTaskAppService { task.getBatchItemId(), DocumentImportBatchItemStage.SPLIT, errorMessage, - failureCode, - canRetryBatchItem(task.getBatchItemId()) + failureCode ); return true; } @@ -1578,8 +1583,7 @@ public class KnowledgeDocumentImportTaskAppService { task.getBatchItemId(), stage, errorMessage, - failureCode, - canRetryBatchItem(task.getBatchItemId())); + failureCode); if (task.getBatchId() != null) { documentImportBatchTracker.markInterrupted(task.getBatchId()); } @@ -1714,9 +1718,7 @@ public class KnowledgeDocumentImportTaskAppService { task.getBatchItemId(), DocumentImportBatchItemStage.PARSE, errorMessage, - errorCode, - isRetryableParseFailure(errorCode) - && canRetryBatchItem(task.getBatchItemId())); + errorCode); return true; } @@ -2177,16 +2179,55 @@ public class KnowledgeDocumentImportTaskAppService { null); } - private void markIndexFailed(DocumentImportTask task, - tech.easyflow.ai.entity.Document document, - String errorMessage) { + /** + * 使用通用索引失败码标记任务失败。 + * + * @param task 索引任务 + * @param document 文档 + * @param errorMessage 用户可见错误信息 + */ + private void markIndexFailed( + DocumentImportTask task, + tech.easyflow.ai.entity.Document document, + String errorMessage) { + markIndexFailed( + task, document, errorMessage, TASK_ERROR_INDEX_FAILED); + } + + /** + * 将索引任务标记为失败并保留阶段特定失败码。 + * + * @param task 索引任务 + * @param document 文档 + * @param errorMessage 用户可见错误信息 + * @param failureCode 稳定失败码 + */ + private void markIndexFailed( + DocumentImportTask task, + tech.easyflow.ai.entity.Document document, + String errorMessage, + String failureCode) { KnowledgeDocumentImportTaskAppService executor = selfProxy == null ? this : selfProxy; - if (!executor.failIndexTask(task, document, errorMessage)) { + if (!executor.failIndexTask(task, document, errorMessage, failureCode)) { LOG.warn("向量化任务所有权已失效，忽略迟到失败: taskId={}, documentId={}", task.getId(), document.getId()); } } + /** + * 生成可持久化的索引失败原因。 + * + * @param error 原始异常 + * @param rollbackSucceeded 外部索引是否完整回滚 + * @return 对用户安全且可执行的失败原因 + */ + private String resolveIndexFailureMessage(Exception error, boolean rollbackSucceeded) { + String message = error instanceof BusinessException && StringUtil.hasText(error.getMessage()) + ? error.getMessage() + : DEFAULT_INDEX_FAILURE_MESSAGE; + return rollbackSucceeded ? message : message + ROLLBACK_FAILURE_SUFFIX; + } + /** * 原子写入向量化任务、文档与批次项失败状态。 * @@ -2196,13 +2237,35 @@ public class KnowledgeDocumentImportTaskAppService { * @return 是否仍持有任务终态写入权 */ @Transactional(propagation = Propagation.REQUIRES_NEW) + public boolean failIndexTask( + DocumentImportTask task, + tech.easyflow.ai.entity.Document document, + String errorMessage) { + return failIndexTask( + task, document, errorMessage, TASK_ERROR_INDEX_FAILED); + } + + /** + * 原子写入带阶段失败码的向量化任务、文档与批次项失败状态。 + * + * @param task 向量化任务 + * @param document 文档实体 + * @param errorMessage 用户可见错误信息 + * @param failureCode 稳定失败码 + * @return 是否仍持有任务终态写入权 + */ + @Transactional(propagation = Propagation.REQUIRES_NEW) public boolean failIndexTask(DocumentImportTask task, tech.easyflow.ai.entity.Document document, - String errorMessage) { + String errorMessage, + String failureCode) { + String resolvedFailureCode = StringUtil.hasText(failureCode) + ? failureCode + : TASK_ERROR_INDEX_FAILED; Date now = new Date(); if (!finishTask( task, now, DocumentImportTaskStatus.FAILED, - errorMessage, TASK_ERROR_INDEX_FAILED)) { + errorMessage, resolvedFailureCode)) { return false; } tech.easyflow.ai.entity.Document current = requireDocument(document.getId()); @@ -2211,7 +2274,7 @@ public class KnowledgeDocumentImportTaskAppService { current.setCompletedChunks(0); current.setFailedChunks(defaultInt(current.getTotalChunks())); current.setProgressPercent(0); - setDocumentTaskError(current, errorMessage, TASK_ERROR_INDEX_FAILED); + setDocumentTaskError(current, errorMessage, resolvedFailureCode); persistDocumentTaskState(current, now); LOG.warn("文档向量化任务失败: taskId={}, documentId={}, processStatus={}, completedChunks={}, totalChunks={}, error={}", task.getId(), @@ -2225,8 +2288,7 @@ public class KnowledgeDocumentImportTaskAppService { task.getBatchItemId(), DocumentImportBatchItemStage.INDEX, errorMessage, - TASK_ERROR_INDEX_FAILED, - canRetryBatchItem(task.getBatchItemId())); + resolvedFailureCode); return true; } @@ -2394,12 +2456,14 @@ public class KnowledgeDocumentImportTaskAppService { } /** - * 按分块失败状态原子领取重试权。 + * 按指定失败状态原子领取重新分块重试权。 * * @param document 文档实体 + * @param expectedStatus 领取前预期的失败状态 * @return 是否领取成功 */ - private boolean claimDocumentSplitRetry(tech.easyflow.ai.entity.Document document) { + private boolean claimDocumentSplitRetry(tech.easyflow.ai.entity.Document document, + String expectedStatus) { Date now = new Date(); tech.easyflow.ai.entity.Document update = new tech.easyflow.ai.entity.Document(); update.setProcessStatus(DocumentProcessStatus.SPLITTING.name()); @@ -2411,7 +2475,7 @@ public class KnowledgeDocumentImportTaskAppService { QueryWrapper.create() .eq(tech.easyflow.ai.entity.Document::getId, document.getId()) .eq(tech.easyflow.ai.entity.Document::getProcessStatus, - DocumentProcessStatus.SPLIT_FAILED.name())); + expectedStatus)); if (updated > 0) { document.setProcessStatus(DocumentProcessStatus.SPLITTING.name()); return true; @@ -2419,6 +2483,28 @@ public class KnowledgeDocumentImportTaskAppService { return false; } + /** + * 移除文档对旧分块快照的引用，确保重试只能使用新分块结果。 + * + * @param document 待重新分块的文档 + * @return 失效的旧快照路径 + */ + private String invalidateChunkSnapshot(tech.easyflow.ai.entity.Document document) { + if (document.getOptions() == null || document.getOptions().isEmpty()) { + return null; + } + String snapshotPath = optionAsString( + document.getOptions(), DocumentImportKeys.KEY_DOCUMENT_CHUNK_SNAPSHOT_PATH); + if (!StringUtil.hasText(snapshotPath)) { + return null; + } + Map options = + new LinkedHashMap(document.getOptions()); + options.remove(DocumentImportKeys.KEY_DOCUMENT_CHUNK_SNAPSHOT_PATH); + document.setOptions(options); + return snapshotPath; + } + private void updateDocumentIndexProgress(BigInteger documentId, int totalChunks, int completedChunks) { int progressPercent = Math.min(100, totalChunks <= 0 ? 0 : (completedChunks * 100) / totalChunks); tech.easyflow.ai.entity.Document document = requireDocument(documentId); @@ -2736,16 +2822,18 @@ public class KnowledgeDocumentImportTaskAppService { String sourceFormat, StrategyConfig strategyConfig, Map parseArtifactSummary) { + List chunks; if ("pptx".equals(sourceFormat)) { - return buildPptxDocumentChunks(document, parseArtifactSummary); - } - if ("xlsx".equals(sourceFormat)) { + chunks = buildPptxDocumentChunks(document, parseArtifactSummary); + } else if ("xlsx".equals(sourceFormat)) { int rowsPerChunk = strategyConfig == null || strategyConfig.getRowsPerChunk() == null ? 10 : Math.max(1, strategyConfig.getRowsPerChunk()); - return buildXlsxDocumentChunks(document, parseArtifactSummary, rowsPerChunk); + chunks = buildXlsxDocumentChunks(document, parseArtifactSummary, rowsPerChunk); + } else { + chunks = new ArrayList(); } - return new ArrayList(); + return enforceDocumentChunkHardLimit(chunks); } @SuppressWarnings("unchecked") @@ -2916,6 +3004,94 @@ public class KnowledgeDocumentImportTaskAppService { return chunk; } + /** + * 对 Office 分块应用统一的 BGE-M3 上下文硬限制。 + * + * @param chunks Office 构建器生成的原始分块 + * @return 内容均处于安全预算内且排序连续的分块 + */ + private List enforceDocumentChunkHardLimit( + List chunks) { + List safeChunks = new ArrayList(); + if (chunks == null || chunks.isEmpty()) { + return safeChunks; + } + int sorting = 1; + for (DocumentChunk chunk : chunks) { + String content = chunk == null || chunk.getContent() == null + ? "" + : chunk.getContent(); + List parts = + BgeM3ChunkSafety.splitToHardLimit(content); + if (parts.isEmpty()) { + continue; + } + for (int index = 0; index < parts.size(); index++) { + BgeM3ChunkSafety.ChunkRange part = parts.get(index); + String partContent = + content.substring(part.getStart(), part.getEnd()).trim(); + if (!StringUtil.hasText(partContent)) { + continue; + } + DocumentChunk safeChunk = parts.size() == 1 + ? chunk + : copyDocumentChunkPart(chunk); + safeChunk.setSorting(sorting++); + safeChunk.setContent(partContent); + Map options = copyOptions(safeChunk.getOptions()); + options.put("charCount", partContent.length()); + options.put("tokenEstimate", Math.max( + 1, BgeM3ChunkSafety.estimateContentTokens(partContent))); + options.put("partNo", index + 1); + options.put("partTotal", parts.size()); + if (parts.size() > 1) { + options.put("hardSplit", Boolean.TRUE); + options.put( + "hardSplitTokenLimit", + RagDefaults.BGE_M3_HARD_CHUNK_TOKEN_LIMIT); + boundSplitRenderMarkdown(options, partContent); + } + safeChunk.setOptions(options); + safeChunks.add(safeChunk); + } + } + return safeChunks; + } + + /** + * 复制一个待继续切分的文档块，并为子块生成独立主键。 + * + * @param source 原始分块 + * @return 不共享可变元数据的子块 + */ + private DocumentChunk copyDocumentChunkPart(DocumentChunk source) { + DocumentChunk copy = new DocumentChunk(); + copy.setId(generateId(copy)); + copy.setDocumentId(source.getDocumentId()); + copy.setDocumentCollectionId(source.getDocumentCollectionId()); + copy.setOptions(copyOptions(source.getOptions())); + return copy; + } + + /** + * 避免强制子块继续复制整页超长渲染正文。 + * + * @param options 子块元数据 + * @param partContent 当前子块正文 + */ + private void boundSplitRenderMarkdown( + Map options, + String partContent) { + Object renderMarkdown = + options.get(DocumentImportKeys.KEY_DOCUMENT_RENDER_MARKDOWN); + if (renderMarkdown instanceof String + && !BgeM3ChunkSafety.isWithinHardLimit((String) renderMarkdown)) { + options.put( + DocumentImportKeys.KEY_DOCUMENT_RENDER_MARKDOWN, + partContent); + } + } + private String buildPptxChunkRenderMarkdown(int slideIndex, String title, String imageName, @@ -3328,6 +3504,8 @@ public class KnowledgeDocumentImportTaskAppService { config.setChunkSize(asInteger(rawSnapshot.get("chunkSize"), RagDefaults.CHUNK_SIZE)); config.setOverlapSize(asInteger(rawSnapshot.get("overlapSize"), RagDefaults.OVERLAP_SIZE)); config.setRegex(asString(rawSnapshot.get("regex"))); + config.setRetainRegexMatch( + asBoolean(rawSnapshot.get("retainRegexMatch"), false)); config.setRowsPerChunk(asInteger(rawSnapshot.get("rowsPerChunk"), config.getRowsPerChunk())); config.setMdSplitterLevel(asInteger(rawSnapshot.get("mdSplitterLevel"), RagDefaults.MD_SPLITTER_LEVEL)); return config; @@ -3534,69 +3712,6 @@ public class KnowledgeDocumentImportTaskAppService { } } - /** - * 判断批次项是否仍未达到最大尝试次数。 - * - * @param itemId 批次文件项 ID - * @return 是否允许继续重试 - */ - private boolean canRetryBatchItem(BigInteger itemId) { - if (itemId == null) { - return false; - } - DocumentImportBatchItem item = - documentImportBatchTracker.requireItem(itemId); - int completedRetries = - item.getAttemptCount() == null ? 0 : item.getAttemptCount(); - return completedRetries - < Math.max(1, bulkProperties.getMaxTaskAttempts()) - 1; - } - - /** - * 判断历史批次失败项能否按当前失败分类恢复重试资格。 - * - *

用于兼容修复前已经写入 {@code retryable=false} 的失败项。 - * 新失败项仍在写入终态时持久化重试资格，避免正常路径重复计算。

- * - * @param item 批次失败项 - * @return 是否可以恢复重试资格 - */ - public boolean isRecoverableBatchFailure(DocumentImportBatchItem item) { - if (item == null - || !DocumentImportBatchItemStatus.FAILED.name().equals(item.getStatus()) - || (item.getAttemptCount() != null - && item.getAttemptCount() - >= Math.max(1, bulkProperties.getMaxTaskAttempts()) - 1)) { - return false; - } - DocumentImportBatchItemStage stage; - try { - stage = DocumentImportBatchItemStage.valueOf(item.getStage()); - } catch (IllegalArgumentException | NullPointerException error) { - return false; - } - return switch (stage) { - case PARSE -> isRetryableParseFailure(item.getFailureCode()); - case SPLIT, INDEX -> true; - case UPLOAD, DONE -> false; - }; - } - - /** - * 判断解析失败是否属于可恢复故障。 - * - *

只有已明确识别的输入或请求问题禁止重试。文档源暂时不可用、 - * 解析服务异常和未识别的系统错误允许人工重试，并继续受最大尝试 - * 次数约束，便于服务恢复或代码修复后沿用原任务继续处理。

- * - * @param failureCode 稳定失败码 - * @return 是否可恢复 - */ - private boolean isRetryableParseFailure(String failureCode) { - return !TASK_ERROR_UNSUPPORTED_DOCUMENT_SOURCE.equals(failureCode) - && !TASK_ERROR_INVALID_PARSE_REQUEST.equals(failureCode); - } - private void updateBatchItem(BigInteger itemId, DocumentImportBatchItemStage stage, DocumentImportBatchItemStatus status, @@ -3613,14 +3728,12 @@ public class KnowledgeDocumentImportTaskAppService { * @param stage 失败阶段 * @param errorSummary 错误摘要 * @param failureCode 稳定失败码 - * @param retryable 是否允许重试 */ private void updateBatchItemFailure( BigInteger itemId, DocumentImportBatchItemStage stage, String errorSummary, - String failureCode, - boolean retryable) { + String failureCode) { if (itemId == null) { return; } @@ -3629,7 +3742,7 @@ public class KnowledgeDocumentImportTaskAppService { stage, DocumentImportBatchItemStatus.FAILED, errorSummary, - retryable, + true, 0, failureCode ); @@ -3694,6 +3807,7 @@ public class KnowledgeDocumentImportTaskAppService { private void storeDocumentChunks(StoreExecutionContext storeContext, List documentChunks) { List documents = new ArrayList(); for (DocumentChunk chunk : documentChunks) { + assertEmbeddingChunkWithinHardLimit(chunk); Document storeDocument = new Document(); storeDocument.setId(chunk.getId()); storeDocument.setContent(chunk.getContent()); @@ -3705,10 +3819,25 @@ public class KnowledgeDocumentImportTaskAppService { if (result == null || !result.isSuccess()) { throw new BusinessException("向量化写入失败"); } - if (storeContext.searcher != null) { - for (Document storeDocument : documents) { - storeContext.searcher.addDocument(storeDocument); - } + if (storeContext.searcher != null && !storeContext.searcher.addDocuments(documents)) { + throw new BusinessException("关键词索引写入失败"); + } + } + + /** + * 在调用向量模型前校验最终分块预算，阻止遗漏入口发送超长正文。 + * + * @param chunk 待向量化分块 + * @throws BusinessException 分块超过 BGE-M3 安全上限 + */ + private void assertEmbeddingChunkWithinHardLimit(DocumentChunk chunk) { + String content = chunk == null ? null : chunk.getContent(); + int tokenEstimate = BgeM3ChunkSafety.estimateContentTokens(content); + if (tokenEstimate > RagDefaults.BGE_M3_HARD_CHUNK_TOKEN_LIMIT) { + BigInteger chunkId = chunk == null ? null : chunk.getId(); + throw new BusinessException( + "分块内容超过向量模型上下文上限，请重新分块：chunkId=" + + chunkId + "，tokenEstimate=" + tokenEstimate); } } @@ -3720,10 +3849,19 @@ public class KnowledgeDocumentImportTaskAppService { } } - private void rollbackStoredChunks(BigInteger taskId, - BigInteger documentId, - StoreExecutionContext storeContext, - List documentChunks) { + /** + * 回滚已写入的向量和关键词索引。 + * + * @param taskId 任务 ID + * @param documentId 文档 ID + * @param storeContext 外部存储上下文 + * @param documentChunks 待回滚分块 + * @return 两类外部索引均回滚成功时返回 {@code true} + */ + private boolean rollbackStoredChunks(BigInteger taskId, + BigInteger documentId, + StoreExecutionContext storeContext, + List documentChunks) { try { Set uniqueIds = new LinkedHashSet(); for (DocumentChunk chunk : documentChunks) { @@ -3732,6 +3870,9 @@ public class KnowledgeDocumentImportTaskAppService { } } List ids = new ArrayList(uniqueIds); + if (ids.isEmpty()) { + return true; + } LOG.warn("开始回滚文档向量化外部索引: taskId={}, documentId={}, knowledgeId={}, chunkCount={}", taskId, documentId, @@ -3742,18 +3883,18 @@ public class KnowledgeDocumentImportTaskAppService { String failReason = deleteResult == null ? "未返回结果" : deleteResult.getFailReason(); throw new IllegalStateException("向量存储回滚失败: " + failReason); } - if (storeContext.searcher != null) { - for (BigInteger id : ids) { - storeContext.searcher.deleteDocument(id); - } + if (storeContext.searcher != null && !storeContext.searcher.deleteDocuments(ids)) { + throw new IllegalStateException("关键词索引回滚失败"); } LOG.warn("文档向量化外部索引回滚完成: taskId={}, documentId={}, knowledgeId={}, chunkCount={}", taskId, documentId, storeContext == null || storeContext.knowledge == null ? null : storeContext.knowledge.getId(), ids.size()); + return true; } catch (Exception e) { LOG.error("回滚文档向量数据失败", e); + return false; } } @@ -4203,6 +4344,18 @@ public class KnowledgeDocumentImportTaskAppService { } } + /** + * 在重试事务提交后删除已经失效的旧分块快照。 + * + * @param snapshotPath 旧分块快照路径 + */ + private void deleteChunkSnapshotAfterCommit(String snapshotPath) { + if (!StringUtil.hasText(snapshotPath)) { + return; + } + runAfterCommit(() -> deleteChunkSnapshotAfterCompletion(snapshotPath)); + } + private StrategyConfig resolveStrategyConfig(DocumentCollection knowledge, StrategyConfig requestConfig, AnalysisResult analysisResult) { @@ -4247,9 +4400,27 @@ public class KnowledgeDocumentImportTaskAppService { if (config.getMdSplitterLevel() == null || config.getMdSplitterLevel() <= 0) { config.setMdSplitterLevel(RagDefaults.MD_SPLITTER_LEVEL); } + if (config.getRetainRegexMatch() == null) { + config.setRetainRegexMatch(Boolean.FALSE); + } + validateStrategyOverlap(config); return config; } + /** + * 校验长度分块参数能够保证分块游标持续前进。 + * + * @param config 最终生效的分块策略 + * @throws BusinessException 重叠大小不小于分块大小 + */ + private void validateStrategyOverlap(StrategyConfig config) { + if (config.getOverlapSize() != null + && config.getChunkSize() != null + && config.getOverlapSize() >= config.getChunkSize()) { + throw new BusinessException("分块重叠大小必须小于分块大小"); + } + } + /** * 任务已被看门狗或其他执行者收口时抛出的内部异常。 */ @@ -4285,6 +4456,10 @@ public class KnowledgeDocumentImportTaskAppService { config.setChunkSize(asInteger(rawProfile.get("chunkSize"), config.getChunkSize())); config.setOverlapSize(asInteger(rawProfile.get("overlapSize"), config.getOverlapSize())); config.setRegex(asString(rawProfile.get("regex"))); + if (rawProfile.containsKey("retainRegexMatch")) { + config.setRetainRegexMatch( + asBoolean(rawProfile.get("retainRegexMatch"), false)); + } config.setRowsPerChunk(asInteger(rawProfile.get("rowsPerChunk"), config.getRowsPerChunk())); config.setMdSplitterLevel(asInteger(rawProfile.get("mdSplitterLevel"), config.getMdSplitterLevel())); return config; @@ -4306,6 +4481,9 @@ public class KnowledgeDocumentImportTaskAppService { if (StringUtil.hasText(source.getRegex())) { target.setRegex(source.getRegex()); } + if (source.getRetainRegexMatch() != null) { + target.setRetainRegexMatch(source.getRetainRegexMatch()); + } if (source.getRowsPerChunk() != null) { target.setRowsPerChunk(source.getRowsPerChunk()); } @@ -4320,6 +4498,8 @@ public class KnowledgeDocumentImportTaskAppService { map.put("chunkSize", strategyConfig.getChunkSize()); map.put("overlapSize", strategyConfig.getOverlapSize()); map.put("regex", strategyConfig.getRegex()); + map.put("retainRegexMatch", Boolean.TRUE.equals( + strategyConfig.getRetainRegexMatch())); map.put("rowsPerChunk", strategyConfig.getRowsPerChunk()); map.put("mdSplitterLevel", strategyConfig.getMdSplitterLevel()); return map; diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/KnowledgeImportBatchFacade.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/KnowledgeImportBatchFacade.java index 19cc49c4..7abe52f8 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/KnowledgeImportBatchFacade.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/KnowledgeImportBatchFacade.java @@ -764,9 +764,9 @@ public class KnowledgeImportBatchFacade { counts.setPending(internal.getPendingCount()); counts.setFailed(internal.getFailedCount()); counts.setSkipped(internal.getSkippedCount()); - counts.setRetryableFailed(internal.getRetryableFailedCount()); + counts.setRetryableFailed(internal.getFailedCount()); response.setCounts(counts); - response.setCanRetry(internal.getRetryableFailedCount() > 0 + response.setCanRetry(internal.getFailedCount() > 0 && (DocumentImportBatchStatus.PARTIAL_SUCCEEDED.name().equals(batch.getStatus()) || DocumentImportBatchStatus.INTERRUPTED.name().equals(batch.getStatus()))); PublicDocumentImportDtos.ItemPage items = @@ -795,7 +795,10 @@ public class KnowledgeImportBatchFacade { record.setStage(item.getStage()); record.setStatus(item.getStatus()); record.setAttemptCount(valueOrZero(item.getAttemptCount())); - record.setRetryable(Boolean.TRUE.equals(item.getRetryable())); + // 人工重试按实际失败状态选择，兼容字段不再受历史 retryable 标记影响。 + record.setRetryable( + DocumentImportBatchItemStatus.FAILED.name().equals(item.getStatus()) + ); if (StringUtil.hasText(item.getFailureCode()) || StringUtil.hasText(item.getErrorSummary())) { PublicDocumentImportDtos.ItemError error = diff --git a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchAppServiceTest.java b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchAppServiceTest.java index 90efbed7..c2156476 100644 --- a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchAppServiceTest.java +++ b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchAppServiceTest.java @@ -16,6 +16,7 @@ import tech.easyflow.ai.entity.DocumentImportBatchItem; import tech.easyflow.ai.enums.DocumentImportBatchItemStage; import tech.easyflow.ai.enums.DocumentImportBatchItemStatus; import tech.easyflow.ai.enums.DocumentImportBatchStatus; +import tech.easyflow.ai.enums.DocumentImportMode; import tech.easyflow.ai.mapper.DocumentImportBatchItemMapper; import tech.easyflow.ai.mapper.DocumentImportBatchMapper; import tech.easyflow.ai.mapper.DocumentMapper; @@ -606,10 +607,10 @@ public class DocumentImportBatchAppServiceTest { } /** - * 验证修复前的可恢复失败项会在重试领取事务中恢复重试资格和批次计数。 + * 验证人工重试直接选择历史失败项，不受旧重试资格字段限制。 */ @Test - public void retryShouldRestoreLegacyRecoverableItem() { + public void retryShouldSelectLegacyFailedItemWithoutRetryableGate() { TestContext context = createContext(); DocumentImportBatch batch = context.batchService.getOne( QueryWrapper.create() @@ -627,17 +628,6 @@ public class DocumentImportBatchAppServiceTest { failed.setErrorSummary("历史代码异常"); Mockito.when(context.itemService.list(Mockito.any(QueryWrapper.class))) .thenReturn(List.of(failed)); - Mockito.when(context.taskAppService.isRecoverableBatchFailure(failed)) - .thenReturn(true); - Mockito.when(context.batchTracker.transitionItem( - Mockito.eq(failed.getId()), - Mockito.eq(DocumentImportBatchItemStage.PARSE), - Mockito.eq(DocumentImportBatchItemStatus.FAILED), - Mockito.eq("历史代码异常"), - Mockito.eq(true), - Mockito.eq(0), - Mockito.eq("parse_failed") - )).thenReturn(true); Mockito.when(context.batchMapper.claimRetry( Mockito.any(), Mockito.anyString(), @@ -661,15 +651,15 @@ public class DocumentImportBatchAppServiceTest { } Assert.assertEquals(Integer.valueOf(1), result.getRetriedCount()); - Assert.assertTrue(failed.getRetryable()); - Mockito.verify(context.batchTracker).transitionItem( - failed.getId(), - DocumentImportBatchItemStage.PARSE, - DocumentImportBatchItemStatus.FAILED, - "历史代码异常", - true, - 0, - "parse_failed" + Assert.assertFalse(failed.getRetryable()); + Mockito.verify(context.batchTracker, Mockito.never()).transitionItem( + Mockito.any(), + Mockito.any(), + Mockito.any(), + Mockito.any(), + Mockito.anyBoolean(), + Mockito.anyInt(), + Mockito.any() ); Mockito.verify(context.batchMapper).claimRetry( Mockito.eq(batch.getId()), @@ -681,6 +671,63 @@ public class DocumentImportBatchAppServiceTest { Mockito.verify(context.lockHandle).release(); } + /** + * 验证管理端继续按真实失败项恢复，不依赖历史可重试失败计数。 + */ + @Test + public void continueShouldResumeFailedItemsWhenCompatibilityCounterIsZero() { + TestContext context = createContext(); + DocumentImportBatch batch = context.batchService.getOne( + QueryWrapper.create() + ); + batch.setImportMode(DocumentImportMode.AUTO.name()); + batch.setStatus(DocumentImportBatchStatus.PARTIAL_SUCCEEDED.name()); + batch.setFailedCount(1); + batch.setRetryableFailedCount(0); + DocumentImportBatchItem failed = uploadedItem( + BigInteger.valueOf(33), + batch.getId() + ); + failed.setStage(DocumentImportBatchItemStage.INDEX.name()); + failed.setStatus(DocumentImportBatchItemStatus.FAILED.name()); + failed.setRetryable(false); + Mockito.when(context.itemService.list(Mockito.any(QueryWrapper.class))) + .thenReturn(List.of(failed)); + Mockito.when(context.batchMapper.updateByQuery( + Mockito.any(DocumentImportBatch.class), + Mockito.any(QueryWrapper.class) + )).thenReturn(1); + Mockito.when(context.batchTracker.toStatusResponse(batch)) + .thenAnswer(invocation -> { + DocumentImportBatchDtos.StatusResponse response = + new DocumentImportBatchDtos.StatusResponse(); + response.setStatus(batch.getStatus()); + return response; + }); + + DocumentImportBatchDtos.StatusResponse response; + beginTransactionSynchronization(); + try { + response = context.service.continueBatch( + batch.getKnowledgeId(), batch.getId() + ); + } finally { + completeTransactionSynchronization( + TransactionSynchronization.STATUS_COMMITTED + ); + } + + Assert.assertEquals(DocumentImportBatchStatus.RUNNING.name(), + response.getStatus()); + Mockito.verify(context.batchTracker, Mockito.never()) + .refreshBatch(batch.getId()); + Mockito.verify(context.taskAppService).retryBatchFailures( + batch.getId(), + Set.of(failed.getClientFileKey()) + ); + Mockito.verify(context.lockHandle).release(); + } + /** * 验证运行中的任务拒绝再次领取，调用方可继续查询原 taskId。 */ diff --git a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchTrackerTest.java b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchTrackerTest.java index 16fa78e1..e5e26aa8 100644 --- a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchTrackerTest.java +++ b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchTrackerTest.java @@ -63,10 +63,10 @@ public class DocumentImportBatchTrackerTest { } /** - * 验证中断批次在汇总失败项后仍保留可继续状态。 + * 验证兼容可重试计数与人工可继续的失败总数保持一致。 */ @Test - public void shouldPreserveInterruptedStatusUntilUserContinues() { + public void shouldExposeAllFailuresAsContinuable() { DocumentImportBatchService batchService = mock(DocumentImportBatchService.class); DocumentImportBatchItemService itemService = mock(DocumentImportBatchItemService.class); DocumentImportBatchMapper batchMapper = mock(DocumentImportBatchMapper.class); @@ -74,7 +74,7 @@ public class DocumentImportBatchTrackerTest { DocumentImportBatch batch = batch(DocumentImportBatchStatus.INTERRUPTED, 2); batch.setCompletedCount(1); batch.setFailedCount(1); - batch.setRetryableFailedCount(1); + batch.setRetryableFailedCount(0); when(batchService.getById(batch.getId())).thenReturn(batch); DocumentImportBatchTracker tracker = diff --git a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/KnowledgeImportBatchFacadeTest.java b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/KnowledgeImportBatchFacadeTest.java index 73195095..882bfb8f 100644 --- a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/KnowledgeImportBatchFacadeTest.java +++ b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/KnowledgeImportBatchFacadeTest.java @@ -11,6 +11,9 @@ import tech.easyflow.ai.documentimport.ImportCallerContext; import tech.easyflow.ai.documentimport.ImportCallerType; import tech.easyflow.ai.documentimport.PublicDocumentImportDtos; import tech.easyflow.ai.entity.DocumentImportBatch; +import tech.easyflow.ai.entity.DocumentImportBatchItem; +import tech.easyflow.ai.enums.DocumentImportBatchItemStage; +import tech.easyflow.ai.enums.DocumentImportBatchItemStatus; import tech.easyflow.ai.enums.DocumentImportBatchStatus; import tech.easyflow.ai.mapper.DocumentImportBatchMapper; import tech.easyflow.ai.service.DocumentImportBatchItemService; @@ -318,6 +321,52 @@ public class KnowledgeImportBatchFacadeTest { Assert.assertEquals(Integer.valueOf(2), response.getRetriedCount()); } + /** + * 验证 Public API 将所有失败文件统一标记为可重试，不受历史标记影响。 + */ + @Test + public void statusShouldExposeEveryFailedItemAsRetryable() { + TestContext context = createContext(); + DocumentImportBatch batch = new DocumentImportBatch(); + batch.setId(BigInteger.ONE); + batch.setKnowledgeId(BigInteger.TWO); + batch.setStatus(DocumentImportBatchStatus.PARTIAL_SUCCEEDED.name()); + Mockito.when(context.batchAppService.requireBatchForCaller( + BigInteger.ONE, context.caller + )).thenReturn(batch); + + DocumentImportBatchDtos.StatusResponse internal = + new DocumentImportBatchDtos.StatusResponse(); + internal.setTotalCount(1); + internal.setFailedCount(1); + internal.setRetryableFailedCount(1); + internal.setProgressPercent(100); + Mockito.when(context.batchTracker.toStatusResponse(batch)) + .thenReturn(internal); + + DocumentImportBatchItem failed = new DocumentImportBatchItem(); + failed.setBatchId(batch.getId()); + failed.setClientFileKey("failed.txt"); + failed.setStage(DocumentImportBatchItemStage.INDEX.name()); + failed.setStatus(DocumentImportBatchItemStatus.FAILED.name()); + failed.setRetryable(false); + Mockito.when(context.itemService.page( + Mockito.any(com.mybatisflex.core.paginate.Page.class), + Mockito.any(com.mybatisflex.core.query.QueryWrapper.class) + )).thenReturn(new com.mybatisflex.core.paginate.Page<>( + List.of(failed), 1, 1, 1 + )); + + PublicDocumentImportDtos.StatusResponse response = context.facade.getStatus( + context.caller, batch.getId(), null, 1, 20 + ); + + Assert.assertTrue(response.getCanRetry()); + Assert.assertEquals(Integer.valueOf(1), + response.getCounts().getRetryableFailed()); + Assert.assertTrue(response.getItems().getRecords().get(0).getRetryable()); + } + /** * 创建单文件元数据。 * @@ -406,6 +455,7 @@ public class KnowledgeImportBatchFacadeTest { return new TestContext( facade, batchAppService, + batchTracker, batchService, itemService, batchMapper, @@ -422,6 +472,7 @@ public class KnowledgeImportBatchFacadeTest { private record TestContext( KnowledgeImportBatchFacade facade, DocumentImportBatchAppService batchAppService, + DocumentImportBatchTracker batchTracker, DocumentImportBatchService batchService, DocumentImportBatchItemService itemService, DocumentImportBatchMapper batchMapper, diff --git a/easyflow-ui-admin/app/src/views/ai/documentCollection/DocumentImportBatchStatus.test.ts b/easyflow-ui-admin/app/src/views/ai/documentCollection/DocumentImportBatchStatus.test.ts index d6c534ca..78533358 100644 --- a/easyflow-ui-admin/app/src/views/ai/documentCollection/DocumentImportBatchStatus.test.ts +++ b/easyflow-ui-admin/app/src/views/ai/documentCollection/DocumentImportBatchStatus.test.ts @@ -79,4 +79,78 @@ describe('documentImportBatchStatus', () => { ); wrapper.unmount(); }); + + it('部分失败批次按 failedCount 继续并发送统一批次参数', async () => { + const failedBatch = { + batchId: 'batch-1', + completedCount: 1, + failedCount: 1, + importMode: 'AUTO', + pendingCount: 0, + processingCount: 0, + progressPercent: 100, + retryableFailedCount: 0, + skippedCount: 0, + status: 'PARTIAL_SUCCEEDED', + totalCount: 2, + }; + apiMocks.get.mockResolvedValue({ + data: failedBatch, + errorCode: 0, + }); + apiMocks.post.mockResolvedValue({ + data: { + ...failedBatch, + failedCount: 0, + pendingCount: 1, + progressPercent: 50, + status: 'RUNNING', + }, + errorCode: 0, + }); + + const wrapper = mount(DocumentImportBatchStatus, { + props: { + knowledgeId: 'knowledge-1', + manageable: true, + }, + }); + await flushPromises(); + + const continueButton = wrapper.find('.batch-status__continue'); + expect(continueButton.exists()).toBe(true); + await continueButton.trigger('click'); + await flushPromises(); + + expect(apiMocks.post).toHaveBeenCalledWith( + '/api/v1/document/import/batch/continue', + { + batchId: 'batch-1', + knowledgeId: 'knowledge-1', + }, + ); + wrapper.unmount(); + }); + + it('没有失败项时不使用旧兼容计数显示继续按钮', async () => { + apiMocks.get.mockResolvedValue({ + data: { + ...createBatch('COMPLETED', 2), + retryableFailedCount: 1, + status: 'PARTIAL_SUCCEEDED', + }, + errorCode: 0, + }); + + const wrapper = mount(DocumentImportBatchStatus, { + props: { + knowledgeId: 'knowledge-1', + manageable: true, + }, + }); + await flushPromises(); + + expect(wrapper.find('.batch-status__continue').exists()).toBe(false); + wrapper.unmount(); + }); }); diff --git a/easyflow-ui-admin/app/src/views/ai/documentCollection/DocumentImportBatchStatus.vue b/easyflow-ui-admin/app/src/views/ai/documentCollection/DocumentImportBatchStatus.vue index c04e1f2f..4dfdbac9 100644 --- a/easyflow-ui-admin/app/src/views/ai/documentCollection/DocumentImportBatchStatus.vue +++ b/easyflow-ui-admin/app/src/views/ai/documentCollection/DocumentImportBatchStatus.vue @@ -15,7 +15,6 @@ interface BatchStatus { pendingCount: number; processingCount: number; progressPercent: number; - retryableFailedCount?: number; skippedCount: number; status: | 'CANCELLED' @@ -53,8 +52,7 @@ const canContinue = computed( props.manageable && (batch.value?.status === 'INTERRUPTED' || batch.value?.status === 'PARTIAL_SUCCEEDED') && - (Number(batch.value?.retryableFailedCount || 0) > 0 || - Number(batch.value?.failedCount || 0) > 0), + Number(batch.value?.failedCount || 0) > 0, ); const processedCount = computed(