feat(knowledge): S6 摄入轮询键 documentId→batch(Dify 按 batch 轮询 indexing-status)

为 muse_knowledge_processing_task 加 runtime_batch_id 列并落库 Dify 上传返回的索引批次(从 uploadResult.summary 的 batch 键抽取)。poll worker 取 runtimeBatchId 非空优先、否则回退 ragflowDocumentId 作轮询键,RAGFlow 无 batch 路径逐字不变;skip gate 放宽为 documentId 与 batch 都空才跳过。

DDL 采用新迁移 V37 ALTER TABLE ADD COLUMN,非回编 V14 建表语句——沿用 sql/muse append-only 约定(V24/V27/V30/V31 同法),避免破坏已应用库(dev/live)的 Flyway 校验和。

验证:mvn -pl muse-module-knowledge-server -am test 目标两类全绿(PollWorkerTest 9/0/0/0、DocumentServiceTest 17/0/0/0,聚合 Tests run 26 Failures 0 Errors 0 Skipped 0,BUILD SUCCESS)。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeDocumentService.java

@@ -292,8 +292,11 @@ public class MuseKnowledgeDocumentService {
                     uploadResult.externalId(), parseResult);
             return new UploadOutcome(ragflowContext, "failed");
         }
+        // Dify 摄入把索引批次 batch 透出到 uploadResult.summary().get("batch")，是其 indexing-status 轮询主键；
+        // RAGFlow 无此字段 → runtimeBatchId 为 null，轮询回退 documentId、行为逐字不变。
+        String runtimeBatchId = uploadResult.summary() == null ? null : (String) uploadResult.summary().get("batch");
         processingTaskService.markRagflowParseAccepted(ragflowContext.version().getId(), ragflowContext.task().getTaskId(),
-                ragflowContext.ragflowDatasetId(), uploadResult.externalId(), parseResult);
+                ragflowContext.ragflowDatasetId(), uploadResult.externalId(), runtimeBatchId, parseResult);
         return new UploadOutcome(ragflowContext, "processing");
     }
 

muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeParseStatusPollWorker.java

