diff --git a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/facade/DifyKnowledgeRuntimeClient.java b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/facade/DifyKnowledgeRuntimeClient.java index df4b960a..fa53f9b5 100644 --- a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/facade/DifyKnowledgeRuntimeClient.java +++ b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/facade/DifyKnowledgeRuntimeClient.java @@ -30,7 +30,7 @@ import java.util.UUID; /** * Dify Datasets API 的 Knowledge 运行时 adapter。 * - *

HTTP 范式对齐 {@code RealDifyMuseAiRuntimeClient} / {@code HttpRagFlowKnowledgeRuntimeClient}: + *

HTTP 范式对齐 {@code RealDifyMuseAiRuntimeClient}: * JDK {@link HttpClient} + 直连 {@link ProxySelector}(内网 Dify 必须绕过系统代理,否则 live 调用被污染成假 502)+ * Bearer(Datasets API key)+ 脱敏审计 + fail-closed 错误映射。契约以 S1 实测的 {@code P1rDifyDatasetsContractLiveIT} 为准。

* diff --git a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/facade/DifyKnowledgeRuntimeClientConfiguration.java b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/facade/DifyKnowledgeRuntimeClientConfiguration.java index 33c8ccf6..71c7fe65 100644 --- a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/facade/DifyKnowledgeRuntimeClientConfiguration.java +++ b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/facade/DifyKnowledgeRuntimeClientConfiguration.java @@ -10,16 +10,17 @@ import org.springframework.util.StringUtils; import java.time.Duration; /** - * Dify Datasets runtime client 的 Spring 装配配置。 + * Dify Datasets runtime client 的 Spring 装配配置(S6d 起为 Knowledge 缺省运行时)。 * - *

与 {@link RagFlowKnowledgeRuntimeClientConfiguration} 共存(RAGFlow 装配 S6d 才摘除)。为确保单人只装一个 - * {@link KnowledgeRuntimeClient},两个装配类按 provider 选择器 {@code muse.knowledge.runtime-provider} 互斥: - * 本类仅在 {@code =dify} 时参与,RAGFlow 装配在 {@code =ragflow} 或缺省(matchIfMissing)时参与。这样 - * {@code @ConditionalOnMissingBean} 的求值顺序不再影响结果,避免组件扫描顺序带来的非确定性冲突。

+ *

S6d 摘除 RAGFlow 运行时后,本类是唯一的具体 {@link KnowledgeRuntimeClient} 装配来源,按 provider 选择器 + * {@code muse.knowledge.runtime-provider} 生效:取 {@code dify} 或缺省({@code matchIfMissing=true},即未显式配置时 + * 默认 Dify)时装 Dify 主 bean;显式配成其它值(如遗留的 {@code ragflow},其运行时已删)时主 bean 让位,由下方 + * {@code @ConditionalOnMissingBean} 兜底装通用的 {@link UnavailableKnowledgeRuntimeClient}。两个 @Bean 同处一类、 + * 兜底声明在主 bean 之后,{@code @ConditionalOnMissingBean} 按声明顺序求值,确保任何 provider 取值下都有且仅有 + * 一个 bean,绝不因无 bean 致启动失败。

* - *

装配范式对齐 RAGFlow(无 {@code .enable} 开关):读 {@code muse.knowledge.dify.*};缺 base-url 或 - * dataset-api-key 一律装 {@link UnavailableKnowledgeRuntimeClient}(fail-closed,即"选了 Dify 但配错=拒绝", - * 绝不静默回退 RAGFlow)。

+ *

读 {@code muse.knowledge.dify.*};缺 base-url 或 dataset-api-key 一律装 {@link UnavailableKnowledgeRuntimeClient} + * (fail-closed,即"选了 Dify 但配错=拒绝",绝不静默降级)。

*/ @Configuration(proxyBeanMethods = false) public class DifyKnowledgeRuntimeClientConfiguration { @@ -30,7 +31,7 @@ public class DifyKnowledgeRuntimeClientConfiguration { @Bean @ConditionalOnMissingBean(KnowledgeRuntimeClient.class) - @ConditionalOnProperty(name = "muse.knowledge.runtime-provider", havingValue = "dify") + @ConditionalOnProperty(name = "muse.knowledge.runtime-provider", havingValue = "dify", matchIfMissing = true) public KnowledgeRuntimeClient difyKnowledgeRuntimeClient(Environment environment) { String baseUrl = environment.getProperty(PROPERTY_PREFIX + "base-url"); String apiKey = environment.getProperty(PROPERTY_PREFIX + "dataset-api-key"); @@ -43,6 +44,17 @@ public class DifyKnowledgeRuntimeClientConfiguration { return new DifyKnowledgeRuntimeClient(baseUrl, apiKey, Duration.ofSeconds(timeoutSeconds), retryBudget); } + /** + * 通用不可用兜底 bean:仅当上面 Dify 主 bean 未装成(provider 显式配成非 dify 值,如遗留 {@code ragflow})时生效。 + * 声明在主 bean 之后,{@code @ConditionalOnMissingBean} 按同类内 @Bean 声明顺序求值,保证 provider 任意取值下 + * 始终恰好一个 {@link KnowledgeRuntimeClient},既不因无 bean 启动失败,也不静默降级(桩返回 CONFIG_MISSING)。 + */ + @Bean + @ConditionalOnMissingBean(KnowledgeRuntimeClient.class) + public KnowledgeRuntimeClient unavailableKnowledgeRuntimeClient() { + return new UnavailableKnowledgeRuntimeClient(); + } + private int positiveInt(Environment environment, String propertyName, int defaultValue) { Integer value = environment.getProperty(PROPERTY_PREFIX + propertyName, Integer.class, defaultValue); return value == null || value <= 0 ? defaultValue : value; diff --git a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/facade/HttpRagFlowKnowledgeRuntimeClient.java b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/facade/HttpRagFlowKnowledgeRuntimeClient.java deleted file mode 100644 index ae9d6248..00000000 --- a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/facade/HttpRagFlowKnowledgeRuntimeClient.java +++ /dev/null @@ -1,582 +0,0 @@ -package cn.iocoder.muse.module.knowledge.application.muse.facade; - -import cn.iocoder.muse.framework.common.util.json.JsonUtils; -import com.fasterxml.jackson.databind.JsonNode; -import org.springframework.util.StringUtils; - -import java.io.ByteArrayOutputStream; -import java.io.IOException; -import java.net.ProxySelector; -import java.net.URLEncoder; -import java.net.URI; -import java.net.http.HttpClient; -import java.net.http.HttpHeaders; -import java.net.http.HttpRequest; -import java.net.http.HttpResponse; -import java.nio.charset.StandardCharsets; -import java.security.MessageDigest; -import java.security.NoSuchAlgorithmException; -import java.time.Duration; -import java.time.Instant; -import java.util.HexFormat; -import java.util.ArrayList; -import java.util.LinkedHashMap; -import java.util.LinkedHashSet; -import java.util.List; -import java.util.Locale; -import java.util.Map; -import java.util.Set; -import java.util.UUID; - -/** - * RAGFlow HTTP runtime client。 - */ -public class HttpRagFlowKnowledgeRuntimeClient implements KnowledgeRuntimeClient { - - private final String baseUrl; - private final String apiKey; - private final Duration timeout; - private final int retryBudget; - private final boolean graphRagAttributionReady; - private final HttpClient httpClient; - private static final ProxySelector DIRECT_PROXY_SELECTOR = ProxySelector.of(null); - - public HttpRagFlowKnowledgeRuntimeClient(String baseUrl, String apiKey, Duration timeout, - int retryBudget, boolean graphRagAttributionReady) { - this.baseUrl = normalizeBaseUrl(baseUrl); - this.apiKey = apiKey; - this.timeout = timeout == null ? Duration.ofSeconds(5) : timeout; - this.retryBudget = Math.max(retryBudget, 0); - this.graphRagAttributionReady = graphRagAttributionReady; - this.httpClient = HttpClient.newBuilder() - .connectTimeout(this.timeout) - // RAGFlow 部署在 tailnet 内网，显式直连可避免系统代理污染 live IT 与运行时调用。 - .proxy(DIRECT_PROXY_SELECTOR) - .build(); - } - - @Override - public RuntimeResult createDataset(CreateDatasetCommand command) { - Map body = new LinkedHashMap<>(); - body.put("name", command.name()); - // RAGFlow 当前 create dataset 接口只接受 name；config 由 updateDatasetConfig 单独提交，避免创建阶段被外部 API 拒绝。 - return executeJson(Operation.CREATE_DATASET, command.correlationId(), command.requestHash(), command.attempt(), - "POST", "/api/v1/datasets", body, "id", null); - } - - @Override - public RuntimeResult uploadDocuments(UploadDocumentsCommand command) { - MultipartPayload multipartPayload = buildMultipartPayload(command.documents()); - if (multipartPayload == null) { - return failure(Operation.UPLOAD_DOCUMENTS, command.correlationId(), command.requestHash(), command.attempt(), - null, FailureClass.FILE_REJECTED, 0L, "document content is not materialized"); - } - return execute(Operation.UPLOAD_DOCUMENTS, command.correlationId(), command.requestHash(), command.attempt(), - "POST", "/api/v1/datasets/" + command.ragflowDatasetId() + "/documents", - multipartPayload.body(), multipartPayload.contentType(), "id", null); - } - - @Override - public RuntimeResult startParseDocuments(StartParseDocumentsCommand command) { - Map body = Map.of("document_ids", command.ragflowDocumentIds()); - RuntimeResult rawResult = executeJson(Operation.START_PARSE_DOCUMENTS, command.correlationId(), command.requestHash(), - command.attempt(), "POST", "/api/v1/datasets/" + command.ragflowDatasetId() + "/chunks", - body, null, command.processingTaskId()); - if (rawResult.status() == Status.FAILED) { - return rawResult; - } - // RAGFlow parse 没有稳定外部 task id；Muse 只向上返回内部 processingTaskId/correlationId。 - return new RuntimeResult(null, Operation.START_PARSE_DOCUMENTS, command.correlationId(), command.requestHash(), - command.attempt(), rawResult.durationMillis(), Status.ACCEPTED, null, command.processingTaskId(), - "{\"status\":\"accepted\",\"processingTaskId\":\"" + escape(command.processingTaskId()) + "\"}", - List.of(), Map.of("processingTaskId", command.processingTaskId())); - } - - @Override - public RuntimeResult pollDocumentStatuses(PollDocumentStatusesCommand command) { - String path = "/api/v1/datasets/" + command.ragflowDatasetId() + "/documents"; - if (command.ragflowDocumentIds() != null && !command.ragflowDocumentIds().isEmpty()) { - if (command.ragflowDocumentIds().size() > 1) { - return singleDocumentPollingValidationFailure(command); - } - // RAGFlow list documents 的真实查询参数是 id=，不是批量 document_ids。 - // 当前 Muse 上传链路按单文档 smoke 验收；多文档批量轮询后续应在调用方拆分，避免生成 RAGFlow 不支持的查询。 - path += "?id=" + encode(command.ragflowDocumentIds().getFirst()); - } - RuntimeResult rawResult = execute(Operation.POLL_DOCUMENT_STATUSES, command.correlationId(), - command.requestHash(), command.attempt(), "GET", path, new byte[0], null, null, command.processingTaskId()); - if (rawResult.status() == Status.FAILED) { - return rawResult; - } - List statuses = normalizeDocumentStatuses(rawResult.summary().get("responseBody"), - command.ragflowDocumentIds()); - return new RuntimeResult(null, Operation.POLL_DOCUMENT_STATUSES, command.correlationId(), command.requestHash(), - command.attempt(), rawResult.durationMillis(), Status.SUCCEEDED, null, command.processingTaskId(), - sanitizeSummary(Map.of("documents", statuses)), statuses, Map.of("documents", statuses)); - } - - private RuntimeResult singleDocumentPollingValidationFailure(PollDocumentStatusesCommand command) { - Map summary = new LinkedHashMap<>(); - summary.put("status", Status.FAILED.name().toLowerCase(Locale.ROOT)); - summary.put("failureClass", FailureClass.VALIDATION_ERROR.name()); - summary.put("operation", Operation.POLL_DOCUMENT_STATUSES.wireName()); - summary.put("reason", "only support single document status polling"); - return new RuntimeResult(null, Operation.POLL_DOCUMENT_STATUSES, command.correlationId(), command.requestHash(), - command.attempt(), 0L, Status.FAILED, FailureClass.VALIDATION_ERROR, command.processingTaskId(), - sanitizeSummary(summary), List.of(), summary); - } - - @Override - public RuntimeResult retrieveChunks(RetrieveChunksCommand command) { - Map body = new LinkedHashMap<>(); - body.put("dataset_ids", command.ragflowDatasetIds()); - // RAGFlow 要求 document_ids 为 list;为 null(不限定具体文档、按授权 dataset 全量检索)时不放入 body, - // 否则 RAGFlow 拒绝 code 102 "`documents` should be a list" → 检索 fail-closed,P-A 上下文检索永不命中。 - if (command.ragflowDocumentIds() != null) { - body.put("document_ids", command.ragflowDocumentIds()); - } - body.put("question", command.question()); - body.put("top_k", command.topK()); - body.put("similarity_threshold", command.threshold()); - // RAGFlow 官方 HTTP 契约字段名是 metadata_condition；旧字段 metadata_filter 会被静默忽略。 - if (command.metadataCondition() != null) { - body.put("metadata_condition", command.metadataCondition()); - } - return executeJson(Operation.RETRIEVE_CHUNKS, command.correlationId(), command.requestHash(), command.attempt(), - "POST", "/api/v1/retrieval", body, null, null); - } - - protected HttpResponsePayload send(HttpRequestPayload request) throws IOException, InterruptedException { - HttpRequest.Builder builder = HttpRequest.newBuilder() - .uri(URI.create(baseUrl + request.path())) - .timeout(timeout) - .header("Accept", "application/json"); - if (StringUtils.hasText(apiKey)) { - builder.header("Authorization", "Bearer " + apiKey); - } - if ("GET".equals(request.method())) { - builder.GET(); - } else { - builder.header("Content-Type", request.contentType() == null ? "application/json" : request.contentType()) - .method(request.method(), HttpRequest.BodyPublishers.ofByteArray(request.body())); - } - HttpResponse response = httpClient.send(builder.build(), HttpResponse.BodyHandlers.ofString()); - return new HttpResponsePayload(response.statusCode(), response.body(), response.headers()); - } - - private RuntimeResult executeJson(Operation operation, String correlationId, String requestHash, int attempt, - String method, String path, Object body, String externalIdField, - String processingTaskId) { - byte[] requestBody = "GET".equals(method) ? new byte[0] - : JsonUtils.toJsonString(body == null ? Map.of() : body).getBytes(StandardCharsets.UTF_8); - return execute(operation, correlationId, requestHash, attempt, method, path, requestBody, - "application/json", externalIdField, processingTaskId); - } - - private RuntimeResult execute(Operation operation, String correlationId, String requestHash, int attempt, - String method, String path, byte[] requestBody, String contentType, - String externalIdField, String processingTaskId) { - Instant startedAt = Instant.now(); - FailureClass configurationFailure = validateConfiguration(); - if (configurationFailure != null) { - return failure(operation, correlationId, requestHash, attempt, processingTaskId, configurationFailure, - elapsedMillis(startedAt), configurationFailure.name()); - } - int maxAttempts = Math.max(1, retryBudget + 1); - RuntimeResult lastFailure = null; - for (int currentAttempt = 1; currentAttempt <= maxAttempts; currentAttempt++) { - try { - HttpResponsePayload response = send(new HttpRequestPayload(method, path, - requestBody == null ? new byte[0] : requestBody, contentType)); - long duration = elapsedMillis(startedAt); - FailureClass httpFailure = classifyHttpFailure(response.statusCode()); - if (httpFailure != null) { - lastFailure = failure(operation, correlationId, requestHash, attempt, processingTaskId, - httpFailure, duration, response.body()); - if (!isRetryable(httpFailure) || currentAttempt == maxAttempts) { - return lastFailure; - } - continue; - } - return success(operation, correlationId, requestHash, attempt, processingTaskId, externalIdField, - response.body(), duration); - } catch (java.net.http.HttpTimeoutException ex) { - lastFailure = failure(operation, correlationId, requestHash, attempt, processingTaskId, - FailureClass.TIMEOUT, elapsedMillis(startedAt), ex.getMessage()); - } catch (IOException ex) { - lastFailure = failure(operation, correlationId, requestHash, attempt, processingTaskId, - FailureClass.RAGFLOW_UNAVAILABLE, elapsedMillis(startedAt), ex.getMessage()); - } catch (InterruptedException ex) { - Thread.currentThread().interrupt(); - return failure(operation, correlationId, requestHash, attempt, processingTaskId, - FailureClass.RAGFLOW_UNAVAILABLE, elapsedMillis(startedAt), ex.getMessage()); - } catch (RuntimeException ex) { - return failure(operation, correlationId, requestHash, attempt, processingTaskId, - FailureClass.UNKNOWN_FAILURE, elapsedMillis(startedAt), ex.getMessage()); - } - if (currentAttempt == maxAttempts) { - return lastFailure; - } - } - return lastFailure == null - ? failure(operation, correlationId, requestHash, attempt, processingTaskId, - FailureClass.UNKNOWN_FAILURE, elapsedMillis(startedAt), "unknown failure") - : lastFailure; - } - - private RuntimeResult success(Operation operation, String correlationId, String requestHash, int attempt, - String processingTaskId, String externalIdField, String responseBody, - long durationMillis) { - JsonNode bodyTree = parseBody(responseBody); - if (bodyTree == null) { - return failure(operation, correlationId, requestHash, attempt, processingTaskId, - FailureClass.RESPONSE_SCHEMA_CHANGED, durationMillis, responseBody); - } - FailureClass envelopeFailure = classifyEnvelopeFailure(bodyTree); - if (envelopeFailure != null) { - return failure(operation, correlationId, requestHash, attempt, processingTaskId, - envelopeFailure, durationMillis, firstText(bodyTree, "message", "msg", "error")); - } - JsonNode data = bodyTree.has("data") ? bodyTree.path("data") : bodyTree; - String externalId = externalIdField == null ? null : findText(data, externalIdField); - if (externalIdField != null && !StringUtils.hasText(externalId)) { - return failure(operation, correlationId, requestHash, attempt, processingTaskId, - FailureClass.RESPONSE_SCHEMA_CHANGED, durationMillis, "required externalId is missing"); - } - Map summary = buildMetadataSummary(operation, bodyTree, externalId, null); - // 后续状态归一仍需要原始响应树，审计摘要只输出 buildMetadataSummary 的保守元数据。 - summary.put("responseBody", bodyTree); - if (externalId != null) { - summary.put("externalId", externalId); - } - return new RuntimeResult(externalId, operation, correlationId, requestHash, attempt, durationMillis, - Status.SUCCEEDED, null, processingTaskId, - sanitizeSummary(buildMetadataSummary(operation, bodyTree, externalId, null)), List.of(), summary); - } - - private RuntimeResult failure(Operation operation, String correlationId, String requestHash, int attempt, - String processingTaskId, FailureClass failureClass, long durationMillis, - String message) { - Map summary = new LinkedHashMap<>(); - summary.put("status", Status.FAILED.name().toLowerCase(Locale.ROOT)); - summary.put("failureClass", failureClass.name()); - summary.put("operation", operation.wireName()); - if (message != null && !message.isBlank()) { - summary.put("providerMessageSummary", safeSummaryNode(message)); - } - return new RuntimeResult(null, operation, correlationId, requestHash, attempt, durationMillis, - Status.FAILED, failureClass, processingTaskId, sanitizeSummary(summary), List.of(), summary); - } - - private JsonNode safeSummaryNode(String message) { - Map summary = new LinkedHashMap<>(); - summary.put("type", "provider_body"); - summary.put("length", message == null ? 0 : message.length()); - summary.put("sha256", sha256(message == null ? "" : message)); - summary.put("redacted", true); - summary.put("json", looksLikeJson(message)); - return JsonUtils.getObjectMapper().valueToTree(summary); - } - - private FailureClass validateConfiguration() { - if (!StringUtils.hasText(baseUrl)) { - return FailureClass.CONFIG_MISSING; - } - if (!StringUtils.hasText(apiKey)) { - return FailureClass.AUTH_MISSING; - } - return null; - } - - private FailureClass classifyHttpFailure(int statusCode) { - if (statusCode >= 200 && statusCode < 300) { - return null; - } - return switch (statusCode) { - case 400 -> FailureClass.VALIDATION_ERROR; - case 401, 403 -> FailureClass.AUTH_FAILED; - case 404 -> FailureClass.TASK_NOT_FOUND; - case 409 -> FailureClass.CONFLICT; - case 413, 415 -> FailureClass.FILE_REJECTED; - case 429 -> FailureClass.RATE_LIMITED; - default -> statusCode >= 500 ? FailureClass.RAGFLOW_UNAVAILABLE : FailureClass.UNKNOWN_FAILURE; - }; - } - - private boolean isRetryable(FailureClass failureClass) { - return failureClass == FailureClass.RAGFLOW_UNAVAILABLE - || failureClass == FailureClass.TIMEOUT - || failureClass == FailureClass.RATE_LIMITED; - } - - private FailureClass classifyEnvelopeFailure(JsonNode bodyTree) { - if (bodyTree == null || bodyTree.isMissingNode() || bodyTree.isNull()) { - return FailureClass.RESPONSE_SCHEMA_CHANGED; - } - if (bodyTree.has("code") && bodyTree.path("code").canConvertToInt() && bodyTree.path("code").asInt() != 0) { - return classifyRagflowCode(bodyTree.path("code").asInt(), firstText(bodyTree, "message", "msg", "error")); - } - if (bodyTree.has("error") || bodyTree.has("errors")) { - return classifyRagflowMessage(firstText(bodyTree, "error", "message", "msg")); - } - String status = firstText(bodyTree, "status"); - if (status != null && Set.of("failed", "error", "unauthorized", "forbidden").contains(status.toLowerCase(Locale.ROOT))) { - return classifyRagflowMessage(firstText(bodyTree, "message", "msg", "error", "status")); - } - return null; - } - - private FailureClass classifyRagflowCode(int code, String message) { - return switch (code) { - case 401, 403, 1001, 1002 -> FailureClass.AUTH_FAILED; - case 400, 422 -> FailureClass.VALIDATION_ERROR; - case 404 -> FailureClass.TASK_NOT_FOUND; - case 409 -> FailureClass.CONFLICT; - case 429 -> FailureClass.RATE_LIMITED; - default -> classifyRagflowMessage(message); - }; - } - - private FailureClass classifyRagflowMessage(String message) { - String normalized = message == null ? "" : message.toLowerCase(Locale.ROOT); - if (normalized.contains("auth") || normalized.contains("unauthor") || normalized.contains("forbidden")) { - return FailureClass.AUTH_FAILED; - } - if (normalized.contains("validation") || normalized.contains("invalid") || normalized.contains("param")) { - return FailureClass.VALIDATION_ERROR; - } - if (normalized.contains("index") && normalized.contains("ready")) { - return FailureClass.INDEX_NOT_READY; - } - if (normalized.contains("chunk") && (normalized.contains("empty") || normalized.contains("no "))) { - return FailureClass.NO_CHUNK; - } - if (normalized.contains("graph") && normalized.contains("ready")) { - return FailureClass.GRAPH_NOT_READY; - } - if (normalized.contains("already") && normalized.contains("running")) { - return FailureClass.TASK_ALREADY_RUNNING; - } - if (normalized.contains("not found")) { - return FailureClass.TASK_NOT_FOUND; - } - return FailureClass.RESPONSE_SCHEMA_CHANGED; - } - - private List normalizeDocumentStatuses(Object responseBody, List requestedDocumentIds) { - JsonNode root = responseBody instanceof JsonNode jsonNode ? jsonNode : JsonUtils.getObjectMapper().valueToTree(responseBody); - JsonNode documents = root.path("data"); - if (documents.isObject() && documents.has("docs")) { - documents = documents.path("docs"); - } - if (documents.isObject() && documents.has("documents")) { - documents = documents.path("documents"); - } - List statuses = new ArrayList<>(); - if (!documents.isArray()) { - return statuses; - } - Set requestedIds = requestedDocumentIds == null ? Set.of() : new LinkedHashSet<>(requestedDocumentIds); - for (JsonNode document : documents) { - String documentId = firstText(document, "id", "document_id"); - if (!requestedIds.isEmpty() && !requestedIds.contains(documentId)) { - continue; - } - String run = document.path("run").asText(""); - Integer progress = document.has("progress") && document.path("progress").canConvertToInt() - ? document.path("progress").asInt() : null; - statuses.add(new DocumentStatus( - documentId, - normalizeMuseStatus(run, progress), - run, - progress, - firstText(document, "progress_msg", "progressMsg") - )); - } - return statuses; - } - - private String normalizeMuseStatus(String run, Integer progress) { - String normalizedRun = run == null ? "" : run.toLowerCase(Locale.ROOT); - if (Set.of("done", "completed", "success").contains(normalizedRun) || Integer.valueOf(100).equals(progress)) { - return "searchable"; - } - if (Set.of("running", "parsing", "indexing", "queued").contains(normalizedRun)) { - return "processing"; - } - if (Set.of("failed", "error", "cancelled").contains(normalizedRun)) { - return "failed"; - } - return "pending_process"; - } - - private JsonNode parseBody(String responseBody) { - try { - return JsonUtils.parseTree(responseBody == null || responseBody.isBlank() ? "{}" : responseBody); - } catch (RuntimeException ex) { - return null; - } - } - - private String findText(JsonNode node, String fieldName) { - if (node == null || node.isMissingNode() || node.isNull()) { - return null; - } - if (node.has(fieldName) && node.path(fieldName).isTextual()) { - return node.path(fieldName).asText(); - } - if (node.isArray() && !node.isEmpty()) { - return findText(node.get(0), fieldName); - } - return null; - } - - private String firstText(JsonNode node, String... fieldNames) { - for (String fieldName : fieldNames) { - if (node.has(fieldName) && node.path(fieldName).isTextual()) { - return node.path(fieldName).asText(); - } - } - return null; - } - - private MultipartPayload buildMultipartPayload(List documents) { - if (documents == null || documents.isEmpty()) { - return null; - } - String boundary = "----MuseRagFlow" + UUID.randomUUID().toString().replace("-", ""); - try { - ByteArrayOutputStream body = new ByteArrayOutputStream(); - for (DocumentUpload document : documents) { - if (document == null || document.content() == null || document.content().length == 0 - || !StringUtils.hasText(document.fileName())) { - return null; - } - String contentType = StringUtils.hasText(document.contentType()) ? document.contentType() : "application/octet-stream"; - writeAscii(body, "--" + boundary + "\r\n"); - writeAscii(body, "Content-Disposition: form-data; name=\"file\"; filename=\"" - + escapeMultipart(document.fileName()) + "\"\r\n"); - writeAscii(body, "Content-Type: " + contentType + "\r\n\r\n"); - // Task 3 只承接小型已材料化资源；Task 5 再接对象存储/流式文件边界。 - body.write(document.content()); - writeAscii(body, "\r\n"); - } - writeAscii(body, "--" + boundary + "--\r\n"); - return new MultipartPayload(body.toByteArray(), "multipart/form-data; boundary=" + boundary); - } catch (IOException ex) { - return null; - } - } - - private Map buildMetadataSummary(Operation operation, JsonNode bodyTree, String externalId, - FailureClass failureClass) { - Map summary = new LinkedHashMap<>(); - summary.put("operation", operation.wireName()); - summary.put("status", failureClass == null ? Status.SUCCEEDED.name().toLowerCase(Locale.ROOT) - : Status.FAILED.name().toLowerCase(Locale.ROOT)); - if (bodyTree != null && bodyTree.has("code")) { - summary.put("code", bodyTree.path("code")); - } - if (StringUtils.hasText(externalId)) { - summary.put("externalId", externalId); - } - if (failureClass != null) { - summary.put("failureClass", failureClass.name()); - } - JsonNode data = bodyTree == null ? null : bodyTree.path("data"); - if (data != null && data.isArray()) { - summary.put("count", data.size()); - } else if (data != null && data.isObject()) { - putCountIfPresent(summary, data, "docs"); - putCountIfPresent(summary, data, "documents"); - putCountIfPresent(summary, data, "chunks"); - putCountIfPresent(summary, data, "nodes"); - putCountIfPresent(summary, data, "edges"); - JsonNode graph = data.path("graph"); - if (graph.isObject()) { - putCountIfPresent(summary, graph, "nodes"); - putCountIfPresent(summary, graph, "edges"); - } - if (data.has("mind_map")) { - summary.put("mindMapPresent", !data.path("mind_map").isNull()); - if (data.path("mind_map").isArray()) { - summary.put("mindMapCount", data.path("mind_map").size()); - } - } else if (data.has("mindMap")) { - summary.put("mindMapPresent", !data.path("mindMap").isNull()); - if (data.path("mindMap").isArray()) { - summary.put("mindMapCount", data.path("mindMap").size()); - } - } - if (data.has("progress")) { - summary.put("progress", data.path("progress")); - } - } - return summary; - } - - private void putCountIfPresent(Map summary, JsonNode data, String fieldName) { - if (data.has(fieldName) && data.path(fieldName).isArray()) { - summary.put(fieldName + "Count", data.path(fieldName).size()); - } - } - - private String sanitizeSummary(Object value) { - return RagFlowKnowledgeRedactor.sanitizeObject(value); - } - - private long elapsedMillis(Instant startedAt) { - return Math.max(0L, Duration.between(startedAt, Instant.now()).toMillis()); - } - - private String escape(String value) { - return value == null ? "" : value.replace("\\", "\\\\").replace("\"", "\\\""); - } - - private String escapeMultipart(String value) { - return escape(value).replace("\r", "").replace("\n", ""); - } - - private String encode(String value) { - return URLEncoder.encode(value == null ? "" : value, StandardCharsets.UTF_8); - } - - private boolean looksLikeJson(String value) { - if (value == null) { - return false; - } - String trimmed = value.trim(); - return (trimmed.startsWith("{") && trimmed.endsWith("}")) - || (trimmed.startsWith("[") && trimmed.endsWith("]")); - } - - private String sha256(String payload) { - try { - MessageDigest digest = MessageDigest.getInstance("SHA-256"); - return HexFormat.of().formatHex(digest.digest(payload.getBytes(StandardCharsets.UTF_8))); - } catch (NoSuchAlgorithmException e) { - throw new IllegalStateException("JDK 缺少 SHA-256 摘要算法", e); - } - } - - private void writeAscii(ByteArrayOutputStream outputStream, String text) throws IOException { - outputStream.write(text.getBytes(StandardCharsets.US_ASCII)); - } - - private static String normalizeBaseUrl(String rawBaseUrl) { - if (rawBaseUrl == null || rawBaseUrl.isBlank()) { - return null; - } - return rawBaseUrl.endsWith("/") ? rawBaseUrl.substring(0, rawBaseUrl.length() - 1) : rawBaseUrl; - } - - protected record HttpRequestPayload(String method, String path, byte[] body, String contentType) { - } - - protected record HttpResponsePayload(int statusCode, String body, HttpHeaders headers) { - } - - private record MultipartPayload(byte[] body, String contentType) { - } - -} diff --git a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/facade/RagFlowKnowledgeRuntimeClientConfiguration.java b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/facade/RagFlowKnowledgeRuntimeClientConfiguration.java deleted file mode 100644 index 093ff405..00000000 --- a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/facade/RagFlowKnowledgeRuntimeClientConfiguration.java +++ /dev/null @@ -1,54 +0,0 @@ -package cn.iocoder.muse.module.knowledge.application.muse.facade; - -import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.core.env.Environment; -import org.springframework.util.StringUtils; - -import java.time.Duration; - -/** - * RAGFlow runtime client 的 Spring 装配配置。 - * - *

与 {@link DifyKnowledgeRuntimeClientConfiguration} 按 provider 选择器 {@code muse.knowledge.runtime-provider} - * 互斥:仅在 {@code =ragflow} 或缺省(matchIfMissing=true,即保持既有部署零变更)时参与,{@code =dify} 时让位给 Dify 装配。 - * S6d 摘除 RAGFlow 时连同本类一并删除。

- */ -@Configuration(proxyBeanMethods = false) -public class RagFlowKnowledgeRuntimeClientConfiguration { - - private static final String PROPERTY_PREFIX = "muse.knowledge.ragflow."; - private static final int DEFAULT_TIMEOUT_SECONDS = 5; - private static final int DEFAULT_RETRY_BUDGET = 0; - - @Bean - @ConditionalOnMissingBean(KnowledgeRuntimeClient.class) - @ConditionalOnProperty(name = "muse.knowledge.runtime-provider", havingValue = "ragflow", matchIfMissing = true) - public KnowledgeRuntimeClient ragFlowKnowledgeRuntimeClient(Environment environment) { - String baseUrl = environment.getProperty(PROPERTY_PREFIX + "base-url"); - String apiKey = environment.getProperty(PROPERTY_PREFIX + "api-key"); - if (!StringUtils.hasText(baseUrl) || !StringUtils.hasText(apiKey)) { - return new UnavailableRagFlowKnowledgeRuntimeClient(); - } - - int timeoutSeconds = positiveInt(environment, "timeout-seconds", DEFAULT_TIMEOUT_SECONDS); - int retryBudget = nonNegativeInt(environment, "retry-budget", DEFAULT_RETRY_BUDGET); - boolean graphRagAttributionReady = Boolean.TRUE.equals(environment.getProperty( - PROPERTY_PREFIX + "graphrag-attribution-ready", Boolean.class, false)); - return new HttpRagFlowKnowledgeRuntimeClient(baseUrl, apiKey, Duration.ofSeconds(timeoutSeconds), - retryBudget, graphRagAttributionReady); - } - - private int positiveInt(Environment environment, String propertyName, int defaultValue) { - Integer value = environment.getProperty(PROPERTY_PREFIX + propertyName, Integer.class, defaultValue); - return value == null || value <= 0 ? defaultValue : value; - } - - private int nonNegativeInt(Environment environment, String propertyName, int defaultValue) { - Integer value = environment.getProperty(PROPERTY_PREFIX + propertyName, Integer.class, defaultValue); - return value == null || value < 0 ? defaultValue : value; - } - -} diff --git a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/facade/UnavailableRagFlowKnowledgeRuntimeClient.java b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/facade/UnavailableRagFlowKnowledgeRuntimeClient.java deleted file mode 100644 index d483a3fe..00000000 --- a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/facade/UnavailableRagFlowKnowledgeRuntimeClient.java +++ /dev/null @@ -1,46 +0,0 @@ -package cn.iocoder.muse.module.knowledge.application.muse.facade; - -import java.util.List; -import java.util.Map; - -/** - * RAGFlow 未配置时的 fail closed client。 - */ -public class UnavailableRagFlowKnowledgeRuntimeClient implements KnowledgeRuntimeClient { - - @Override - public RuntimeResult createDataset(CreateDatasetCommand command) { - return unavailable(Operation.CREATE_DATASET, command.correlationId(), command.requestHash(), command.attempt(), null); - } - - @Override - public RuntimeResult uploadDocuments(UploadDocumentsCommand command) { - return unavailable(Operation.UPLOAD_DOCUMENTS, command.correlationId(), command.requestHash(), command.attempt(), null); - } - - @Override - public RuntimeResult startParseDocuments(StartParseDocumentsCommand command) { - return unavailable(Operation.START_PARSE_DOCUMENTS, command.correlationId(), command.requestHash(), - command.attempt(), command.processingTaskId()); - } - - @Override - public RuntimeResult pollDocumentStatuses(PollDocumentStatusesCommand command) { - return unavailable(Operation.POLL_DOCUMENT_STATUSES, command.correlationId(), command.requestHash(), - command.attempt(), command.processingTaskId()); - } - - @Override - public RuntimeResult retrieveChunks(RetrieveChunksCommand command) { - return unavailable(Operation.RETRIEVE_CHUNKS, command.correlationId(), command.requestHash(), command.attempt(), null); - } - - private RuntimeResult unavailable(Operation operation, String correlationId, String requestHash, - int attempt, String processingTaskId) { - return new RuntimeResult(null, operation, correlationId, requestHash, attempt, 0L, - Status.FAILED, FailureClass.CONFIG_MISSING, processingTaskId, - "{\"status\":\"failed\",\"failureClass\":\"CONFIG_MISSING\"}", - List.of(), Map.of("failureClass", FailureClass.CONFIG_MISSING.name())); - } - -} diff --git a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/test/java/cn/iocoder/muse/module/knowledge/application/muse/facade/DifyKnowledgeRuntimeClientConfigurationTest.java b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/test/java/cn/iocoder/muse/module/knowledge/application/muse/facade/DifyKnowledgeRuntimeClientConfigurationTest.java index 53f2fe84..a5c02ef1 100644 --- a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/test/java/cn/iocoder/muse/module/knowledge/application/muse/facade/DifyKnowledgeRuntimeClientConfigurationTest.java +++ b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/test/java/cn/iocoder/muse/module/knowledge/application/muse/facade/DifyKnowledgeRuntimeClientConfigurationTest.java @@ -8,17 +8,17 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertInstanceOf; /** - * Dify Knowledge 运行时 client 装配测试。 + * Dify Knowledge 运行时 client 装配测试(S6d 起 Dify 为缺省运行时)。 * - *

两层验证:①装配方法本身(配 dify.* 全→Dify、缺→Unavailable);②与 RAGFlow 装配共存时按 - * {@code muse.knowledge.runtime-provider} 选择器确定性互斥——任何组合下只装一个 {@link KnowledgeRuntimeClient}, - * 不依赖组件扫描/Bean 求值顺序。

+ *

两层验证:①装配方法本身(配 dify.* 全→Dify、缺→Unavailable);②provider 选择器 + * {@code muse.knowledge.runtime-provider} 语义——取 {@code dify} 或缺省装 Dify,显式配成其它值(如遗留 + * {@code ragflow},运行时已删)由兜底装通用 {@link UnavailableKnowledgeRuntimeClient};任何取值下都有且仅有 + * 一个 {@link KnowledgeRuntimeClient}。

*/ class DifyKnowledgeRuntimeClientConfigurationTest { private final ApplicationContextRunner contextRunner = new ApplicationContextRunner() - .withUserConfiguration(DifyKnowledgeRuntimeClientConfiguration.class, - RagFlowKnowledgeRuntimeClientConfiguration.class); + .withUserConfiguration(DifyKnowledgeRuntimeClientConfiguration.class); // ==================== 装配方法本身 ==================== @@ -58,7 +58,7 @@ class DifyKnowledgeRuntimeClientConfigurationTest { assertInstanceOf(UnavailableKnowledgeRuntimeClient.class, client); } - // ==================== 与 RAGFlow 装配共存:确定性互斥 ==================== + // ==================== provider 选择器:缺省 Dify + 非 dify 值兜底 Unavailable ==================== @Test void should_installDifyOnlyWhenProviderIsDify() { @@ -85,37 +85,36 @@ class DifyKnowledgeRuntimeClientConfigurationTest { } @Test - void should_installRagFlowWhenProviderIsRagflow() { - contextRunner.withPropertyValues( - "muse.knowledge.runtime-provider=ragflow", - "muse.knowledge.ragflow.base-url=https://ragflow.example", - "muse.knowledge.ragflow.api-key=ragflow-key") + void should_installUnavailableWhenProviderIsRagflow() { + // 遗留 provider=ragflow(运行时 S6d 已删):Dify 主 bean 让位,兜底装通用 UnavailableKnowledgeRuntimeClient, + // 既不因无 bean 启动失败,也不静默降级。 + contextRunner.withPropertyValues("muse.knowledge.runtime-provider=ragflow") .run(context -> { assertEquals(1, context.getBeansOfType(KnowledgeRuntimeClient.class).size()); - assertInstanceOf(HttpRagFlowKnowledgeRuntimeClient.class, + assertInstanceOf(UnavailableKnowledgeRuntimeClient.class, context.getBean(KnowledgeRuntimeClient.class)); }); } @Test - void should_defaultToRagFlowWhenProviderUnset() { - // 缺省 provider(既有部署零变更):matchIfMissing 让 RAGFlow 装配继续生效,Dify 让位。 + void should_defaultToDifyWhenProviderUnset() { + // 缺省 provider(未显式配置):matchIfMissing 让 Dify 成缺省运行时。 contextRunner.withPropertyValues( - "muse.knowledge.ragflow.base-url=https://ragflow.example", - "muse.knowledge.ragflow.api-key=ragflow-key") + "muse.knowledge.dify.base-url=http://dify.example/v1", + "muse.knowledge.dify.dataset-api-key=dataset-key") .run(context -> { assertEquals(1, context.getBeansOfType(KnowledgeRuntimeClient.class).size()); - assertInstanceOf(HttpRagFlowKnowledgeRuntimeClient.class, + assertInstanceOf(DifyKnowledgeRuntimeClient.class, context.getBean(KnowledgeRuntimeClient.class)); }); } @Test void should_installSingleUnavailableWhenProviderUnsetAndNothingConfigured() { + // 缺省 provider + 未配 dify.*:缺省进 Dify 装配但配缺 → fail-closed 通用 Unavailable 桩,仍确保只有一个 bean。 contextRunner.run(context -> { - // 仍确保只有一个 bean(RAGFlow 的 Unavailable 桩),不因两装配类共存而重复注册。 assertEquals(1, context.getBeansOfType(KnowledgeRuntimeClient.class).size()); - assertInstanceOf(UnavailableRagFlowKnowledgeRuntimeClient.class, + assertInstanceOf(UnavailableKnowledgeRuntimeClient.class, context.getBean(KnowledgeRuntimeClient.class)); }); } diff --git a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/test/java/cn/iocoder/muse/module/knowledge/application/muse/facade/KnowledgeRuntimeClientTest.java b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/test/java/cn/iocoder/muse/module/knowledge/application/muse/facade/KnowledgeRuntimeClientTest.java deleted file mode 100644 index 9cdb5a5d..00000000 --- a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/test/java/cn/iocoder/muse/module/knowledge/application/muse/facade/KnowledgeRuntimeClientTest.java +++ /dev/null @@ -1,398 +0,0 @@ -package cn.iocoder.muse.module.knowledge.application.muse.facade; - -import org.junit.jupiter.api.Test; - -import java.io.IOException; -import java.net.http.HttpHeaders; -import java.time.Duration; -import java.util.Arrays; -import java.nio.charset.StandardCharsets; -import java.util.ArrayList; -import java.util.List; -import java.util.Map; - -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertNull; -import static org.junit.jupiter.api.Assertions.assertTrue; - -/** - * Knowledge 运行时 client 基础行为测试。 - */ -class KnowledgeRuntimeClientTest { - - @Test - void unavailableClientShouldFailClosedWhenConfigMissing() { - KnowledgeRuntimeClient client = new UnavailableRagFlowKnowledgeRuntimeClient(); - - KnowledgeRuntimeClient.RuntimeResult result = client.createDataset( - new KnowledgeRuntimeClient.CreateDatasetCommand(100L, 2001L, 4001L, - "KB", Map.of(), "corr-1", "hash-1", 1)); - - assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status()); - assertEquals(KnowledgeRuntimeClient.FailureClass.CONFIG_MISSING, result.failureClass()); - assertEquals("createDataset", result.operation().wireName()); - assertFalse(result.redactedSummary().isBlank()); - } - - @Test - void startParseShouldExposeOnlyMuseProcessingTaskAndCorrelation() { - HttpRagFlowKnowledgeRuntimeClient client = new StubHttpRagFlowKnowledgeRuntimeClient( - "https://ragflow.example", "configured-api-key", """ - {"code":0,"data":{"task_id":"external-task-should-not-leak","id":"external-task-should-not-leak"}} - """); - - KnowledgeRuntimeClient.RuntimeResult result = client.startParseDocuments( - new KnowledgeRuntimeClient.StartParseDocumentsCommand(100L, 2001L, 4001L, - "dataset-1", List.of("doc-1"), "muse-task-1", "corr-parse-1", "hash-parse-1", 1)); - - assertEquals(KnowledgeRuntimeClient.Status.ACCEPTED, result.status()); - assertNull(result.externalId()); - assertEquals("muse-task-1", result.processingTaskId()); - assertEquals("corr-parse-1", result.correlationId()); - assertFalse(result.redactedSummary().contains("external-task-should-not-leak")); - } - - @Test - void redactionShouldNotLeakCredentialsFromSummaries() { - HttpRagFlowKnowledgeRuntimeClient client = new StubHttpRagFlowKnowledgeRuntimeClient( - "https://ragflow.example", "configured-api-key", "{\"code\":0,\"data\":{\"id\":\"dataset-1\"}}"); - - KnowledgeRuntimeClient.RuntimeResult result = client.createDataset( - new KnowledgeRuntimeClient.CreateDatasetCommand(100L, 2001L, 4001L, - "KB", Map.of("apiKey", "secret-value", "authorization", "Bearer abcdefgh"), - "corr-1", "hash-1", 1)); - - assertFalse(result.redactedSummary().contains("secret-value")); - assertFalse(result.redactedSummary().toLowerCase().contains("bearer abcdefgh")); - } - - @Test - void createDatasetShouldSendOnlyNameWithoutConfigPayload() { - CapturingHttpRagFlowKnowledgeRuntimeClient client = new CapturingHttpRagFlowKnowledgeRuntimeClient( - "https://ragflow.example", "configured-api-key", "{\"code\":0,\"data\":{\"id\":\"dataset-1\"}}"); - - KnowledgeRuntimeClient.RuntimeResult result = client.createDataset( - new KnowledgeRuntimeClient.CreateDatasetCommand(100L, 2001L, 4001L, - "KB", Map.of("chunk_method", "naive", "apiKey", "secret-value"), - "corr-create-name-only", "hash-create-name-only", 1)); - - assertEquals(KnowledgeRuntimeClient.Status.SUCCEEDED, result.status()); - String requestBody = new String(client.requests.get(0).body(), StandardCharsets.UTF_8); - assertTrue(requestBody.contains("KB")); - assertFalse(requestBody.contains("config")); - assertFalse(requestBody.contains("chunk_method")); - assertFalse(requestBody.contains("secret-value")); - } - - @Test - void ragflowEnvelopeNonZeroCodeShouldFailClosedInsteadOfSucceeded() { - HttpRagFlowKnowledgeRuntimeClient client = new StubHttpRagFlowKnowledgeRuntimeClient( - "https://ragflow.example", "configured-api-key", "{\"code\":401,\"message\":\"auth failed\"}"); - - KnowledgeRuntimeClient.RuntimeResult result = client.createDataset( - new KnowledgeRuntimeClient.CreateDatasetCommand(100L, 2001L, 4001L, - "KB", Map.of(), "corr-envelope-1", "hash-envelope-1", 1)); - - assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status()); - assertEquals(KnowledgeRuntimeClient.FailureClass.AUTH_FAILED, result.failureClass()); - assertNull(result.externalId()); - } - - @Test - void createDatasetShouldFailClosedWhenRequiredExternalIdMissing() { - HttpRagFlowKnowledgeRuntimeClient client = new StubHttpRagFlowKnowledgeRuntimeClient( - "https://ragflow.example", "configured-api-key", "{\"code\":0,\"data\":{\"name\":\"KB\"}}"); - - KnowledgeRuntimeClient.RuntimeResult result = client.createDataset( - new KnowledgeRuntimeClient.CreateDatasetCommand(100L, 2001L, 4001L, - "KB", Map.of(), "corr-envelope-2", "hash-envelope-2", 1)); - - assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status()); - assertEquals(KnowledgeRuntimeClient.FailureClass.RESPONSE_SCHEMA_CHANGED, result.failureClass()); - } - - @Test - void failureSummaryShouldNotContainPlainProviderErrorBody() { - String providerError = "plain provider unavailable body"; - HttpRagFlowKnowledgeRuntimeClient client = new StubHttpRagFlowKnowledgeRuntimeClient( - "https://ragflow.example", "configured-api-key", 500, providerError); - - KnowledgeRuntimeClient.RuntimeResult result = client.createDataset( - new KnowledgeRuntimeClient.CreateDatasetCommand(100L, 2001L, 4001L, - "KB", Map.of(), "corr-fail-body-1", "hash-fail-body-1", 1)); - - assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status()); - assertEquals(KnowledgeRuntimeClient.FailureClass.RAGFLOW_UNAVAILABLE, result.failureClass()); - assertFalse(result.redactedSummary().contains(providerError)); - assertTrue(result.redactedSummary().contains("\"redacted\":true")); - assertTrue(result.redactedSummary().contains("\"sha256\"")); - } - - @Test - void failureSummaryShouldNotContainJsonProviderMessageOrErrorFields() { - String providerMessage = "plain provider raw error"; - String providerError = "raw error field"; - HttpRagFlowKnowledgeRuntimeClient client = new StubHttpRagFlowKnowledgeRuntimeClient( - "https://ragflow.example", "configured-api-key", 500, - "{\"message\":\"" + providerMessage + "\",\"error\":\"" + providerError + "\"}"); - - KnowledgeRuntimeClient.RuntimeResult result = client.createDataset( - new KnowledgeRuntimeClient.CreateDatasetCommand(100L, 2001L, 4001L, - "KB", Map.of(), "corr-fail-body-json-1", "hash-fail-body-json-1", 1)); - - assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status()); - assertEquals(KnowledgeRuntimeClient.FailureClass.RAGFLOW_UNAVAILABLE, result.failureClass()); - assertFalse(result.redactedSummary().contains(providerMessage)); - assertFalse(result.redactedSummary().contains(providerError)); - assertTrue(result.redactedSummary().contains("\"redacted\":true")); - assertTrue(result.redactedSummary().contains("\"sha256\"")); - } - - @Test - void failureSummaryShouldNotContainJsonProviderErrorsArray() { - String providerError = "array raw error"; - HttpRagFlowKnowledgeRuntimeClient client = new StubHttpRagFlowKnowledgeRuntimeClient( - "https://ragflow.example", "configured-api-key", 400, - "{\"errors\":[{\"message\":\"" + providerError + "\"}]}"); - - KnowledgeRuntimeClient.RuntimeResult result = client.createDataset( - new KnowledgeRuntimeClient.CreateDatasetCommand(100L, 2001L, 4001L, - "KB", Map.of(), "corr-fail-body-json-2", "hash-fail-body-json-2", 1)); - - assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status()); - assertEquals(KnowledgeRuntimeClient.FailureClass.VALIDATION_ERROR, result.failureClass()); - assertFalse(result.redactedSummary().contains(providerError)); - assertTrue(result.redactedSummary().contains("\"redacted\":true")); - assertTrue(result.redactedSummary().contains("\"sha256\"")); - } - - @Test - void nonZeroEnvelopeFailureSummaryShouldNotContainPlainMessage() { - String providerMessage = "business rule rejected this dataset"; - HttpRagFlowKnowledgeRuntimeClient client = new StubHttpRagFlowKnowledgeRuntimeClient( - "https://ragflow.example", "configured-api-key", - "{\"code\":422,\"message\":\"" + providerMessage + "\"}"); - - KnowledgeRuntimeClient.RuntimeResult result = client.createDataset( - new KnowledgeRuntimeClient.CreateDatasetCommand(100L, 2001L, 4001L, - "KB", Map.of(), "corr-fail-body-2", "hash-fail-body-2", 1)); - - assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status()); - assertEquals(KnowledgeRuntimeClient.FailureClass.VALIDATION_ERROR, result.failureClass()); - assertFalse(result.redactedSummary().contains(providerMessage)); - assertTrue(result.redactedSummary().contains("\"redacted\":true")); - assertTrue(result.redactedSummary().contains("\"sha256\"")); - } - - @Test - void uploadDocumentsShouldUseMultipartAndRejectMissingMaterializedBytes() { - CapturingHttpRagFlowKnowledgeRuntimeClient client = new CapturingHttpRagFlowKnowledgeRuntimeClient( - "https://ragflow.example", "configured-api-key", "{\"code\":0,\"data\":[{\"id\":\"doc-1\"}]}"); - - KnowledgeRuntimeClient.RuntimeResult rejected = client.uploadDocuments( - new KnowledgeRuntimeClient.UploadDocumentsCommand(100L, 2001L, 4001L, - "dataset-1", List.of(new KnowledgeRuntimeClient.DocumentUpload( - 5001L, 6001L, "file-ref-1", "novel.txt", "text/plain", 10L, null)), - "corr-upload-1", "hash-upload-1", 1)); - - assertEquals(KnowledgeRuntimeClient.Status.FAILED, rejected.status()); - assertEquals(KnowledgeRuntimeClient.FailureClass.FILE_REJECTED, rejected.failureClass()); - - KnowledgeRuntimeClient.RuntimeResult uploaded = client.uploadDocuments( - new KnowledgeRuntimeClient.UploadDocumentsCommand(100L, 2001L, 4001L, - "dataset-1", List.of(new KnowledgeRuntimeClient.DocumentUpload( - 5001L, 6001L, "file-ref-1", "novel.txt", "text/plain", 5L, - "hello".getBytes(StandardCharsets.UTF_8))), - "corr-upload-2", "hash-upload-2", 1)); - - assertEquals(KnowledgeRuntimeClient.Status.SUCCEEDED, uploaded.status()); - assertEquals("POST", client.requests.get(0).method()); - assertTrue(client.requests.get(0).contentType().startsWith("multipart/form-data; boundary=")); - assertTrue(new String(client.requests.get(0).body(), StandardCharsets.ISO_8859_1).contains("filename=\"novel.txt\"")); - assertFalse(new String(client.requests.get(0).body(), StandardCharsets.ISO_8859_1).contains("\"documents\"")); - } - - @Test - void uploadDocumentsShouldPreserveBinaryBytesInsideMultipartPayload() { - CapturingHttpRagFlowKnowledgeRuntimeClient client = new CapturingHttpRagFlowKnowledgeRuntimeClient( - "https://ragflow.example", "configured-api-key", "{\"code\":0,\"data\":[{\"id\":\"doc-1\"}]}"); - byte[] binaryContent = new byte[]{0x25, 0x50, 0x44, 0x46, 0x00, (byte) 0x80, (byte) 0xFF, 0x01}; - - KnowledgeRuntimeClient.RuntimeResult uploaded = client.uploadDocuments( - new KnowledgeRuntimeClient.UploadDocumentsCommand(100L, 2001L, 4001L, - "dataset-1", List.of(new KnowledgeRuntimeClient.DocumentUpload( - 5001L, 6001L, "file-ref-1", "binary.pdf", "application/pdf", - (long) binaryContent.length, binaryContent)), - "corr-upload-bin", "hash-upload-bin", 1)); - - assertEquals(KnowledgeRuntimeClient.Status.SUCCEEDED, uploaded.status()); - assertTrue(containsSubsequence(client.requests.get(0).body(), binaryContent)); - } - - @Test - void redactedSummaryShouldNotLeakChunkQuestionPromptOrResponseContent() { - String responseBody = """ - {"code":0,"data":{"chunks":[{"id":"chunk-1","content":"secret chunk content"}],"response":"raw answer text"}} - """; - HttpRagFlowKnowledgeRuntimeClient client = new StubHttpRagFlowKnowledgeRuntimeClient( - "https://ragflow.example", "configured-api-key", responseBody); - - KnowledgeRuntimeClient.RuntimeResult result = client.retrieveChunks( - new KnowledgeRuntimeClient.RetrieveChunksCommand(100L, 2001L, 4001L, - List.of("dataset-1"), List.of("doc-1"), "private retrieval question", - 3, 0.5, Map.of("prompt", "private prompt"), "corr-redact-1", "hash-redact-1", 1)); - - assertFalse(result.redactedSummary().contains("secret chunk content")); - assertFalse(result.redactedSummary().contains("raw answer text")); - assertFalse(result.redactedSummary().contains("private retrieval question")); - assertFalse(result.redactedSummary().contains("private prompt")); - } - - @Test - void retrieveChunksShouldSucceedWithoutTopLevelIdAndKeepOnlyMetadataSummary() { - HttpRagFlowKnowledgeRuntimeClient client = new StubHttpRagFlowKnowledgeRuntimeClient( - "https://ragflow.example", "configured-api-key", """ - {"code":0,"data":{"chunks":[{"content":"secret chunk content","score":0.91}],"response":"raw answer text"}} - """); - - KnowledgeRuntimeClient.RuntimeResult result = client.retrieveChunks( - new KnowledgeRuntimeClient.RetrieveChunksCommand(100L, 2001L, 4001L, - List.of("dataset-1"), List.of("doc-1"), "private retrieval question", - 3, 0.5, Map.of("prompt", "private prompt"), "corr-retrieve-1", "hash-retrieve-1", 1)); - - assertEquals(KnowledgeRuntimeClient.Status.SUCCEEDED, result.status()); - assertNull(result.externalId()); - assertTrue(result.redactedSummary().contains("\"chunksCount\":1")); - assertFalse(result.redactedSummary().contains("secret chunk content")); - assertFalse(result.redactedSummary().contains("raw answer text")); - assertFalse(result.redactedSummary().contains("private retrieval question")); - assertFalse(result.redactedSummary().contains("private prompt")); - } - - @Test - void retrieveChunksShouldOmitDocumentIdsFromBodyWhenNull() { - // 回归:document_ids 为 null(P-A 按授权 dataset 全量检索、不限定具体文档)时不得放入 body; - // RAGFlow 要求其为 list,传 null 会被拒 code 102 "`documents` should be a list" 致检索 fail-closed → P-A 永不命中。 - CapturingHttpRagFlowKnowledgeRuntimeClient client = new CapturingHttpRagFlowKnowledgeRuntimeClient( - "https://ragflow.example", "configured-api-key", - "{\"code\":0,\"data\":{\"chunks\":[{\"content\":\"c\",\"score\":0.9}]}}"); - - KnowledgeRuntimeClient.RuntimeResult result = client.retrieveChunks( - new KnowledgeRuntimeClient.RetrieveChunksCommand(100L, 2001L, 4001L, - List.of("dataset-1"), null, "q", 5, 0.2, null, "corr-null-doc", "hash-null-doc", 1)); - - assertEquals(KnowledgeRuntimeClient.Status.SUCCEEDED, result.status()); - String requestBody = new String(client.requests.get(0).body(), StandardCharsets.UTF_8); - assertFalse(requestBody.contains("document_ids"), "document_ids 为 null 时不应出现在请求 body"); - assertTrue(requestBody.contains("dataset_ids")); - } - - @Test - void retrieveChunksShouldSendMetadataConditionWithOfficialRagflowFieldName() { - // 回归:RAGFlow 检索元数据过滤的官方字段是 metadata_condition； - // metadata_filter 会被外部服务静默忽略，不能作为任何隔离或筛选证据。 - CapturingHttpRagFlowKnowledgeRuntimeClient client = new CapturingHttpRagFlowKnowledgeRuntimeClient( - "https://ragflow.example", "configured-api-key", - "{\"code\":0,\"data\":{\"chunks\":[{\"content\":\"c\",\"score\":0.9}]}}"); - - KnowledgeRuntimeClient.RuntimeResult result = client.retrieveChunks( - new KnowledgeRuntimeClient.RetrieveChunksCommand(100L, 2001L, 4001L, - List.of("dataset-1"), List.of("doc-1"), "q", 5, 0.2, - Map.of("logic", "and", "conditions", List.of(Map.of( - "name", "source", "comparison_operator", "=", "value", "public"))), - "corr-metadata-condition", "hash-metadata-condition", 1)); - - assertEquals(KnowledgeRuntimeClient.Status.SUCCEEDED, result.status()); - String requestBody = new String(client.requests.get(0).body(), StandardCharsets.UTF_8); - assertTrue(requestBody.contains("\"metadata_condition\"")); - assertFalse(requestBody.contains("\"metadata_filter\"")); - } - - @Test - void pollDocumentStatusesShouldUseRagflowIdQueryForSingleDocumentSmoke() { - CapturingHttpRagFlowKnowledgeRuntimeClient client = new CapturingHttpRagFlowKnowledgeRuntimeClient( - "https://ragflow.example", "configured-api-key", """ - {"code":0,"data":{"docs":[ - {"id":"doc-1","run":"DONE","progress":100,"progress_msg":"done"}, - {"id":"doc-2","run":"RUNNING","progress":50,"progress_msg":"running"}, - {"id":"doc-3","run":"DONE","progress":100,"progress_msg":"done"} - ]}} - """); - - KnowledgeRuntimeClient.RuntimeResult result = client.pollDocumentStatuses( - new KnowledgeRuntimeClient.PollDocumentStatusesCommand(100L, 2001L, 4001L, - "dataset-1", List.of("doc-1"), "task-1", "corr-poll-1", "hash-poll-1", 1)); - - assertEquals(KnowledgeRuntimeClient.Status.SUCCEEDED, result.status()); - assertEquals(List.of("doc-1"), result.documentStatuses().stream() - .map(KnowledgeRuntimeClient.DocumentStatus::ragflowDocumentId).toList()); - assertTrue(client.requests.get(0).path().contains("id=doc-1")); - assertFalse(client.requests.get(0).path().contains("document_ids=")); - } - - @Test - void pollDocumentStatusesShouldRejectMultipleDocumentIdsBeforeCallingRagflow() { - CapturingHttpRagFlowKnowledgeRuntimeClient client = new CapturingHttpRagFlowKnowledgeRuntimeClient( - "https://ragflow.example", "configured-api-key", "{\"code\":0,\"data\":{\"docs\":[]}}"); - - KnowledgeRuntimeClient.RuntimeResult result = client.pollDocumentStatuses( - new KnowledgeRuntimeClient.PollDocumentStatusesCommand(100L, 2001L, 4001L, - "dataset-1", List.of("doc-1", "doc-2"), "task-1", "corr-poll-many", "hash-poll-many", 1)); - - assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status()); - assertEquals(KnowledgeRuntimeClient.FailureClass.VALIDATION_ERROR, result.failureClass()); - assertTrue(result.redactedSummary().contains("single document status polling")); - assertEquals(0, client.requests.size()); - } - - private static boolean containsSubsequence(byte[] payload, byte[] expected) { - for (int start = 0; start <= payload.length - expected.length; start++) { - if (Arrays.equals(Arrays.copyOfRange(payload, start, start + expected.length), expected)) { - return true; - } - } - return false; - } - - private static final class StubHttpRagFlowKnowledgeRuntimeClient extends HttpRagFlowKnowledgeRuntimeClient { - - private final String body; - private final int statusCode; - - private StubHttpRagFlowKnowledgeRuntimeClient(String baseUrl, String apiKey, String body) { - this(baseUrl, apiKey, 200, body); - } - - private StubHttpRagFlowKnowledgeRuntimeClient(String baseUrl, String apiKey, int statusCode, String body) { - super(baseUrl, apiKey, Duration.ofSeconds(1), 0, true); - this.statusCode = statusCode; - this.body = body; - } - - @Override - protected HttpResponsePayload send(HttpRequestPayload request) throws IOException, InterruptedException { - return new HttpResponsePayload(statusCode, body, HttpHeaders.of(Map.of(), (name, value) -> true)); - } - - } - - private static final class CapturingHttpRagFlowKnowledgeRuntimeClient extends HttpRagFlowKnowledgeRuntimeClient { - - private final String body; - private final List requests = new ArrayList<>(); - - private CapturingHttpRagFlowKnowledgeRuntimeClient(String baseUrl, String apiKey, String body) { - super(baseUrl, apiKey, Duration.ofSeconds(1), 0, true); - this.body = body; - } - - @Override - protected HttpResponsePayload send(HttpRequestPayload request) throws IOException, InterruptedException { - requests.add(request); - return new HttpResponsePayload(200, body, HttpHeaders.of(Map.of(), (name, value) -> true)); - } - - } - -} diff --git a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/test/java/cn/iocoder/muse/module/knowledge/application/muse/facade/P1rRagFlowLiveAcceptanceIT.java b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/test/java/cn/iocoder/muse/module/knowledge/application/muse/facade/P1rRagFlowLiveAcceptanceIT.java deleted file mode 100644 index 94f5f6d0..00000000 --- a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/test/java/cn/iocoder/muse/module/knowledge/application/muse/facade/P1rRagFlowLiveAcceptanceIT.java +++ /dev/null @@ -1,237 +0,0 @@ -package cn.iocoder.muse.module.knowledge.application.muse.facade; - -import cn.iocoder.muse.framework.common.util.json.JsonUtils; -import com.fasterxml.jackson.databind.JsonNode; -import org.junit.jupiter.api.Test; - -import java.net.URI; -import java.nio.charset.StandardCharsets; -import java.security.MessageDigest; -import java.security.NoSuchAlgorithmException; -import java.time.Duration; -import java.time.Instant; -import java.util.HexFormat; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; - -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertNotNull; -import static org.junit.jupiter.api.Assertions.assertTrue; -import static org.junit.jupiter.api.Assumptions.assumeTrue; - -/** - * P1R-5 RAGFlow 外部 live acceptance 入口。 - * - *

默认必须跳过，只有显式设置 MUSE_P1R_EXTERNAL_ACCEPTANCE=true 才会真实调用外部 RAGFlow。 - * 本测试会保留新建 dataset 作为验收证据，不做删除；输出证据只打印 endpoint 与 key 指纹。

- */ -class P1rRagFlowLiveAcceptanceIT { - - private static final String ACCEPTANCE_ENV = "MUSE_P1R_EXTERNAL_ACCEPTANCE"; - private static final String BASE_URL_ENV = "MUSE_KNOWLEDGE_RAGFLOW_BASE_URL"; - private static final String API_KEY_ENV = "MUSE_KNOWLEDGE_RAGFLOW_API_KEY"; - private static final String GRAPHRAG_READY_ENV = "MUSE_KNOWLEDGE_RAGFLOW_GRAPHRAG_ATTRIBUTION_READY"; - - @Test - void shouldCallRagFlowThroughMuseRuntimeClientAndPrintRedactedEvidence() throws Exception { - assumeTrue(externalAcceptanceEnabled(), "P1R 外部验收未启用，设置 MUSE_P1R_EXTERNAL_ACCEPTANCE=true 后才运行"); - - String baseUrl = requiredEnv(BASE_URL_ENV); - String apiKey = requiredEnv(API_KEY_ENV); - boolean graphRagReady = Boolean.parseBoolean(envOrDefault(GRAPHRAG_READY_ENV, "false")); - HttpRagFlowKnowledgeRuntimeClient client = client(baseUrl, apiKey, graphRagReady); - String suffix = String.valueOf(Instant.now().toEpochMilli()); - String datasetName = "p1r-live-" + suffix; - String correlationPrefix = "p1r5-ragflow-live-" + suffix; - - KnowledgeRuntimeClient.RuntimeResult createDataset = client.createDataset( - new KnowledgeRuntimeClient.CreateDatasetCommand(1L, 1L, 1L, datasetName, - Map.of(), correlationPrefix + "-dataset", hash(datasetName), 1)); - assertSucceeded(createDataset, "createDataset"); - assertNotNull(createDataset.externalId(), "RAGFlow createDataset 必须返回 dataset id"); - String datasetId = createDataset.externalId(); - - byte[] content = liveDocumentContent(suffix); - KnowledgeRuntimeClient.RuntimeResult upload = client.uploadDocuments( - new KnowledgeRuntimeClient.UploadDocumentsCommand(1L, 1L, 1L, datasetId, - List.of(new KnowledgeRuntimeClient.DocumentUpload(1L, 1L, - "p1r-live-" + suffix, datasetName + ".txt", "text/plain", - (long) content.length, content)), - correlationPrefix + "-upload", hash(datasetId + "-upload"), 1)); - assertSucceeded(upload, "uploadDocuments"); - assertNotNull(upload.externalId(), "RAGFlow uploadDocuments 必须返回 document id"); - String documentId = upload.externalId(); - - KnowledgeRuntimeClient.RuntimeResult parse = client.startParseDocuments( - new KnowledgeRuntimeClient.StartParseDocumentsCommand(1L, 1L, 1L, datasetId, - List.of(documentId), correlationPrefix + "-parse-task", - correlationPrefix + "-parse", hash(documentId + "-parse"), 1)); - assertEquals(KnowledgeRuntimeClient.Status.ACCEPTED, parse.status(), - () -> "startParseDocuments 未进入 accepted：" + parse.redactedSummary()); - - KnowledgeRuntimeClient.RuntimeResult poll = waitUntilSearchable(client, datasetId, documentId, - correlationPrefix); - KnowledgeRuntimeClient.RuntimeResult retrieval = client.retrieveChunks( - new KnowledgeRuntimeClient.RetrieveChunksCommand(1L, 1L, 1L, - List.of(datasetId), List.of(documentId), "P1R live acceptance marker 是什么？", - 3, 0.0, Map.of(), correlationPrefix + "-retrieval", - hash(documentId + "-retrieval"), 1)); - assertSucceeded(retrieval, "retrieveChunks"); - assertTrue(countFromRedactedSummary(retrieval.redactedSummary(), "chunksCount") > 0, - () -> "retrieveChunks 没有返回 chunk，不能作为 retrieval 正向证据：" + retrieval.redactedSummary()); - - Map evidence = redactedEvidence(baseUrl, apiKey, datasetName, datasetId, documentId, - createDataset, upload, parse, poll, retrieval, graphRagReady); - System.out.println(JsonUtils.toJsonString(evidence)); - } - - private HttpRagFlowKnowledgeRuntimeClient client(String baseUrl, String apiKey, boolean graphRagReady) { - return new HttpRagFlowKnowledgeRuntimeClient(baseUrl, apiKey, - Duration.ofSeconds(intEnv("MUSE_KNOWLEDGE_RAGFLOW_TIMEOUT_SECONDS", 30)), - intEnv("MUSE_KNOWLEDGE_RAGFLOW_RETRY_BUDGET", 0), graphRagReady); - } - - private KnowledgeRuntimeClient.RuntimeResult waitUntilSearchable( - HttpRagFlowKnowledgeRuntimeClient client, String datasetId, String documentId, String correlationPrefix) - throws InterruptedException { - long deadline = System.nanoTime() - + Duration.ofSeconds(intEnv("MUSE_P1R_RAGFLOW_PARSE_WAIT_SECONDS", 120)).toNanos(); - KnowledgeRuntimeClient.RuntimeResult last = null; - int attempt = 1; - while (System.nanoTime() < deadline) { - last = client.pollDocumentStatuses(new KnowledgeRuntimeClient.PollDocumentStatusesCommand( - 1L, 1L, 1L, datasetId, List.of(documentId), correlationPrefix + "-parse-task", - correlationPrefix + "-poll-" + attempt, hash(documentId + "-poll-" + attempt), attempt)); - assertSucceeded(last, "pollDocumentStatuses"); - if (last.documentStatuses().stream().anyMatch(status -> "searchable".equals(status.museStatus()))) { - return last; - } - Thread.sleep(2_000L); - attempt++; - } - throw new AssertionError("RAGFlow document parse 未在等待窗口内进入 searchable，last=" - + (last == null ? "none" : last.redactedSummary())); - } - - private void assertSucceeded(KnowledgeRuntimeClient.RuntimeResult result, String operation) { - assertNotNull(result, operation + " result 不能为空"); - assertEquals(KnowledgeRuntimeClient.Status.SUCCEEDED, result.status(), - () -> operation + " failed: " + result.redactedSummary()); - } - - private byte[] liveDocumentContent(String suffix) { - String text = """ - P1R live acceptance marker %s. - This document is intentionally small and kept in RAGFlow as external acceptance evidence. - Muse Knowledge RAGFlow smoke verifies dataset creation, document upload, parse polling, chunks listing and retrieval. - """.formatted(suffix); - return text.getBytes(StandardCharsets.UTF_8); - } - - private Map redactedEvidence(String baseUrl, String apiKey, String datasetName, - String datasetId, String documentId, - KnowledgeRuntimeClient.RuntimeResult createDataset, - KnowledgeRuntimeClient.RuntimeResult upload, - KnowledgeRuntimeClient.RuntimeResult parse, - KnowledgeRuntimeClient.RuntimeResult poll, - KnowledgeRuntimeClient.RuntimeResult retrieval, - boolean graphRagReady) { - Map evidence = new LinkedHashMap<>(); - evidence.put("acceptance", "p1r5-ragflow-live"); - evidence.put("endpoint", endpointSummary(baseUrl, "/api/v1")); - evidence.put("apiKey", secretFingerprint(apiKey)); - evidence.put("datasetName", datasetName); - evidence.put("datasetId", datasetId); - evidence.put("documentId", documentId); - evidence.put("createDataset", operationSummary(createDataset)); - evidence.put("uploadDocuments", operationSummary(upload)); - evidence.put("startParseDocuments", operationSummary(parse)); - evidence.put("pollDocumentStatuses", operationSummary(poll)); - evidence.put("documentStatuses", poll.documentStatuses()); - evidence.put("retrieveChunks", operationSummary(retrieval)); - evidence.put("retrievalChunksCount", countFromRedactedSummary(retrieval.redactedSummary(), "chunksCount")); - evidence.put("graphRagAttributionReady", graphRagReady); - evidence.put("datasetRetained", true); - return evidence; - } - - private Map operationSummary(KnowledgeRuntimeClient.RuntimeResult result) { - Map summary = new LinkedHashMap<>(); - summary.put("operation", result.operation().wireName()); - summary.put("status", result.status()); - summary.put("failureClass", result.failureClass()); - summary.put("durationMillis", result.durationMillis()); - summary.put("externalIdPresent", result.externalId() != null && !result.externalId().isBlank()); - return summary; - } - - private int countFromRedactedSummary(String redactedSummary, String fieldName) { - JsonNode node = JsonUtils.parseTree(redactedSummary == null || redactedSummary.isBlank() ? "{}" : redactedSummary); - return node.has(fieldName) && node.path(fieldName).canConvertToInt() ? node.path(fieldName).asInt() : 0; - } - - private Map endpointSummary(String baseUrl, String path) { - URI uri = URI.create(trimRight(baseUrl, "/") + path); - Map endpoint = new LinkedHashMap<>(); - endpoint.put("scheme", uri.getScheme()); - endpoint.put("host", uri.getHost()); - endpoint.put("port", uri.getPort()); - endpoint.put("path", uri.getPath()); - return endpoint; - } - - private Map secretFingerprint(String secret) { - Map fingerprint = new LinkedHashMap<>(); - fingerprint.put("length", secret == null ? 0 : secret.length()); - fingerprint.put("sha256Prefix", sha256Prefix(secret == null ? "" : secret, 12)); - return fingerprint; - } - - private boolean externalAcceptanceEnabled() { - return "true".equalsIgnoreCase(System.getenv(ACCEPTANCE_ENV)); - } - - private String requiredEnv(String name) { - String value = System.getenv(name); - assertTrue(value != null && !value.isBlank(), "缺少外部验收环境变量：" + name); - return value; - } - - private String envOrDefault(String name, String defaultValue) { - String value = System.getenv(name); - return value == null || value.isBlank() ? defaultValue : value; - } - - private int intEnv(String name, int defaultValue) { - String value = System.getenv(name); - if (value == null || value.isBlank()) { - return defaultValue; - } - return Integer.parseInt(value); - } - - private String hash(String value) { - return sha256Prefix(value == null ? "" : value, 64); - } - - private String sha256Prefix(String value, int length) { - try { - MessageDigest digest = MessageDigest.getInstance("SHA-256"); - String hex = HexFormat.of().formatHex(digest.digest(value.getBytes(StandardCharsets.UTF_8))); - return hex.substring(0, Math.min(length, hex.length())); - } catch (NoSuchAlgorithmException e) { - throw new IllegalStateException("JDK 缺少 SHA-256 摘要算法", e); - } - } - - private String trimRight(String value, String suffix) { - String result = value == null ? "" : value.trim(); - while (result.endsWith(suffix)) { - result = result.substring(0, result.length() - suffix.length()); - } - return result; - } - -} diff --git a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/test/java/cn/iocoder/muse/module/knowledge/application/muse/facade/RagFlowKnowledgeRuntimeClientConfigurationTest.java b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/test/java/cn/iocoder/muse/module/knowledge/application/muse/facade/RagFlowKnowledgeRuntimeClientConfigurationTest.java deleted file mode 100644 index c91ad205..00000000 --- a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/test/java/cn/iocoder/muse/module/knowledge/application/muse/facade/RagFlowKnowledgeRuntimeClientConfigurationTest.java +++ /dev/null @@ -1,37 +0,0 @@ -package cn.iocoder.muse.module.knowledge.application.muse.facade; - -import org.junit.jupiter.api.Test; -import org.springframework.mock.env.MockEnvironment; - -import static org.junit.jupiter.api.Assertions.assertInstanceOf; - -/** - * RAGFlow runtime client Spring 装配测试。 - */ -class RagFlowKnowledgeRuntimeClientConfigurationTest { - - @Test - void should_registerUnavailableClientWhenRequiredConfigMissing() { - RagFlowKnowledgeRuntimeClientConfiguration configuration = new RagFlowKnowledgeRuntimeClientConfiguration(); - - KnowledgeRuntimeClient client = configuration.ragFlowKnowledgeRuntimeClient(new MockEnvironment()); - - assertInstanceOf(UnavailableRagFlowKnowledgeRuntimeClient.class, client); - } - - @Test - void should_registerHttpClientWhenBaseUrlAndApiKeyAreConfigured() { - MockEnvironment environment = new MockEnvironment() - .withProperty("muse.knowledge.ragflow.base-url", "https://ragflow.example") - .withProperty("muse.knowledge.ragflow.api-key", "test-api-key") - .withProperty("muse.knowledge.ragflow.timeout-seconds", "7") - .withProperty("muse.knowledge.ragflow.retry-budget", "2") - .withProperty("muse.knowledge.ragflow.graphrag-attribution-ready", "true"); - RagFlowKnowledgeRuntimeClientConfiguration configuration = new RagFlowKnowledgeRuntimeClientConfiguration(); - - KnowledgeRuntimeClient client = configuration.ragFlowKnowledgeRuntimeClient(environment); - - assertInstanceOf(HttpRagFlowKnowledgeRuntimeClient.class, client); - } - -} diff --git a/muse-cloud/muse-server/src/test/java/cn/iocoder/muse/server/framework/api/P1rKnowledgeRuntimeEndToEndLiveAcceptanceIT.java b/muse-cloud/muse-server/src/test/java/cn/iocoder/muse/server/framework/api/P1rKnowledgeRuntimeEndToEndLiveAcceptanceIT.java deleted file mode 100644 index 082c0899..00000000 --- a/muse-cloud/muse-server/src/test/java/cn/iocoder/muse/server/framework/api/P1rKnowledgeRuntimeEndToEndLiveAcceptanceIT.java +++ /dev/null @@ -1,1061 +0,0 @@ -package cn.iocoder.muse.server.framework.api; - -import cn.hutool.extra.spring.SpringUtil; -import cn.iocoder.muse.framework.common.enums.UserTypeEnum; -import cn.iocoder.muse.framework.common.util.json.JsonUtils; -import cn.iocoder.muse.framework.datasource.config.MuseDataSourceAutoConfiguration; -import cn.iocoder.muse.framework.mybatis.config.MuseMybatisAutoConfiguration; -import cn.iocoder.muse.framework.security.core.LoginUser; -import cn.iocoder.muse.framework.security.core.util.SecurityFrameworkUtils; -import cn.iocoder.muse.framework.tenant.core.context.TenantContextHolder; -import cn.iocoder.muse.framework.tenant.core.util.TenantUtils; -import cn.iocoder.muse.module.infra.api.file.FileApi; -import cn.iocoder.muse.module.knowledge.application.muse.MuseKnowledgeAuditService; -import cn.iocoder.muse.module.knowledge.application.muse.MuseKnowledgeCommandService; -import cn.iocoder.muse.module.knowledge.application.muse.MuseKnowledgeDocumentService; -import cn.iocoder.muse.module.knowledge.application.muse.MuseKnowledgeMaterializationService; -import cn.iocoder.muse.module.knowledge.application.muse.MuseKnowledgeProcessingTaskService; -import cn.iocoder.muse.module.knowledge.application.muse.facade.HttpRagFlowKnowledgeRuntimeClient; -import cn.iocoder.muse.module.knowledge.application.muse.facade.KnowledgeFileFacade; -import cn.iocoder.muse.module.knowledge.application.muse.facade.KnowledgeRuntimeClient; -import cn.iocoder.muse.module.knowledge.controller.app.muse.vo.AppKnowledgeDocumentVO; -import cn.iocoder.muse.module.knowledge.dal.dataobject.muse.MuseKnowledgeBaseDO; -import cn.iocoder.muse.module.knowledge.dal.mysql.muse.MuseKnowledgeBaseMapper; -import com.baomidou.mybatisplus.autoconfigure.MybatisPlusAutoConfiguration; -import com.github.yulichang.autoconfigure.MybatisPlusJoinAutoConfiguration; -import com.fasterxml.jackson.databind.JsonNode; -import org.flywaydb.core.Flyway; -import org.flywaydb.core.api.MigrationInfo; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.Test; -import org.slf4j.LoggerFactory; -import org.springframework.beans.factory.ObjectProvider; -import org.springframework.boot.WebApplicationType; -import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration; -import org.springframework.boot.autoconfigure.jdbc.DataSourceTransactionManagerAutoConfiguration; -import org.springframework.boot.builder.SpringApplicationBuilder; -import org.springframework.boot.test.context.TestConfiguration; -import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Import; -import org.springframework.context.annotation.Primary; -import org.springframework.core.env.MapPropertySource; -import org.springframework.mock.web.MockHttpServletRequest; -import org.springframework.security.core.context.SecurityContextHolder; -import org.springframework.util.StringUtils; - -import javax.sql.DataSource; -import java.net.URI; -import java.nio.charset.StandardCharsets; -import java.nio.file.Files; -import java.nio.file.Path; -import java.security.MessageDigest; -import java.security.NoSuchAlgorithmException; -import java.sql.Connection; -import java.sql.DriverManager; -import java.sql.PreparedStatement; -import java.sql.ResultSet; -import java.sql.SQLException; -import java.sql.Statement; -import java.time.Duration; -import java.time.Instant; -import java.util.HexFormat; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Locale; -import java.util.Map; -import java.util.Objects; -import java.util.Set; - -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertNotNull; -import static org.junit.jupiter.api.Assertions.assertThrows; -import static org.junit.jupiter.api.Assertions.assertTrue; -import static org.junit.jupiter.api.Assumptions.assumeTrue; - -/** - * P1R-5 Knowledge 主业务入口到真实 RAGFlow / PostgreSQL 的 opt-in live acceptance。 - * - *

默认必须跳过，避免普通测试误清 PostgreSQL 或外呼真实 RAGFlow。只有显式设置 - * {@code MUSE_P1R_EXTERNAL_ACCEPTANCE=true} 时，本测试才会清理 PostgreSQL {@code _test} 库、 - * 迁移到 V14、seed 最小用户 KB，并通过 {@link MuseKnowledgeDocumentService#uploadAppDocument} - * 触发真实 {@link HttpRagFlowKnowledgeRuntimeClient}。

- */ -class P1rKnowledgeRuntimeEndToEndLiveAcceptanceIT { - - private static final String ACCEPTANCE_ENV = "MUSE_P1R_EXTERNAL_ACCEPTANCE"; - private static final String JDBC_URL_PROPERTY = "p1r.flyway.url"; - private static final String JDBC_USER_PROPERTY = "p1r.flyway.user"; - private static final String JDBC_URL_ENV = "P1R_FLYWAY_URL"; - private static final String JDBC_USER_ENV = "P1R_FLYWAY_USER"; - private static final String FLYWAY_LOCATION_PROPERTY = "p1r.flyway.locations"; - private static final String POSTGRESQL_JDBC_PREFIX = "jdbc:postgresql://"; - private static final String JDBC_URI_PREFIX = "jdbc:"; - private static final String TARGET_VERSION = "14"; - private static final Long TENANT_ID = 100L; - private static final Long OWNER_USER_ID = 2001L; - private static final String API_VERSION = "1"; - private static final String RAGFLOW_API_PATH = "/api/v1"; - private static final Set CREDENTIAL_QUERY_KEYS = Set.of( - "user", "username", "password", "pass", "pwd", "sslpassword", "ssl_password", - "token", "secret", "api_key", "apikey", "bearer", "access_token", "refresh_token"); - - @AfterEach - void clearRuntimeContexts() { - // live acceptance 在同一 JVM 内手工设置租户和登录态，测试结束必须清理，避免污染后续测试。 - SecurityContextHolder.clearContext(); - TenantContextHolder.clear(); - } - - @Test - void shouldMaskNonPostgresqlJdbcAuthorityInFailureOutput() { - String masked = maskedUrl("jdbc:mysql://user:password@host:3306/db_test?password=x"); - - assertFalse(masked.contains("user"), () -> "非 PostgreSQL JDBC URL 失败摘要不能泄露 userinfo: " + masked); - assertFalse(masked.contains("password"), () -> "非 PostgreSQL JDBC URL 失败摘要不能泄露密码: " + masked); - assertFalse(masked.contains("host"), () -> "非 PostgreSQL JDBC URL 失败摘要不能泄露真实 host: " + masked); - } - - @Test - void shouldMaskMalformedPostgresqlJdbcParseFailureOutput() { - String url = "jdbc:postgresql://user:password@bad host/muse_local_test?password=x"; - - AssertionError error = assertThrows(AssertionError.class, () -> assertSafeJdbcUrl(url)); - String message = error.getMessage(); - - assertTrue(message.contains("") || message.contains(""), - () -> "malformed PostgreSQL JDBC URL 解析失败消息必须包含脱敏摘要: " + message); - assertFalse(message.contains("user"), () -> "解析失败消息不能泄露 userinfo: " + message); - assertFalse(message.contains("password"), () -> "解析失败消息不能泄露密码或凭据参数: " + message); - assertFalse(message.contains("bad host"), () -> "解析失败消息不能泄露真实 host: " + message); - } - - @Test - void shouldUploadKnowledgeDocumentThroughMuseBusinessServiceAndPersistDbFactsWithRealPostgresqlAndRagFlow() - throws Exception { - assumeTrue(externalAcceptanceEnabled(), - "P1R 外部端到端验收未启用，设置 MUSE_P1R_EXTERNAL_ACCEPTANCE=true 后才运行"); - - LiveSettings settings = LiveSettings.fromEnvironment(); - DatabaseEngineFacts dbEngine = migrateIsolatedTestDatabase(settings); - - try (ConfigurableApplicationContext context = liveContext(settings)) { - setLiveLoginUser(); - LiveResult result = TenantUtils.execute(TENANT_ID, () -> executeMuseKnowledgeBusinessFlow(context)); - DbFacts dbFacts = readDbFacts(settings, result); - assertDbFacts(result, dbFacts); - ExternalVerification externalVerification = verifyExternalRagFlowSearchable(context, result, dbFacts); - - System.out.println(JsonUtils.toJsonString(redactedEvidence(settings, result, dbFacts, - externalVerification, dbEngine))); - } - } - - private void setLiveLoginUser() { - // 当前测试直接调用 application service，没有 HTTP 认证过滤器；这里补齐审计字段依赖的登录态。 - SecurityFrameworkUtils.setLoginUser(liveLoginUser(), new MockHttpServletRequest()); - } - - private LoginUser liveLoginUser() { - LoginUser loginUser = new LoginUser(); - loginUser.setId(OWNER_USER_ID); - loginUser.setUserType(UserTypeEnum.MEMBER.getValue()); - loginUser.setTenantId(TENANT_ID); - loginUser.setVisitTenantId(TENANT_ID); - return loginUser; - } - - private LiveResult executeMuseKnowledgeBusinessFlow(ConfigurableApplicationContext context) { - Long kbId = seedUserKnowledgeBase(context); - String suffix = String.valueOf(Instant.now().toEpochMilli()); - String commandId = "p1r5-knowledge-e2e-live-" + suffix; - - AppKnowledgeDocumentVO.UploadReqVO request = new AppKnowledgeDocumentVO.UploadReqVO(); - request.setCommandId(commandId); - request.setEntryContent(liveDocumentContent(suffix)); - request.setSourceDescription("P1R-5 Knowledge 主业务 live acceptance"); - - AppKnowledgeDocumentVO.UploadRespVO response = context.getBean(MuseKnowledgeDocumentService.class) - .uploadAppDocument(OWNER_USER_ID, API_VERSION, kbId, request); - assertNotNull(response.getDocumentId(), "uploadAppDocument 必须返回 documentId"); - assertTrue(StringUtils.hasText(response.getScanTaskId()), "uploadAppDocument 必须返回 processing task id"); - assertEquals("pending_scan", response.getProcessingStatus(), - "上传响应不是最终 RAGFlow 证据，最终判断必须读取 PostgreSQL facts"); - return new LiveResult(kbId, commandId, response.getDocumentId(), response.getScanTaskId(), suffix); - } - - private Long seedUserKnowledgeBase(ConfigurableApplicationContext context) { - restartKnowledgeBaseIdentity(context.getBean(DataSource.class)); - MuseKnowledgeBaseDO kb = new MuseKnowledgeBaseDO(); - kb.setName("P1R-5 live user KB"); - kb.setDescription("P1R-5 Knowledge / RAGFlow 主业务端到端验收 seed KB"); - kb.setKbType("user"); - kb.setOwnerUserId(OWNER_USER_ID); - kb.setStatus("active"); - kb.setActiveVersion(1); - kb.setVisibilityPolicy(JsonUtils.toJsonString(Map.of("scope", "p1r-live", "owner", "user"))); - kb.setCommandId("p1r5-live-kb-" + Instant.now().toEpochMilli()); - kb.setRevision(1); - kb.setTenantId(TENANT_ID); - context.getBean(MuseKnowledgeBaseMapper.class).insert(kb); - assertNotNull(kb.getId(), "seed KB 必须返回 id"); - return kb.getId(); - } - - private void restartKnowledgeBaseIdentity(DataSource dataSource) { - long restartWith = Instant.now().toEpochMilli(); - try (Connection connection = dataSource.getConnection(); - Statement statement = connection.createStatement()) { - // RAGFlow datasetName 由业务服务按 kbId + activeVersion 生成；重置到时间戳区间可避免 live 重跑撞上旧外部 dataset。 - statement.execute("ALTER TABLE muse_knowledge_base ALTER COLUMN id RESTART WITH " + restartWith); - } catch (SQLException ex) { - throw new IllegalStateException("设置 P1R live seed KB identity 起点失败", ex); - } - } - - private String liveDocumentContent(String suffix) { - return """ - P1R-5 Knowledge live acceptance marker %s. - Muse Knowledge business service uploads this entryContent through uploadAppDocument. - The live assertion proves Muse DB facts for KB, document, version, processing task, bindings, RAGFlow calls and command. - """.formatted(suffix); - } - - private void assertDbFacts(LiveResult result, DbFacts facts) { - assertEquals(1, facts.knowledgeBaseCount(), "必须存在 1 行 muse_knowledge_base seed KB"); - assertEquals(1, facts.documentCount(), "必须存在 1 行 muse_knowledge_document"); - assertEquals(1, facts.documentVersionCount(), "必须存在 1 行 muse_knowledge_document_version"); - assertEquals(1, facts.processingTaskCount(), "必须存在 1 行 muse_knowledge_processing_task"); - assertEquals(1, facts.commandCount(), "必须存在 1 行 muse_knowledge_command"); - assertEquals(2, facts.bindingCount(), "必须存在 dataset binding + document version binding"); - assertEquals(1, facts.datasetBindingCount(), "必须存在 1 行 dataset binding"); - assertEquals(1, facts.documentBindingCount(), "必须存在 1 行 document version binding"); - assertTrue(facts.ragflowCallCount() >= 3, "必须至少记录 createDataset/uploadDocuments/startParseDocuments 3 次 RAGFlow 调用"); - - assertEquals("user", facts.kbType(), "seed KB 必须是用户 KB"); - assertEquals("active", facts.kbStatus(), "seed KB 必须是 active"); - assertEquals(1, facts.kbActiveVersion(), "seed KB activeVersion 必须是 1"); - assertEquals("completed", facts.commandStatus(), "upload command 必须完成，证明可回放快照已落库"); - assertEquals("passed", facts.documentVersionScanStatus(), "test-only scanner 必须只把扫描结果标记为 passed"); - assertEquals("processing", facts.documentVersionProcessingStatus(), "RAGFlow parse accepted 后资料版本应处于 processing"); - assertEquals("parsing", facts.processingTaskStatus(), "processing_task 必须进入 parsing 语义"); - assertEquals("materialized", facts.materializationStatus(), "entryContent 必须已材料化为可上传 bytes"); - assertEquals("passed", facts.processingTaskScanStatus(), "processing_task 必须记录扫描已通过"); - assertEquals("processing", facts.processingTaskParseStatus(), "processing_task 必须记录 parse processing"); - assertEquals("pending", facts.processingTaskIndexStatus(), "parse accepted 后 index 仍应等待轮询推进"); - assertEquals(result.scanTaskId(), facts.processingTaskId(), "响应 scanTaskId 必须对应 DB processing_task.task_id"); - assertTrue(StringUtils.hasText(facts.ragflowDatasetId()), "processing_task 必须写入 ragflow_dataset_id"); - assertTrue(StringUtils.hasText(facts.ragflowDocumentId()), "processing_task 必须写入 ragflow_document_id"); - assertTrue(facts.processingTaskStarted(), "processing_task 必须有 started_at"); - assertFalse(facts.processingTaskFinished(), "parse accepted 后 processing_task 不应被伪造为 finished"); - assertTrue(StringUtils.hasText(facts.processingTaskResultSummary()) - && !"{}".equals(facts.processingTaskResultSummary().trim()), - "processing_task 必须有非空 result_summary"); - - assertTrue(facts.createDatasetSucceeded(), "ragflow_call 必须记录 createDataset succeeded"); - assertTrue(facts.uploadDocumentsSucceeded(), "ragflow_call 必须记录 uploadDocuments succeeded"); - assertTrue(facts.startParseAccepted(), "ragflow_call 必须记录 startParseDocuments accepted"); - assertEquals(facts.ragflowDatasetId(), facts.datasetBindingDatasetId(), - "dataset binding 必须和 processing_task datasetId 一致"); - assertEquals(facts.ragflowDatasetId(), facts.documentBindingDatasetId(), - "document binding 必须和 processing_task datasetId 一致"); - assertEquals(facts.ragflowDocumentId(), facts.documentBindingDocumentId(), - "document binding 必须和 processing_task documentId 一致"); - assertTrue(facts.allRagflowResponseSummariesNonEmpty(), - "每条 RAGFlow business call 都必须有非空 response_summary"); - } - - private ExternalVerification verifyExternalRagFlowSearchable(ConfigurableApplicationContext context, - LiveResult result, - DbFacts facts) throws InterruptedException { - KnowledgeRuntimeClient client = context.getBean(KnowledgeRuntimeClient.class); - String correlationPrefix = "p1r5-knowledge-e2e-external-" + result.suffix(); - KnowledgeRuntimeClient.RuntimeResult poll = waitUntilSearchable(client, facts.ragflowDatasetId(), - facts.ragflowDocumentId(), facts.processingTaskId(), correlationPrefix); - KnowledgeRuntimeClient.RuntimeResult retrieval = client.retrieveChunks( - new KnowledgeRuntimeClient.RetrieveChunksCommand(TENANT_ID, OWNER_USER_ID, result.kbId(), - List.of(facts.ragflowDatasetId()), List.of(facts.ragflowDocumentId()), - "P1R-5 Knowledge live acceptance marker 是什么？", 3, 0.0, Map.of(), - correlationPrefix + "-retrieval", hash(facts.ragflowDocumentId() + "-retrieval"), 1)); - assertSucceeded(retrieval, "retrieveChunks"); - int retrievalChunksCount = countFromRedactedSummary(retrieval.redactedSummary(), "chunksCount"); - assertTrue(retrievalChunksCount > 0, - () -> "retrieveChunks 没有返回 chunk，不能作为外部检索证据：" + retrieval.redactedSummary()); - return new ExternalVerification(poll, retrieval, retrievalChunksCount); - } - - private KnowledgeRuntimeClient.RuntimeResult waitUntilSearchable(KnowledgeRuntimeClient client, - String datasetId, - String documentId, - String processingTaskId, - String correlationPrefix) - throws InterruptedException { - long deadline = System.nanoTime() - + Duration.ofSeconds(intEnv("MUSE_P1R_RAGFLOW_PARSE_WAIT_SECONDS", 120)).toNanos(); - KnowledgeRuntimeClient.RuntimeResult last = null; - int attempt = 1; - while (System.nanoTime() < deadline) { - last = client.pollDocumentStatuses(new KnowledgeRuntimeClient.PollDocumentStatusesCommand( - TENANT_ID, OWNER_USER_ID, 0L, datasetId, List.of(documentId), processingTaskId, - correlationPrefix + "-poll-" + attempt, hash(documentId + "-poll-" + attempt), attempt)); - assertSucceeded(last, "pollDocumentStatuses"); - if (last.documentStatuses().stream().anyMatch(status -> "searchable".equals(status.museStatus()))) { - return last; - } - Thread.sleep(2_000L); - attempt++; - } - throw new AssertionError("RAGFlow document parse 未在等待窗口内进入 searchable，last=" - + (last == null ? "none" : last.redactedSummary())); - } - - private void assertSucceeded(KnowledgeRuntimeClient.RuntimeResult result, String operation) { - assertNotNull(result, operation + " result 不能为空"); - assertEquals(KnowledgeRuntimeClient.Status.SUCCEEDED, result.status(), - () -> operation + " failed: " + result.redactedSummary()); - } - - private DbFacts readDbFacts(LiveSettings settings, LiveResult result) { - try (Connection connection = DriverManager.getConnection(settings.jdbcUrl(), settings.jdbcUser(), - settings.jdbcPassword())) { - return new DbFacts( - count(connection, "muse_knowledge_base", "id = ?", result.kbId()), - count(connection, "muse_knowledge_document", "id = ?", result.documentId()), - count(connection, "muse_knowledge_document_version", "document_id = ?", result.documentId()), - count(connection, "muse_knowledge_processing_task", "task_id = ?", result.scanTaskId()), - count(connection, "muse_knowledge_command", "command_id = ?", result.commandId()), - count(connection, "muse_knowledge_ragflow_binding", "kb_id = ?", result.kbId()), - count(connection, "muse_knowledge_ragflow_binding", "kb_id = ? AND document_id IS NULL", result.kbId()), - count(connection, "muse_knowledge_ragflow_binding", - "kb_id = ? AND document_version_id = (SELECT id FROM muse_knowledge_document_version WHERE document_id = ?)", - result.kbId(), result.documentId()), - count(connection, "muse_knowledge_ragflow_call", "command_id = ?", result.commandId()), - stringValue(connection, "SELECT kb_type FROM muse_knowledge_base WHERE id = ?", result.kbId()), - stringValue(connection, "SELECT status FROM muse_knowledge_base WHERE id = ?", result.kbId()), - intValue(connection, "SELECT active_version FROM muse_knowledge_base WHERE id = ?", result.kbId()), - stringValue(connection, "SELECT status FROM muse_knowledge_command WHERE command_id = ?", - result.commandId()), - stringValue(connection, """ - SELECT scan_status FROM muse_knowledge_document_version - WHERE document_id = ? - ORDER BY version DESC LIMIT 1 - """, result.documentId()), - stringValue(connection, """ - SELECT processing_status FROM muse_knowledge_document_version - WHERE document_id = ? - ORDER BY version DESC LIMIT 1 - """, result.documentId()), - stringValue(connection, """ - SELECT task_id FROM muse_knowledge_processing_task - WHERE task_id = ? - """, result.scanTaskId()), - stringValue(connection, """ - SELECT status FROM muse_knowledge_processing_task - WHERE task_id = ? - """, result.scanTaskId()), - stringValue(connection, """ - SELECT materialization_status FROM muse_knowledge_processing_task - WHERE task_id = ? - """, result.scanTaskId()), - stringValue(connection, """ - SELECT scan_status FROM muse_knowledge_processing_task - WHERE task_id = ? - """, result.scanTaskId()), - stringValue(connection, """ - SELECT parse_status FROM muse_knowledge_processing_task - WHERE task_id = ? - """, result.scanTaskId()), - stringValue(connection, """ - SELECT index_status FROM muse_knowledge_processing_task - WHERE task_id = ? - """, result.scanTaskId()), - stringValue(connection, """ - SELECT ragflow_dataset_id FROM muse_knowledge_processing_task - WHERE task_id = ? - """, result.scanTaskId()), - stringValue(connection, """ - SELECT ragflow_document_id FROM muse_knowledge_processing_task - WHERE task_id = ? - """, result.scanTaskId()), - booleanValue(connection, """ - SELECT started_at IS NOT NULL FROM muse_knowledge_processing_task - WHERE task_id = ? - """, result.scanTaskId()), - booleanValue(connection, """ - SELECT finished_at IS NOT NULL FROM muse_knowledge_processing_task - WHERE task_id = ? - """, result.scanTaskId()), - stringValue(connection, """ - SELECT result_summary::text FROM muse_knowledge_processing_task - WHERE task_id = ? - """, result.scanTaskId()), - booleanValue(connection, """ - SELECT EXISTS ( - SELECT 1 FROM muse_knowledge_ragflow_call - WHERE tenant_id = ? AND deleted = FALSE AND command_id = ? - AND operation = 'createDataset' AND status = 'succeeded' - ) - """, TENANT_ID, result.commandId()), - booleanValue(connection, """ - SELECT EXISTS ( - SELECT 1 FROM muse_knowledge_ragflow_call - WHERE tenant_id = ? AND deleted = FALSE AND command_id = ? - AND operation = 'uploadDocuments' AND status = 'succeeded' - ) - """, TENANT_ID, result.commandId()), - booleanValue(connection, """ - SELECT EXISTS ( - SELECT 1 FROM muse_knowledge_ragflow_call - WHERE tenant_id = ? AND deleted = FALSE AND command_id = ? - AND operation = 'startParseDocuments' AND status = 'accepted' - ) - """, TENANT_ID, result.commandId()), - stringValue(connection, """ - SELECT ragflow_dataset_id FROM muse_knowledge_ragflow_binding - WHERE tenant_id = ? AND deleted = FALSE AND kb_id = ? AND document_id IS NULL - ORDER BY id DESC LIMIT 1 - """, TENANT_ID, result.kbId()), - stringValue(connection, """ - SELECT ragflow_dataset_id FROM muse_knowledge_ragflow_binding - WHERE tenant_id = ? AND deleted = FALSE AND document_id = ? - ORDER BY id DESC LIMIT 1 - """, TENANT_ID, result.documentId()), - stringValue(connection, """ - SELECT ragflow_document_id FROM muse_knowledge_ragflow_binding - WHERE tenant_id = ? AND deleted = FALSE AND document_id = ? - ORDER BY id DESC LIMIT 1 - """, TENANT_ID, result.documentId()), - booleanValue(connection, """ - SELECT NOT EXISTS ( - SELECT 1 FROM muse_knowledge_ragflow_call - WHERE tenant_id = ? AND deleted = FALSE AND command_id = ? - AND (response_summary IS NULL OR response_summary = '{}'::jsonb) - ) - """, TENANT_ID, result.commandId())); - } catch (SQLException ex) { - throw new IllegalStateException("读取 P1R Knowledge live acceptance DB 事实失败", ex); - } - } - - private long count(Connection connection, String table, String where, Object... args) throws SQLException { - String sql = "SELECT COUNT(*) FROM " + table + " WHERE tenant_id = ? AND deleted = FALSE AND " + where; - try (PreparedStatement statement = connection.prepareStatement(sql)) { - statement.setLong(1, TENANT_ID); - bind(statement, 2, args); - try (ResultSet resultSet = statement.executeQuery()) { - assertTrue(resultSet.next(), "必须能读取计数: " + table); - return resultSet.getLong(1); - } - } - } - - private String stringValue(Connection connection, String sql, Object... args) throws SQLException { - try (PreparedStatement statement = connection.prepareStatement(sql)) { - bind(statement, 1, args); - try (ResultSet resultSet = statement.executeQuery()) { - return resultSet.next() ? resultSet.getString(1) : null; - } - } - } - - private int intValue(Connection connection, String sql, Object... args) throws SQLException { - try (PreparedStatement statement = connection.prepareStatement(sql)) { - bind(statement, 1, args); - try (ResultSet resultSet = statement.executeQuery()) { - return resultSet.next() ? resultSet.getInt(1) : 0; - } - } - } - - private boolean booleanValue(Connection connection, String sql, Object... args) throws SQLException { - try (PreparedStatement statement = connection.prepareStatement(sql)) { - bind(statement, 1, args); - try (ResultSet resultSet = statement.executeQuery()) { - return resultSet.next() && resultSet.getBoolean(1); - } - } - } - - private void bind(PreparedStatement statement, int startIndex, Object... args) throws SQLException { - for (int i = 0; i < args.length; i++) { - Object value = args[i]; - if (value instanceof Long longValue) { - statement.setLong(startIndex + i, longValue); - } else if (value instanceof Integer intValue) { - statement.setInt(startIndex + i, intValue); - } else { - statement.setString(startIndex + i, String.valueOf(value)); - } - } - } - - private ConfigurableApplicationContext liveContext(LiveSettings settings) { - Map properties = new LinkedHashMap<>(); - properties.put("muse.info.base-package", "cn.iocoder.muse.module.knowledge"); - properties.put("spring.datasource.url", settings.jdbcUrl()); - properties.put("spring.datasource.username", settings.jdbcUser()); - properties.put("spring.datasource.password", settings.jdbcPassword()); - properties.put("spring.datasource.driver-class-name", "org.postgresql.Driver"); - properties.put("spring.main.banner-mode", "off"); - properties.put("spring.main.lazy-initialization", "true"); - properties.put("mybatis-plus.global-config.db-config.id-type", "AUTO"); - properties.put("muse.knowledge.ragflow.base-url", settings.ragflowBaseUrl()); - properties.put("muse.knowledge.ragflow.api-key", settings.ragflowApiKey()); - properties.put("muse.knowledge.ragflow.timeout-seconds", settings.ragflowTimeoutSeconds()); - properties.put("muse.knowledge.ragflow.retry-budget", settings.ragflowRetryBudget()); - properties.put("muse.knowledge.ragflow.graphrag-attribution-ready", "false"); - - return new SpringApplicationBuilder(LiveAcceptanceConfiguration.class) - .web(WebApplicationType.NONE) - .initializers(applicationContext -> applicationContext.getEnvironment().getPropertySources() - .addFirst(new MapPropertySource("p1r-knowledge-live-acceptance", properties))) - .properties(properties) - .run(); - } - - private DatabaseEngineFacts migrateIsolatedTestDatabase(LiveSettings settings) { - silenceFlywayInfoLogs(); - assertSafeJdbcUrl(settings.jdbcUrl()); - DatabaseEngineFacts dbEngine = assertPostgresqlTestDatabase(settings); - System.out.println(JsonUtils.toJsonString(Map.of( - "acceptance", "p1r5-knowledge-runtime-e2e-live", - "preCleanDbEngine", dbEngineEvidence(dbEngine)))); - Flyway flyway = Flyway.configure() - .dataSource(settings.jdbcUrl(), settings.jdbcUser(), settings.jdbcPassword()) - .locations(resolveMuseSqlLocation(settings.flywayLocations())) - .schemas("public") - .defaultSchema("public") - .target(TARGET_VERSION) - .cleanDisabled(false) - .load(); - flyway.clean(); - flyway.migrate(); - MigrationInfo current = flyway.info().current(); - assertEquals(TARGET_VERSION, Objects.requireNonNull(current, "必须存在当前 Flyway 版本") - .getVersion().getVersion(), "P1R-5 Knowledge live acceptance 必须迁移到 V14 schema"); - return dbEngine; - } - - private DatabaseEngineFacts assertPostgresqlTestDatabase(LiveSettings settings) { - try (Connection connection = DriverManager.getConnection(settings.jdbcUrl(), settings.jdbcUser(), - settings.jdbcPassword()); - PreparedStatement statement = connection.prepareStatement("SELECT current_database(), version()"); - ResultSet resultSet = statement.executeQuery()) { - assertTrue(resultSet.next(), "Flyway clean 前必须能读取 PostgreSQL engine 信息"); - String databaseName = resultSet.getString(1); - String version = resultSet.getString(2); - assertTrue(StringUtils.hasText(databaseName) && databaseName.endsWith("_test"), - "Flyway clean 前 current_database() 必须是 _test 隔离库: " + databaseName); - assertTrue(StringUtils.hasText(version) && version.contains("PostgreSQL"), - "Flyway clean 前 version() 必须来自 PostgreSQL: " + versionSummary(version)); - return new DatabaseEngineFacts(databaseName, version); - } catch (SQLException ex) { - throw new IllegalStateException("Flyway clean 前检查 PostgreSQL _test 数据库失败: " - + maskedUrl(settings.jdbcUrl()), ex); - } - } - - private Map redactedEvidence(LiveSettings settings, LiveResult result, DbFacts facts, - ExternalVerification externalVerification, - DatabaseEngineFacts dbEngine) { - Map evidence = new LinkedHashMap<>(); - evidence.put("acceptance", "p1r5-knowledge-runtime-e2e-live"); - evidence.put("endpoint", endpointSummary(settings.ragflowBaseUrl(), RAGFLOW_API_PATH)); - evidence.put("apiKey", secretFingerprint(settings.ragflowApiKey())); - evidence.put("jdbcUrl", maskedUrl(settings.jdbcUrl())); - evidence.put("flywayTargetVersion", TARGET_VERSION); - evidence.put("dbEngine", dbEngineEvidence(dbEngine)); - evidence.put("request", requestEvidence(result, facts)); - evidence.put("response", responseEvidence(result, facts)); - evidence.put("db", dbEvidence(facts)); - evidence.put("externalVerification", externalEvidence(externalVerification)); - return evidence; - } - - private Map requestEvidence(LiveResult result, DbFacts facts) { - Map request = new LinkedHashMap<>(); - request.put("tenantId", TENANT_ID); - request.put("ownerUserId", OWNER_USER_ID); - request.put("kbId", result.kbId()); - request.put("commandIdSha256Prefix", sha256Prefix(result.commandId(), 12)); - request.put("documentId", result.documentId()); - request.put("processingTaskId", result.scanTaskId()); - request.put("entryContentLength", length(liveDocumentContent(result.suffix()))); - request.put("entryContentSha256Prefix", sha256Prefix(liveDocumentContent(result.suffix()), 12)); - request.put("ragflowDatasetId", facts.ragflowDatasetId()); - request.put("ragflowDocumentId", facts.ragflowDocumentId()); - return request; - } - - private Map responseEvidence(LiveResult result, DbFacts facts) { - Map response = new LinkedHashMap<>(); - response.put("uploadResponseProcessingStatus", "pending_scan"); - response.put("dbTaskStatus", facts.processingTaskStatus()); - response.put("dbParseStatus", facts.processingTaskParseStatus()); - response.put("dbIndexStatus", facts.processingTaskIndexStatus()); - response.put("commandStatus", facts.commandStatus()); - response.put("processingTaskStarted", facts.processingTaskStarted()); - response.put("processingTaskFinished", facts.processingTaskFinished()); - response.put("processingTaskResultSummaryLength", length(facts.processingTaskResultSummary())); - response.put("processingTaskResultSummarySha256Prefix", - sha256Prefix(defaultValue(facts.processingTaskResultSummary(), ""), 12)); - response.put("documentId", result.documentId()); - return response; - } - - private Map dbEvidence(DbFacts facts) { - Map db = new LinkedHashMap<>(); - db.put("knowledgeBaseCount", facts.knowledgeBaseCount()); - db.put("documentCount", facts.documentCount()); - db.put("documentVersionCount", facts.documentVersionCount()); - db.put("processingTaskCount", facts.processingTaskCount()); - db.put("commandCount", facts.commandCount()); - db.put("bindingCount", facts.bindingCount()); - db.put("datasetBindingCount", facts.datasetBindingCount()); - db.put("documentBindingCount", facts.documentBindingCount()); - db.put("ragflowCallCount", facts.ragflowCallCount()); - db.put("createDatasetSucceeded", facts.createDatasetSucceeded()); - db.put("uploadDocumentsSucceeded", facts.uploadDocumentsSucceeded()); - db.put("startParseAccepted", facts.startParseAccepted()); - db.put("allRagflowResponseSummariesNonEmpty", facts.allRagflowResponseSummariesNonEmpty()); - return db; - } - - private Map externalEvidence(ExternalVerification verification) { - Map evidence = new LinkedHashMap<>(); - evidence.put("scope", "external_verification_not_business_persisted"); - evidence.put("pollDocumentStatuses", operationSummary(verification.poll())); - evidence.put("documentStatuses", verification.poll().documentStatuses()); - evidence.put("retrieveChunks", operationSummary(verification.retrieval())); - evidence.put("retrievalChunksCount", verification.retrievalChunksCount()); - return evidence; - } - - private Map operationSummary(KnowledgeRuntimeClient.RuntimeResult result) { - Map summary = new LinkedHashMap<>(); - summary.put("operation", result.operation().wireName()); - summary.put("status", result.status()); - summary.put("failureClass", result.failureClass()); - summary.put("durationMillis", result.durationMillis()); - summary.put("externalIdPresent", result.externalId() != null && !result.externalId().isBlank()); - return summary; - } - - private Map dbEngineEvidence(DatabaseEngineFacts facts) { - Map engine = new LinkedHashMap<>(); - engine.put("databaseNameSha256Prefix", sha256Prefix(facts.databaseName(), 12)); - engine.put("databaseNameSuffix", facts.databaseName().endsWith("_test") ? "_test" : ""); - engine.put("versionSummary", versionSummary(facts.version())); - return engine; - } - - private Map endpointSummary(String baseUrl, String path) { - URI uri = URI.create(trimRight(baseUrl, "/") + path); - Map endpoint = new LinkedHashMap<>(); - endpoint.put("scheme", uri.getScheme()); - endpoint.put("host", uri.getHost()); - endpoint.put("port", uri.getPort()); - endpoint.put("path", uri.getPath()); - return endpoint; - } - - private Map secretFingerprint(String secret) { - Map fingerprint = new LinkedHashMap<>(); - fingerprint.put("length", secret == null ? 0 : secret.length()); - fingerprint.put("sha256Prefix", sha256Prefix(secret == null ? "" : secret, 12)); - return fingerprint; - } - - private int countFromRedactedSummary(String redactedSummary, String fieldName) { - JsonNode node = JsonUtils.parseTree(redactedSummary == null || redactedSummary.isBlank() ? "{}" : redactedSummary); - return node.has(fieldName) && node.path(fieldName).canConvertToInt() ? node.path(fieldName).asInt() : 0; - } - - private static String requiredPropertyOrEnv(String propertyName, String envName) { - String value = System.getProperty(propertyName); - if (!StringUtils.hasText(value)) { - value = System.getenv(envName); - } - assertTrue(StringUtils.hasText(value), "缺少必需系统属性或环境变量: " + propertyName + " / " + envName); - return value.trim(); - } - - private static String requiredEnv(String name) { - String value = System.getenv(name); - assertTrue(StringUtils.hasText(value), "缺少必需环境变量: " + name); - return value.trim(); - } - - private static String requiredPasswordEnvironment() { - String password = firstNonBlankEnvironment("P1R_FLYWAY_PASSWORD", "MUSE_POSTGRES_PASSWORD"); - assertTrue(password != null, "缺少必需数据库密码环境变量: P1R_FLYWAY_PASSWORD 或 MUSE_POSTGRES_PASSWORD"); - return password; - } - - private static String firstNonBlankEnvironment(String... names) { - // 数据库密码不允许从 system property 读取，避免被 Surefire XML 或命令历史记录。 - for (String name : names) { - String value = System.getenv(name); - if (StringUtils.hasText(value)) { - return value; - } - } - return null; - } - - private static int intEnv(String name, int defaultValue) { - String value = System.getenv(name); - if (!StringUtils.hasText(value)) { - return defaultValue; - } - return Integer.parseInt(value.trim()); - } - - private static String resolveMuseSqlLocation(String requestedLocations) { - assertEquals("filesystem:sql/muse", requestedLocations, - "P1R live acceptance 要求显式使用 filesystem:sql/muse"); - Path current = Path.of(System.getProperty("user.dir")).toAbsolutePath(); - for (Path cursor = current; cursor != null; cursor = cursor.getParent()) { - Path candidate = cursor.resolve("sql/muse"); - if (Files.isDirectory(candidate)) { - return "filesystem:" + candidate; - } - } - throw new IllegalStateException("无法从当前目录向上找到 sql/muse: " + current); - } - - private static void assertSafeJdbcUrl(String url) { - assertTrue(url.startsWith(POSTGRESQL_JDBC_PREFIX), - "JDBC URL 必须显式使用 jdbc:postgresql://，避免 Flyway clean 误连其它数据库: " + maskedUrl(url)); - URI uri = parseJdbcUriForAssertion(url); - assertEquals("postgresql", uri.getScheme(), - "JDBC URL 必须解析为 PostgreSQL scheme: " + maskedUrl(url)); - assertTrue(StringUtils.hasText(uri.getHost()), - "JDBC URL 必须显式包含 host: " + maskedUrl(url)); - assertFalse(StringUtils.hasText(uri.getRawUserInfo()), - "JDBC URL 不能携带 authority/userinfo，用户名走属性/环境变量，密码只走环境变量: " + maskedUrl(url)); - assertNoCredentialQuery(url); - String databaseName = jdbcDatabaseName(uri); - assertTrue(databaseName.endsWith("_test"), - "JDBC URL 必须指向 _test 后缀隔离库，避免清理非测试库: " + maskedUrl(url)); - assertFalse("muse_local".equals(databaseName), - "P1R live acceptance 禁止默认连接 muse_local"); - } - - private static URI parseJdbcUriForAssertion(String url) { - try { - return URI.create(url.substring(JDBC_URI_PREFIX.length())); - } catch (IllegalArgumentException ex) { - // JDK URI 解析异常会回显原始 authority/query；断言失败只允许输出脱敏后的 JDBC 摘要。 - throw new AssertionError("JDBC URL 必须可解析为 PostgreSQL URI: " + maskedUrl(url)); - } - } - - private static String jdbcDatabaseName(URI uri) { - String path = uri.getRawPath(); - assertTrue(StringUtils.hasText(path) && path.length() > 1, - "JDBC URL 必须包含数据库名"); - String databaseName = path.substring(path.lastIndexOf('/') + 1); - assertTrue(StringUtils.hasText(databaseName), - "JDBC URL 必须包含数据库名"); - return databaseName; - } - - private static void assertNoCredentialQuery(String url) { - int queryStart = url.indexOf('?'); - if (queryStart < 0) { - return; - } - String query = url.substring(queryStart + 1); - for (String parameter : query.split("&")) { - String key = parameter; - int equalsStart = key.indexOf('='); - if (equalsStart >= 0) { - key = key.substring(0, equalsStart); - } - assertFalse(isCredentialQueryKey(key), - "JDBC URL 不能携带凭据 query 参数；用户名走属性/环境变量，密码只走环境变量"); - } - } - - private static boolean isCredentialQueryKey(String rawKey) { - String key = rawKey.trim().toLowerCase(Locale.ROOT).replace('-', '_'); - return CREDENTIAL_QUERY_KEYS.contains(key) - || key.endsWith("_token") - || key.endsWith("_secret") - || key.endsWith("_password"); - } - - private static String maskedUrl(String url) { - if (!StringUtils.hasText(url)) { - return ""; - } - // 失败消息只能暴露 JDBC 类型、占位 authority、测试库名和 query 是否存在；解析失败也不能回显原始串。 - String trimmed = url.trim(); - if (!trimmed.startsWith(JDBC_URI_PREFIX)) { - return ""; - } - - String jdbcBody = trimmed.substring(JDBC_URI_PREFIX.length()); - String scheme = jdbcScheme(jdbcBody); - if (!StringUtils.hasText(scheme)) { - return JDBC_URI_PREFIX + ""; - } - if (!jdbcBody.regionMatches(true, scheme.length(), "://", 0, 3)) { - return JDBC_URI_PREFIX + scheme + ":"; - } - if ("postgresql".equals(scheme)) { - return maskedPostgresqlUrl(jdbcBody); - } - return maskedHierarchicalJdbcUrl(scheme, jdbcBody); - } - - private static String jdbcScheme(String jdbcBody) { - int schemeEnd = jdbcBody.indexOf(':'); - if (schemeEnd <= 0) { - return ""; - } - String scheme = jdbcBody.substring(0, schemeEnd).trim().toLowerCase(Locale.ROOT); - return scheme.matches("[a-z][a-z0-9+.-]*") ? scheme : ""; - } - - private static String maskedPostgresqlUrl(String jdbcBody) { - try { - URI uri = URI.create(jdbcBody); - String database = safeJdbcDatabaseName(uri.getRawPath(), ""); - String port = uri.getPort() < 0 ? "" : ":" + uri.getPort(); - String suffix = StringUtils.hasText(uri.getRawQuery()) ? "?" : ""; - return POSTGRESQL_JDBC_PREFIX + "" + port + "/" + database + suffix; - } catch (RuntimeException ignored) { - return POSTGRESQL_JDBC_PREFIX + "/" + fallbackJdbcDatabaseName(jdbcBody, "") - + querySuffix(jdbcBody); - } - } - - private static String maskedHierarchicalJdbcUrl(String scheme, String jdbcBody) { - try { - URI uri = URI.create(jdbcBody); - String database = safeJdbcDatabaseName(uri.getRawPath(), ""); - String databaseSuffix = StringUtils.hasText(database) ? "/" + database : ""; - String querySuffix = StringUtils.hasText(uri.getRawQuery()) ? "?" : ""; - return JDBC_URI_PREFIX + scheme + "://" + databaseSuffix + querySuffix; - } catch (RuntimeException ignored) { - String database = fallbackJdbcDatabaseName(jdbcBody, ""); - String databaseSuffix = StringUtils.hasText(database) ? "/" + database : ""; - return JDBC_URI_PREFIX + scheme + "://" + databaseSuffix + querySuffix(jdbcBody); - } - } - - private static String safeJdbcDatabaseName(String rawPath, String fallback) { - if (!StringUtils.hasText(rawPath) || rawPath.length() <= 1) { - return fallback; - } - String database = rawPath.substring(rawPath.lastIndexOf('/') + 1); - if (!StringUtils.hasText(database)) { - return fallback; - } - return database.matches("[A-Za-z0-9._%-]+") ? database : fallback; - } - - private static String fallbackJdbcDatabaseName(String jdbcBody, String fallback) { - int authorityStart = jdbcBody.indexOf("://"); - if (authorityStart < 0) { - return fallback; - } - int pathStart = jdbcBody.indexOf('/', authorityStart + 3); - if (pathStart < 0) { - return fallback; - } - int pathEnd = firstIndexOf(jdbcBody, pathStart + 1, '?', '#'); - String path = jdbcBody.substring(pathStart, pathEnd); - return safeJdbcDatabaseName(path, fallback); - } - - private static int firstIndexOf(String value, int start, char... candidates) { - int result = value.length(); - for (char candidate : candidates) { - int index = value.indexOf(candidate, start); - if (index >= 0 && index < result) { - result = index; - } - } - return result; - } - - private static String querySuffix(String value) { - return value.indexOf('?') < 0 ? "" : "?"; - } - - private static String versionSummary(String version) { - if (!StringUtils.hasText(version)) { - return ""; - } - String normalized = version.replaceAll("\\s+", " ").trim(); - return normalized.length() > 96 ? normalized.substring(0, 96) : normalized; - } - - private static void silenceFlywayInfoLogs() { - try { - Object flywayLogger = LoggerFactory.getLogger("org.flywaydb"); - Class levelClass = Class.forName("ch.qos.logback.classic.Level"); - Object warnLevel = levelClass.getField("WARN").get(null); - flywayLogger.getClass().getMethod("setLevel", levelClass).invoke(flywayLogger, warnLevel); - } catch (ReflectiveOperationException | LinkageError ignored) { - // 日志实现不是 logback 时不影响验收；测试自身仍只输出 masked URL 和摘要。 - } - } - - private static boolean externalAcceptanceEnabled() { - return "true".equalsIgnoreCase(System.getenv(ACCEPTANCE_ENV)); - } - - private static String hash(String value) { - return sha256Prefix(value == null ? "" : value, 64); - } - - private static String sha256Prefix(String value, int length) { - try { - MessageDigest digest = MessageDigest.getInstance("SHA-256"); - String hex = HexFormat.of().formatHex(digest.digest(value.getBytes(StandardCharsets.UTF_8))); - return hex.substring(0, Math.min(length, hex.length())); - } catch (NoSuchAlgorithmException e) { - throw new IllegalStateException("JDK 缺少 SHA-256 摘要算法", e); - } - } - - private static String trimRight(String value, String suffix) { - String result = value == null ? "" : value.trim(); - while (result.endsWith(suffix)) { - result = result.substring(0, result.length() - suffix.length()); - } - return result; - } - - private static String defaultValue(String value, String fallback) { - return value == null ? fallback : value; - } - - private static int length(String value) { - return value == null ? 0 : value.length(); - } - - private record LiveSettings(String jdbcUrl, - String jdbcUser, - String jdbcPassword, - String flywayLocations, - String ragflowBaseUrl, - String ragflowApiKey, - int ragflowTimeoutSeconds, - int ragflowRetryBudget) { - - private static LiveSettings fromEnvironment() { - String jdbcUrl = requiredPropertyOrEnv(JDBC_URL_PROPERTY, JDBC_URL_ENV); - assertSafeJdbcUrl(jdbcUrl); - return new LiveSettings( - jdbcUrl, - requiredPropertyOrEnv(JDBC_USER_PROPERTY, JDBC_USER_ENV), - requiredPasswordEnvironment(), - System.getProperty(FLYWAY_LOCATION_PROPERTY, "filesystem:sql/muse"), - requiredEnv("MUSE_KNOWLEDGE_RAGFLOW_BASE_URL"), - requiredEnv("MUSE_KNOWLEDGE_RAGFLOW_API_KEY"), - intEnv("MUSE_KNOWLEDGE_RAGFLOW_TIMEOUT_SECONDS", 30), - intEnv("MUSE_KNOWLEDGE_RAGFLOW_RETRY_BUDGET", 0)); - } - } - - private record LiveResult(Long kbId, - String commandId, - Long documentId, - String scanTaskId, - String suffix) { - } - - private record DbFacts(long knowledgeBaseCount, - long documentCount, - long documentVersionCount, - long processingTaskCount, - long commandCount, - long bindingCount, - long datasetBindingCount, - long documentBindingCount, - long ragflowCallCount, - String kbType, - String kbStatus, - int kbActiveVersion, - String commandStatus, - String documentVersionScanStatus, - String documentVersionProcessingStatus, - String processingTaskId, - String processingTaskStatus, - String materializationStatus, - String processingTaskScanStatus, - String processingTaskParseStatus, - String processingTaskIndexStatus, - String ragflowDatasetId, - String ragflowDocumentId, - boolean processingTaskStarted, - boolean processingTaskFinished, - String processingTaskResultSummary, - boolean createDatasetSucceeded, - boolean uploadDocumentsSucceeded, - boolean startParseAccepted, - String datasetBindingDatasetId, - String documentBindingDatasetId, - String documentBindingDocumentId, - boolean allRagflowResponseSummariesNonEmpty) { - } - - private record ExternalVerification(KnowledgeRuntimeClient.RuntimeResult poll, - KnowledgeRuntimeClient.RuntimeResult retrieval, - int retrievalChunksCount) { - } - - private record DatabaseEngineFacts(String databaseName, - String version) { - } - - @TestConfiguration - @Import({ - MuseDataSourceAutoConfiguration.class, - DataSourceAutoConfiguration.class, - DataSourceTransactionManagerAutoConfiguration.class, - MuseMybatisAutoConfiguration.class, - MybatisPlusAutoConfiguration.class, - MybatisPlusJoinAutoConfiguration.class, - MuseKnowledgeDocumentService.class, - MuseKnowledgeCommandService.class, - MuseKnowledgeAuditService.class, - MuseKnowledgeMaterializationService.class, - MuseKnowledgeProcessingTaskService.class, - SpringUtil.class - }) - static class LiveAcceptanceConfiguration { - - @Bean - KnowledgeRuntimeClient ragFlowKnowledgeRuntimeClient() { - return new HttpRagFlowKnowledgeRuntimeClient(requiredEnv("MUSE_KNOWLEDGE_RAGFLOW_BASE_URL"), - requiredEnv("MUSE_KNOWLEDGE_RAGFLOW_API_KEY"), - Duration.ofSeconds(intEnv("MUSE_KNOWLEDGE_RAGFLOW_TIMEOUT_SECONDS", 30)), - intEnv("MUSE_KNOWLEDGE_RAGFLOW_RETRY_BUDGET", 0), false); - } - - @Bean - @Primary - KnowledgeFileFacade p1rLiveKnowledgeFileFacade(ObjectProvider fileApiProvider) { - // 补足扫描服务后的新构造签名；本 IT override materializeText 走 test-only passed，不经 scanService 判定。 - return new KnowledgeFileFacade(fileApiProvider, - new cn.iocoder.muse.module.knowledge.application.muse.facade.KnowledgeContentScanService()) { - - @Override - public MaterializedFile materializeText(Long kbId, String fileName, String entryContent) { - byte[] content = entryContent.getBytes(StandardCharsets.UTF_8); - // test-only passed scanner 只替代“扫描服务已通过”这一外部前置条件；RAGFlow adapter 和 Muse DB 写入仍走真实主业务链路。 - return MaterializedFile.materialized("p1r-live-file-ref-" + kbId + "-" + sha256Prefix(fileName, 8), - fileName, (long) content.length, "text/plain; charset=UTF-8", - sha256Prefix(entryContent, 64), "passed", content); - } - }; - } - } -}