feat(kg): 补全知识图谱实体同步和关系构建

新增功能：
- 补全 4 类实体同步：Workflow、Job、LabelTask、KnowledgeSet
- 补全 7 类关系构建：USES_DATASET、PRODUCES、ASSIGNED_TO、TRIGGERS、DEPENDS_ON、IMPACTS、SOURCED_FROM
- 新增 39 个测试用例，总计 111 个测试

问题修复（三轮迭代）：
第一轮（6 个问题）：
- toStringList null/blank 过滤
- mergeUsesDatasetRelations 统一逻辑
- fetchAllPaged 去重抽取
- IMPACTS 占位标记
- 测试断言增强
- syncAll 固定索引改 Map

第二轮（2 个问题）：
- 活跃 ID 空值/空白归一化（两层防御）
- 关系构建 N+1 查询消除（预加载 Map）

第三轮（1 个问题）：
- 空元素 NPE 防御（GraphSyncService 12 处 + GraphSyncStepService 6 处）

代码变更：+1936 行，-101 行
测试结果：111 tests, 0 failures

已知 P3 问题（非阻塞）：
- 安全注释与实现不一致（待权限过滤任务一起处理）
- 测试覆盖缺口（可后续补充）

backend/services/knowledge-graph-service/src/main/java/com/datamate/knowledgegraph/application/GraphSyncService.java

@@ -5,6 +5,10 @@ import com.datamate.common.infrastructure.exception.SystemErrorCode;
 import com.datamate.knowledgegraph.domain.model.SyncResult;
 import com.datamate.knowledgegraph.infrastructure.client.DataManagementClient;
 import com.datamate.knowledgegraph.infrastructure.client.DataManagementClient.DatasetDTO;
+import com.datamate.knowledgegraph.infrastructure.client.DataManagementClient.WorkflowDTO;
+import com.datamate.knowledgegraph.infrastructure.client.DataManagementClient.JobDTO;
+import com.datamate.knowledgegraph.infrastructure.client.DataManagementClient.LabelTaskDTO;
+import com.datamate.knowledgegraph.infrastructure.client.DataManagementClient.KnowledgeSetDTO;
 import com.datamate.knowledgegraph.infrastructure.exception.KnowledgeGraphErrorCode;
 import com.datamate.knowledgegraph.infrastructure.neo4j.KnowledgeGraphProperties;
 import lombok.RequiredArgsConstructor;
