diff --git a/src/main/java/com/gxwebsoft/ai/config/WebSocketServer.java b/src/main/java/com/gxwebsoft/ai/config/WebSocketServer.java index cb1d647..5f302c0 100644 --- a/src/main/java/com/gxwebsoft/ai/config/WebSocketServer.java +++ b/src/main/java/com/gxwebsoft/ai/config/WebSocketServer.java @@ -1,88 +1,106 @@ -package com.gxwebsoft.ai.config; - -import org.springframework.stereotype.Component; - -import javax.websocket.OnClose; -import javax.websocket.OnOpen; -import javax.websocket.Session; -import javax.websocket.server.PathParam; -import javax.websocket.server.ServerEndpoint; -import java.io.IOException; -import java.util.concurrent.ConcurrentHashMap; - - -@ServerEndpoint(value = "/chat/{userId}") -@Component -public class WebSocketServer { - - /** - * concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。 - */ - private static ConcurrentHashMap webSocketMap = new ConcurrentHashMap<>(); - /** - * 与某个客户端的连接会话，需要通过它来给客户端发送数据 - */ - private Session session; - /** - * 接收userId - */ - private String userId = ""; - - /** - * 连接建立成功调用的方法 - */ - @OnOpen - public void onOpen(Session session, @PathParam("userId") String userId) { - this.session = session; - this.userId = userId; - if (webSocketMap.containsKey(userId)) { - webSocketMap.remove(userId); - webSocketMap.put(userId, this); - //加入set中 - } else { - webSocketMap.put(userId, this); - } - - try { - sendMessage(userId, "连接成功"); - } catch (IOException e) { - - } - } - - /** - * 连接关闭调用的方法 - */ - @OnClose - public void onClose() { - if (webSocketMap.containsKey(userId)) { - webSocketMap.remove(userId); - } - } - - - /** - * 实现服务器主动推送 - */ - public void sendMessage(String userId, String message) throws IOException { - System.out.println("userId:" + userId); - if (webSocketMap.containsKey(userId)) { - Session session1 = webSocketMap.get(userId).session; - System.out.println("session1:" + session1); - if (session1 != null) session1.getBasicRemote().sendText(message); - } - } - - - /** - * 实现服务器主动推送 - */ - public void sendAllMessage(String message) throws IOException { - ConcurrentHashMap.KeySetView userIds = webSocketMap.keySet(); - for (String userId : userIds) { - WebSocketServer webSocketServer = webSocketMap.get(userId); - webSocketServer.session.getBasicRemote().sendText(message); - } - } - -} +package com.gxwebsoft.ai.config; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Component; + +import javax.websocket.OnClose; +import javax.websocket.OnOpen; +import javax.websocket.Session; +import javax.websocket.server.PathParam; +import javax.websocket.server.ServerEndpoint; +import java.io.IOException; +import java.util.concurrent.ConcurrentHashMap; + + +@ServerEndpoint(value = "/chat/{userId}") +@Component +public class WebSocketServer { + + private static final Logger logger = LoggerFactory.getLogger(WebSocketServer.class); + + /** + * concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。 + */ + private static ConcurrentHashMap webSocketMap = new ConcurrentHashMap<>(); + /** + * 与某个客户端的连接会话，需要通过它来给客户端发送数据 + */ + private Session session; + /** + * 接收userId + */ + private String userId = ""; + + /** + * 连接建立成功调用的方法 + * + *

这里才是值得打日志的位置：一次连接一条，低频且信息量大 + * （排查「大纲/创意生成时端上收不到流式内容」时，第一件事就是确认端到底连没连上）。 + * 推送路径 {@link #sendMessage(String, String)} 相反，单次请求会被调用上千次，必须静默。 + */ + @OnOpen + public void onOpen(Session session, @PathParam("userId") String userId) { + this.session = session; + this.userId = userId; + if (webSocketMap.containsKey(userId)) { + webSocketMap.remove(userId); + webSocketMap.put(userId, this); + //加入set中 + } else { + webSocketMap.put(userId, this); + } + logger.info("websocket connected, userId={}", userId); + + try { + sendMessage(userId, "连接成功"); + } catch (IOException e) { + logger.warn("websocket 握手回应发送失败, userId={}", userId, e); + } + } + + /** + * 连接关闭调用的方法 + */ + @OnClose + public void onClose() { + if (webSocketMap.containsKey(userId)) { + webSocketMap.remove(userId); + logger.info("websocket closed, userId={}", userId); + } + } + + + /** + * 实现服务器主动推送 + * + *

两条 AI 流式链路（创意助手、活动策划大纲）**每个 SSE 事件**都会走到这里： + * 实测一次大纲生成 3062 个事件，即单次请求就会调 3000+ 次。 + * 因此这里**一行日志都不能打**——原先的两行 {@code System.out.println} + * （{@code userId} 与 {@code session1}）在控制台刷出上万行「userId:833」， + * 既刷屏、又绕过日志级别、还会淹没真正的告警。 + * + *

连接与断开另见 {@link #onOpen(Session, String)} / {@link #onClose()}，那里是低频的。 + */ + public void sendMessage(String userId, String message) throws IOException { + if (webSocketMap.containsKey(userId)) { + Session session1 = webSocketMap.get(userId).session; + if (session1 != null) { + session1.getBasicRemote().sendText(message); + } + } + } + + + /** + * 实现服务器主动推送 + */ + public void sendAllMessage(String message) throws IOException { + ConcurrentHashMap.KeySetView userIds = webSocketMap.keySet(); + for (String userId : userIds) { + WebSocketServer webSocketServer = webSocketMap.get(userId); + webSocketServer.session.getBasicRemote().sendText(message); + } + } + +} diff --git a/src/test/java/com/gxwebsoft/ai/config/WebSocketServerSilentPushTest.java b/src/test/java/com/gxwebsoft/ai/config/WebSocketServerSilentPushTest.java new file mode 100644 index 0000000..4457623 --- /dev/null +++ b/src/test/java/com/gxwebsoft/ai/config/WebSocketServerSilentPushTest.java @@ -0,0 +1,82 @@ +package com.gxwebsoft.ai.config; + +import org.junit.jupiter.api.Test; + +import javax.websocket.RemoteEndpoint; +import javax.websocket.Session; +import java.io.ByteArrayOutputStream; +import java.io.PrintStream; +import java.lang.reflect.Proxy; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.Callable; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * WebSocket 推送是**高频路径**，不得往标准输出打印。 + * + *

两条 AI 流式链路（创意助手、活动策划大纲）每个 SSE 事件都会调一次 + * {@link WebSocketServer#sendMessage(String, String)}：实测一次大纲生成 3062 个事件， + * 即单次请求就会调用 3000+ 次。最初这里有两行 {@code System.out.println} + * （{@code userId} 与 {@code session1}），于是一次生成就在控制台刷出上万行 + * 「userId:833」，既刷屏又绕过日志级别（见 .scratch/ai-assistant/issues/12）。 + * + *

两个分支（目标已连接 / 未连接）各锁一条：改回 {@code println} 立刻失败。 + */ +class WebSocketServerSilentPushTest { + + @Test + void 目标未连接时推送不得打印() throws Exception { + String output = captureStdout(() -> { + // 未连接的 userId：走"查不到会话"的分支，也正是刷屏时最常见的那条 + new WebSocketServer().sendMessage("silent-push-absent", "{\"answer\":\"测试\"}"); + return null; + }); + + assertEquals("", output, "推送是高频路径，不能刷标准输出"); + } + + @Test + void 目标已连接时推送也不得打印() throws Exception { + WebSocketServer server = new WebSocketServer(); + // onOpen 会打一条 INFO（连接日志，低频且应当保留），故放在捕获之外 + server.onOpen(fakeSession(), "silent-push-open"); + + String output = captureStdout(() -> { + server.sendMessage("silent-push-open", "{\"answer\":\"测试\"}"); + return null; + }); + + assertEquals("", output, "推送是高频路径，不能刷标准输出"); + } + + /** + * 捕获一段代码往标准输出写的内容。推送是 void 且抛受检异常，故用 {@link Callable}。 + */ + private static String captureStdout(Callable action) throws Exception { + PrintStream original = System.out; + ByteArrayOutputStream captured = new ByteArrayOutputStream(); + try { + System.setOut(new PrintStream(captured, true, StandardCharsets.UTF_8.name())); + action.call(); + } finally { + System.setOut(original); + } + return captured.toString(StandardCharsets.UTF_8.name()); + } + + /** + * 一个什么都不做的 {@link Session}：只为让 {@code sendMessage} 走进"已连接"分支。 + */ + private static Session fakeSession() { + return (Session) Proxy.newProxyInstance(Session.class.getClassLoader(), + new Class[]{Session.class}, (proxy, method, args) -> { + if ("getBasicRemote".equals(method.getName())) { + return Proxy.newProxyInstance(RemoteEndpoint.Basic.class.getClassLoader(), + new Class[]{RemoteEndpoint.Basic.class}, + (endpoint, call, callArgs) -> null); + } + return null; + }); + } +} diff --git a/src/test/java/com/gxwebsoft/ai/controller/AiCreativeRetryIntegrationTest.java b/src/test/java/com/gxwebsoft/ai/controller/AiCreativeRetryIntegrationTest.java new file mode 100644 index 0000000..4ed9b44 --- /dev/null +++ b/src/test/java/com/gxwebsoft/ai/controller/AiCreativeRetryIntegrationTest.java @@ -0,0 +1,220 @@ +package com.gxwebsoft.ai.controller; + +import com.alibaba.fastjson.JSONObject; +import com.gxwebsoft.ai.config.DifyProperties; +import com.gxwebsoft.common.core.web.ApiResult; +import com.sun.net.httpserver.HttpServer; +import org.junit.jupiter.api.Test; + +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * 创意助手自动重试的端到端验证（见 .scratch/ai-assistant/issues/11）。 + * + *

真起一个本地 HTTP server 扮演 Dify：第一次返回上游容量不足的失败流， + * 第二次返回成功流，然后确认 controller **真的重跑了工作流**并把第二次的产出回给端上。 + * + *

单元测试只能证明"判定函数返回 true"；这条用例证明的是"重试确实发生了、且结果被采用"， + * 也就是用户看到的"有时候能出图、有时候失败"是否真的被修掉。 + */ +class AiCreativeRetryIntegrationTest { + + /** + * 上游容量不足的原文（2026-10-13 内网实测，request_id 为真实值）。 + */ + private static final String CAPACITY_ERROR = + "An error occurred in the shaoruidong/volcengine_ai/volcengine_ai, error type:" + + " ToolProviderCredentialValidationError, error details: b'{\"code\":50500," + + "\"data\":{\"algorithm_base_resp\":{\"status_code\":110001,\"status_message\":" + + "\"no more instances to retry: last error: \"}},\"message\":\"Internal Error\"," + + "\"request_id\":\"202610131604021580A5E0F3C983DD3D63\",\"status\":50500}'"; + + /** + * 凭证失效：重试无用，必须只跑一次。 + */ + private static final String ACCESS_DENIED_ERROR = + "req_id: dc688258de PluginInvokeError: API request failed with status code 50400," + + " message: Access Denied"; + + @Test + void 上游容量不足_自动重跑一次并把第二次产出回给端上() throws Exception { + AtomicInteger hits = new AtomicInteger(); + try (MockDify dify = MockDify.start(exchange -> { + int attempt = hits.incrementAndGet(); + String stream = attempt == 1 ? capacityFailureStream() : succeededStream(); + MockDify.reply(exchange, stream); + })) { + ApiResult result = executeCreative(dify.baseUrl()); + + assertEquals(2, hits.get(), "第一次撞上上游容量不足后应自动重跑一次"); + assertEquals(0, result.getCode(), "重试成功后对端仍是正常返回"); + + Map payload = (Map) result.getData(); + assertEquals("succeeded", payload.get("status"), "第二次的状态应透出"); + assertEquals("迎新晚会海报已生成", payload.get("answer"), "第二次的产出应回给端上"); + } + } + + @Test + void 凭证失效_不重试且只跑一次() throws Exception { + AtomicInteger hits = new AtomicInteger(); + try (MockDify dify = MockDify.start(exchange -> { + hits.incrementAndGet(); + MockDify.reply(exchange, failureStream(ACCESS_DENIED_ERROR)); + })) { + ApiResult result = executeCreative(dify.baseUrl()); + + assertEquals(1, hits.get(), "凭证问题重试无用，不应重跑"); + Map payload = (Map) result.getData(); + assertEquals("failed", payload.get("status")); + assertEquals("生成失败，请稍后重试", payload.get("answer")); + assertTrue(!payload.containsKey("raw"), "失败态仍不得下发整包"); + } + } + + // ------------------------------------------------------------------ 辅助 + + private static ApiResult executeCreative(String baseUrl) throws Exception { + DifyProperties properties = new DifyProperties(); + properties.setBaseUrl(baseUrl); + properties.getApps().setCreativeAssistant("app-integration-test"); + + AiController controller = new AiController(); + Field propertiesField = AiController.class.getDeclaredField("difyProperties"); + propertiesField.setAccessible(true); + propertiesField.set(controller, properties); + + Map requestBody = new HashMap<>(); + requestBody.put("user_content", "生成一张迎新晚会海报"); + + Method method = AiController.class.getDeclaredMethod( + "executeCreativeAssistant", Map.class, Integer.class); + method.setAccessible(true); + return (ApiResult) method.invoke(controller, requestBody, 1); + } + + /** + * 模拟真实出图分支：前面有个成功的节点，紧接着「即梦AI-图像生成」节点失败。 + */ + private static String capacityFailureStream() { + return node("即梦AI-图像生成", "failed", CAPACITY_ERROR) + + failureBody(CAPACITY_ERROR); + } + + private static String failureStream(String error) { + return failureBody(error); + } + + private static String failureBody(String error) { + JSONObject data = new JSONObject(true); + data.put("status", "failed"); + data.put("outputs", new JSONObject(true)); + data.put("error", error); + + JSONObject finished = new JSONObject(true); + finished.put("event", "workflow_finished"); + finished.put("workflow_run_id", "run-1"); + finished.put("task_id", "task-1"); + finished.put("data", data); + return sse(finished); + } + + private static String succeededStream() { + JSONObject outputs = new JSONObject(true); + outputs.put("photo_message", "迎新晚会海报已生成"); + + JSONObject data = new JSONObject(true); + data.put("status", "succeeded"); + data.put("outputs", outputs); + + JSONObject finished = new JSONObject(true); + finished.put("event", "workflow_finished"); + finished.put("workflow_run_id", "run-2"); + finished.put("task_id", "task-2"); + finished.put("data", data); + return node("即梦AI-图像生成", "succeeded", null) + sse(finished); + } + + private static String node(String title, String status, String error) { + JSONObject data = new JSONObject(true); + data.put("title", title); + data.put("type", "tool"); + data.put("status", status); + if (error != null) { + data.put("error", error); + } + JSONObject node = new JSONObject(true); + node.put("event", "node_finished"); + node.put("data", data); + return sse(node); + } + + private static String sse(JSONObject payload) { + return "data: " + payload.toJSONString() + "\n\n"; + } + + /** + * 允许抛 IOException 的单次请求处理器（{@code Consumer} 不能抛受检异常）。 + */ + @FunctionalInterface + private interface ExchangeHandler { + + void handle(com.sun.net.httpserver.HttpExchange exchange) throws java.io.IOException; + } + + /** + * 扮演 Dify 的极简 SSE 服务端。 + */ + private static final class MockDify implements AutoCloseable { + + private final HttpServer server; + + private MockDify(HttpServer server) { + this.server = server; + } + + private static MockDify start(ExchangeHandler handler) throws Exception { + HttpServer server = HttpServer.create(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0), 0); + server.createContext("/v1/workflows/run", exchange -> { + drain(exchange); + handler.handle(exchange); + }); + server.start(); + return new MockDify(server); + } + + private String baseUrl() { + return "http://127.0.0.1:" + server.getAddress().getPort() + "/v1"; + } + + private static void reply(com.sun.net.httpserver.HttpExchange exchange, String body) throws java.io.IOException { + byte[] bytes = body.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().add("Content-Type", "text/event-stream"); + exchange.sendResponseHeaders(200, bytes.length); + exchange.getResponseBody().write(bytes); + exchange.close(); + } + + private static void drain(com.sun.net.httpserver.HttpExchange exchange) throws java.io.IOException { + byte[] buffer = new byte[1024]; + while (exchange.getRequestBody().read(buffer) != -1) { + // 丢掉请求体，确保客户端把请求写完 + } + } + + @Override + public void close() { + server.stop(0); + } + } +} diff --git a/src/test/java/com/gxwebsoft/ai/controller/AiCreativeRetryTest.java b/src/test/java/com/gxwebsoft/ai/controller/AiCreativeRetryTest.java new file mode 100644 index 0000000..426de46 --- /dev/null +++ b/src/test/java/com/gxwebsoft/ai/controller/AiCreativeRetryTest.java @@ -0,0 +1,124 @@ +package com.gxwebsoft.ai.controller; + +import com.alibaba.fastjson.JSONObject; +import com.gxwebsoft.common.core.utils.JSONUtil; +import org.junit.jupiter.api.Test; + +import java.lang.reflect.Method; +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.assertTrue; + +/** + * 上游瞬时故障的重试判定与失败文案（见 .scratch/ai-assistant/issues/11）。 + * + *

背景：内网实例上「即梦AI-图像生成」偶发 + * {@code {"code":50500,...,"status_code":110001,"status_message":"no more instances to retry"}}， + * 2026-10-13 同一 prompt 连跑 6 次有 2 次失败，整节点 3 秒即返回——是上游算力不足， + * 与内容、参数、凭证都无关，重试一次通常就成。 + * + *

而凭证失效、模型停用、审核拒绝重试无用，必须排除， + * 否则会把用户的一次失败拖成两次等待。 + */ +class AiCreativeRetryTest { + + /** + * 2026-10-13 内网实测的上游原文（保留关键结构，request_id 为真实值）。 + */ + private static final String CAPACITY_ERROR = + "An error occurred in the shaoruidong/volcengine_ai/volcengine_ai, please contact the" + + " author of shaoruidong/volcengine_ai/volcengine_ai for help, error type:" + + " ToolProviderCredentialValidationError, error details: b'{\"code\":50500," + + "\"data\":{\"algorithm_base_resp\":{\"status_code\":110001,\"status_message\":" + + "\"no more instances to retry: last error: \"},\"binary_data_base64\":[]}," + + "\"message\":\"Internal Error\"," + + "\"request_id\":\"202610131604021580A5E0F3C983DD3D63\",\"status\":50500}'"; + + private static final String ACCESS_DENIED_ERROR = + "req_id: dc688258de PluginInvokeError: [models] Error: API request failed with status" + + " code 50400, message: Access Denied"; + + private static final String MODEL_DISABLED_ERROR = + "req_id: dc688258de PluginInvokeError: Model disabled."; + + private static JSONObject failedEvent(String error) { + JSONObject data = new JSONObject(true); + data.put("status", "failed"); + data.put("outputs", new JSONObject(true)); + data.put("error", error); + + JSONObject event = new JSONObject(true); + event.put("event", "workflow_finished"); + event.put("workflow_run_id", "b2e18332-68bd-49e6-90e2-bd9d16d428ff"); + event.put("task_id", "0b1fdd88-cc6f-45b8-989d-be4f85bcff74"); + event.put("data", data); + return event; + } + + private static Map buildFailedResult(String error) throws Exception { + // 直接构造 JSONObject 作为报文体，别绕 JSONUtil.toJSONString——它按值序列化， + // 对字符串不加引号，拼出来的报文非法，parseObject 会静默返回 null。 + JSONObject responseData = failedEvent(error); + Method method = AiController.class.getDeclaredMethod( + "buildCreativeAssistantResult", Map.class, String.class); + method.setAccessible(true); + //noinspection unchecked + return (Map) method.invoke(new AiController(), responseData, ""); + } + + private static boolean retryable(String reason) throws Exception { + Method method = AiController.class.getDeclaredMethod("isRetryableCreativeFailure", String.class); + method.setAccessible(true); + return (Boolean) method.invoke(null, reason); + } + + @Test + void 上游容量不足_值得重试() throws Exception { + assertTrue(retryable(CAPACITY_ERROR), "no more instances to retry 属于瞬时故障，应重试"); + } + + @Test + void 上游容量不足_文案换成服务繁忙() throws Exception { + Map result = buildFailedResult(CAPACITY_ERROR); + assertEquals("failed", result.get("status"), "状态仍须可被判为失败"); + assertEquals("生成服务繁忙，请稍后重试", result.get("answer")); + } + + @Test + void 上游容量不足_失败态仍不得下发上游原文() throws Exception { + String json = JSONUtil.toJSONString(buildFailedResult(CAPACITY_ERROR)); + assertFalse(json.contains("50500"), "不得下发上游错误码：" + json); + assertFalse(json.contains("no more instances"), "不得下发上游错误原文：" + json); + assertFalse(json.contains("request_id"), "不得下发 req_id：" + json); + assertFalse(json.contains("volcengine"), "不得下发上游插件名：" + json); + assertFalse(json.contains("raw"), "失败态不得带 raw：" + json); + } + + @Test + void 凭证失效_不重试且文案不变() throws Exception { + assertFalse(retryable(ACCESS_DENIED_ERROR), "凭证问题重试无用"); + assertEquals("生成失败，请稍后重试", buildFailedResult(ACCESS_DENIED_ERROR).get("answer")); + } + + @Test + void 模型停用_不重试且文案不变() throws Exception { + assertFalse(retryable(MODEL_DISABLED_ERROR), "模型被停用重试无用"); + assertEquals("生成失败，请稍后重试", buildFailedResult(MODEL_DISABLED_ERROR).get("answer")); + } + + @Test + void 没有错误信息_不重试() throws Exception { + String blank = ""; + assertFalse(retryable(null), "错误信息缺失时无从判断，保守不重试"); + assertFalse(retryable(blank), "错误信息为空时不重试"); + } + + @Test + void 网关瞬时故障_值得重试() throws Exception { + assertTrue(retryable("502 Bad Gateway")); + assertTrue(retryable("connection reset by peer")); + assertTrue(retryable("Read timed out")); + } +}