@@ -76,8 +76,11 @@ public class MuseKnowledgeParseStatusPollWorker {
         int advanced = 0;
         for (MuseKnowledgeProcessingTaskDO task : tasks) {
             // 缺租户/外部 id 的任务无法安全轮询，跳过（避免空指针与跨租户写串）。
+            // 轮询键防御性放宽：ragflowDocumentId 与 runtimeBatchId 都空才跳过——Dify 场景两者都会有值，
+            // RAGFlow 只有 documentId，任一非空即可据以轮询。
             if (task.getTenantId() == null || !StringUtils.hasText(task.getRagflowDatasetId())
-                    || !StringUtils.hasText(task.getRagflowDocumentId())) {
+                    || (!StringUtils.hasText(task.getRagflowDocumentId())
+                            && !StringUtils.hasText(task.getRuntimeBatchId()))) {
                 continue;
             }
             advanced += TenantUtils.execute(task.getTenantId(), () -> pollOne(task));
@@ -87,10 +90,14 @@ public class MuseKnowledgeParseStatusPollWorker {
 
     private int pollOne(MuseKnowledgeProcessingTaskDO task) {
         KnowledgeRuntimeClient.RuntimeResult result;
+        // 轮询主键择键：Dify 上传落 runtimeBatchId（indexing-status 按 batch 查）→ 优先用之；
+        // RAGFlow 无 batch → 回退 ragflowDocumentId，保持原逐字行为。skip gate 已保证两者至少一非空，pollKey 不为 null。
+        String pollKey = StringUtils.hasText(task.getRuntimeBatchId())
+                ? task.getRuntimeBatchId() : task.getRagflowDocumentId();
         try {
             result = ragFlowClient.pollDocumentStatuses(new KnowledgeRuntimeClient.PollDocumentStatusesCommand(
                     task.getTenantId(), task.getOwnerUserId(), task.getKbId(), task.getRagflowDatasetId(),
-                    List.of(task.getRagflowDocumentId()), task.getTaskId(),
+                    List.of(pollKey), task.getTaskId(),
                     pollCorrelationId(task), pollRequestHash(task), 1));
         } catch (RuntimeException ex) {
             // RAGFlow 临时不可达：保持 parsing，下轮重试，不误判失败。

muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeProcessingTaskService.java

@@ -20,6 +20,7 @@ import jakarta.annotation.Resource;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
+import org.springframework.util.StringUtils;
 
 import java.time.LocalDateTime;
 import java.util.LinkedHashMap;
@@ -178,7 +179,8 @@ public class MuseKnowledgeProcessingTaskService {
 
     @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
     public void markRagflowParseAccepted(Long documentVersionId, String taskId, String ragflowDatasetId,
-                                         String ragflowDocumentId, KnowledgeRuntimeClient.RuntimeResult result) {
+                                         String ragflowDocumentId, String runtimeBatchId,
+                                         KnowledgeRuntimeClient.RuntimeResult result) {
         MuseKnowledgeProcessingTaskDO task = processingTaskMapper.selectByTaskId(taskId);
         if (task != null) {
             task.setStatus("parsing");
@@ -187,6 +189,11 @@ public class MuseKnowledgeProcessingTaskService {
             // 新建 dataset 的上传链路必须把外部 datasetId 同步固化到 task，后续轮询、补偿和 live 取证都依赖这条事实。
             task.setRagflowDatasetId(ragflowDatasetId);
             task.setRagflowDocumentId(ragflowDocumentId);
+            // Dify 上传返回的索引批次是其 indexing-status 轮询主键，必须固化；RAGFlow 无 batch → runtimeBatchId 为空，
+            // 用 hasText 判定避免把已落值覆盖成 null（poll worker 据此在 batch / documentId 间择键）。
+            if (StringUtils.hasText(runtimeBatchId)) {
+                task.setRuntimeBatchId(runtimeBatchId);
+            }
             task.setRetryable(false);
             task.setProgressMessage("RAGFlow parse accepted; polling required");
             task.setResultSummary(JsonUtils.toJsonString(Map.of("ragflow", safeSummary(result))));

muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/dal/dataobject/muse/MuseKnowledgeProcessingTaskDO.java

@@ -42,6 +42,12 @@ public class MuseKnowledgeProcessingTaskDO extends TenantBaseDO {
     private String status;
     private String ragflowDatasetId;
     private String ragflowDocumentId;
+    /**
+     * 运行时索引批次 id：Dify Datasets 摄入返回的顶层 batch，是其 indexing-status 轮询主键
+     * （GET /v1/datasets/{id}/documents/{batch}/indexing-status）。RAGFlow 上传不产生 batch，留空，
+     * 轮询回退到 ragflowDocumentId，行为逐字不变。命名沿用 provider 无关口径，避免与 ragflowDocumentId 混淆。
+     */
+    private String runtimeBatchId;
     /** RAGFlow document run/progress 轮询得到的解析进度。 */
     private Integer parseProgress;
     private String progressMessage;

muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/test/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeDocumentServiceTest.java

@@ -323,11 +323,12 @@ class MuseKnowledgeDocumentServiceTest extends BaseMockitoUnitTest {
         });
         when(processingTaskMapper.selectByTaskId("knowledge-processing-6001")).thenReturn(processingTask(6001L));
         when(processingTaskMapper.insert(any(MuseKnowledgeProcessingTaskDO.class))).thenAnswer(invocation -> 1);
+        // 模拟 Dify 上传把索引批次 batch 透出到 summary，验证其被抽取并落库到 task.runtimeBatchId（RAGFlow 无此字段）。
         when(ragFlowClient.uploadDocuments(any())).thenReturn(new KnowledgeRuntimeClient.RuntimeResult(
                 "rag-doc-1", KnowledgeRuntimeClient.Operation.UPLOAD_DOCUMENTS,
                 "uploadGlobalKBDocument:cmd-upload-file-success", "hash-file-success", 1, 12L,
                 KnowledgeRuntimeClient.Status.SUCCEEDED, null, null,
-                "{\"status\":\"succeeded\"}", List.of(), java.util.Map.of()));
+                "{\"status\":\"succeeded\"}", List.of(), java.util.Map.of("batch", "dify-batch-1")));
         when(ragFlowClient.startParseDocuments(any())).thenReturn(new KnowledgeRuntimeClient.RuntimeResult(
                 null, KnowledgeRuntimeClient.Operation.START_PARSE_DOCUMENTS,
                 "uploadGlobalKBDocument:cmd-upload-file-success", "hash-file-success", 1, 8L,
@@ -380,6 +381,7 @@ class MuseKnowledgeDocumentServiceTest extends BaseMockitoUnitTest {
                 "parsing".equals(task.getStatus())
                         && "rag-dataset-1".equals(task.getRagflowDatasetId())
                         && "rag-doc-1".equals(task.getRagflowDocumentId())
+                        && "dify-batch-1".equals(task.getRuntimeBatchId())
                         && "processing".equals(task.getResultSummary()) == false));
     }
 

muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/test/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeParseStatusPollWorkerTest.java

@@ -9,6 +9,7 @@ import cn.iocoder.muse.module.knowledge.application.muse.facade.KnowledgeRuntime
 import cn.iocoder.muse.module.knowledge.dal.dataobject.muse.MuseKnowledgeProcessingTaskDO;
 import cn.iocoder.muse.module.knowledge.dal.mysql.muse.MuseKnowledgeProcessingTaskMapper;
 import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 
 import java.util.List;
@@ -129,6 +130,7 @@ class MuseKnowledgeParseStatusPollWorkerTest extends BaseMockitoUnitTest {
 
     @Test
     void should_skip_whenMissingRagflowIds() {
+        // documentId 与 runtimeBatchId 都空 → 无可用轮询键 → 跳过。
         MuseKnowledgeProcessingTaskDO t = parsingTask();
         t.setRagflowDocumentId(null);
         when(processingTaskMapper.selectParsingForPoll(anyInt())).thenReturn(List.of(t));
@@ -138,4 +140,37 @@ class MuseKnowledgeParseStatusPollWorkerTest extends BaseMockitoUnitTest {
         assertEquals(0, advanced);
         verifyNoInteractions(ragFlowClient);
     }
+
+    @Test
+    void should_pollByRuntimeBatchId_whenDifyBatchPresent() {
+        // Dify 场景：上传落 runtimeBatchId（=索引 batch）与 ragflowDocumentId（=document.id）；轮询主键取 batch。
+        MuseKnowledgeProcessingTaskDO t = parsingTask();
+        t.setRuntimeBatchId("dify-batch-9");
+        when(processingTaskMapper.selectParsingForPoll(anyInt())).thenReturn(List.of(t));
+        when(ragFlowClient.pollDocumentStatuses(any())).thenReturn(
+                poll(Status.SUCCEEDED, new DocumentStatus("doc-1", "processing", "RUNNING", 50, "running")));
+
+        worker(true).pollOnce();
+
+        ArgumentCaptor<KnowledgeRuntimeClient.PollDocumentStatusesCommand> captor =
+                ArgumentCaptor.forClass(KnowledgeRuntimeClient.PollDocumentStatusesCommand.class);
+        verify(ragFlowClient).pollDocumentStatuses(captor.capture());
+        assertEquals(List.of("dify-batch-9"), captor.getValue().ragflowDocumentIds());
+    }
+
+    @Test
+    void should_pollByDocumentId_whenRuntimeBatchIdAbsent() {
+        // RAGFlow 场景：无 batch，runtimeBatchId 留空 → 轮询回退 ragflowDocumentId，逐字不变。
+        MuseKnowledgeProcessingTaskDO t = parsingTask();
+        when(processingTaskMapper.selectParsingForPoll(anyInt())).thenReturn(List.of(t));
+        when(ragFlowClient.pollDocumentStatuses(any())).thenReturn(
+                poll(Status.SUCCEEDED, new DocumentStatus("doc-1", "processing", "RUNNING", 50, "running")));
+
+        worker(true).pollOnce();
+
+        ArgumentCaptor<KnowledgeRuntimeClient.PollDocumentStatusesCommand> captor =
+                ArgumentCaptor.forClass(KnowledgeRuntimeClient.PollDocumentStatusesCommand.class);
+        verify(ragFlowClient).pollDocumentStatuses(captor.capture());
+        assertEquals(List.of("doc-1"), captor.getValue().ragflowDocumentIds());
+    }
 }

muse-cloud/sql/muse/V37__add_knowledge_processing_task_runtime_batch_id.sql

@@ -0,0 +1,12 @@
+-- S6 知识摄入轮询键从 documentId 泛化到 batch：为 Dify Datasets 索引状态轮询补运行时批次列。
+--
+-- 背景：KnowledgeRuntimeClient 是 provider 无关端口（muse.knowledge.runtime-provider 切换 ragflow / dify）。
+--   Dify uploadDocuments 返回 document.id（落 ragflow_document_id）+ 顶层 batch，其索引状态轮询打
+--   GET /v1/datasets/{id}/documents/{batch}/indexing-status，轮询主键=batch；RAGFlow 仍按 documentId 轮询、
+--   不产生 batch。此前摄入 caller 未持久化 batch，poll worker 只能传 documentId → Dify 轮询打错端点必坏。
+--   本迁移新增可空列承载 Dify 批次：Dify 上传落 batch → 按 batch 轮询；RAGFlow 留空 → 回退 documentId、行为逐字不变。
+--
+-- 迁移方式：沿用本目录 append-only 约定（V24/V27/V30/V31 均以新迁移 ALTER 既有表），不回编 V14 建表语句，
+--   避免破坏已应用库（dev / live）的 Flyway 校验和。
+ALTER TABLE muse_knowledge_processing_task
+    ADD COLUMN runtime_batch_id VARCHAR(64);