@@ -66,46 +70,110 @@ public class GraphSyncService {
 
             // 一次拉取，全程共享
             List<DatasetDTO> datasets = fetchDatasetsWithRetry(syncId);
+            List<WorkflowDTO> workflows = fetchWithRetry(syncId, "workflows",
+                    () -> dataManagementClient.listAllWorkflows());
+            List<JobDTO> jobs = fetchWithRetry(syncId, "jobs",
+                    () -> dataManagementClient.listAllJobs());
+            List<LabelTaskDTO> labelTasks = fetchWithRetry(syncId, "label-tasks",
+                    () -> dataManagementClient.listAllLabelTasks());
+            List<KnowledgeSetDTO> knowledgeSets = fetchWithRetry(syncId, "knowledge-sets",
+                    () -> dataManagementClient.listAllKnowledgeSets());
 
-            List<SyncResult> results = new ArrayList<>();
+            // 使用 LinkedHashMap 按 syncType 存取，保持插入顺序且避免固定下标
+            Map<String, SyncResult> resultMap = new LinkedHashMap<>();
 
             // 实体同步
-            results.add(stepService.upsertDatasetEntities(graphId, datasets, syncId));
+            resultMap.put("Dataset", stepService.upsertDatasetEntities(graphId, datasets, syncId));
-            results.add(stepService.upsertFieldEntities(graphId, datasets, syncId));
+            resultMap.put("Field", stepService.upsertFieldEntities(graphId, datasets, syncId));
 
-            Set<String> usernames = extractUsernames(datasets);
+            Set<String> usernames = extractUsernames(datasets, workflows, jobs, labelTasks, knowledgeSets);
-            results.add(stepService.upsertUserEntities(graphId, usernames, syncId));
+            resultMap.put("User", stepService.upsertUserEntities(graphId, usernames, syncId));
-            results.add(stepService.upsertOrgEntities(graphId, syncId));
+            resultMap.put("Org", stepService.upsertOrgEntities(graphId, syncId));
+            resultMap.put("Workflow", stepService.upsertWorkflowEntities(graphId, workflows, syncId));
+            resultMap.put("Job", stepService.upsertJobEntities(graphId, jobs, syncId));
+            resultMap.put("LabelTask", stepService.upsertLabelTaskEntities(graphId, labelTasks, syncId));
+            resultMap.put("KnowledgeSet", stepService.upsertKnowledgeSetEntities(graphId, knowledgeSets, syncId));
 
-            // 全量对账：删除 MySQL 已移除的记录，并回填 purged 到对应 SyncResult
+            // 全量对账：删除 MySQL 已移除的记录（按 syncType 查找，无需固定下标）
             Set<String> activeDatasetIds = datasets.stream()
+                    .filter(Objects::nonNull)
                     .map(DatasetDTO::getId)
+                    .filter(Objects::nonNull)
+                    .filter(id -> !id.isBlank())
                     .collect(Collectors.toSet());
-            results.get(0).setPurged(
+            resultMap.get("Dataset").setPurged(
                     stepService.purgeStaleEntities(graphId, "Dataset", activeDatasetIds, syncId));
 
             Set<String> activeFieldIds = new HashSet<>();
             for (DatasetDTO dto : datasets) {
-                if (dto.getTags() != null) {
+                if (dto == null || dto.getTags() == null) {
+                    continue;
+                }
                 for (DataManagementClient.TagDTO tag : dto.getTags()) {
+                    if (tag == null || tag.getName() == null) {
+                        continue;
+                    }
                     activeFieldIds.add(dto.getId() + ":tag:" + tag.getName());
                 }
             }
-            }
+            resultMap.get("Field").setPurged(
-            results.get(1).setPurged(
                     stepService.purgeStaleEntities(graphId, "Field", activeFieldIds, syncId));
 
             Set<String> activeUserIds = usernames.stream()
                     .map(u -> "user:" + u)
                     .collect(Collectors.toSet());
-            results.get(2).setPurged(
+            resultMap.get("User").setPurged(
                     stepService.purgeStaleEntities(graphId, "User", activeUserIds, syncId));
 
-            // 关系构建（MERGE 幂等）
+            Set<String> activeWorkflowIds = workflows.stream()
-            results.add(stepService.mergeHasFieldRelations(graphId, syncId));
+                    .filter(Objects::nonNull)
-            results.add(stepService.mergeDerivedFromRelations(graphId, syncId));
+                    .map(WorkflowDTO::getId)
-            results.add(stepService.mergeBelongsToRelations(graphId, syncId));
+                    .filter(Objects::nonNull)
+                    .filter(id -> !id.isBlank())
+                    .collect(Collectors.toSet());
+            resultMap.get("Workflow").setPurged(
+                    stepService.purgeStaleEntities(graphId, "Workflow", activeWorkflowIds, syncId));
 
+            Set<String> activeJobIds = jobs.stream()
+                    .filter(Objects::nonNull)
+                    .map(JobDTO::getId)
+                    .filter(Objects::nonNull)
+                    .filter(id -> !id.isBlank())
+                    .collect(Collectors.toSet());
+            resultMap.get("Job").setPurged(
+                    stepService.purgeStaleEntities(graphId, "Job", activeJobIds, syncId));
+
+            Set<String> activeLabelTaskIds = labelTasks.stream()
+                    .filter(Objects::nonNull)
+                    .map(LabelTaskDTO::getId)
+                    .filter(Objects::nonNull)
+                    .filter(id -> !id.isBlank())
+                    .collect(Collectors.toSet());
+            resultMap.get("LabelTask").setPurged(
+                    stepService.purgeStaleEntities(graphId, "LabelTask", activeLabelTaskIds, syncId));
+
+            Set<String> activeKnowledgeSetIds = knowledgeSets.stream()
+                    .filter(Objects::nonNull)
+                    .map(KnowledgeSetDTO::getId)
+                    .filter(Objects::nonNull)
+                    .filter(id -> !id.isBlank())
+                    .collect(Collectors.toSet());
+            resultMap.get("KnowledgeSet").setPurged(
+                    stepService.purgeStaleEntities(graphId, "KnowledgeSet", activeKnowledgeSetIds, syncId));
+
+            // 关系构建（MERGE 幂等）
+            resultMap.put("HAS_FIELD", stepService.mergeHasFieldRelations(graphId, syncId));
+            resultMap.put("DERIVED_FROM", stepService.mergeDerivedFromRelations(graphId, syncId));
+            resultMap.put("BELONGS_TO", stepService.mergeBelongsToRelations(graphId, syncId));
+            resultMap.put("USES_DATASET", stepService.mergeUsesDatasetRelations(graphId, syncId));
+            resultMap.put("PRODUCES", stepService.mergeProducesRelations(graphId, syncId));
+            resultMap.put("ASSIGNED_TO", stepService.mergeAssignedToRelations(graphId, syncId));
+            resultMap.put("TRIGGERS", stepService.mergeTriggersRelations(graphId, syncId));
+            resultMap.put("DEPENDS_ON", stepService.mergeDependsOnRelations(graphId, syncId));
+            resultMap.put("IMPACTS", stepService.mergeImpactsRelations(graphId, syncId));
+            resultMap.put("SOURCED_FROM", stepService.mergeSourcedFromRelations(graphId, syncId));
+
+            List<SyncResult> results = new ArrayList<>(resultMap.values());
             log.info("[{}] Full sync completed for graphId={}. Summary: {}", syncId, graphId,
                     results.stream()
                             .map(r -> r.getSyncType() + "(+" + r.getCreated() + "/~" + r.getUpdated() + "/-" + r.getFailed() + ")")
@@ -132,7 +200,10 @@ public class GraphSyncService {
         try {
             List<DatasetDTO> datasets = fetchDatasetsWithRetry(syncId);
             SyncResult result = stepService.upsertDatasetEntities(graphId, datasets, syncId);
-            Set<String> activeIds = datasets.stream().map(DatasetDTO::getId).collect(Collectors.toSet());
+            Set<String> activeIds = datasets.stream()
+                    .filter(Objects::nonNull).map(DatasetDTO::getId)
+                    .filter(Objects::nonNull).filter(id -> !id.isBlank())
+                    .collect(Collectors.toSet());
             int purged = stepService.purgeStaleEntities(graphId, "Dataset", activeIds, syncId);
             result.setPurged(purged);
             return result;
@@ -155,10 +226,14 @@ public class GraphSyncService {
             SyncResult result = stepService.upsertFieldEntities(graphId, datasets, syncId);
             Set<String> activeFieldIds = new HashSet<>();
             for (DatasetDTO dto : datasets) {
-                if (dto.getTags() != null) {
+                if (dto == null || dto.getTags() == null) {
-                    for (DataManagementClient.TagDTO tag : dto.getTags()) {
+                    continue;
-                        activeFieldIds.add(dto.getId() + ":tag:" + tag.getName());
                 }
+                for (DataManagementClient.TagDTO tag : dto.getTags()) {
+                    if (tag == null || tag.getName() == null) {
+                        continue;
+                    }
+                    activeFieldIds.add(dto.getId() + ":tag:" + tag.getName());
                 }
             }
             result.setPurged(stepService.purgeStaleEntities(graphId, "Field", activeFieldIds, syncId));
@@ -179,7 +254,15 @@ public class GraphSyncService {
         ReentrantLock lock = acquireLock(graphId, syncId);
         try {
             List<DatasetDTO> datasets = fetchDatasetsWithRetry(syncId);
-            Set<String> usernames = extractUsernames(datasets);
+            List<WorkflowDTO> workflows = fetchWithRetry(syncId, "workflows",
+                    () -> dataManagementClient.listAllWorkflows());
+            List<JobDTO> jobs = fetchWithRetry(syncId, "jobs",
+                    () -> dataManagementClient.listAllJobs());
+            List<LabelTaskDTO> labelTasks = fetchWithRetry(syncId, "label-tasks",
+                    () -> dataManagementClient.listAllLabelTasks());
+            List<KnowledgeSetDTO> knowledgeSets = fetchWithRetry(syncId, "knowledge-sets",
+                    () -> dataManagementClient.listAllKnowledgeSets());
+            Set<String> usernames = extractUsernames(datasets, workflows, jobs, labelTasks, knowledgeSets);
             SyncResult result = stepService.upsertUserEntities(graphId, usernames, syncId);
             Set<String> activeUserIds = usernames.stream().map(u -> "user:" + u).collect(Collectors.toSet());
             result.setPurged(stepService.purgeStaleEntities(graphId, "User", activeUserIds, syncId));
@@ -261,6 +344,229 @@ public class GraphSyncService {
         }
     }
 
+    // -----------------------------------------------------------------------
+    // 新增实体同步
+    // -----------------------------------------------------------------------
+
+    public SyncResult syncWorkflows(String graphId) {
+        validateGraphId(graphId);
+        String syncId = UUID.randomUUID().toString().substring(0, 8);
+        ReentrantLock lock = acquireLock(graphId, syncId);
+        try {
+            List<WorkflowDTO> workflows = fetchWithRetry(syncId, "workflows",
+                    () -> dataManagementClient.listAllWorkflows());
+            SyncResult result = stepService.upsertWorkflowEntities(graphId, workflows, syncId);
+            Set<String> activeIds = workflows.stream()
+                    .filter(Objects::nonNull).map(WorkflowDTO::getId)
+                    .filter(Objects::nonNull).filter(id -> !id.isBlank())
+                    .collect(Collectors.toSet());
+            result.setPurged(stepService.purgeStaleEntities(graphId, "Workflow", activeIds, syncId));
+            return result;
+        } catch (BusinessException e) {
+            throw e;
+        } catch (Exception e) {
+            log.error("[{}] Workflow sync failed for graphId={}", syncId, graphId, e);
+            throw BusinessException.of(KnowledgeGraphErrorCode.SYNC_FAILED, "工作流同步失败，syncId=" + syncId);
+        } finally {
+            releaseLock(graphId, lock);
+        }
+    }
+
+    public SyncResult syncJobs(String graphId) {
+        validateGraphId(graphId);
+        String syncId = UUID.randomUUID().toString().substring(0, 8);
+        ReentrantLock lock = acquireLock(graphId, syncId);
+        try {
+            List<JobDTO> jobs = fetchWithRetry(syncId, "jobs",
+                    () -> dataManagementClient.listAllJobs());
+            SyncResult result = stepService.upsertJobEntities(graphId, jobs, syncId);
+            Set<String> activeIds = jobs.stream()
+                    .filter(Objects::nonNull).map(JobDTO::getId)
+                    .filter(Objects::nonNull).filter(id -> !id.isBlank())
+                    .collect(Collectors.toSet());
+            result.setPurged(stepService.purgeStaleEntities(graphId, "Job", activeIds, syncId));
+            return result;
+        } catch (BusinessException e) {
+            throw e;
+        } catch (Exception e) {
+            log.error("[{}] Job sync failed for graphId={}", syncId, graphId, e);
+            throw BusinessException.of(KnowledgeGraphErrorCode.SYNC_FAILED, "作业同步失败，syncId=" + syncId);
+        } finally {
+            releaseLock(graphId, lock);
+        }
+    }
+
+    public SyncResult syncLabelTasks(String graphId) {
+        validateGraphId(graphId);
+        String syncId = UUID.randomUUID().toString().substring(0, 8);
+        ReentrantLock lock = acquireLock(graphId, syncId);
+        try {
+            List<LabelTaskDTO> tasks = fetchWithRetry(syncId, "label-tasks",
+                    () -> dataManagementClient.listAllLabelTasks());
+            SyncResult result = stepService.upsertLabelTaskEntities(graphId, tasks, syncId);
+            Set<String> activeIds = tasks.stream()
+                    .filter(Objects::nonNull).map(LabelTaskDTO::getId)
+                    .filter(Objects::nonNull).filter(id -> !id.isBlank())
+                    .collect(Collectors.toSet());
+            result.setPurged(stepService.purgeStaleEntities(graphId, "LabelTask", activeIds, syncId));
+            return result;
+        } catch (BusinessException e) {
+            throw e;
+        } catch (Exception e) {
+            log.error("[{}] LabelTask sync failed for graphId={}", syncId, graphId, e);
+            throw BusinessException.of(KnowledgeGraphErrorCode.SYNC_FAILED, "标注任务同步失败，syncId=" + syncId);
+        } finally {
+            releaseLock(graphId, lock);
+        }
+    }
+
+    public SyncResult syncKnowledgeSets(String graphId) {
+        validateGraphId(graphId);
+        String syncId = UUID.randomUUID().toString().substring(0, 8);
+        ReentrantLock lock = acquireLock(graphId, syncId);
+        try {
+            List<KnowledgeSetDTO> knowledgeSets = fetchWithRetry(syncId, "knowledge-sets",
+                    () -> dataManagementClient.listAllKnowledgeSets());
+            SyncResult result = stepService.upsertKnowledgeSetEntities(graphId, knowledgeSets, syncId);
+            Set<String> activeIds = knowledgeSets.stream()
+                    .filter(Objects::nonNull).map(KnowledgeSetDTO::getId)
+                    .filter(Objects::nonNull).filter(id -> !id.isBlank())
+                    .collect(Collectors.toSet());
+            result.setPurged(stepService.purgeStaleEntities(graphId, "KnowledgeSet", activeIds, syncId));
+            return result;
+        } catch (BusinessException e) {
+            throw e;
+        } catch (Exception e) {
+            log.error("[{}] KnowledgeSet sync failed for graphId={}", syncId, graphId, e);
+            throw BusinessException.of(KnowledgeGraphErrorCode.SYNC_FAILED, "知识集同步失败，syncId=" + syncId);
+        } finally {
+            releaseLock(graphId, lock);
+        }
+    }
+
+    // -----------------------------------------------------------------------
+    // 新增关系构建
+    // -----------------------------------------------------------------------
+
+    public SyncResult buildUsesDatasetRelations(String graphId) {
+        validateGraphId(graphId);
+        String syncId = UUID.randomUUID().toString().substring(0, 8);
+        ReentrantLock lock = acquireLock(graphId, syncId);
+        try {
+            return stepService.mergeUsesDatasetRelations(graphId, syncId);
+        } catch (BusinessException e) {
+            throw e;
+        } catch (Exception e) {
+            log.error("[{}] USES_DATASET relation build failed for graphId={}", syncId, graphId, e);
+            throw BusinessException.of(KnowledgeGraphErrorCode.SYNC_FAILED,
+                    "USES_DATASET 关系构建失败，syncId=" + syncId);
+        } finally {
+            releaseLock(graphId, lock);
+        }
+    }
+
+    public SyncResult buildProducesRelations(String graphId) {
+        validateGraphId(graphId);
+        String syncId = UUID.randomUUID().toString().substring(0, 8);
+        ReentrantLock lock = acquireLock(graphId, syncId);
+        try {
+            return stepService.mergeProducesRelations(graphId, syncId);
+        } catch (BusinessException e) {
+            throw e;
+        } catch (Exception e) {
+            log.error("[{}] PRODUCES relation build failed for graphId={}", syncId, graphId, e);
+            throw BusinessException.of(KnowledgeGraphErrorCode.SYNC_FAILED,
+                    "PRODUCES 关系构建失败，syncId=" + syncId);
+        } finally {
+            releaseLock(graphId, lock);
+        }
+    }
+
+    public SyncResult buildAssignedToRelations(String graphId) {
+        validateGraphId(graphId);
+        String syncId = UUID.randomUUID().toString().substring(0, 8);
+        ReentrantLock lock = acquireLock(graphId, syncId);
+        try {
+            return stepService.mergeAssignedToRelations(graphId, syncId);
+        } catch (BusinessException e) {
+            throw e;
+        } catch (Exception e) {
+            log.error("[{}] ASSIGNED_TO relation build failed for graphId={}", syncId, graphId, e);
+            throw BusinessException.of(KnowledgeGraphErrorCode.SYNC_FAILED,
+                    "ASSIGNED_TO 关系构建失败，syncId=" + syncId);
+        } finally {
+            releaseLock(graphId, lock);
+        }
+    }
+
+    public SyncResult buildTriggersRelations(String graphId) {
+        validateGraphId(graphId);
+        String syncId = UUID.randomUUID().toString().substring(0, 8);
+        ReentrantLock lock = acquireLock(graphId, syncId);
+        try {
+            return stepService.mergeTriggersRelations(graphId, syncId);
+        } catch (BusinessException e) {
+            throw e;
+        } catch (Exception e) {
+            log.error("[{}] TRIGGERS relation build failed for graphId={}", syncId, graphId, e);
+            throw BusinessException.of(KnowledgeGraphErrorCode.SYNC_FAILED,
+                    "TRIGGERS 关系构建失败，syncId=" + syncId);
+        } finally {
+            releaseLock(graphId, lock);
+        }
+    }
+
+    public SyncResult buildDependsOnRelations(String graphId) {
+        validateGraphId(graphId);
+        String syncId = UUID.randomUUID().toString().substring(0, 8);
+        ReentrantLock lock = acquireLock(graphId, syncId);
+        try {
+            return stepService.mergeDependsOnRelations(graphId, syncId);
+        } catch (BusinessException e) {
+            throw e;
+        } catch (Exception e) {
+            log.error("[{}] DEPENDS_ON relation build failed for graphId={}", syncId, graphId, e);
+            throw BusinessException.of(KnowledgeGraphErrorCode.SYNC_FAILED,
+                    "DEPENDS_ON 关系构建失败，syncId=" + syncId);
+        } finally {
+            releaseLock(graphId, lock);
+        }
+    }
+
+    public SyncResult buildImpactsRelations(String graphId) {
+        validateGraphId(graphId);
+        String syncId = UUID.randomUUID().toString().substring(0, 8);
+        ReentrantLock lock = acquireLock(graphId, syncId);
+        try {
+            return stepService.mergeImpactsRelations(graphId, syncId);
+        } catch (BusinessException e) {
+            throw e;
+        } catch (Exception e) {
+            log.error("[{}] IMPACTS relation build failed for graphId={}", syncId, graphId, e);
+            throw BusinessException.of(KnowledgeGraphErrorCode.SYNC_FAILED,
+                    "IMPACTS 关系构建失败，syncId=" + syncId);
+        } finally {
+            releaseLock(graphId, lock);
+        }
+    }
+
+    public SyncResult buildSourcedFromRelations(String graphId) {
+        validateGraphId(graphId);
+        String syncId = UUID.randomUUID().toString().substring(0, 8);
+        ReentrantLock lock = acquireLock(graphId, syncId);
+        try {
+            return stepService.mergeSourcedFromRelations(graphId, syncId);
+        } catch (BusinessException e) {
+            throw e;
+        } catch (Exception e) {
+            log.error("[{}] SOURCED_FROM relation build failed for graphId={}", syncId, graphId, e);
+            throw BusinessException.of(KnowledgeGraphErrorCode.SYNC_FAILED,
+                    "SOURCED_FROM 关系构建失败，syncId=" + syncId);
+        } finally {
+            releaseLock(graphId, lock);
+        }
+    }
+
     // -----------------------------------------------------------------------
     // 内部方法
     // -----------------------------------------------------------------------
@@ -289,16 +595,24 @@ public class GraphSyncService {
     }
 
     private List<DatasetDTO> fetchDatasetsWithRetry(String syncId) {
+        return fetchWithRetry(syncId, "datasets", () -> dataManagementClient.listAllDatasets());
+    }
+
+    /**
+     * 通用带重试的数据拉取方法。
+     */
+    private <T> List<T> fetchWithRetry(String syncId, String resourceName,
+                                        java.util.function.Supplier<List<T>> fetcher) {
         int maxRetries = properties.getSync().getMaxRetries();
         long retryInterval = properties.getSync().getRetryInterval();
         Exception lastException = null;
 
         for (int attempt = 1; attempt <= maxRetries; attempt++) {
             try {
-                return dataManagementClient.listAllDatasets();
+                return fetcher.get();
             } catch (Exception e) {
                 lastException = e;
-                log.warn("[{}] Dataset fetch attempt {}/{} failed: {}", syncId, attempt, maxRetries, e.getMessage());
+                log.warn("[{}] {} fetch attempt {}/{} failed: {}", syncId, resourceName, attempt, maxRetries, e.getMessage());
                 if (attempt < maxRetries) {
                     try {
                         Thread.sleep(retryInterval * attempt);
@@ -309,24 +623,54 @@ public class GraphSyncService {
                 }
             }
         }
-        log.error("[{}] All {} fetch attempts failed", syncId, maxRetries, lastException);
+        log.error("[{}] All {} fetch attempts for {} failed", syncId, maxRetries, resourceName, lastException);
         throw BusinessException.of(KnowledgeGraphErrorCode.SYNC_FAILED,
-                "拉取数据集失败（已重试 " + maxRetries + " 次），syncId=" + syncId);
+                "拉取" + resourceName + "失败（已重试 " + maxRetries + " 次），syncId=" + syncId);
     }
 
-    private static Set<String> extractUsernames(List<DatasetDTO> datasets) {
+    /**
+     * 从所有实体类型中提取用户名。
+     */
+    private static Set<String> extractUsernames(List<DatasetDTO> datasets,
+                                                 List<WorkflowDTO> workflows,
+                                                 List<JobDTO> jobs,
+                                                 List<LabelTaskDTO> labelTasks,
+                                                 List<KnowledgeSetDTO> knowledgeSets) {
         Set<String> usernames = new LinkedHashSet<>();
         for (DatasetDTO dto : datasets) {
-            if (dto.getCreatedBy() != null && !dto.getCreatedBy().isBlank()) {
+            if (dto == null) { continue; }
-                usernames.add(dto.getCreatedBy());
+            addIfPresent(usernames, dto.getCreatedBy());
+            addIfPresent(usernames, dto.getUpdatedBy());
         }
-            if (dto.getUpdatedBy() != null && !dto.getUpdatedBy().isBlank()) {
+        for (WorkflowDTO dto : workflows) {
-                usernames.add(dto.getUpdatedBy());
+            if (dto == null) { continue; }
+            addIfPresent(usernames, dto.getCreatedBy());
+            addIfPresent(usernames, dto.getUpdatedBy());
         }
+        for (JobDTO dto : jobs) {
+            if (dto == null) { continue; }
+            addIfPresent(usernames, dto.getCreatedBy());
+            addIfPresent(usernames, dto.getUpdatedBy());
+        }
+        for (LabelTaskDTO dto : labelTasks) {
+            if (dto == null) { continue; }
+            addIfPresent(usernames, dto.getCreatedBy());
+            addIfPresent(usernames, dto.getUpdatedBy());
+        }
+        for (KnowledgeSetDTO dto : knowledgeSets) {
+            if (dto == null) { continue; }
+            addIfPresent(usernames, dto.getCreatedBy());
+            addIfPresent(usernames, dto.getUpdatedBy());
         }
         return usernames;
     }
 
+    private static void addIfPresent(Set<String> set, String value) {
+        if (value != null && !value.isBlank()) {
+            set.add(value);
+        }
+    }
+
     private void validateGraphId(String graphId) {
         if (graphId == null || !UUID_PATTERN.matcher(graphId).matches()) {
             throw BusinessException.of(SystemErrorCode.INVALID_PARAMETER, "graphId 格式无效");

backend/services/knowledge-graph-service/src/main/java/com/datamate/knowledgegraph/application/GraphSyncStepService.java

@@ -5,6 +5,10 @@ import com.datamate.knowledgegraph.domain.model.SyncResult;
 import com.datamate.knowledgegraph.domain.repository.GraphEntityRepository;
 import com.datamate.knowledgegraph.infrastructure.client.DataManagementClient.DatasetDTO;
 import com.datamate.knowledgegraph.infrastructure.client.DataManagementClient.TagDTO;
+import com.datamate.knowledgegraph.infrastructure.client.DataManagementClient.WorkflowDTO;
+import com.datamate.knowledgegraph.infrastructure.client.DataManagementClient.JobDTO;
+import com.datamate.knowledgegraph.infrastructure.client.DataManagementClient.LabelTaskDTO;
+import com.datamate.knowledgegraph.infrastructure.client.DataManagementClient.KnowledgeSetDTO;
 import com.datamate.knowledgegraph.infrastructure.neo4j.KnowledgeGraphProperties;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -14,6 +18,7 @@ import org.springframework.transaction.annotation.Transactional;
 
 import java.time.LocalDateTime;
 import java.util.*;
+import java.util.stream.Collectors;
 
 /**
  * 同步步骤执行器（事务边界）。
@@ -48,6 +53,11 @@ public class GraphSyncStepService {
 
         for (int i = 0; i < datasets.size(); i++) {
             DatasetDTO dto = datasets.get(i);
+            if (dto == null) {
+                result.incrementSkipped();
+                continue;
+            }
+            String sourceId = dto.getId();
             try {
                 Map<String, Object> props = new HashMap<>();
                 props.put("dataset_type", dto.getDatasetType());
@@ -70,8 +80,8 @@ public class GraphSyncStepService {
                     log.debug("[{}] Processed {}/{} datasets", syncId, i + 1, datasets.size());
                 }
             } catch (Exception e) {
-                log.warn("[{}] Failed to upsert dataset: sourceId={}", syncId, dto.getId(), e);
+                log.warn("[{}] Failed to upsert dataset: sourceId={}", syncId, sourceId, e);
-                result.addError("dataset:" + dto.getId());
+                result.addError("dataset:" + sourceId);
             }
         }
         return endResult(result);
@@ -82,25 +92,30 @@ public class GraphSyncStepService {
         SyncResult result = beginResult("Field", syncId);
 
         for (DatasetDTO dto : datasets) {
-            if (dto.getTags() == null || dto.getTags().isEmpty()) {
+            if (dto == null || dto.getTags() == null || dto.getTags().isEmpty()) {
                 continue;
             }
+            String dtoId = dto.getId();
             for (TagDTO tag : dto.getTags()) {
+                if (tag == null) {
+                    continue;
+                }
+                String tagName = tag.getName();
                 try {
-                    String fieldSourceId = dto.getId() + ":tag:" + tag.getName();
+                    String fieldSourceId = dtoId + ":tag:" + tagName;
                     Map<String, Object> props = new HashMap<>();
                     props.put("data_type", "TAG");
-                    props.put("dataset_source_id", dto.getId());
+                    props.put("dataset_source_id", dtoId);
                     if (tag.getColor() != null) {
                         props.put("color", tag.getColor());
                     }
 
-                    upsertEntity(graphId, fieldSourceId, "Field", tag.getName(),
+                    upsertEntity(graphId, fieldSourceId, "Field", tagName,
                             "数据集[" + dto.getName() + "]的标签字段", props, result);
                 } catch (Exception e) {
                     log.warn("[{}] Failed to upsert field: dataset={}, tag={}",
-                            syncId, dto.getId(), tag.getName(), e);
+                            syncId, dtoId, tagName, e);
-                    result.addError("field:" + dto.getId() + ":" + tag.getName());
+                    result.addError("field:" + dtoId + ":" + tagName);
                 }
             }
         }
@@ -141,6 +156,213 @@ public class GraphSyncStepService {
         return endResult(result);
     }
 
+    @Transactional
+    public SyncResult upsertWorkflowEntities(String graphId, List<WorkflowDTO> workflows, String syncId) {
+        SyncResult result = beginResult("Workflow", syncId);
+        int batchSize = properties.getImportBatchSize();
+
+        for (int i = 0; i < workflows.size(); i++) {
+            WorkflowDTO dto = workflows.get(i);
+            if (dto == null) {
+                result.incrementSkipped();
+                continue;
+            }
+            String sourceId = dto.getId();
+            try {
+                Map<String, Object> props = new HashMap<>();
+                props.put("workflow_type", dto.getWorkflowType());
+                props.put("status", dto.getStatus());
+                if (dto.getVersion() != null) {
+                    props.put("version", dto.getVersion());
+                }
+                if (dto.getOperatorCount() != null) {
+                    props.put("operator_count", dto.getOperatorCount());
+                }
+                if (dto.getSchedule() != null) {
+                    props.put("schedule", dto.getSchedule());
+                }
+                if (dto.getInputDatasetIds() != null) {
+                    props.put("input_dataset_ids", dto.getInputDatasetIds());
+                }
+
+                upsertEntity(graphId, dto.getId(), "Workflow",
+                        dto.getName(), dto.getDescription(), props, result);
+
+                if ((i + 1) % batchSize == 0) {
+                    log.debug("[{}] Processed {}/{} workflows", syncId, i + 1, workflows.size());
+                }
+            } catch (Exception e) {
+                log.warn("[{}] Failed to upsert workflow: sourceId={}", syncId, sourceId, e);
+                result.addError("workflow:" + sourceId);
+            }
+        }
+        return endResult(result);
+    }
+
+    @Transactional
+    public SyncResult upsertJobEntities(String graphId, List<JobDTO> jobs, String syncId) {
+        SyncResult result = beginResult("Job", syncId);
+        int batchSize = properties.getImportBatchSize();
+
+        for (int i = 0; i < jobs.size(); i++) {
+            JobDTO dto = jobs.get(i);
+            if (dto == null) {
+                result.incrementSkipped();
+                continue;
+            }
+            String sourceId = dto.getId();
+            try {
+                Map<String, Object> props = new HashMap<>();
+                props.put("job_type", dto.getJobType());
+                props.put("status", dto.getStatus());
+                if (dto.getStartedAt() != null) {
+                    props.put("started_at", dto.getStartedAt());
+                }
+                if (dto.getCompletedAt() != null) {
+                    props.put("completed_at", dto.getCompletedAt());
+                }
+                if (dto.getDurationSeconds() != null) {
+                    props.put("duration_seconds", dto.getDurationSeconds());
+                }
+                if (dto.getInputCount() != null) {
+                    props.put("input_count", dto.getInputCount());
+                }
+                if (dto.getOutputCount() != null) {
+                    props.put("output_count", dto.getOutputCount());
+                }
+                if (dto.getErrorMessage() != null) {
+                    props.put("error_message", dto.getErrorMessage());
+                }
+                if (dto.getInputDatasetId() != null) {
+                    props.put("input_dataset_id", dto.getInputDatasetId());
+                }
+                if (dto.getOutputDatasetId() != null) {
+                    props.put("output_dataset_id", dto.getOutputDatasetId());
+                }
+                if (dto.getWorkflowId() != null) {
+                    props.put("workflow_id", dto.getWorkflowId());
+                }
+                if (dto.getDependsOnJobId() != null) {
+                    props.put("depends_on_job_id", dto.getDependsOnJobId());
+                }
+                if (dto.getCreatedBy() != null) {
+                    props.put("created_by", dto.getCreatedBy());
+                }
+
+                upsertEntity(graphId, dto.getId(), "Job",
+                        dto.getName(), dto.getDescription(), props, result);
+
+                if ((i + 1) % batchSize == 0) {
+                    log.debug("[{}] Processed {}/{} jobs", syncId, i + 1, jobs.size());
+                }
+            } catch (Exception e) {
+                log.warn("[{}] Failed to upsert job: sourceId={}", syncId, sourceId, e);
+                result.addError("job:" + sourceId);
+            }
+        }
+        return endResult(result);
+    }
+
+    @Transactional
+    public SyncResult upsertLabelTaskEntities(String graphId, List<LabelTaskDTO> tasks, String syncId) {
+        SyncResult result = beginResult("LabelTask", syncId);
+        int batchSize = properties.getImportBatchSize();
+
+        for (int i = 0; i < tasks.size(); i++) {
+            LabelTaskDTO dto = tasks.get(i);
+            if (dto == null) {
+                result.incrementSkipped();
+                continue;
+            }
+            String sourceId = dto.getId();
+            try {
+                Map<String, Object> props = new HashMap<>();
+                props.put("task_mode", dto.getTaskMode());
+                props.put("status", dto.getStatus());
+                if (dto.getDataType() != null) {
+                    props.put("data_type", dto.getDataType());
+                }
+                if (dto.getLabelingType() != null) {
+                    props.put("labeling_type", dto.getLabelingType());
+                }
+                if (dto.getProgress() != null) {
+                    props.put("progress", dto.getProgress());
+                }
+                if (dto.getTemplateName() != null) {
+                    props.put("template_name", dto.getTemplateName());
+                }
+                if (dto.getDatasetId() != null) {
+                    props.put("dataset_id", dto.getDatasetId());
+                }
+                if (dto.getCreatedBy() != null) {
+                    props.put("created_by", dto.getCreatedBy());
+                }
+
+                upsertEntity(graphId, dto.getId(), "LabelTask",
+                        dto.getName(), dto.getDescription(), props, result);
+
+                if ((i + 1) % batchSize == 0) {
+                    log.debug("[{}] Processed {}/{} label tasks", syncId, i + 1, tasks.size());
+                }
+            } catch (Exception e) {
+                log.warn("[{}] Failed to upsert label task: sourceId={}", syncId, sourceId, e);
+                result.addError("label_task:" + sourceId);
+            }
+        }
+        return endResult(result);
+    }
+
+    @Transactional
+    public SyncResult upsertKnowledgeSetEntities(String graphId, List<KnowledgeSetDTO> knowledgeSets, String syncId) {
+        SyncResult result = beginResult("KnowledgeSet", syncId);
+        int batchSize = properties.getImportBatchSize();
+
+        for (int i = 0; i < knowledgeSets.size(); i++) {
+            KnowledgeSetDTO dto = knowledgeSets.get(i);
+            if (dto == null) {
+                result.incrementSkipped();
+                continue;
+            }
+            String sourceId = dto.getId();
+            try {
+                Map<String, Object> props = new HashMap<>();
+                props.put("status", dto.getStatus());
+                if (dto.getDomain() != null) {
+                    props.put("domain", dto.getDomain());
+                }
+                if (dto.getBusinessLine() != null) {
+                    props.put("business_line", dto.getBusinessLine());
+                }
+                if (dto.getSensitivity() != null) {
+                    props.put("sensitivity", dto.getSensitivity());
+                }
+                if (dto.getItemCount() != null) {
+                    props.put("item_count", dto.getItemCount());
+                }
+                if (dto.getValidFrom() != null) {
+                    props.put("valid_from", dto.getValidFrom());
+                }
+                if (dto.getValidTo() != null) {
+                    props.put("valid_to", dto.getValidTo());
+                }
+                if (dto.getSourceDatasetIds() != null) {
+                    props.put("source_dataset_ids", dto.getSourceDatasetIds());
+                }
+
+                upsertEntity(graphId, dto.getId(), "KnowledgeSet",
+                        dto.getName(), dto.getDescription(), props, result);
+
+                if ((i + 1) % batchSize == 0) {
+                    log.debug("[{}] Processed {}/{} knowledge sets", syncId, i + 1, knowledgeSets.size());
+                }
+            } catch (Exception e) {
+                log.warn("[{}] Failed to upsert knowledge set: sourceId={}", syncId, sourceId, e);
+                result.addError("knowledge_set:" + sourceId);
+            }
+        }
+        return endResult(result);
+    }
+
     // -----------------------------------------------------------------------
     // 全量对账删除
     // -----------------------------------------------------------------------
@@ -154,7 +376,14 @@ public class GraphSyncStepService {
      */
     @Transactional
     public int purgeStaleEntities(String graphId, String type, Set<String> activeSourceIds, String syncId) {
-        if (activeSourceIds.isEmpty()) {
+        // 防御式过滤：移除 null / 空白 ID，防止 Cypher 三值逻辑导致 IN 判断失效
+        Set<String> sanitized = activeSourceIds.stream()
+                .filter(Objects::nonNull)
+                .map(String::trim)
+                .filter(s -> !s.isEmpty())
+                .collect(Collectors.toSet());
+
+        if (sanitized.isEmpty()) {
             if (!properties.getSync().isAllowPurgeOnEmptySnapshot()) {
                 log.warn("[{}] Empty snapshot protection: active source IDs empty for type={}, " +
                         "purge BLOCKED (set allowPurgeOnEmptySnapshot=true to override)", syncId, type);
@@ -167,7 +396,7 @@ public class GraphSyncStepService {
         String cypher;
         Map<String, Object> params;
 
-        if (activeSourceIds.isEmpty()) {
+        if (sanitized.isEmpty()) {
             cypher = "MATCH (e:Entity {graph_id: $graphId, type: $type, source_type: 'SYNC'}) " +
                     "DETACH DELETE e " +
                     "RETURN count(*) AS deleted";
@@ -180,7 +409,7 @@ public class GraphSyncStepService {
             params = Map.of(
                     "graphId", graphId,
                     "type", type,
-                    "activeSourceIds", new ArrayList<>(activeSourceIds)
+                    "activeSourceIds", new ArrayList<>(sanitized)
             );
         }
 
@@ -205,6 +434,7 @@ public class GraphSyncStepService {
     public SyncResult mergeHasFieldRelations(String graphId, String syncId) {
         SyncResult result = beginResult("HAS_FIELD", syncId);
 
+        Map<String, String> datasetMap = buildSourceIdToEntityIdMap(graphId, "Dataset");
         List<GraphEntity> fields = entityRepository.findByGraphIdAndType(graphId, "Field");
 
         for (GraphEntity field : fields) {
@@ -215,14 +445,13 @@ public class GraphSyncStepService {
                     continue;
                 }
 
-                Optional<GraphEntity> datasetOpt = entityRepository.findByGraphIdAndSourceIdAndType(
+                String datasetEntityId = datasetMap.get(datasetSourceId.toString());
-                        graphId, datasetSourceId.toString(), "Dataset");
+                if (datasetEntityId == null) {
-                if (datasetOpt.isEmpty()) {
                     result.incrementSkipped();
                     continue;
                 }
 
-                boolean created = mergeRelation(graphId, datasetOpt.get().getId(), field.getId(),
+                boolean created = mergeRelation(graphId, datasetEntityId, field.getId(),
                         "HAS_FIELD", "{}", syncId);
                 if (created) {
                     result.incrementCreated();
@@ -241,6 +470,7 @@ public class GraphSyncStepService {
     public SyncResult mergeDerivedFromRelations(String graphId, String syncId) {
         SyncResult result = beginResult("DERIVED_FROM", syncId);
 
+        Map<String, String> datasetMap = buildSourceIdToEntityIdMap(graphId, "Dataset");
         List<GraphEntity> datasets = entityRepository.findByGraphIdAndType(graphId, "Dataset");
 
         for (GraphEntity dataset : datasets) {
@@ -250,14 +480,13 @@ public class GraphSyncStepService {
                     continue;
                 }
 
-                Optional<GraphEntity> parentOpt = entityRepository.findByGraphIdAndSourceIdAndType(
+                String parentEntityId = datasetMap.get(parentId.toString());
-                        graphId, parentId.toString(), "Dataset");
+                if (parentEntityId == null) {
-                if (parentOpt.isEmpty()) {
                     result.incrementSkipped();
                     continue;
                 }
 
-                boolean created = mergeRelation(graphId, dataset.getId(), parentOpt.get().getId(),
+                boolean created = mergeRelation(graphId, dataset.getId(), parentEntityId,
                         "DERIVED_FROM", "{\"derivation_type\":\"VERSION\"}", syncId);
                 if (created) {
                     result.incrementCreated();
@@ -311,6 +540,255 @@ public class GraphSyncStepService {
         return endResult(result);
     }
 
+    /**
+     * 构建 USES_DATASET 关系：Job/LabelTask/Workflow → Dataset。
+     * <p>
+     * 通过实体扩展属性中的外键字段查找关联 Dataset：
+     * - Job.input_dataset_id → Dataset
+     * - LabelTask.dataset_id → Dataset
+     * - Workflow.input_dataset_ids → Dataset（多值）
+     */
+    @Transactional
+    public SyncResult mergeUsesDatasetRelations(String graphId, String syncId) {
+        SyncResult result = beginResult("USES_DATASET", syncId);
+
+        Map<String, String> datasetMap = buildSourceIdToEntityIdMap(graphId, "Dataset");
+
+        // Job → Dataset (via input_dataset_id)
+        for (GraphEntity job : entityRepository.findByGraphIdAndType(graphId, "Job")) {
+            mergeEntityToDatasets(graphId, job, "input_dataset_id", datasetMap, result, syncId);
+        }
+
+        // LabelTask → Dataset (via dataset_id)
+        for (GraphEntity task : entityRepository.findByGraphIdAndType(graphId, "LabelTask")) {
+            mergeEntityToDatasets(graphId, task, "dataset_id", datasetMap, result, syncId);
+        }
+
+        // Workflow → Dataset (via input_dataset_ids, multi-value)
+        for (GraphEntity workflow : entityRepository.findByGraphIdAndType(graphId, "Workflow")) {
+            mergeEntityToDatasets(graphId, workflow, "input_dataset_ids", datasetMap, result, syncId);
+        }
+
+        return endResult(result);
+    }
+
+    /**
+     * 统一处理实体到 Dataset 的 USES_DATASET 关系构建。
+     * 通过 {@link #toStringList} 兼容单值（String）和多值（List）属性。
+     * 使用预加载的 datasetMap 避免 N+1 查询。
+     */
+    private void mergeEntityToDatasets(String graphId, GraphEntity entity, String propertyKey,
+                                       Map<String, String> datasetMap,
+                                       SyncResult result, String syncId) {
+        try {
+            Object value = entity.getProperties().get(propertyKey);
+            if (value == null) {
+                return;
+            }
+            List<String> datasetIds = toStringList(value);
+            for (String dsId : datasetIds) {
+                String datasetEntityId = datasetMap.get(dsId);
+                if (datasetEntityId == null) {
+                    result.incrementSkipped();
+                    continue;
+                }
+                boolean created = mergeRelation(graphId, entity.getId(), datasetEntityId,
+                        "USES_DATASET", "{\"usage_role\":\"INPUT\"}", syncId);
+                if (created) { result.incrementCreated(); } else { result.incrementSkipped(); }
+            }
+        } catch (Exception e) {
+            log.warn("[{}] Failed to merge USES_DATASET for entity: id={}", syncId, entity.getId(), e);
+            result.addError("uses_dataset:" + entity.getId());
+        }
+    }
+
+    /**
+     * 构建 PRODUCES 关系：Job → Dataset（通过 output_dataset_id）。
+     */
+    @Transactional
+    public SyncResult mergeProducesRelations(String graphId, String syncId) {
+        SyncResult result = beginResult("PRODUCES", syncId);
+
+        Map<String, String> datasetMap = buildSourceIdToEntityIdMap(graphId, "Dataset");
+
+        for (GraphEntity job : entityRepository.findByGraphIdAndType(graphId, "Job")) {
+            try {
+                Object outputDatasetId = job.getProperties().get("output_dataset_id");
+                if (outputDatasetId == null || outputDatasetId.toString().isBlank()) {
+                    continue;
+                }
+                String datasetEntityId = datasetMap.get(outputDatasetId.toString());
+                if (datasetEntityId == null) {
+                    result.incrementSkipped();
+                    continue;
+                }
+                boolean created = mergeRelation(graphId, job.getId(), datasetEntityId,
+                        "PRODUCES", "{\"output_type\":\"PRIMARY\"}", syncId);
+                if (created) { result.incrementCreated(); } else { result.incrementSkipped(); }
+            } catch (Exception e) {
+                log.warn("[{}] Failed to merge PRODUCES for job: id={}", syncId, job.getId(), e);
+                result.addError("produces:" + job.getId());
+            }
+        }
+        return endResult(result);
+    }
+
+    /**
+     * 构建 ASSIGNED_TO 关系：LabelTask/Job → User（通过 createdBy 字段）。
+     */
+    @Transactional
+    public SyncResult mergeAssignedToRelations(String graphId, String syncId) {
+        SyncResult result = beginResult("ASSIGNED_TO", syncId);
+
+        Map<String, String> userMap = buildSourceIdToEntityIdMap(graphId, "User");
+
+        // LabelTask → User
+        for (GraphEntity task : entityRepository.findByGraphIdAndType(graphId, "LabelTask")) {
+            mergeCreatorAssignment(graphId, task, "label_task", userMap, result, syncId);
+        }
+
+        // Job → User
+        for (GraphEntity job : entityRepository.findByGraphIdAndType(graphId, "Job")) {
+            mergeCreatorAssignment(graphId, job, "job", userMap, result, syncId);
+        }
+
+        return endResult(result);
+    }
+
+    private void mergeCreatorAssignment(String graphId, GraphEntity entity, String entityLabel,
+                                        Map<String, String> userMap,
+                                        SyncResult result, String syncId) {
+        try {
+            Object createdBy = entity.getProperties().get("created_by");
+            if (createdBy == null || createdBy.toString().isBlank()) {
+                return;
+            }
+            String userSourceId = "user:" + createdBy;
+            String userEntityId = userMap.get(userSourceId);
+            if (userEntityId == null) {
+                result.incrementSkipped();
+                return;
+            }
+            boolean created = mergeRelation(graphId, entity.getId(), userEntityId,
+                    "ASSIGNED_TO", "{\"role\":\"OWNER\"}", syncId);
+            if (created) { result.incrementCreated(); } else { result.incrementSkipped(); }
+        } catch (Exception e) {
+            log.warn("[{}] Failed to merge ASSIGNED_TO for {}: id={}", syncId, entityLabel, entity.getId(), e);
+            result.addError("assigned_to:" + entityLabel + ":" + entity.getId());
+        }
+    }
+
+    /**
+     * 构建 TRIGGERS 关系：Workflow → Job（通过 Job.workflow_id）。
+     */
+    @Transactional
+    public SyncResult mergeTriggersRelations(String graphId, String syncId) {
+        SyncResult result = beginResult("TRIGGERS", syncId);
+
+        Map<String, String> workflowMap = buildSourceIdToEntityIdMap(graphId, "Workflow");
+
+        for (GraphEntity job : entityRepository.findByGraphIdAndType(graphId, "Job")) {
+            try {
+                Object workflowId = job.getProperties().get("workflow_id");
+                if (workflowId == null || workflowId.toString().isBlank()) {
+                    continue;
+                }
+                String workflowEntityId = workflowMap.get(workflowId.toString());
+                if (workflowEntityId == null) {
+                    result.incrementSkipped();
+                    continue;
+                }
+                // 方向：Workflow → Job
+                boolean created = mergeRelation(graphId, workflowEntityId, job.getId(),
+                        "TRIGGERS", "{\"trigger_type\":\"MANUAL\"}", syncId);
+                if (created) { result.incrementCreated(); } else { result.incrementSkipped(); }
+            } catch (Exception e) {
+                log.warn("[{}] Failed to merge TRIGGERS for job: id={}", syncId, job.getId(), e);
+                result.addError("triggers:" + job.getId());
+            }
+        }
+        return endResult(result);
+    }
+
+    /**
+     * 构建 DEPENDS_ON 关系：Job → Job（通过 Job.depends_on_job_id）。
+     */
+    @Transactional
+    public SyncResult mergeDependsOnRelations(String graphId, String syncId) {
+        SyncResult result = beginResult("DEPENDS_ON", syncId);
+
+        Map<String, String> jobMap = buildSourceIdToEntityIdMap(graphId, "Job");
+
+        for (GraphEntity job : entityRepository.findByGraphIdAndType(graphId, "Job")) {
+            try {
+                Object depJobId = job.getProperties().get("depends_on_job_id");
+                if (depJobId == null || depJobId.toString().isBlank()) {
+                    continue;
+                }
+                String depJobEntityId = jobMap.get(depJobId.toString());
+                if (depJobEntityId == null) {
+                    result.incrementSkipped();
+                    continue;
+                }
+                boolean created = mergeRelation(graphId, job.getId(), depJobEntityId,
+                        "DEPENDS_ON", "{\"dependency_type\":\"STRICT\"}", syncId);
+                if (created) { result.incrementCreated(); } else { result.incrementSkipped(); }
+            } catch (Exception e) {
+                log.warn("[{}] Failed to merge DEPENDS_ON for job: id={}", syncId, job.getId(), e);
+                result.addError("depends_on:" + job.getId());
+            }
+        }
+        return endResult(result);
+    }
+
+    /**
+     * 构建 IMPACTS 关系：Field → Field。
+     * <p>
+     * TODO: 字段影响关系来源于 LLM 抽取或规则引擎，而非简单外键关联。
+     * 当前 MVP 阶段为占位实现，后续由抽取模块填充。
+     */
+    @Transactional
+    public SyncResult mergeImpactsRelations(String graphId, String syncId) {
+        SyncResult result = beginResult("IMPACTS", syncId);
+        result.setPlaceholder(true);
+        log.debug("[{}] IMPACTS relations require extraction data, skipping in sync phase", syncId);
+        return endResult(result);
+    }
+
+    /**
+     * 构建 SOURCED_FROM 关系：KnowledgeSet → Dataset（通过 source_dataset_ids）。
+     */
+    @Transactional
+    public SyncResult mergeSourcedFromRelations(String graphId, String syncId) {
+        SyncResult result = beginResult("SOURCED_FROM", syncId);
+
+        Map<String, String> datasetMap = buildSourceIdToEntityIdMap(graphId, "Dataset");
+
+        for (GraphEntity ks : entityRepository.findByGraphIdAndType(graphId, "KnowledgeSet")) {
+            try {
+                Object sourceIds = ks.getProperties().get("source_dataset_ids");
+                if (sourceIds == null) {
+                    continue;
+                }
+                List<String> datasetIds = toStringList(sourceIds);
+                for (String dsId : datasetIds) {
+                    String datasetEntityId = datasetMap.get(dsId);
+                    if (datasetEntityId == null) {
+                        result.incrementSkipped();
+                        continue;
+                    }
+                    boolean created = mergeRelation(graphId, ks.getId(), datasetEntityId,
+                            "SOURCED_FROM", "{}", syncId);
+                    if (created) { result.incrementCreated(); } else { result.incrementSkipped(); }
+                }
+            } catch (Exception e) {
+                log.warn("[{}] Failed to merge SOURCED_FROM for knowledge set: id={}", syncId, ks.getId(), e);
+                result.addError("sourced_from:" + ks.getId());
+            }
+        }
+        return endResult(result);
+    }
+
     // -----------------------------------------------------------------------
     // 内部方法
     // -----------------------------------------------------------------------
@@ -395,8 +873,11 @@ public class GraphSyncStepService {
             if (list.isEmpty()) {
                 return List.of();
             }
-            // Neo4j 要求列表元素类型一致，统一转为 String 列表
+            // Neo4j 要求列表元素类型一致，统一转为 String 列表；过滤 null 防止脏数据
-            return list.stream().map(Object::toString).toList();
+            return list.stream()
+                    .filter(Objects::nonNull)
+                    .map(Object::toString)
+                    .toList();
         }
         return value;
     }
@@ -406,6 +887,24 @@ public class GraphSyncStepService {
      *
      * @return true 如果是新创建的关系，false 如果已存在
      */
+    /**
+     * 将属性值（可能是 List 或单个 String）安全转换为 String 列表。
+     */
+    @SuppressWarnings("unchecked")
+    private static List<String> toStringList(Object value) {
+        if (value instanceof List<?> list) {
+            return list.stream()
+                    .filter(Objects::nonNull)
+                    .map(Object::toString)
+                    .filter(s -> !s.isBlank())
+                    .toList();
+        }
+        if (value instanceof String str && !str.isBlank()) {
+            return List.of(str);
+        }
+        return List.of();
+    }
+
     private boolean mergeRelation(String graphId, String sourceEntityId, String targetEntityId,
                                   String relationType, String propertiesJson, String syncId) {
         String newId = UUID.randomUUID().toString();
@@ -448,4 +947,13 @@ public class GraphSyncStepService {
         result.setCompletedAt(LocalDateTime.now());
         return result;
     }
+
+    /**
+     * 预加载指定类型的 sourceId → entityId 映射，消除关系构建中的 N+1 查询。
+     */
+    private Map<String, String> buildSourceIdToEntityIdMap(String graphId, String type) {
+        return entityRepository.findByGraphIdAndType(graphId, type).stream()
+                .filter(e -> e.getSourceId() != null)
+                .collect(Collectors.toMap(GraphEntity::getSourceId, GraphEntity::getId, (a, b) -> a));
+    }
 }

backend/services/knowledge-graph-service/src/main/java/com/datamate/knowledgegraph/domain/model/SyncResult.java

@@ -40,6 +40,10 @@ public class SyncResult {
     @Builder.Default
     private int purged = 0;
 
+    /** 标记为占位符的步骤（功能尚未实现，结果无实际数据） */
+    @Builder.Default
+    private boolean placeholder = false;
+
     @Builder.Default
     private List<String> errors = new ArrayList<>();
 

backend/services/knowledge-graph-service/src/main/java/com/datamate/knowledgegraph/infrastructure/client/DataManagementClient.java

@@ -28,6 +28,7 @@ public class DataManagementClient {
 
     private final RestTemplate restTemplate;
     private final String baseUrl;
+    private final String annotationBaseUrl;
     private final int pageSize;
 
     public DataManagementClient(
@@ -35,6 +36,7 @@ public class DataManagementClient {
             KnowledgeGraphProperties properties) {
         this.restTemplate = restTemplate;
         this.baseUrl = properties.getSync().getDataManagementUrl();
+        this.annotationBaseUrl = properties.getSync().getAnnotationServiceUrl();
         this.pageSize = properties.getSync().getPageSize();
     }
 
@@ -42,40 +44,90 @@ public class DataManagementClient {
      * 拉取所有数据集（自动分页）。
      */
     public List<DatasetDTO> listAllDatasets() {
-        List<DatasetDTO> allDatasets = new ArrayList<>();
+        return fetchAllPaged(
+                baseUrl + "/data-management/datasets",
+                new ParameterizedTypeReference<PagedResult<DatasetDTO>>() {},
+                "datasets");
+    }
+
+    /**
+     * 拉取所有工作流（自动分页）。
+     */
+    public List<WorkflowDTO> listAllWorkflows() {
+        return fetchAllPaged(
+                baseUrl + "/data-management/workflows",
+                new ParameterizedTypeReference<PagedResult<WorkflowDTO>>() {},
+                "workflows");
+    }
+
+    /**
+     * 拉取所有作业（自动分页）。
+     */
+    public List<JobDTO> listAllJobs() {
+        return fetchAllPaged(
+                baseUrl + "/data-management/jobs",
+                new ParameterizedTypeReference<PagedResult<JobDTO>>() {},
+                "jobs");
+    }
+
+    /**
+     * 拉取所有标注任务（自动分页，从标注服务）。
+     */
+    public List<LabelTaskDTO> listAllLabelTasks() {
+        return fetchAllPaged(
+                annotationBaseUrl + "/annotation/label-tasks",
+                new ParameterizedTypeReference<PagedResult<LabelTaskDTO>>() {},
+                "label-tasks");
+    }
+
+    /**
+     * 拉取所有知识集（自动分页）。
+     */
+    public List<KnowledgeSetDTO> listAllKnowledgeSets() {
+        return fetchAllPaged(
+                baseUrl + "/data-management/knowledge-sets",
+                new ParameterizedTypeReference<PagedResult<KnowledgeSetDTO>>() {},
+                "knowledge-sets");
+    }
+
+    /**
+     * 通用自动分页拉取方法。
+     */
+    private <T> List<T> fetchAllPaged(String baseEndpoint,
+                                       ParameterizedTypeReference<PagedResult<T>> typeRef,
+                                       String resourceName) {
+        List<T> allItems = new ArrayList<>();
         int page = 0;
 
         while (true) {
-            String url = baseUrl + "/data-management/datasets?page=" + page + "&size=" + pageSize;
+            String url = baseEndpoint + "?page=" + page + "&size=" + pageSize;
-            log.debug("Fetching datasets: page={}, size={}", page, pageSize);
+            log.debug("Fetching {}: page={}, size={}", resourceName, page, pageSize);
 
             try {
-                ResponseEntity<PagedResult<DatasetDTO>> response = restTemplate.exchange(
+                ResponseEntity<PagedResult<T>> response = restTemplate.exchange(
-                        url, HttpMethod.GET, null,
+                        url, HttpMethod.GET, null, typeRef);
-                        new ParameterizedTypeReference<>() {}
-                );
 
-                PagedResult<DatasetDTO> body = response.getBody();
+                PagedResult<T> body = response.getBody();
                 if (body == null || body.getContent() == null || body.getContent().isEmpty()) {
                     break;
                 }
 
-                allDatasets.addAll(body.getContent());
+                allItems.addAll(body.getContent());
-                log.debug("Fetched {} datasets (page {}), total so far: {}",
+                log.debug("Fetched {} {} (page {}), total so far: {}",
-                        body.getContent().size(), page, allDatasets.size());
+                        body.getContent().size(), resourceName, page, allItems.size());
 
                 if (page >= body.getTotalPages() - 1) {
                     break;
                 }
                 page++;
             } catch (RestClientException e) {
-                log.error("Failed to fetch datasets from data-management-service: page={}, url={}", page, url, e);
+                log.error("Failed to fetch {} : page={}, url={}", resourceName, page, url, e);
                 throw e;
             }
         }
 
-        log.info("Fetched {} datasets in total from data-management-service", allDatasets.size());
+        log.info("Fetched {} {} in total", allItems.size(), resourceName);
-        return allDatasets;
+        return allItems;
     }
 
     // -----------------------------------------------------------------------
@@ -123,4 +175,104 @@ public class DataManagementClient {
         private String color;
         private String description;
     }
+
+    /**
+     * 与 data-management-service / data-cleaning-service 的 Workflow 对齐。
+     */
+    @Data
+    @JsonIgnoreProperties(ignoreUnknown = true)
+    public static class WorkflowDTO {
+        private String id;
+        private String name;
+        private String description;
+        private String workflowType;
+        private String status;
+        private String version;
+        private Integer operatorCount;
+        private String schedule;
+        private String createdBy;
+        private String updatedBy;
+        private LocalDateTime createdAt;
+        private LocalDateTime updatedAt;
+        /** 工作流使用的输入数据集 ID 列表 */
+        private List<String> inputDatasetIds;
+    }
+
+    /**
+     * 与 data-management-service 的 Job / CleaningTask / DataSynthInstance 等对齐。
+     */
+    @Data
+    @JsonIgnoreProperties(ignoreUnknown = true)
+    public static class JobDTO {
+        private String id;
+        private String name;
+        private String description;
+        private String jobType;
+        private String status;
+        private String startedAt;
+        private String completedAt;
+        private Long durationSeconds;
+        private Long inputCount;
+        private Long outputCount;
+        private String errorMessage;
+        private String createdBy;
+        private String updatedBy;
+        private LocalDateTime createdAt;
+        private LocalDateTime updatedAt;
+        /** 输入数据集 ID */
+        private String inputDatasetId;
+        /** 输出数据集 ID */
+        private String outputDatasetId;
+        /** 所属工作流 ID（TRIGGERS 关系） */
+        private String workflowId;
+        /** 依赖的作业 ID（DEPENDS_ON 关系） */
+        private String dependsOnJobId;
+    }
+
+    /**
+     * 与 data-annotation-service 的 LabelingProject / AutoAnnotationTask 对齐。
+     */
+    @Data
+    @JsonIgnoreProperties(ignoreUnknown = true)
+    public static class LabelTaskDTO {
+        private String id;
+        private String name;
+        private String description;
+        private String taskMode;
+        private String dataType;
+        private String labelingType;
+        private String status;
+        private Double progress;
+        private String templateName;
+        private String createdBy;
+        private String updatedBy;
+        private LocalDateTime createdAt;
+        private LocalDateTime updatedAt;
+        /** 标注使用的数据集 ID（USES_DATASET 关系） */
+        private String datasetId;
+    }
+
+    /**
+     * 与 data-management-service 的 KnowledgeSet 对齐。
+     */
+    @Data
+    @JsonIgnoreProperties(ignoreUnknown = true)
+    public static class KnowledgeSetDTO {
+        private String id;
+        private String name;
+        private String description;
+        private String status;
+        private String domain;
+        private String businessLine;
+        private String sensitivity;
+        private Integer itemCount;
+        private String validFrom;
+        private String validTo;
+        private String createdBy;
+        private String updatedBy;
+        private LocalDateTime createdAt;
+        private LocalDateTime updatedAt;
+        /** 来源数据集 ID 列表（SOURCED_FROM 关系） */
+        private List<String> sourceDatasetIds;
+    }
 }

backend/services/knowledge-graph-service/src/main/java/com/datamate/knowledgegraph/infrastructure/neo4j/KnowledgeGraphProperties.java

@@ -31,6 +31,9 @@ public class KnowledgeGraphProperties {
         /** 数据管理服务基础 URL */
         private String dataManagementUrl = "http://localhost:8080";
 
+        /** 标注服务基础 URL */
+        private String annotationServiceUrl = "http://localhost:8081";
+
         /** 同步每页拉取数量 */
         private int pageSize = 200;
 

backend/services/knowledge-graph-service/src/main/java/com/datamate/knowledgegraph/interfaces/dto/SyncResultVO.java

@@ -29,6 +29,8 @@ public class SyncResultVO {
     private int purged;
     private int total;
     private long durationMillis;
+    /** 标记为占位符的步骤（功能尚未实现） */
+    private boolean placeholder;
     /** 错误数量（不暴露具体错误信息） */
     private int errorCount;
     private LocalDateTime startedAt;
@@ -45,6 +47,7 @@ public class SyncResultVO {
                 .purged(result.getPurged())
                 .total(result.total())
                 .durationMillis(result.durationMillis())
+                .placeholder(result.isPlaceholder())
                 .errorCount(result.getErrors() != null ? result.getErrors().size() : 0)
                 .startedAt(result.getStartedAt())
                 .completedAt(result.getCompletedAt())

backend/services/knowledge-graph-service/src/main/java/com/datamate/knowledgegraph/interfaces/rest/GraphSyncController.java

@@ -104,4 +104,111 @@ public class GraphSyncController {
             @PathVariable @Pattern(regexp = UUID_REGEX, message = "graphId 格式无效") String graphId) {
         return SyncResultVO.from(syncService.buildBelongsToRelations(graphId));
     }
+
+    // -----------------------------------------------------------------------
+    // 新增实体同步端点
+    // -----------------------------------------------------------------------
+
+    /**
+     * 同步工作流实体。
+     */
+    @PostMapping("/workflows")
+    public SyncResultVO syncWorkflows(
+            @PathVariable @Pattern(regexp = UUID_REGEX, message = "graphId 格式无效") String graphId) {
+        return SyncResultVO.from(syncService.syncWorkflows(graphId));
+    }
+
+    /**
+     * 同步作业实体。
+     */
+    @PostMapping("/jobs")
+    public SyncResultVO syncJobs(
+            @PathVariable @Pattern(regexp = UUID_REGEX, message = "graphId 格式无效") String graphId) {
+        return SyncResultVO.from(syncService.syncJobs(graphId));
+    }
+
+    /**
+     * 同步标注任务实体。
+     */
+    @PostMapping("/label-tasks")
+    public SyncResultVO syncLabelTasks(
+            @PathVariable @Pattern(regexp = UUID_REGEX, message = "graphId 格式无效") String graphId) {
+        return SyncResultVO.from(syncService.syncLabelTasks(graphId));
+    }
+
+    /**
+     * 同步知识集实体。
+     */
+    @PostMapping("/knowledge-sets")
+    public SyncResultVO syncKnowledgeSets(
+            @PathVariable @Pattern(regexp = UUID_REGEX, message = "graphId 格式无效") String graphId) {
+        return SyncResultVO.from(syncService.syncKnowledgeSets(graphId));
+    }
+
+    // -----------------------------------------------------------------------
+    // 新增关系构建端点
+    // -----------------------------------------------------------------------
+
+    /**
+     * 构建 USES_DATASET 关系。
+     */
+    @PostMapping("/relations/uses-dataset")
+    public SyncResultVO buildUsesDatasetRelations(
+            @PathVariable @Pattern(regexp = UUID_REGEX, message = "graphId 格式无效") String graphId) {
+        return SyncResultVO.from(syncService.buildUsesDatasetRelations(graphId));
+    }
+
+    /**
+     * 构建 PRODUCES 关系。
+     */
+    @PostMapping("/relations/produces")
+    public SyncResultVO buildProducesRelations(
+            @PathVariable @Pattern(regexp = UUID_REGEX, message = "graphId 格式无效") String graphId) {
+        return SyncResultVO.from(syncService.buildProducesRelations(graphId));
+    }
+
+    /**
+     * 构建 ASSIGNED_TO 关系。
+     */
+    @PostMapping("/relations/assigned-to")
+    public SyncResultVO buildAssignedToRelations(
+            @PathVariable @Pattern(regexp = UUID_REGEX, message = "graphId 格式无效") String graphId) {
+        return SyncResultVO.from(syncService.buildAssignedToRelations(graphId));
+    }
+
+    /**
+     * 构建 TRIGGERS 关系。
+     */
+    @PostMapping("/relations/triggers")
+    public SyncResultVO buildTriggersRelations(
+            @PathVariable @Pattern(regexp = UUID_REGEX, message = "graphId 格式无效") String graphId) {
+        return SyncResultVO.from(syncService.buildTriggersRelations(graphId));
+    }
+
+    /**
+     * 构建 DEPENDS_ON 关系。
+     */
+    @PostMapping("/relations/depends-on")
+    public SyncResultVO buildDependsOnRelations(
+            @PathVariable @Pattern(regexp = UUID_REGEX, message = "graphId 格式无效") String graphId) {
+        return SyncResultVO.from(syncService.buildDependsOnRelations(graphId));
+    }
+
+    /**
+     * 构建 IMPACTS 关系。
+     */
+    @PostMapping("/relations/impacts")
+    public SyncResultVO buildImpactsRelations(
+            @PathVariable @Pattern(regexp = UUID_REGEX, message = "graphId 格式无效") String graphId) {
+        return SyncResultVO.from(syncService.buildImpactsRelations(graphId));
+    }
+
+    /**
+     * 构建 SOURCED_FROM 关系。
+     */
+    @PostMapping("/relations/sourced-from")
+    public SyncResultVO buildSourcedFromRelations(
+            @PathVariable @Pattern(regexp = UUID_REGEX, message = "graphId 格式无效") String graphId) {
+        return SyncResultVO.from(syncService.buildSourcedFromRelations(graphId));
+    }
 }

backend/services/knowledge-graph-service/src/main/resources/application-knowledgegraph.yml

@@ -27,6 +27,8 @@ datamate:
     sync:
       # 数据管理服务地址
       data-management-url: ${DATA_MANAGEMENT_URL:http://localhost:8080}
+      # 标注服务地址
+      annotation-service-url: ${ANNOTATION_SERVICE_URL:http://localhost:8081}
       # 每页拉取数量
       page-size: ${KG_SYNC_PAGE_SIZE:200}
       # HTTP 连接超时（毫秒）

backend/services/knowledge-graph-service/src/test/java/com/datamate/knowledgegraph/application/GraphSyncServiceTest.java

@@ -4,8 +4,13 @@ import com.datamate.common.infrastructure.exception.BusinessException;
 import com.datamate.knowledgegraph.domain.model.SyncResult;
 import com.datamate.knowledgegraph.infrastructure.client.DataManagementClient;
 import com.datamate.knowledgegraph.infrastructure.client.DataManagementClient.DatasetDTO;
+import com.datamate.knowledgegraph.infrastructure.client.DataManagementClient.WorkflowDTO;
+import com.datamate.knowledgegraph.infrastructure.client.DataManagementClient.JobDTO;
+import com.datamate.knowledgegraph.infrastructure.client.DataManagementClient.LabelTaskDTO;
+import com.datamate.knowledgegraph.infrastructure.client.DataManagementClient.KnowledgeSetDTO;
 import com.datamate.knowledgegraph.infrastructure.neo4j.KnowledgeGraphProperties;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.InjectMocks;
@@ -13,6 +18,9 @@ import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 
 import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -68,12 +76,36 @@ class GraphSyncServiceTest {
                 .isInstanceOf(BusinessException.class);
     }
 
+    @Test
+    void syncWorkflows_invalidGraphId_throwsBusinessException() {
+        assertThatThrownBy(() -> syncService.syncWorkflows(INVALID_GRAPH_ID))
+                .isInstanceOf(BusinessException.class);
+    }
+
+    @Test
+    void syncJobs_invalidGraphId_throwsBusinessException() {
+        assertThatThrownBy(() -> syncService.syncJobs(INVALID_GRAPH_ID))
+                .isInstanceOf(BusinessException.class);
+    }
+
+    @Test
+    void syncLabelTasks_invalidGraphId_throwsBusinessException() {
+        assertThatThrownBy(() -> syncService.syncLabelTasks(INVALID_GRAPH_ID))
+                .isInstanceOf(BusinessException.class);
+    }
+
+    @Test
+    void syncKnowledgeSets_invalidGraphId_throwsBusinessException() {
+        assertThatThrownBy(() -> syncService.syncKnowledgeSets(INVALID_GRAPH_ID))
+                .isInstanceOf(BusinessException.class);
+    }
+
     // -----------------------------------------------------------------------
-    // syncAll — 正常流程
+    // syncAll — 正常流程（8 实体 + 10 关系 = 18 结果）
     // -----------------------------------------------------------------------
 
     @Test
-    void syncAll_success_returnsResultList() {
+    void syncAll_success_returnsAllResults() {
         when(properties.getSync()).thenReturn(syncConfig);
 
         DatasetDTO dto = new DatasetDTO();
@@ -81,40 +113,75 @@ class GraphSyncServiceTest {
         dto.setName("Test");
         dto.setCreatedBy("admin");
         when(dataManagementClient.listAllDatasets()).thenReturn(List.of(dto));
+        when(dataManagementClient.listAllWorkflows()).thenReturn(List.of());
+        when(dataManagementClient.listAllJobs()).thenReturn(List.of());
+        when(dataManagementClient.listAllLabelTasks()).thenReturn(List.of());
+        when(dataManagementClient.listAllKnowledgeSets()).thenReturn(List.of());
 
-        SyncResult entityResult = SyncResult.builder().syncType("Dataset").build();
+        // 8 entity upsert results
-        SyncResult fieldResult = SyncResult.builder().syncType("Field").build();
-        SyncResult userResult = SyncResult.builder().syncType("User").build();
-        SyncResult orgResult = SyncResult.builder().syncType("Org").build();
-        SyncResult hasFieldResult = SyncResult.builder().syncType("HAS_FIELD").build();
-        SyncResult derivedFromResult = SyncResult.builder().syncType("DERIVED_FROM").build();
-        SyncResult belongsToResult = SyncResult.builder().syncType("BELONGS_TO").build();
-
         when(stepService.upsertDatasetEntities(eq(GRAPH_ID), anyList(), anyString()))
-                .thenReturn(entityResult);
+                .thenReturn(SyncResult.builder().syncType("Dataset").build());
         when(stepService.upsertFieldEntities(eq(GRAPH_ID), anyList(), anyString()))
-                .thenReturn(fieldResult);
+                .thenReturn(SyncResult.builder().syncType("Field").build());
         when(stepService.upsertUserEntities(eq(GRAPH_ID), anySet(), anyString()))
-                .thenReturn(userResult);
+                .thenReturn(SyncResult.builder().syncType("User").build());
         when(stepService.upsertOrgEntities(eq(GRAPH_ID), anyString()))
-                .thenReturn(orgResult);
+                .thenReturn(SyncResult.builder().syncType("Org").build());
+        when(stepService.upsertWorkflowEntities(eq(GRAPH_ID), anyList(), anyString()))
+                .thenReturn(SyncResult.builder().syncType("Workflow").build());
+        when(stepService.upsertJobEntities(eq(GRAPH_ID), anyList(), anyString()))
+                .thenReturn(SyncResult.builder().syncType("Job").build());
+        when(stepService.upsertLabelTaskEntities(eq(GRAPH_ID), anyList(), anyString()))
+                .thenReturn(SyncResult.builder().syncType("LabelTask").build());
+        when(stepService.upsertKnowledgeSetEntities(eq(GRAPH_ID), anyList(), anyString()))
+                .thenReturn(SyncResult.builder().syncType("KnowledgeSet").build());
+
         when(stepService.purgeStaleEntities(eq(GRAPH_ID), anyString(), anySet(), anyString()))
                 .thenReturn(0);
+
+        // 10 relation merge results
         when(stepService.mergeHasFieldRelations(eq(GRAPH_ID), anyString()))
-                .thenReturn(hasFieldResult);
+                .thenReturn(SyncResult.builder().syncType("HAS_FIELD").build());
         when(stepService.mergeDerivedFromRelations(eq(GRAPH_ID), anyString()))
-                .thenReturn(derivedFromResult);
+                .thenReturn(SyncResult.builder().syncType("DERIVED_FROM").build());
         when(stepService.mergeBelongsToRelations(eq(GRAPH_ID), anyString()))
-                .thenReturn(belongsToResult);
+                .thenReturn(SyncResult.builder().syncType("BELONGS_TO").build());
+        when(stepService.mergeUsesDatasetRelations(eq(GRAPH_ID), anyString()))
+                .thenReturn(SyncResult.builder().syncType("USES_DATASET").build());
+        when(stepService.mergeProducesRelations(eq(GRAPH_ID), anyString()))
+                .thenReturn(SyncResult.builder().syncType("PRODUCES").build());
+        when(stepService.mergeAssignedToRelations(eq(GRAPH_ID), anyString()))
+                .thenReturn(SyncResult.builder().syncType("ASSIGNED_TO").build());
+        when(stepService.mergeTriggersRelations(eq(GRAPH_ID), anyString()))
+                .thenReturn(SyncResult.builder().syncType("TRIGGERS").build());
+        when(stepService.mergeDependsOnRelations(eq(GRAPH_ID), anyString()))
+                .thenReturn(SyncResult.builder().syncType("DEPENDS_ON").build());
+        when(stepService.mergeImpactsRelations(eq(GRAPH_ID), anyString()))
+                .thenReturn(SyncResult.builder().syncType("IMPACTS").build());
+        when(stepService.mergeSourcedFromRelations(eq(GRAPH_ID), anyString()))
+                .thenReturn(SyncResult.builder().syncType("SOURCED_FROM").build());
 
         List<SyncResult> results = syncService.syncAll(GRAPH_ID);
 
-        assertThat(results).hasSize(7);
+        // 8 entities + 10 relations = 18
-        assertThat(results.get(0).getSyncType()).isEqualTo("Dataset");
+        assertThat(results).hasSize(18);
+
+        // 按 syncType 建立索引，避免依赖固定下标
+        Map<String, SyncResult> byType = results.stream()
+                .collect(Collectors.toMap(SyncResult::getSyncType, Function.identity()));
+
+        // 验证所有 8 个实体类型都存在
+        assertThat(byType).containsKeys("Dataset", "Field", "User", "Org",
+                "Workflow", "Job", "LabelTask", "KnowledgeSet");
+
+        // 验证所有 10 个关系类型都存在
+        assertThat(byType).containsKeys("HAS_FIELD", "DERIVED_FROM", "BELONGS_TO",
+                "USES_DATASET", "PRODUCES", "ASSIGNED_TO", "TRIGGERS",
+                "DEPENDS_ON", "IMPACTS", "SOURCED_FROM");
     }
 
     // -----------------------------------------------------------------------
-    // syncAll — 正常流程
+    // 重试耗尽
     // -----------------------------------------------------------------------
 
     @Test
@@ -124,6 +191,148 @@ class GraphSyncServiceTest {
 
         assertThatThrownBy(() -> syncService.syncDatasets(GRAPH_ID))
                 .isInstanceOf(BusinessException.class)
-                .hasMessageContaining("拉取数据集失败");
+                .hasMessageContaining("datasets");
+    }
+
+    // -----------------------------------------------------------------------
+    // 新增实体单步同步
+    // -----------------------------------------------------------------------
+
+    @Nested
+    class NewEntitySyncTest {
+
+        @Test
+        void syncWorkflows_success() {
+            when(properties.getSync()).thenReturn(syncConfig);
+
+            WorkflowDTO wf = new WorkflowDTO();
+            wf.setId("wf-001");
+            wf.setName("Clean Pipeline");
+            when(dataManagementClient.listAllWorkflows()).thenReturn(List.of(wf));
+            when(stepService.upsertWorkflowEntities(eq(GRAPH_ID), anyList(), anyString()))
+                    .thenReturn(SyncResult.builder().syncType("Workflow").build());
+            when(stepService.purgeStaleEntities(eq(GRAPH_ID), eq("Workflow"), anySet(), anyString()))
+                    .thenReturn(0);
+
+            SyncResult result = syncService.syncWorkflows(GRAPH_ID);
+
+            assertThat(result.getSyncType()).isEqualTo("Workflow");
+            verify(stepService).upsertWorkflowEntities(eq(GRAPH_ID), anyList(), anyString());
+        }
+
+        @Test
+        void syncJobs_success() {
+            when(properties.getSync()).thenReturn(syncConfig);
+
+            JobDTO job = new JobDTO();
+            job.setId("job-001");
+            job.setName("Clean Job");
+            when(dataManagementClient.listAllJobs()).thenReturn(List.of(job));
+            when(stepService.upsertJobEntities(eq(GRAPH_ID), anyList(), anyString()))
+                    .thenReturn(SyncResult.builder().syncType("Job").build());
+            when(stepService.purgeStaleEntities(eq(GRAPH_ID), eq("Job"), anySet(), anyString()))
+                    .thenReturn(0);
+
+            SyncResult result = syncService.syncJobs(GRAPH_ID);
+
+            assertThat(result.getSyncType()).isEqualTo("Job");
+            verify(stepService).upsertJobEntities(eq(GRAPH_ID), anyList(), anyString());
+        }
+
+        @Test
+        void syncLabelTasks_success() {
+            when(properties.getSync()).thenReturn(syncConfig);
+
+            LabelTaskDTO task = new LabelTaskDTO();
+            task.setId("lt-001");
+            task.setName("Label Task");
+            when(dataManagementClient.listAllLabelTasks()).thenReturn(List.of(task));
+            when(stepService.upsertLabelTaskEntities(eq(GRAPH_ID), anyList(), anyString()))
+                    .thenReturn(SyncResult.builder().syncType("LabelTask").build());
+            when(stepService.purgeStaleEntities(eq(GRAPH_ID), eq("LabelTask"), anySet(), anyString()))
+                    .thenReturn(0);
+
+            SyncResult result = syncService.syncLabelTasks(GRAPH_ID);
+
+            assertThat(result.getSyncType()).isEqualTo("LabelTask");
+        }
+
+        @Test
+        void syncKnowledgeSets_success() {
+            when(properties.getSync()).thenReturn(syncConfig);
+
+            KnowledgeSetDTO ks = new KnowledgeSetDTO();
+            ks.setId("ks-001");
+            ks.setName("Knowledge Set");
+            when(dataManagementClient.listAllKnowledgeSets()).thenReturn(List.of(ks));
+            when(stepService.upsertKnowledgeSetEntities(eq(GRAPH_ID), anyList(), anyString()))
+                    .thenReturn(SyncResult.builder().syncType("KnowledgeSet").build());
+            when(stepService.purgeStaleEntities(eq(GRAPH_ID), eq("KnowledgeSet"), anySet(), anyString()))
+                    .thenReturn(0);
+
+            SyncResult result = syncService.syncKnowledgeSets(GRAPH_ID);
+
+            assertThat(result.getSyncType()).isEqualTo("KnowledgeSet");
+        }
+
+        @Test
+        void syncWorkflows_fetchFailed_throwsBusinessException() {
+            when(properties.getSync()).thenReturn(syncConfig);
+            when(dataManagementClient.listAllWorkflows()).thenThrow(new RuntimeException("timeout"));
+
+            assertThatThrownBy(() -> syncService.syncWorkflows(GRAPH_ID))
+                    .isInstanceOf(BusinessException.class)
+                    .hasMessageContaining("workflows");
+        }
+    }
+
+    // -----------------------------------------------------------------------
+    // 新增关系构建
+    // -----------------------------------------------------------------------
+
+    @Nested
+    class NewRelationBuildTest {
+
+        @Test
+        void buildUsesDatasetRelations_invalidGraphId_throws() {
+            assertThatThrownBy(() -> syncService.buildUsesDatasetRelations(INVALID_GRAPH_ID))
+                    .isInstanceOf(BusinessException.class);
+        }
+
+        @Test
+        void buildProducesRelations_invalidGraphId_throws() {
+            assertThatThrownBy(() -> syncService.buildProducesRelations(INVALID_GRAPH_ID))
+                    .isInstanceOf(BusinessException.class);
+        }
+
+        @Test
+        void buildAssignedToRelations_invalidGraphId_throws() {
+            assertThatThrownBy(() -> syncService.buildAssignedToRelations(INVALID_GRAPH_ID))
+                    .isInstanceOf(BusinessException.class);
+        }
+
+        @Test
+        void buildTriggersRelations_invalidGraphId_throws() {
+            assertThatThrownBy(() -> syncService.buildTriggersRelations(INVALID_GRAPH_ID))
+                    .isInstanceOf(BusinessException.class);
+        }
+
+        @Test
+        void buildDependsOnRelations_invalidGraphId_throws() {
+            assertThatThrownBy(() -> syncService.buildDependsOnRelations(INVALID_GRAPH_ID))
+                    .isInstanceOf(BusinessException.class);
+        }
+
+        @Test
+        void buildImpactsRelations_invalidGraphId_throws() {
+            assertThatThrownBy(() -> syncService.buildImpactsRelations(INVALID_GRAPH_ID))
+                    .isInstanceOf(BusinessException.class);
+        }
+
+        @Test
+        void buildSourcedFromRelations_invalidGraphId_throws() {
+            assertThatThrownBy(() -> syncService.buildSourcedFromRelations(INVALID_GRAPH_ID))
+                    .isInstanceOf(BusinessException.class);
+        }
     }
 }

backend/services/knowledge-graph-service/src/test/java/com/datamate/knowledgegraph/application/GraphSyncStepServiceTest.java

@@ -5,6 +5,10 @@ import com.datamate.knowledgegraph.domain.model.SyncResult;
 import com.datamate.knowledgegraph.domain.repository.GraphEntityRepository;
 import com.datamate.knowledgegraph.infrastructure.client.DataManagementClient.DatasetDTO;
 import com.datamate.knowledgegraph.infrastructure.client.DataManagementClient.TagDTO;
+import com.datamate.knowledgegraph.infrastructure.client.DataManagementClient.WorkflowDTO;
+import com.datamate.knowledgegraph.infrastructure.client.DataManagementClient.JobDTO;
+import com.datamate.knowledgegraph.infrastructure.client.DataManagementClient.LabelTaskDTO;
+import com.datamate.knowledgegraph.infrastructure.client.DataManagementClient.KnowledgeSetDTO;
 import com.datamate.knowledgegraph.infrastructure.neo4j.KnowledgeGraphProperties;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Nested;
@@ -22,6 +26,8 @@ import org.springframework.data.neo4j.core.Neo4jClient.RecordFetchSpec;
 import org.springframework.data.neo4j.core.Neo4jClient.MappingSpec;
 
 import java.util.*;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.*;
@@ -56,9 +62,6 @@ class GraphSyncStepServiceTest {
     // Neo4jClient mock chain helper
     // -----------------------------------------------------------------------
 
-    /**
-     * Build a full mock chain for neo4jClient.query(...).bindAll(...).fetchAs(Long).mappedBy(...).one()
-     */
     @SuppressWarnings("unchecked")
     private void setupNeo4jQueryChain(Class<?> fetchType, Object returnValue) {
         UnboundRunnableSpec unboundSpec = mock(UnboundRunnableSpec.class);
@@ -90,7 +93,6 @@ class GraphSyncStepServiceTest {
                     GRAPH_ID, "Dataset", Collections.emptySet(), SYNC_ID);
 
             assertThat(deleted).isEqualTo(0);
-            // Should NOT execute any Cypher query
             verifyNoInteractions(neo4jClient);
         }
 
@@ -133,7 +135,7 @@ class GraphSyncStepServiceTest {
     }
 
     // -----------------------------------------------------------------------
-    // upsertDatasetEntities — P2-5 单条 Cypher 优化
+    // upsertDatasetEntities
     // -----------------------------------------------------------------------
 
     @Nested
@@ -200,12 +202,10 @@ class GraphSyncStepServiceTest {
 
             stepService.upsertDatasetEntities(GRAPH_ID, List.of(dto), SYNC_ID);
 
-            // Verify the MERGE Cypher includes property SET clauses (N+1 fix)
             verify(neo4jClient).query(cypherCaptor.capture());
             String cypher = cypherCaptor.getValue();
             assertThat(cypher).contains("MERGE");
             assertThat(cypher).contains("properties.");
-            // Verify NO separate find+save (N+1 eliminated)
             verifyNoInteractions(entityRepository);
         }
 
@@ -270,7 +270,205 @@ class GraphSyncStepServiceTest {
     }
 
     // -----------------------------------------------------------------------
-    // 关系构建
+    // upsertWorkflowEntities
+    // -----------------------------------------------------------------------
+
+    @Nested
+    class UpsertWorkflowEntitiesTest {
+
+        @Test
+        void upsert_newWorkflow_incrementsCreated() {
+            when(properties.getImportBatchSize()).thenReturn(100);
+            setupNeo4jQueryChain(Boolean.class, true);
+
+            WorkflowDTO dto = new WorkflowDTO();
+            dto.setId("wf-001");
+            dto.setName("Clean Pipeline");
+            dto.setWorkflowType("CLEANING");
+            dto.setStatus("ACTIVE");
+            dto.setVersion("2.1");
+            dto.setOperatorCount(3);
+
+            SyncResult result = stepService.upsertWorkflowEntities(
+                    GRAPH_ID, List.of(dto), SYNC_ID);
+
+            assertThat(result.getCreated()).isEqualTo(1);
+            assertThat(result.getSyncType()).isEqualTo("Workflow");
+        }
+
+        @Test
+        void upsert_emptyList_returnsZero() {
+            when(properties.getImportBatchSize()).thenReturn(100);
+
+            SyncResult result = stepService.upsertWorkflowEntities(
+                    GRAPH_ID, List.of(), SYNC_ID);
+
+            assertThat(result.getCreated()).isEqualTo(0);
+            verifyNoInteractions(neo4jClient);
+        }
+
+        @Test
+        void upsert_withInputDatasetIds_storesProperty() {
+            when(properties.getImportBatchSize()).thenReturn(100);
+            setupNeo4jQueryChain(Boolean.class, true);
+
+            WorkflowDTO dto = new WorkflowDTO();
+            dto.setId("wf-001");
+            dto.setName("Pipeline");
+            dto.setWorkflowType("CLEANING");
+            dto.setInputDatasetIds(List.of("ds-001", "ds-002"));
+
+            stepService.upsertWorkflowEntities(GRAPH_ID, List.of(dto), SYNC_ID);
+
+            verify(neo4jClient).query(cypherCaptor.capture());
+            assertThat(cypherCaptor.getValue()).contains("properties.input_dataset_ids");
+        }
+    }
+
+    // -----------------------------------------------------------------------
+    // upsertJobEntities
+    // -----------------------------------------------------------------------
+
+    @Nested
+    class UpsertJobEntitiesTest {
+
+        @Test
+        void upsert_newJob_incrementsCreated() {
+            when(properties.getImportBatchSize()).thenReturn(100);
+            setupNeo4jQueryChain(Boolean.class, true);
+
+            JobDTO dto = new JobDTO();
+            dto.setId("job-001");
+            dto.setName("Clean Job");
+            dto.setJobType("CLEANING");
+            dto.setStatus("COMPLETED");
+            dto.setDurationSeconds(2100L);
+            dto.setInputDatasetId("ds-001");
+            dto.setOutputDatasetId("ds-002");
+            dto.setWorkflowId("wf-001");
+            dto.setCreatedBy("admin");
+
+            SyncResult result = stepService.upsertJobEntities(
+                    GRAPH_ID, List.of(dto), SYNC_ID);
+
+            assertThat(result.getCreated()).isEqualTo(1);
+            assertThat(result.getSyncType()).isEqualTo("Job");
+        }
+
+        @Test
+        void upsert_jobWithDependency_storesProperty() {
+            when(properties.getImportBatchSize()).thenReturn(100);
+            setupNeo4jQueryChain(Boolean.class, true);
+
+            JobDTO dto = new JobDTO();
+            dto.setId("job-002");
+            dto.setName("Downstream Job");
+            dto.setJobType("SYNTHESIS");
+            dto.setStatus("PENDING");
+            dto.setDependsOnJobId("job-001");
+
+            stepService.upsertJobEntities(GRAPH_ID, List.of(dto), SYNC_ID);
+
+            verify(neo4jClient).query(cypherCaptor.capture());
+            assertThat(cypherCaptor.getValue()).contains("properties.depends_on_job_id");
+        }
+
+        @Test
+        void upsert_emptyList_returnsZero() {
+            when(properties.getImportBatchSize()).thenReturn(100);
+
+            SyncResult result = stepService.upsertJobEntities(
+                    GRAPH_ID, List.of(), SYNC_ID);
+
+            assertThat(result.getCreated()).isEqualTo(0);
+            verifyNoInteractions(neo4jClient);
+        }
+    }
+
+    // -----------------------------------------------------------------------
+    // upsertLabelTaskEntities
+    // -----------------------------------------------------------------------
+
+    @Nested
+    class UpsertLabelTaskEntitiesTest {
+
+        @Test
+        void upsert_newLabelTask_incrementsCreated() {
+            when(properties.getImportBatchSize()).thenReturn(100);
+            setupNeo4jQueryChain(Boolean.class, true);
+
+            LabelTaskDTO dto = new LabelTaskDTO();
+            dto.setId("lt-001");
+            dto.setName("Label Task");
+            dto.setTaskMode("MANUAL");
+            dto.setStatus("IN_PROGRESS");
+            dto.setDataType("image");
+            dto.setLabelingType("object_detection");
+            dto.setProgress(45.5);
+            dto.setDatasetId("ds-001");
+            dto.setCreatedBy("admin");
+
+            SyncResult result = stepService.upsertLabelTaskEntities(
+                    GRAPH_ID, List.of(dto), SYNC_ID);
+
+            assertThat(result.getCreated()).isEqualTo(1);
+            assertThat(result.getSyncType()).isEqualTo("LabelTask");
+        }
+
+        @Test
+        void upsert_emptyList_returnsZero() {
+            when(properties.getImportBatchSize()).thenReturn(100);
+
+            SyncResult result = stepService.upsertLabelTaskEntities(
+                    GRAPH_ID, List.of(), SYNC_ID);
+
+            assertThat(result.getCreated()).isEqualTo(0);
+            verifyNoInteractions(neo4jClient);
+        }
+    }
+
+    // -----------------------------------------------------------------------
+    // upsertKnowledgeSetEntities
+    // -----------------------------------------------------------------------
+
+    @Nested
+    class UpsertKnowledgeSetEntitiesTest {
+
+        @Test
+        void upsert_newKnowledgeSet_incrementsCreated() {
+            when(properties.getImportBatchSize()).thenReturn(100);
+            setupNeo4jQueryChain(Boolean.class, true);
+
+            KnowledgeSetDTO dto = new KnowledgeSetDTO();
+            dto.setId("ks-001");
+            dto.setName("Medical Knowledge");
+            dto.setStatus("PUBLISHED");
+            dto.setDomain("medical");
+            dto.setSensitivity("INTERNAL");
+            dto.setItemCount(320);
+            dto.setSourceDatasetIds(List.of("ds-001", "ds-002"));
+
+            SyncResult result = stepService.upsertKnowledgeSetEntities(
+                    GRAPH_ID, List.of(dto), SYNC_ID);
+
+            assertThat(result.getCreated()).isEqualTo(1);
+            assertThat(result.getSyncType()).isEqualTo("KnowledgeSet");
+        }
+
+        @Test
+        void upsert_emptyList_returnsZero() {
+            when(properties.getImportBatchSize()).thenReturn(100);
+
+            SyncResult result = stepService.upsertKnowledgeSetEntities(
+                    GRAPH_ID, List.of(), SYNC_ID);
+
+            assertThat(result.getCreated()).isEqualTo(0);
+            verifyNoInteractions(neo4jClient);
+        }
+    }
+
+    // -----------------------------------------------------------------------
+    // 关系构建 - 已有
     // -----------------------------------------------------------------------
 
     @Nested
@@ -280,6 +478,8 @@ class GraphSyncStepServiceTest {
         void mergeHasField_noFields_returnsEmptyResult() {
             when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Field"))
                     .thenReturn(List.of());
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Dataset"))
+                    .thenReturn(List.of());
 
             SyncResult result = stepService.mergeHasFieldRelations(GRAPH_ID, SYNC_ID);
 
@@ -293,7 +493,7 @@ class GraphSyncStepServiceTest {
                     .id("entity-1")
                     .type("Dataset")
                     .graphId(GRAPH_ID)
-                    .properties(new HashMap<>()) // no parent_dataset_id
+                    .properties(new HashMap<>())
                     .build();
 
             when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Dataset"))
@@ -315,4 +515,307 @@ class GraphSyncStepServiceTest {
             assertThat(result.getErrors()).contains("belongs_to:org_missing");
         }
     }
+
+    // -----------------------------------------------------------------------
+    // 新增关系构建
+    // -----------------------------------------------------------------------
+
+    @Nested
+    class NewMergeRelationsTest {
+
+        @Test
+        void mergeUsesDataset_noJobs_noTasks_noWorkflows_returnsZero() {
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Job")).thenReturn(List.of());
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "LabelTask")).thenReturn(List.of());
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Workflow")).thenReturn(List.of());
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Dataset")).thenReturn(List.of());
+
+            SyncResult result = stepService.mergeUsesDatasetRelations(GRAPH_ID, SYNC_ID);
+
+            assertThat(result.getSyncType()).isEqualTo("USES_DATASET");
+            assertThat(result.getCreated()).isEqualTo(0);
+        }
+
+        @Test
+        void mergeUsesDataset_jobWithInputDataset_createsRelation() {
+            setupNeo4jQueryChain(String.class, "new-rel-id");
+
+            GraphEntity job = GraphEntity.builder()
+                    .id("job-entity-1").type("Job").graphId(GRAPH_ID)
+                    .properties(new HashMap<>(Map.of("input_dataset_id", "ds-001")))
+                    .build();
+            GraphEntity dataset = GraphEntity.builder()
+                    .id("ds-entity-1").sourceId("ds-001").type("Dataset").graphId(GRAPH_ID)
+                    .build();
+
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Job")).thenReturn(List.of(job));
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "LabelTask")).thenReturn(List.of());
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Workflow")).thenReturn(List.of());
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Dataset")).thenReturn(List.of(dataset));
+
+            SyncResult result = stepService.mergeUsesDatasetRelations(GRAPH_ID, SYNC_ID);
+
+            assertThat(result.getSyncType()).isEqualTo("USES_DATASET");
+            verify(neo4jClient).query(cypherCaptor.capture());
+            String cypher = cypherCaptor.getValue();
+            // 验证关系类型和方向：source → target
+            assertThat(cypher).contains("RELATED_TO");
+            assertThat(cypher).contains("relation_type: $relationType");
+        }
+
+        @Test
+        void mergeUsesDataset_workflowWithSingleStringInput_handledCorrectly() {
+            setupNeo4jQueryChain(String.class, "new-rel-id");
+
+            // 单值 String 而非 List，验证 toStringList 统一处理
+            GraphEntity workflow = GraphEntity.builder()
+                    .id("wf-entity-1").type("Workflow").graphId(GRAPH_ID)
+                    .properties(new HashMap<>(Map.of("input_dataset_ids", "ds-single")))
+                    .build();
+            GraphEntity dataset = GraphEntity.builder()
+                    .id("ds-entity-s").sourceId("ds-single").type("Dataset").graphId(GRAPH_ID)
+                    .build();
+
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Job")).thenReturn(List.of());
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "LabelTask")).thenReturn(List.of());
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Workflow")).thenReturn(List.of(workflow));
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Dataset")).thenReturn(List.of(dataset));
+
+            SyncResult result = stepService.mergeUsesDatasetRelations(GRAPH_ID, SYNC_ID);
+
+            verify(neo4jClient, atLeastOnce()).query(anyString());
+        }
+
+        @Test
+        void mergeUsesDataset_listWithNullElements_filtersNulls() {
+            setupNeo4jQueryChain(String.class, "new-rel-id");
+
+            List<Object> listWithNulls = new ArrayList<>();
+            listWithNulls.add("ds-good");
+            listWithNulls.add(null);
+            listWithNulls.add("");
+            listWithNulls.add("  ");
+
+            GraphEntity workflow = GraphEntity.builder()
+                    .id("wf-entity-2").type("Workflow").graphId(GRAPH_ID)
+                    .properties(new HashMap<>(Map.of("input_dataset_ids", listWithNulls)))
+                    .build();
+            GraphEntity dataset = GraphEntity.builder()
+                    .id("ds-entity-g").sourceId("ds-good").type("Dataset").graphId(GRAPH_ID)
+                    .build();
+
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Job")).thenReturn(List.of());
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "LabelTask")).thenReturn(List.of());
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Workflow")).thenReturn(List.of(workflow));
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Dataset")).thenReturn(List.of(dataset));
+
+            SyncResult result = stepService.mergeUsesDatasetRelations(GRAPH_ID, SYNC_ID);
+
+            // 只有 "ds-good" 应被处理（null、空、空白已过滤），验证只发起一次 mergeRelation
+            verify(neo4jClient, times(1)).query(anyString());
+        }
+
+        @Test
+        void mergeProduces_noJobs_returnsZero() {
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Job")).thenReturn(List.of());
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Dataset")).thenReturn(List.of());
+
+            SyncResult result = stepService.mergeProducesRelations(GRAPH_ID, SYNC_ID);
+
+            assertThat(result.getSyncType()).isEqualTo("PRODUCES");
+            assertThat(result.getCreated()).isEqualTo(0);
+        }
+
+        @Test
+        void mergeProduces_jobWithoutOutput_skips() {
+            GraphEntity job = GraphEntity.builder()
+                    .id("job-entity-1").type("Job").graphId(GRAPH_ID)
+                    .properties(new HashMap<>())
+                    .build();
+
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Job")).thenReturn(List.of(job));
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Dataset")).thenReturn(List.of());
+
+            SyncResult result = stepService.mergeProducesRelations(GRAPH_ID, SYNC_ID);
+
+            assertThat(result.getCreated()).isEqualTo(0);
+        }
+
+        @Test
+        void mergeAssignedTo_noTasksOrJobs_returnsZero() {
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "LabelTask")).thenReturn(List.of());
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Job")).thenReturn(List.of());
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "User")).thenReturn(List.of());
+
+            SyncResult result = stepService.mergeAssignedToRelations(GRAPH_ID, SYNC_ID);
+
+            assertThat(result.getSyncType()).isEqualTo("ASSIGNED_TO");
+            assertThat(result.getCreated()).isEqualTo(0);
+        }
+
+        @Test
+        void mergeAssignedTo_taskWithCreatedBy_verifiesUserLookup() {
+            setupNeo4jQueryChain(String.class, "new-rel-id");
+
+            GraphEntity task = GraphEntity.builder()
+                    .id("lt-entity-1").type("LabelTask").graphId(GRAPH_ID)
+                    .properties(new HashMap<>(Map.of("created_by", "admin")))
+                    .build();
+            GraphEntity user = GraphEntity.builder()
+                    .id("user-entity-1").sourceId("user:admin").type("User").graphId(GRAPH_ID)
+                    .build();
+
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "LabelTask")).thenReturn(List.of(task));
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Job")).thenReturn(List.of());
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "User")).thenReturn(List.of(user));
+
+            SyncResult result = stepService.mergeAssignedToRelations(GRAPH_ID, SYNC_ID);
+
+            assertThat(result.getSyncType()).isEqualTo("ASSIGNED_TO");
+            // 验证通过预加载的 userMap 查找 User 实体（不再有 N+1 查询）
+            verify(neo4jClient).query(cypherCaptor.capture());
+            assertThat(cypherCaptor.getValue()).contains("RELATED_TO");
+        }
+
+        @Test
+        void mergeTriggers_noJobs_returnsZero() {
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Job")).thenReturn(List.of());
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Workflow")).thenReturn(List.of());
+
+            SyncResult result = stepService.mergeTriggersRelations(GRAPH_ID, SYNC_ID);
+
+            assertThat(result.getSyncType()).isEqualTo("TRIGGERS");
+            assertThat(result.getCreated()).isEqualTo(0);
+        }
+
+        @Test
+        void mergeTriggers_jobWithWorkflowId_createsRelationWithCorrectDirection() {
+            setupNeo4jQueryChain(String.class, "new-rel-id");
+
+            GraphEntity job = GraphEntity.builder()
+                    .id("job-entity-1").type("Job").graphId(GRAPH_ID)
+                    .properties(new HashMap<>(Map.of("workflow_id", "wf-001")))
+                    .build();
+            GraphEntity workflow = GraphEntity.builder()
+                    .id("wf-entity-1").sourceId("wf-001").type("Workflow").graphId(GRAPH_ID)
+                    .build();
+
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Job")).thenReturn(List.of(job));
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Workflow")).thenReturn(List.of(workflow));
+
+            SyncResult result = stepService.mergeTriggersRelations(GRAPH_ID, SYNC_ID);
+
+            assertThat(result.getSyncType()).isEqualTo("TRIGGERS");
+            verify(neo4jClient).query(cypherCaptor.capture());
+            String cypher = cypherCaptor.getValue();
+            // 验证 MERGE 使用 RELATED_TO 关系类型
+            assertThat(cypher).contains("RELATED_TO");
+            // 验证 Cypher 参数绑定：source 应为 Workflow，target 应为 Job
+            assertThat(cypher).contains("$sourceEntityId");
+            assertThat(cypher).contains("$targetEntityId");
+        }
+
+        @Test
+        void mergeDependsOn_noJobs_returnsZero() {
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Job")).thenReturn(List.of());
+
+            SyncResult result = stepService.mergeDependsOnRelations(GRAPH_ID, SYNC_ID);
+
+            assertThat(result.getSyncType()).isEqualTo("DEPENDS_ON");
+            assertThat(result.getCreated()).isEqualTo(0);
+        }
+
+        @Test
+        void mergeDependsOn_jobWithDependency_verifiesCypherParams() {
+            setupNeo4jQueryChain(String.class, "new-rel-id");
+
+            GraphEntity job = GraphEntity.builder()
+                    .id("job-entity-2").sourceId("job-002").type("Job").graphId(GRAPH_ID)
+                    .properties(new HashMap<>(Map.of("depends_on_job_id", "job-001")))
+                    .build();
+            GraphEntity depJob = GraphEntity.builder()
+                    .id("job-entity-1").sourceId("job-001").type("Job").graphId(GRAPH_ID)
+                    .build();
+
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Job")).thenReturn(List.of(job, depJob));
+
+            SyncResult result = stepService.mergeDependsOnRelations(GRAPH_ID, SYNC_ID);
+
+            assertThat(result.getSyncType()).isEqualTo("DEPENDS_ON");
+            verify(neo4jClient).query(cypherCaptor.capture());
+            String cypher = cypherCaptor.getValue();
+            assertThat(cypher).contains("RELATED_TO");
+            assertThat(cypher).contains("$relationType");
+        }
+
+        @Test
+        void mergeImpacts_returnsPlaceholderResult() {
+            SyncResult result = stepService.mergeImpactsRelations(GRAPH_ID, SYNC_ID);
+
+            assertThat(result.getSyncType()).isEqualTo("IMPACTS");
+            assertThat(result.getCreated()).isEqualTo(0);
+            assertThat(result.isPlaceholder()).isTrue();
+            verifyNoInteractions(neo4jClient);
+            verifyNoInteractions(entityRepository);
+        }
+
+        @Test
+        void mergeSourcedFrom_noKnowledgeSets_returnsZero() {
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "KnowledgeSet")).thenReturn(List.of());
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Dataset")).thenReturn(List.of());
+
+            SyncResult result = stepService.mergeSourcedFromRelations(GRAPH_ID, SYNC_ID);
+
+            assertThat(result.getSyncType()).isEqualTo("SOURCED_FROM");
+            assertThat(result.getCreated()).isEqualTo(0);
+        }
+
+        @Test
+        void mergeSourcedFrom_ksWithSources_verifiesCypherAndLookup() {
+            setupNeo4jQueryChain(String.class, "new-rel-id");
+
+            GraphEntity ks = GraphEntity.builder()
+                    .id("ks-entity-1").type("KnowledgeSet").graphId(GRAPH_ID)
+                    .properties(new HashMap<>(Map.of("source_dataset_ids", List.of("ds-001"))))
+                    .build();
+            GraphEntity dataset = GraphEntity.builder()
+                    .id("ds-entity-1").sourceId("ds-001").type("Dataset").graphId(GRAPH_ID)
+                    .build();
+
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "KnowledgeSet")).thenReturn(List.of(ks));
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Dataset")).thenReturn(List.of(dataset));
+
+            SyncResult result = stepService.mergeSourcedFromRelations(GRAPH_ID, SYNC_ID);
+
+            assertThat(result.getSyncType()).isEqualTo("SOURCED_FROM");
+            verify(neo4jClient).query(cypherCaptor.capture());
+            assertThat(cypherCaptor.getValue()).contains("RELATED_TO");
+        }
+
+        @Test
+        void mergeSourcedFrom_listWithNullElements_filtersNulls() {
+            setupNeo4jQueryChain(String.class, "new-rel-id");
+
+            List<Object> listWithNulls = new ArrayList<>();
+            listWithNulls.add("ds-valid");
+            listWithNulls.add(null);
+            listWithNulls.add("");
+
+            GraphEntity ks = GraphEntity.builder()
+                    .id("ks-entity-2").type("KnowledgeSet").graphId(GRAPH_ID)
+                    .properties(new HashMap<>(Map.of("source_dataset_ids", listWithNulls)))
+                    .build();
+            GraphEntity dataset = GraphEntity.builder()
+                    .id("ds-entity-v").sourceId("ds-valid").type("Dataset").graphId(GRAPH_ID)
+                    .build();
+
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "KnowledgeSet")).thenReturn(List.of(ks));
+            when(entityRepository.findByGraphIdAndType(GRAPH_ID, "Dataset")).thenReturn(List.of(dataset));
+
+            SyncResult result = stepService.mergeSourcedFromRelations(GRAPH_ID, SYNC_ID);
+
+            // 只有 "ds-valid" 应被处理，null 和空字符串已过滤；验证只发起一次 mergeRelation
+            verify(neo4jClient, times(1)).query(anyString());
+        }
+    }
 }