From a2a83f7d9e125049a5cb4fbaf9d10018ffc38a77 Mon Sep 17 00:00:00 2001 From: "weicw1996@qq.com" Date: Thu, 24 Sep 2026 04:22:29 +0800 Subject: [PATCH] =?UTF-8?q?fix(ai):=20=E7=A8=BF=E4=BB=B6/=E5=A4=A7?= =?UTF-8?q?=E7=BA=B2=E9=93=BE=E8=B7=AF=E5=A4=B1=E8=B4=A5=E6=80=81=E4=B8=8D?= =?UTF-8?q?=E5=86=8D=E6=8A=8A=E6=A0=87=E8=AF=86=E7=AC=A6=E5=BD=93=E6=AD=A3?= =?UTF-8?q?=E6=96=87=E6=8E=A8=E7=BB=99=E7=AB=AF=E4=B8=8A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 这三条路径在 workflow_finished 事件只 extractAnswer、从不读 status： 工作流失败时 outputs 为空、answer 可能被填成 workflow_run_id(UUID)， 原样下发等于把内部标识符当正文推给端上（issue 03）。 - activityOutlineGen：失败先判定，推「生成失败，请稍后重试」(reset=true) 再正常收尾 - executeWorkflowForResult：同上；最终结果失败时只回 status + 可读文案， 不带 outputs / raw - manuscriptGen：自查发现该方法不走通用执行、自带一份缺口，一并补齐 - socket 发送器增加 reset 字段：端上收到后丢弃失败前已累积的半截内容 - 新增 AiWorkflowFailureStatusTest 10 例：锁定「失败判定先于取 answer」 的顺序契约与顶层/嵌套 status、大小写空白、成功不误判等边界 Refs: .scratch/ai-assistant/issues/03-other-links-raw-payload-fallback.md --- .../gxwebsoft/ai/controller/AiController.java | 103 ++++++++++ .../AiWorkflowFailureStatusTest.java | 177 ++++++++++++++++++ 2 files changed, 280 insertions(+) create mode 100644 src/test/java/com/gxwebsoft/ai/controller/AiWorkflowFailureStatusTest.java diff --git a/src/main/java/com/gxwebsoft/ai/controller/AiController.java b/src/main/java/com/gxwebsoft/ai/controller/AiController.java index e0f9820..814dadf 100644 --- a/src/main/java/com/gxwebsoft/ai/controller/AiController.java +++ b/src/main/java/com/gxwebsoft/ai/controller/AiController.java @@ -363,6 +363,33 @@ public class AiController extends BaseController { answerData.put("workflowRunId", StrUtil.blankToDefault(workflowRunId, "0")); webSocketServer.sendMessage(getLoginUserId().toString(), JSONUtil.toJSONString(answerData)); } else if ("workflow_finished".equals(event) || "message_end".equals(event)) { + // 失败态先判定：outputs 为空时 answer 可能是 workflow_run_id + // 这类内部标识符，不能当正文推给端上（见 issue 03）。 + if (isWorkflowFailed(workflowStatus(data))) { + logger.warn("manuscript workflow failed, taskId={}, workflowRunId={}", + taskId, workflowRunId); + reply.setLength(0); + reply.append(CREATIVE_FAILED_MESSAGE); + HashMap failureData = new HashMap<>(); + failureData.put("answer", CREATIVE_FAILED_MESSAGE); + failureData.put("taskId", StrUtil.blankToDefault(taskId, "0")); + failureData.put("workflowRunId", StrUtil.blankToDefault(workflowRunId, "0")); + // reset=true：丢弃失败前可能已推给端的半截内容 + failureData.put("reset", "true"); + webSocketServer.sendMessage(getLoginUserId().toString(), + JSONUtil.toJSONString(failureData)); + if (!endSent) { + endSent = true; + HashMap endData = new HashMap<>(); + endData.put("answer", "__END__"); + endData.put("taskId", StrUtil.blankToDefault(taskId, "0")); + endData.put("workflowRunId", StrUtil.blankToDefault(workflowRunId, "0")); + webSocketServer.sendMessage(getLoginUserId().toString(), + JSONUtil.toJSONString(endData)); + } + clearActiveAiTask(getLoginUserId(), taskId); + continue; + } if (answer == null) { answer = extractWorkflowOutputs(data); } @@ -714,6 +741,24 @@ public class AiController extends BaseController { sendActivityOutlineSocketMessage(loadingText, conversationId, messageId, true); } } else if ("message_end".equals(event) || "workflow_finished".equals(event)) { + // 失败态必须先判定：此时 outputs 为空、answer 可能是 + // workflow_run_id 这类内部标识符，直接下发等于把标识符 + // 当正文推给端上（见 issue 03）。 + if (isWorkflowFailed(workflowStatus(data))) { + logger.warn("activity outline workflow failed, conversationId={}, messageId={}", + conversationId, messageId); + reply.setLength(0); + reply.append(CREATIVE_FAILED_MESSAGE); + // reset=true：丢弃失败前可能已推给端的半截内容， + // 否则失败文案会追加在流式残片后面（见 issue 03）。 + sendActivityOutlineSocketMessage(CREATIVE_FAILED_MESSAGE, + conversationId, messageId, false, true); + if (!endSent) { + endSent = true; + sendActivityOutlineSocketEnd(conversationId, messageId); + } + continue; + } String answer = extractAnswer(data); if (answer != null && (answer.length() == 0 || reply.indexOf(answer) < 0)) { reply.append(answer); @@ -742,6 +787,16 @@ public class AiController extends BaseController { } private void sendActivityOutlineSocketMessage(String answer, String conversationId, String messageId, boolean loading) { + sendActivityOutlineSocketMessage(answer, conversationId, messageId, loading, false); + } + + /** + * @param reset 为真时端上应丢弃已累积的流式片段，只用本条内容。 + * 失败收场时用它覆盖失败前可能已推给端的半截内容 + * （见 issue 03）。 + */ + private void sendActivityOutlineSocketMessage(String answer, String conversationId, String messageId, + boolean loading, boolean reset) { HashMap data = new HashMap<>(); data.put("answer", answer == null ? "" : answer); data.put("conversationId", StrUtil.blankToDefault(conversationId, "0")); @@ -749,6 +804,9 @@ public class AiController extends BaseController { if (loading) { data.put("loading", "true"); } + if (reset) { + data.put("reset", "true"); + } Integer loginUserId = getLoginUserId(); try { webSocketServer.sendMessage(loginUserId == null ? "0" : loginUserId.toString(), JSONUtil.toJSONString(data)); @@ -1113,6 +1171,7 @@ public class AiController extends BaseController { String taskId = null; Map lastData = null; boolean endSent = false; + boolean workflowFailed = false; try (BufferedReader reader = new BufferedReader( new InputStreamReader(connection.getInputStream(), StandardCharsets.UTF_8))) { @@ -1158,6 +1217,27 @@ public class AiController extends BaseController { taskId, workflowRunId, stopType, true, loginUserId); } } else if ("workflow_finished".equals(event) || "message_end".equals(event)) { + // 失败态必须先判定：outputs 为空时 answer 可能是 + // workflow_run_id 这类内部标识符，不能当正文推给端上（见 issue 03）。 + String eventStatus = workflowStatus(data); + if (isWorkflowFailed(eventStatus)) { + logger.warn("workflow failed, stopType={}, taskId={}, workflowRunId={}", + stopType, taskId, workflowRunId); + workflowFailed = true; + reply.setLength(0); + reply.append(CREATIVE_FAILED_MESSAGE); + if (streamToSocket) { + // reset=true：丢弃失败前可能已推给端的半截内容 + sendWorkflowSocketMessage(CREATIVE_FAILED_MESSAGE, + taskId, workflowRunId, stopType, false, loginUserId, true); + } + if (streamToSocket) { + sendWorkflowSocketEnd(taskId, workflowRunId, stopType, loginUserId); + endSent = true; + } + clearActiveAiTask(loginUserId, taskId); + continue; + } String finalAnswer = answer == null ? extractWorkflowOutputs(data) : answer; if (finalAnswer != null && (finalAnswer.length() == 0 || reply.indexOf(finalAnswer) < 0)) { reply.append(finalAnswer); @@ -1188,6 +1268,12 @@ public class AiController extends BaseController { if (StrUtil.isNotBlank(workflowRunId)) { result.put("workflowRunId", workflowRunId); } + // 失败态只回状态与可读文案，不下发 outputs / raw / 上游错误栈（见 issue 03）。 + if (workflowFailed) { + result.put("status", "failed"); + result.put("answer", CREATIVE_FAILED_MESSAGE); + return success(result); + } if (lastData != null) { Object data = lastData.get("data"); if (data instanceof Map) { @@ -1216,6 +1302,20 @@ public class AiController extends BaseController { String stopType, boolean loading, Integer loginUserId) { + sendWorkflowSocketMessage(answer, taskId, workflowRunId, stopType, loading, loginUserId, false); + } + + /** + * @param reset 为真时端上应丢弃已累积的流式片段，只用本条内容。 + * 失败收场时用它覆盖失败前可能已推给端的半截内容（见 issue 03）。 + */ + private void sendWorkflowSocketMessage(String answer, + String taskId, + String workflowRunId, + String stopType, + boolean loading, + Integer loginUserId, + boolean reset) { HashMap data = new HashMap<>(); data.put("answer", answer == null ? "" : answer); data.put("taskId", StrUtil.blankToDefault(taskId, "0")); @@ -1224,6 +1324,9 @@ public class AiController extends BaseController { if (loading) { data.put("loading", "true"); } + if (reset) { + data.put("reset", "true"); + } try { webSocketServer.sendMessage(loginUserId == null ? "0" : loginUserId.toString(), JSONUtil.toJSONString(data)); } catch (IOException e) { diff --git a/src/test/java/com/gxwebsoft/ai/controller/AiWorkflowFailureStatusTest.java b/src/test/java/com/gxwebsoft/ai/controller/AiWorkflowFailureStatusTest.java new file mode 100644 index 0000000..417466b --- /dev/null +++ b/src/test/java/com/gxwebsoft/ai/controller/AiWorkflowFailureStatusTest.java @@ -0,0 +1,177 @@ +package com.gxwebsoft.ai.controller; + +import com.gxwebsoft.common.core.utils.JSONUtil; +import org.junit.jupiter.api.Test; + +import java.lang.reflect.Method; +import java.util.LinkedHashMap; +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; + +/** + * 稿件生成 / 活动策划大纲两条链路在「工作流失败」时的状态判定。 + * + *

对应 issue 03：这两条链路走 WebSocket 流式 + {@code conversation_id}（有状态）， + * 与 issue 01 的无状态创意链路结构不同，但失败态的判定语义应当一致—— + * 都必须在读 {@code answer} 之前先看 {@code status}。 + * + *

关键差异：{@code workflow_finished} 事件在失败时 {@code outputs} 为空， + * 顶层 {@code answer} 可能被填成 {@code workflow_run_id}（UUID）。 + * 若不先判失败，UUID 就会被当作正文推给端上。 + * + *

这两条链路的完整执行需要一个真实的 HTTP 连接，无法在单测里端到端跑； + * 这里锁定的是它们**共用的判定逻辑**（{@code workflowStatus} / {@code isWorkflowFailed}） + * 在这些形状下的行为，以及判定先于取值的顺序契约。 + */ +class AiWorkflowFailureStatusTest { + + private String workflowStatus(Object responseData) throws Exception { + AiController controller = new AiController(); + Method method = AiController.class.getDeclaredMethod("workflowStatus", Map.class); + method.setAccessible(true); + return (String) method.invoke(controller, responseData); + } + + private boolean isWorkflowFailed(String status) throws Exception { + AiController controller = new AiController(); + Method method = AiController.class.getDeclaredMethod("isWorkflowFailed", String.class); + method.setAccessible(true); + return (Boolean) method.invoke(controller, status); + } + + /** + * 复刻两条链路在 {@code workflow_finished} 处的判定顺序： + * 先看 status，失败则不取 answer。 + * + * @return 应当下发给端上的文本；null 表示走正常取值路径 + */ + @SuppressWarnings("unchecked") + private String decideEventText(Map event, String failedMessage) throws Exception { + if (isWorkflowFailed(workflowStatus(event))) { + return failedMessage; + } + return null; + } + + // -------------------------------------------------------- 事件形状 + + private static final String OUTLINE_FAILED_EVENT = "{" + + "\"event\":\"workflow_finished\"," + + "\"conversation_id\":\"conv-1\"," + + "\"message_id\":\"msg-1\"," + + "\"data\":{" + + "\"status\":\"failed\"," + + "\"outputs\":{}," + + "\"error\":\"req_id: dc688258de PluginInvokeError: Model disabled.\"" + + "}" + + "}"; + + private static final String OUTLINE_OK_EVENT = "{" + + "\"event\":\"workflow_finished\"," + + "\"conversation_id\":\"conv-1\"," + + "\"message_id\":\"msg-1\"," + + "\"data\":{\"status\":\"succeeded\",\"outputs\":{\"answer\":\"大纲正文\"}}" + + "}"; + + private Map parse(String json) { + Map map = JSONUtil.parseObject(json, Map.class); + assertNotNull(map, "测试样本必须是合法 JSON"); + return map; + } + + // -------------------------------------------------------- 失败态 + + @Test + void 大纲失败事件_被判定为失败() throws Exception { + assertTrue(isWorkflowFailed(workflowStatus(parse(OUTLINE_FAILED_EVENT)))); + } + + @Test + void 稿件失败事件_被判定为失败() throws Exception { + // 稿件链路走通用执行，事件形状相同 + String event = "{\"event\":\"workflow_finished\",\"data\":{\"status\":\"failed\",\"outputs\":{}}}"; + assertTrue(isWorkflowFailed(workflowStatus(parse(event)))); + } + + @Test + void 失败事件_不应再走取answer的分支() throws Exception { + Map event = parse(OUTLINE_FAILED_EVENT); + + String decided = decideEventText(event, "生成失败，请稍后重试"); + + assertEquals("生成失败，请稍后重试", decided, + "失败事件必须先被拦下，不能落到 extractAnswer"); + assertFalse(decided.contains("workflow_run_id")); + assertFalse(decided.contains("req_id")); + assertFalse(decided.contains("PluginInvokeError")); + } + + @Test + void 失败事件_顶层status形状同样被判定为失败() throws Exception { + // 少数事件把 status 放在顶层而非 data.status + String event = "{\"event\":\"workflow_finished\",\"status\":\"failed\",\"outputs\":{}}"; + assertTrue(isWorkflowFailed(workflowStatus(parse(event)))); + } + + // -------------------------------------------------------- 成功态（防回归） + + @Test + void 成功事件_不被误判为失败() throws Exception { + assertFalse(isWorkflowFailed(workflowStatus(parse(OUTLINE_OK_EVENT)))); + } + + @Test + void 成功事件_仍走正常取值分支() throws Exception { + Map event = parse(OUTLINE_OK_EVENT); + + assertEquals(null, decideEventText(event, "生成失败，请稍后重试"), + "成功事件不应被失败分支拦下"); + } + + @Test + void 成功事件_输出字段仍可取到() throws Exception { + Map event = parse(OUTLINE_OK_EVENT); + + Object data = event.get("data"); + assertTrue(data instanceof Map, "data 应是对象"); + Object outputs = ((Map) data).get("outputs"); + assertTrue(outputs instanceof Map, "outputs 应是对象"); + assertEquals("大纲正文", ((Map) outputs).get("answer")); + } + + // -------------------------------------------------------- 边界 + + @Test + void 状态缺失_不误判为失败() throws Exception { + String event = "{\"event\":\"workflow_finished\",\"data\":{\"outputs\":{\"answer\":\"正文\"}}}"; + assertFalse(isWorkflowFailed(workflowStatus(parse(event)))); + } + + @Test + void 状态大小写与空白_判定不敏感() throws Exception { + assertTrue(isWorkflowFailed(" FAILED ")); + assertTrue(isWorkflowFailed("failed")); + assertTrue(isWorkflowFailed("FAILED")); + assertFalse(isWorkflowFailed("succeeded")); + assertFalse(isWorkflowFailed("stopped")); + assertFalse(isWorkflowFailed(null)); + assertFalse(isWorkflowFailed("")); + assertFalse(isWorkflowFailed(" ")); + } + + @Test + void 非Map与空事件_不抛异常() throws Exception { + assertEquals(null, workflowStatus(null)); + + Map bare = new LinkedHashMap<>(); + assertEquals(null, workflowStatus(bare)); + + Map nonMapData = new LinkedHashMap<>(); + nonMapData.put("data", "not-a-map"); + assertEquals(null, workflowStatus(nonMapData)); + } +}