xieyinlu/ruoyi-ai
1
0
Fork 0
mirror of https://gitcode.com/ageerle/ruoyi-ai.git synced 2026-09-13 00:14:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(sse): SSE 改为按会话隔离 + 修 trace 租户 bug + 清理冗余

Browse Source 
SSE 串台修复：
- SseEmitterManager 新增按 sessionId 维度的 connect/sendEvent/disconnect，
  每个会话一个 SSE 连接，替代原 userId+token 维度（同用户多会话串台）
- SseMessageDto 增加 sessionId + eventDto 字段，跨实例按会话路由
- SseTopicListener 优先按 sessionId 路由，回退原 userId/群发逻辑
- SseMessageUtils 增加 sessionId 重载，对话链路全部切换
- ChatServiceFacade / MyMcpClientListener / WorkflowStarter 切到 sessionId
- 通知/全局 /sse 端点保留 userId 模式（通知按用户）

trace 租户 bug 修复：
- trace_run / trace_node 加入 tenant.excludes，监控表跨租户全局可见，
  绕过异步线程租户上下文不传播导致 node 查不到的问题

冗余清理：
- 删除零使用的 @TraceNode 注解 + TraceNodeAspect 切面 + 对应单测
- TraceProperties 删除 recordDetail/maxInputLength/maxOutputLength
  （仅对已删切面生效，编程式埋点用 RagTracePayloadBuilder 不受影响）
- TracePayloadUtils 删除 input/output/asString 方法
- TraceConstants 删除无引用的 NODE_METHOD/HTTP/DB/CACHE/TASK/STREAM
- TraceContext 删除预留的 copyNodeStack
- TraceNodeVo 删除未用的 children 字段
- TraceRecordServiceImpl.getDetail 删除重复 enrich
- application.yml 删除 trace.payload 无效配置项
- RagTraceNodeTypes 删除无引用的 NODE_STREAM

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
ageerle
2026-07-23 15:35:10 +08:00
parent 11bb1dba0f
commit d1a820728f
21 changed files with 259 additions and 405 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
ruoyi-admin/src/main/resources/application.yml
View File

@@ -146,6 +146,9 @@ tenant:
- sys_client
- sys_oss_config
- flow_spel
# 链路追踪监控表：运维需跨租户全局查看，且 trace_node 在异步线程写入、租户上下文不传播，故排除租户过滤
- trace_run
- trace_node
# MyBatisPlus配置
# https://baomidou.com/config/
@@ -238,15 +241,6 @@ trace:
# 关闭后所有埋点代码会直接透传业务逻辑，不写库、不创建上下文，零性能开销
enabled: true
payload:
# 是否记录方法真实入参和返回值，默认 false（仅存摘要）
# 仅对 @TraceNode 注解方式生效，编程式埋点使用 TracePayloadBuilder 手动构建的摘要不受此开关影响
# true → 自动序列化原始入参/返回值，截断后写入 input_payload / output_payload
# false → 只存 nodeName / 耗时 / 状态等元信息，input_payload / output_payload 为空
record-detail: false
# input payload 最大字符长度，超过部分会被截断丢弃
max-input-length: 1000
# output payload 最大字符长度，超过部分会被截断丢弃
max-output-length: 2000
# 错误信息最大字符长度（格式: "异常类名: 异常消息"），超过部分会被截断丢弃
max-error-length: 1000

125
ruoyi-common/ruoyi-common-sse/src/main/java/org/ruoyi/common/sse/core/SseEmitterManager.java
View File

@@ -32,6 +32,16 @@ public class SseEmitterManager {
*/
private final static String SSE_TOPIC = "global:sse";
/**
* 按会话维度管理：每个会话一个 SSE 连接，用于对话流式响应
* Key: sessionId
*/
private final static Map<String, SseEmitter> SESSION_EMITTERS = new ConcurrentHashMap<>();
/**
* 按用户维度管理：全局通知、站内信等场景，一个用户可有多个连接（按 token 区分）
* Key: userId, Value: token -> SseEmitter
*/
private final static Map<Long, Map<String, SseEmitter>> USER_TOKEN_EMITTERS = new ConcurrentHashMap<>();
public SseEmitterManager() {
@@ -40,6 +50,98 @@ public class SseEmitterManager {
.scheduleWithFixedDelay(this::sseMonitor, 60L, 60L, TimeUnit.SECONDS);
}
// ======================== 会话维度（对话流式响应） ========================
/**
* 建立与指定会话的 SSE 连接，每个会话仅保留一个连接，重复建连会替换旧连接。
*
* @param sessionId 会话ID
* @return SseEmitter 实例
*/
public SseEmitter connect(String sessionId) {
if (sessionId == null) {
throw new IllegalArgumentException("sessionId 不能为空");
}
// 关闭已存在的 SseEmitter，保证每个会话只有一个活跃连接
SseEmitter oldEmitter = SESSION_EMITTERS.remove(sessionId);
if (oldEmitter != null) {
oldEmitter.complete();
}
SseEmitter emitter = new SseEmitter(86400000L);
SESSION_EMITTERS.put(sessionId, emitter);
emitter.onCompletion(() -> removeSessionEmitter(sessionId, emitter));
emitter.onTimeout(() -> removeSessionEmitter(sessionId, emitter));
emitter.onError(e -> removeSessionEmitter(sessionId, emitter));
try {
emitter.send(SseEmitter.event().comment("connected"));
} catch (IOException e) {
SESSION_EMITTERS.remove(sessionId);
}
return emitter;
}
/**
* 断开指定会话的 SSE 连接
*
* @param sessionId 会话ID
*/
public void disconnect(String sessionId) {
if (sessionId == null) {
return;
}
SseEmitter emitter = SESSION_EMITTERS.remove(sessionId);
if (emitter != null) {
try {
emitter.send(SseEmitter.event().comment("disconnected"));
} catch (Exception exception) {
log.error(exception.getMessage());
}
emitter.complete();
}
}
/**
* 向指定会话发送结构化事件
*
* @param sessionId 会话ID
* @param eventDto SSE事件对象
*/
public void sendEvent(String sessionId, SseEventDto eventDto) {
if (sessionId == null || eventDto == null) {
return;
}
SseEmitter emitter = SESSION_EMITTERS.get(sessionId);
if (emitter == null) {
log.warn("【SSE发送失败】sessionId: {} 没有活跃的SSE连接", sessionId);
return;
}
try {
log.debug("【SSE发送】sessionId: {}, event: {}", sessionId, eventDto.getEvent());
emitter.send(SseEmitter.event()
.name(eventDto.getEvent())
.data(JSONUtil.toJsonStr(eventDto)));
} catch (Exception e) {
log.error("【SSE发送失败】sessionId: {}, error: {}", sessionId, e.getMessage());
removeSessionEmitter(sessionId, emitter);
}
}
private void removeSessionEmitter(String sessionId, SseEmitter emitter) {
boolean removed = SESSION_EMITTERS.remove(sessionId, emitter);
if (removed) {
try {
emitter.complete();
} catch (Exception ignore) {
// 忽略重复关闭异常
}
}
}
// ======================== 用户维度（全局通知） ========================
/**
* 建立与指定用户的 SSE 连接
*
@@ -154,6 +256,23 @@ public class SseEmitterManager {
// 循环结束后统一清理空用户，避免并发修改异常
toRemoveUsers.forEach(USER_TOKEN_EMITTERS::remove);
// 会话维度心跳：发送失败的连接移除
if (!SESSION_EMITTERS.isEmpty()) {
SESSION_EMITTERS.entrySet().removeIf(entry -> {
try {
entry.getValue().send(heartbeat);
return false;
} catch (Exception ex) {
try {
entry.getValue().complete();
} catch (Exception ignore) {
// 忽略重复关闭异常
}
return true;
}
});
}
}
/**
@@ -243,9 +362,11 @@ public class SseEmitterManager {
SseMessageDto broadcastMessage = new SseMessageDto();
broadcastMessage.setMessage(sseMessageDto.getMessage());
broadcastMessage.setUserIds(sseMessageDto.getUserIds());
broadcastMessage.setSessionId(sseMessageDto.getSessionId());
broadcastMessage.setEventDto(sseMessageDto.getEventDto());
RedisUtils.publish(SSE_TOPIC, broadcastMessage, consumer -> {
log.info("SSE发送主题订阅消息topic:{} session keys:{} message:{}",
SSE_TOPIC, sseMessageDto.getUserIds(), sseMessageDto.getMessage());
log.info("SSE发送主题订阅消息topic:{} session:{} session keys:{} message:{}",
SSE_TOPIC, sseMessageDto.getSessionId(), sseMessageDto.getUserIds(), sseMessageDto.getMessage());
});
}

10
ruoyi-common/ruoyi-common-sse/src/main/java/org/ruoyi/common/sse/dto/SseMessageDto.java
View File

@@ -26,4 +26,14 @@ public class SseMessageDto implements Serializable {
* 需要发送的消息
*/
private String message;
/**
* 按会话定向推送的会话ID（非空时优先按会话路由，忽略 userIds）
*/
private String sessionId;
/**
* 结构化事件（按会话定向推送时使用，message 为兼容旧逻辑保留）
*/
private SseEventDto eventDto;
}

18
ruoyi-common/ruoyi-common-sse/src/main/java/org/ruoyi/common/sse/listener/SseTopicListener.java
View File

@@ -1,8 +1,10 @@
package org.ruoyi.common.sse.listener;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import lombok.extern.slf4j.Slf4j;
import org.ruoyi.common.sse.core.SseEmitterManager;
import org.ruoyi.common.sse.dto.SseMessageDto;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
@@ -28,8 +30,20 @@ public class SseTopicListener implements ApplicationRunner, Ordered {
@Override
public void run(ApplicationArguments args) throws Exception {
sseEmitterManager.subscribeMessage((message) -> {
log.info("SSE主题订阅收到消息session keys={} message={}", message.getUserIds(), message.getMessage());
// 如果key不为空就按照key发消息 如果为空就群发
log.info("SSE主题订阅收到消息session:{} session keys={} message={}",
message.getSessionId(), message.getUserIds(), message.getMessage());
// 优先按会话路由（对话流式响应）
if (StrUtil.isNotBlank(message.getSessionId())) {
if (message.getEventDto() != null) {
sseEmitterManager.sendEvent(message.getSessionId(), message.getEventDto());
} else if (message.getMessage() != null) {
// 兼容按会话发纯文本的场景
sseEmitterManager.sendEvent(message.getSessionId(),
org.ruoyi.common.sse.dto.SseEventDto.content(message.getMessage()));
}
return;
}
// 否则按用户/群发路由（全局通知）
if (CollUtil.isNotEmpty(message.getUserIds())) {
message.getUserIds().forEach(key -> {
sseEmitterManager.sendMessage(key, message.getMessage());

64
ruoyi-common/ruoyi-common-sse/src/main/java/org/ruoyi/common/sse/utils/SseMessageUtils.java
View File

@@ -93,6 +93,15 @@ public class SseMessageUtils {
MANAGER.disconnect(userId, tokenValue);
}
/**
* 完成指定会话的SSE连接
*
* @param sessionId 会话ID
*/
public static void completeConnection(String sessionId) {
MANAGER.disconnect(sessionId);
}
/**
* 向指定的SSE会话发送结构化事件
*
@@ -106,6 +115,22 @@ public class SseMessageUtils {
MANAGER.sendEvent(userId, eventDto);
}
/**
* 向指定会话发送结构化事件（通过 Redis 广播，跨实例可达）
*
* @param sessionId 会话ID
* @param eventDto SSE事件对象
*/
public static void sendEvent(String sessionId, SseEventDto eventDto) {
if (!isEnable() || sessionId == null) {
return;
}
SseMessageDto dto = new SseMessageDto();
dto.setSessionId(sessionId);
dto.setEventDto(eventDto);
MANAGER.publishMessage(dto);
}
/**
* 发送内容事件
*
@@ -116,6 +141,16 @@ public class SseMessageUtils {
sendEvent(userId, SseEventDto.content(content));
}
/**
* 向指定会话发送内容事件
*
* @param sessionId 会话ID
* @param content 内容
*/
public static void sendContent(String sessionId, String content) {
sendEvent(sessionId, SseEventDto.content(content));
}
/**
* 发送推理内容事件
*
@@ -126,6 +161,16 @@ public class SseMessageUtils {
sendEvent(userId, SseEventDto.reasoning(reasoningContent));
}
/**
* 向指定会话发送推理内容事件
*
* @param sessionId 会话ID
* @param reasoningContent 推理内容
*/
public static void sendReasoning(String sessionId, String reasoningContent) {
sendEvent(sessionId, SseEventDto.reasoning(reasoningContent));
}
/**
* 发送完成事件
*
@@ -135,6 +180,15 @@ public class SseMessageUtils {
sendEvent(userId, SseEventDto.done());
}
/**
* 向指定会话发送完成事件
*
* @param sessionId 会话ID
*/
public static void sendDone(String sessionId) {
sendEvent(sessionId, SseEventDto.done());
}
/**
* 发送错误事件
*
@@ -145,6 +199,16 @@ public class SseMessageUtils {
sendEvent(userId, SseEventDto.error(error));
}
/**
* 向指定会话发送错误事件
*
* @param sessionId 会话ID
* @param error 错误信息
*/
public static void sendError(String sessionId, String error) {
sendEvent(sessionId, SseEventDto.error(error));
}
/**
* 是否开启
*/

31
ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/annotation/TraceNode.java
View File

@@ -1,31 +0,0 @@
package org.ruoyi.common.trace.annotation;
import org.ruoyi.common.trace.constant.TraceConstants;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
/**
* Trace 方法节点标记。
*/
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface TraceNode {
/**
* 节点名称。
*/
String name() default "";
/**
* 节点类型。
*/
String type() default TraceConstants.NODE_METHOD;
/**
* 安全输入摘要。
*/
String input() default "";
}

89
ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/aspect/TraceNodeAspect.java
View File

@@ -1,89 +0,0 @@
package org.ruoyi.common.trace.aspect;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.reflect.MethodSignature;
import org.ruoyi.common.trace.annotation.TraceNode;
import org.ruoyi.common.trace.config.TraceProperties;
import org.ruoyi.common.trace.constant.TraceConstants;
import org.ruoyi.common.trace.core.TraceContext;
import org.ruoyi.common.trace.service.TraceRecordService;
import org.ruoyi.common.trace.util.TracePayloadUtils;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;
import java.lang.reflect.Method;
import java.util.Date;
/**
* Trace 方法节点采集切面。
*/
@Slf4j
@Aspect
@Order(Ordered.HIGHEST_PRECEDENCE + 10)
@RequiredArgsConstructor
public class TraceNodeAspect {
private final TraceRecordService traceRecordService;
private final TraceProperties traceProperties;
@Around("@annotation(traceNode)")
public Object aroundNode(ProceedingJoinPoint joinPoint, TraceNode traceNode) throws Throwable {
if (!traceProperties.isEnabled() || StrUtil.isBlank(TraceContext.getTraceId())) {
return joinPoint.proceed();
}
Method method = ((MethodSignature) joinPoint.getSignature()).getMethod();
String nodeId = IdUtil.getSnowflakeNextIdStr();
String traceId = TraceContext.getTraceId();
long startMillis = System.currentTimeMillis();
Date startTime = new Date();
org.ruoyi.common.trace.domain.TraceNode node = new org.ruoyi.common.trace.domain.TraceNode();
node.setTraceId(traceId);
node.setNodeId(nodeId);
node.setParentNodeId(TraceContext.currentNodeId());
node.setDepth(TraceContext.depth());
node.setNodeName(StrUtil.blankToDefault(traceNode.name(), method.getName()));
node.setNodeType(StrUtil.blankToDefault(traceNode.type(), TraceConstants.NODE_METHOD));
node.setClassName(method.getDeclaringClass().getName());
node.setMethodName(method.getName());
node.setStatus(TraceConstants.STATUS_RUNNING);
node.setStartTime(startTime);
node.setInputPayload(StrUtil.isBlank(traceNode.input()) ? null : traceNode.input());
try {
traceRecordService.startNode(node);
} catch (Exception ex) {
log.warn("写入 trace 节点失败，traceId={}", traceId, ex);
return joinPoint.proceed();
}
TraceContext.pushNode(nodeId);
try {
Object result = joinPoint.proceed();
safeFinishNode(traceId, nodeId, TraceConstants.STATUS_SUCCESS, null,
TracePayloadUtils.output(result, traceProperties), startMillis);
return result;
} catch (Throwable ex) {
safeFinishNode(traceId, nodeId, TraceConstants.STATUS_ERROR,
TracePayloadUtils.error(ex, traceProperties), null, startMillis);
throw ex;
} finally {
TraceContext.popNode();
}
}
private void safeFinishNode(String traceId, String nodeId, String status, String errorMessage, String outputPayload, long startMillis) {
try {
traceRecordService.finishNode(traceId, nodeId, status, errorMessage, outputPayload,
new Date(), System.currentTimeMillis() - startMillis);
} catch (Exception ex) {
log.warn("更新 trace 节点失败，traceId={}, nodeId={}", traceId, nodeId, ex);
}
}
}

14
ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/config/TraceAutoConfiguration.java
View File

@@ -1,24 +1,14 @@
package org.ruoyi.common.trace.config;
import org.ruoyi.common.trace.aspect.TraceNodeAspect;
import org.ruoyi.common.trace.service.TraceRecordService;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
/**
* 通用链路追踪自动配置。
* <p>
* 仅注册配置属性；节点采集通过 {@code TraceNodeTemplate} / {@code DefaultTraceStreamSpan} 编程式埋点完成。
*/
@AutoConfiguration
@EnableConfigurationProperties(TraceProperties.class)
public class TraceAutoConfiguration {
@Bean
@ConditionalOnMissingBean
@ConditionalOnProperty(prefix = "trace", name = "enabled", havingValue = "true", matchIfMissing = true)
public TraceNodeAspect traceNodeAspect(TraceRecordService traceRecordService, TraceProperties traceProperties) {
return new TraceNodeAspect(traceRecordService, traceProperties);
}
}

15
ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/config/TraceProperties.java
View File

@@ -23,21 +23,6 @@ public class TraceProperties {
@Data
public static class Payload {
/**
* 是否记录截断后的详情。默认 false，只记录摘要。
*/
private boolean recordDetail = false;
/**
* input payload 最大长度。
*/
private int maxInputLength = 1000;
/**
* output payload 最大长度。
*/
private int maxOutputLength = 2000;
/**
* 错误信息最大长度。
*/

7
ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/constant/TraceConstants.java
View File

@@ -12,11 +12,4 @@ public final class TraceConstants {
public static final String STATUS_SUCCESS = "SUCCESS";
public static final String STATUS_ERROR = "ERROR";
public static final String STATUS_CANCELLED = "CANCELLED";
public static final String NODE_METHOD = "METHOD";
public static final String NODE_HTTP = "HTTP";
public static final String NODE_DB = "DB";
public static final String NODE_CACHE = "CACHE";
public static final String NODE_TASK = "TASK";
public static final String NODE_STREAM = "STREAM";
}

8
ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/core/TraceContext.java
View File

@@ -96,14 +96,6 @@ public final class TraceContext {
}
}
/**
* 为未来的跨线程上下文传播保留的拷贝入口，避免共享可变栈。
*/
public static Deque<String> copyNodeStack() {
Deque<String> stack = NODE_STACK.get();
return stack == null ? new ArrayDeque<>() : new ArrayDeque<>(stack);
}
public static void clear() {
TRACE_ID.remove();
BUSINESS_TYPE.remove();

9
ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/domain/vo/TraceNodeVo.java
View File

@@ -8,7 +8,6 @@ import org.ruoyi.common.trace.domain.TraceNode;
import java.io.Serial;
import java.io.Serializable;
import java.util.Date;
import java.util.List;
import java.util.Map;
/**
@@ -39,17 +38,15 @@ public class TraceNodeVo implements Serializable {
private Long durationMs;
private String errorMessage;
/** 原始 payload 字符串（兼容旧版），新代码请使用 parsedInput */
/** 原始 input payload 字符串，parsedInput 解析失败时回退使用 */
private String inputPayload;
/** 原始 payload 字符串（兼容旧版），新代码请使用 parsedOutput */
/** 原始 output payload 字符串，parsedOutput 解析失败时回退使用 */
private String outputPayload;
/** 原始 metadata 字符串（兼容旧版），新代码请使用 parsedMetadata */
/** 原始 metadata 字符串，parsedMetadata 解析失败时回退使用 */
private String metadata;
private List<TraceNodeVo> children;
// ======================== 展示用计算字段 ========================
/** 节点类型中文标签，如 "知识检索"、"LLM 调用" */

3
ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/service/impl/TraceRecordServiceImpl.java
View File

@@ -134,9 +134,6 @@ public class TraceRecordServiceImpl implements TraceRecordService {
detail.setRun(run);
List<TraceNodeVo> nodes = listNodes(traceId);
if (nodes != null) {
nodes.forEach(this::enrichNodeVo);
}
// 返回扁平列表，前端自行按 parentNodeId 建树
detail.setNodes(nodes != null ? nodes : new ArrayList<>());

18
ruoyi-common/ruoyi-common-trace/src/main/java/org/ruoyi/common/trace/util/TracePayloadUtils.java
View File

@@ -33,20 +33,6 @@ public final class TracePayloadUtils {
private TracePayloadUtils() {
}
public static String input(Object value, TraceProperties properties) {
if (value == null || properties == null || !properties.getPayload().isRecordDetail()) {
return null;
}
return truncate(asString(value), properties.getPayload().getMaxInputLength());
}
public static String output(Object value, TraceProperties properties) {
if (value == null || properties == null || !properties.getPayload().isRecordDetail()) {
return null;
}
return truncate(asString(value), properties.getPayload().getMaxOutputLength());
}
public static String error(Throwable throwable, TraceProperties properties) {
if (throwable == null) {
return null;
@@ -93,8 +79,4 @@ public final class TracePayloadUtils {
return Collections.singletonMap("_raw", json);
}
}
private static String asString(Object value) {
return value instanceof String str ? str : toJson(value);
}
}

134
ruoyi-common/ruoyi-common-trace/src/test/java/org/ruoyi/common/trace/aspect/TraceNodeAspectTest.java
View File

@@ -1,134 +0,0 @@
package org.ruoyi.common.trace.aspect;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.reflect.MethodSignature;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.ruoyi.common.trace.annotation.TraceNode;
import org.ruoyi.common.trace.config.TraceProperties;
import org.ruoyi.common.trace.constant.TraceConstants;
import org.ruoyi.common.trace.core.TraceContext;
import org.ruoyi.common.trace.service.TraceRecordService;
import java.lang.reflect.Method;
import java.util.Date;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
class TraceNodeAspectTest {
@Mock
private TraceRecordService traceRecordService;
@Mock
private ProceedingJoinPoint joinPoint;
@Mock
private MethodSignature methodSignature;
@AfterEach
void tearDown() {
TraceContext.clear();
}
@Test
void shouldProceedWithoutRecordingWhenNoTraceContext() throws Throwable {
TraceNodeAspect aspect = new TraceNodeAspect(traceRecordService, new TraceProperties());
Method method = SampleService.class.getDeclaredMethod("sample");
when(joinPoint.proceed()).thenReturn("ok");
Object result = aspect.aroundNode(joinPoint, method.getAnnotation(TraceNode.class));
assertEquals("ok", result);
verify(traceRecordService, never()).startNode(any());
}
@Test
void shouldRecordSuccessNode() throws Throwable {
TraceProperties properties = new TraceProperties();
properties.getPayload().setRecordDetail(true);
TraceNodeAspect aspect = new TraceNodeAspect(traceRecordService, properties);
Method method = SampleService.class.getDeclaredMethod("sample");
prepareSignature(method);
when(joinPoint.proceed()).thenReturn("ok");
try (var ignored = TraceContext.begin("trace-1", "TEST", "biz-1", 1L, "000000")) {
Object result = aspect.aroundNode(joinPoint, method.getAnnotation(TraceNode.class));
assertEquals("ok", result);
assertNull(TraceContext.currentNodeId());
}
ArgumentCaptor<org.ruoyi.common.trace.domain.TraceNode> nodeCaptor =
ArgumentCaptor.forClass(org.ruoyi.common.trace.domain.TraceNode.class);
verify(traceRecordService).startNode(nodeCaptor.capture());
org.ruoyi.common.trace.domain.TraceNode node = nodeCaptor.getValue();
assertEquals("trace-1", node.getTraceId());
assertEquals("sample-node", node.getNodeName());
assertEquals(TraceConstants.NODE_METHOD, node.getNodeType());
assertEquals("safe-input", node.getInputPayload());
verify(traceRecordService).finishNode(eq("trace-1"), eq(node.getNodeId()), eq(TraceConstants.STATUS_SUCCESS),
isNull(), eq("ok"), any(Date.class), anyLong());
}
@Test
void shouldRethrowBusinessExceptionAfterErrorRecord() throws Throwable {
TraceNodeAspect aspect = new TraceNodeAspect(traceRecordService, new TraceProperties());
Method method = SampleService.class.getDeclaredMethod("sample");
prepareSignature(method);
IllegalArgumentException error = new IllegalArgumentException("bad");
when(joinPoint.proceed()).thenThrow(error);
try (var ignored = TraceContext.begin("trace-1", "TEST", "biz-1", 1L, "000000")) {
assertThrows(IllegalArgumentException.class,
() -> aspect.aroundNode(joinPoint, method.getAnnotation(TraceNode.class)));
}
ArgumentCaptor<org.ruoyi.common.trace.domain.TraceNode> nodeCaptor =
ArgumentCaptor.forClass(org.ruoyi.common.trace.domain.TraceNode.class);
verify(traceRecordService).startNode(nodeCaptor.capture());
verify(traceRecordService).finishNode(eq("trace-1"), eq(nodeCaptor.getValue().getNodeId()), eq(TraceConstants.STATUS_ERROR),
eq("IllegalArgumentException: bad"), isNull(), any(Date.class), anyLong());
}
@Test
void writeFailureShouldNotAffectBusinessReturn() throws Throwable {
TraceNodeAspect aspect = new TraceNodeAspect(traceRecordService, new TraceProperties());
Method method = SampleService.class.getDeclaredMethod("sample");
prepareSignature(method);
when(joinPoint.proceed()).thenReturn("ok");
doThrow(new IllegalStateException("db down")).when(traceRecordService).startNode(any());
try (var ignored = TraceContext.begin("trace-1", "TEST", "biz-1", 1L, "000000")) {
assertEquals("ok", aspect.aroundNode(joinPoint, method.getAnnotation(TraceNode.class)));
}
}
private void prepareSignature(Method method) {
when(joinPoint.getSignature()).thenReturn(methodSignature);
when(methodSignature.getMethod()).thenReturn(method);
}
private static class SampleService {
@TraceNode(name = "sample-node", input = "safe-input")
String sample() {
return "ok";
}
}
}

12
ruoyi-common/ruoyi-common-trace/src/test/java/org/ruoyi/common/trace/core/TraceContextTest.java
View File

@@ -3,8 +3,6 @@ package org.ruoyi.common.trace.core;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import java.util.Deque;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
@@ -44,14 +42,4 @@ class TraceContextTest {
assertNull(TraceContext.currentNodeId());
assertEquals(0, TraceContext.depth());
}
@Test
void copyNodeStackShouldReturnIndependentDeque() {
TraceContext.pushNode("root");
Deque<String> copy = TraceContext.copyNodeStack();
copy.push("child");
assertEquals("root", TraceContext.currentNodeId());
assertEquals("child", copy.peek());
}
}

12
ruoyi-common/ruoyi-common-trace/src/test/java/org/ruoyi/common/trace/util/TracePayloadUtilsTest.java
View File

@@ -1,7 +1,6 @@
package org.ruoyi.common.trace.util;
import org.junit.jupiter.api.Test;
import org.ruoyi.common.trace.config.TraceProperties;
import java.util.Map;
@@ -26,15 +25,4 @@ class TracePayloadUtilsTest {
String json = TracePayloadUtils.toJson(Map.of("count", 2));
assertTrue(json.contains("\"count\":2"));
}
@Test
void detailEnabledShouldTruncateInputAndOutput() {
TraceProperties properties = new TraceProperties();
properties.getPayload().setRecordDetail(true);
properties.getPayload().setMaxInputLength(4);
properties.getPayload().setMaxOutputLength(5);
assertEquals("1234", TracePayloadUtils.input("123456", properties));
assertEquals("12345", TracePayloadUtils.output("1234567", properties));
}
}

6
ruoyi-modules/ruoyi-aiflow/src/main/java/org/ruoyi/workflow/workflow/WorkflowStarter.java
View File

@@ -56,10 +56,10 @@ public class WorkflowStarter implements IWorkFlowStarterService {
public SseEmitter streaming(User user, String workflowUuid, List<ObjectNode> userInputs, Long sessionId) {
// 获取用户ID
Long userId = LoginHelper.getUserId();
// 获取登录Token
// 获取登录Token（仅透传给 WfState，工作流 SSE 通过 emitter 直发，不串台）
String tokenValue = StpUtil.getTokenValue();
// 根据用户ID和Token连接SSE对象
SseEmitter sseEmitter = sseEmitterManager.connect(userId, tokenValue);
// 根据会话ID连接SSE对象（每会话一个连接，避免同用户多会话串台）
SseEmitter sseEmitter = sseEmitterManager.connect(String.valueOf(sessionId));
if (!sseEmitterHelper.checkOrComplete(user, sseEmitter)) {
return sseEmitter;
}

14
ruoyi-modules/ruoyi-chat/src/main/java/org/ruoyi/observability/MyMcpClientListener.java
View File

@@ -41,14 +41,14 @@ public class MyMcpClientListener implements McpClientListener {
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
private final Long userId;
private final String sessionId;
public MyMcpClientListener(Long userId) {
this.userId = userId;
public MyMcpClientListener(String sessionId) {
this.sessionId = sessionId;
}
public MyMcpClientListener() {
this.userId = null;
this.sessionId = null;
}
// ==================== 工具执行 ====================
@@ -146,8 +146,8 @@ public class MyMcpClientListener implements McpClientListener {
* 推送 MCP 事件到前端
*/
private void pushMcpEvent(String name, String status, String result) {
if (userId == null) {
log.warn("userId 为空，无法推送 MCP 事件");
if (sessionId == null) {
log.warn("sessionId 为空，无法推送 MCP 事件");
return;
}
try {
@@ -157,7 +157,7 @@ public class MyMcpClientListener implements McpClientListener {
content.put("result", result);
String json = OBJECT_MAPPER.writeValueAsString(content);
SseMessageUtils.sendEvent(userId, SseEventDto.builder()
SseMessageUtils.sendEvent(sessionId, SseEventDto.builder()
.event("mcp")
.content(json)
.build());

62
ruoyi-modules/ruoyi-chat/src/main/java/org/ruoyi/service/chat/impl/ChatServiceFacade.java
View File

@@ -72,7 +72,6 @@ import org.ruoyi.domain.vo.agent.AgentVo;
import org.ruoyi.domain.vo.knowledge.KnowledgeInfoVo;
import org.ruoyi.factory.ChatServiceFactory;
import org.ruoyi.mcp.service.core.LangChain4jMcpToolProviderService;
import org.ruoyi.mcp.service.core.ToolProviderFactory;
import org.ruoyi.observability.*;
import org.ruoyi.service.agent.IAgentService;
import org.ruoyi.service.chat.AbstractChatService;
@@ -81,7 +80,6 @@ import org.ruoyi.service.chat.impl.memory.PersistentChatMemoryStore;
import org.ruoyi.service.knowledge.IKnowledgeInfoService;
import org.ruoyi.service.retrieval.KnowledgeRetrievalService;
import org.ruoyi.service.knowledge.retriever.CustomVectorRetriever;
import org.ruoyi.service.vector.VectorStoreService;
import org.ruoyi.trace.RagTraceNodeTypes;
import org.ruoyi.trace.RagTracePayloadBuilder;
import org.springframework.stereotype.Service;
@@ -120,8 +118,6 @@ public class ChatServiceFacade implements IChatService {
private final IKnowledgeInfoService knowledgeInfoService;
private final VectorStoreService vectorStoreService;
private final KnowledgeRetrievalService knowledgeRetrievalService;
private final SseEmitterManager sseEmitterManager;
@@ -130,7 +126,6 @@ public class ChatServiceFacade implements IChatService {
private final IWorkFlowStarterService workFlowStarterService;
private final ToolProviderFactory toolProviderFactory;
private final IAgentService agentService;
@@ -156,13 +151,13 @@ public class ChatServiceFacade implements IChatService {
*/
public SseEmitter sseChat(ChatRequest chatRequest) {
// 4. 具体的服务实现
// 具体的服务实现
Long userId = LoginHelper.getUserId();
String tokenValue = StpUtil.getTokenValue();
SseEmitter emitter = sseEmitterManager.connect(userId, tokenValue);
// 每个会话一个 SSE 连接，避免同用户多会话串台
SseEmitter emitter = sseEmitterManager.connect(String.valueOf(chatRequest.getSessionId()));
// 0. 智能体解析：传入 agentId 时按智能体绑定的模型覆盖 model 字段
// （前端默认走智能体；enableThinking 不再作为对话模式开关，Supervisor 多 Agent 编排成为默认智能体路径）
// 智能体解析：传入 agentId 时按智能体绑定的模型覆盖 model 字段
AgentVo agentVo = null;
if (chatRequest.getAgentId() != null) {
agentVo = agentService.queryById(chatRequest.getAgentId());
@@ -176,14 +171,14 @@ public class ChatServiceFacade implements IChatService {
}
}
// 1. 根据模型名称查询完整配置
// 根据模型名称查询完整配置
ChatModelVo chatModelVo = chatModelService.selectModelByName(chatRequest.getModel());
if (chatModelVo == null) {
throw new IllegalArgumentException("模型不存在: " + chatRequest.getModel());
}
// 2. 构建上下文消息列表（系统提示词 + 历史消息 + 当前用户消息）
// 注意：RAG 检索增强统一在 handleAgentChat 中执行一次，此处不再重复检索
// 构建上下文消息列表（系统提示词 + 历史消息 + 当前用户消息）
// 注意：RAG 检索增强统一在 handleAgentChat 中执行一次，此处不再重复检索
List<ChatMessage> contextMessages = buildContextMessages(chatRequest, agentVo);
chatRequest.setEmitter(emitter);
@@ -242,13 +237,14 @@ public class ChatServiceFacade implements IChatService {
ChatModel plannerModel = chatService.buildChatModel(chatModelVo);
Long userId = chatRequest.getUserId();
String sessionId = String.valueOf(chatRequest.getSessionId());
// 工具装配：智能体有关联工具ID时按ID装配，否则回退到原有硬编码 MCP 客户端
ToolProvider toolProvider;
if (agentVo != null && agentVo.getMcpToolIds() != null && !agentVo.getMcpToolIds().isEmpty()) {
toolProvider = langChain4jMcpToolProviderService.getToolProvider(agentVo.getMcpToolIds());
} else {
toolProvider = buildDefaultMcpToolProvider(userId);
toolProvider = buildDefaultMcpToolProvider(sessionId);
}
// Skills 装配：智能体有勾选技能名时按名过滤磁盘 skills，否则加载全部
@@ -321,16 +317,14 @@ public class ChatServiceFacade implements IChatService {
promptBuilder.append(augmentedInput);
String prompt = promptBuilder.toString();
String tokenValue = chatRequest.getTokenValue();
// 异步执行 supervisor，避免阻塞 HTTP 请求线程导致 SSE 事件被缓冲
CompletableFuture.runAsync(() -> {
TraceStreamSpan llmSpan = null;
try (TraceScope ignored = openTraceScope(traceRun, userId)) {
llmSpan = startLlmCallSpan(traceRun, chatRequest);
String result = supervisor.invoke(prompt);
SseMessageUtils.sendContent(userId, result);
SseMessageUtils.sendDone(userId);
SseMessageUtils.sendContent(sessionId, result);
SseMessageUtils.sendDone(sessionId);
// 保存助手回复到数据库（智能体对话为默认路径后，需在此落库以保留历史）
if (StringUtils.isNotBlank(result)) {
chatMessageService.saveChatMessage(userId, chatRequest.getSessionId(),
@@ -347,12 +341,12 @@ public class ChatServiceFacade implements IChatService {
}
finishTraceRun(traceRun, TraceConstants.STATUS_ERROR, e);
log.error("Supervisor 执行失败", e);
SseMessageUtils.sendError(userId, e.getMessage());
SseMessageUtils.sendError(sessionId, e.getMessage());
} finally {
if (llmSpan != null) {
llmSpan.detach();
}
SseMessageUtils.completeConnection(userId, tokenValue);
SseMessageUtils.completeConnection(sessionId);
}
});
return chatRequest.getEmitter();
@@ -461,8 +455,10 @@ public class ChatServiceFacade implements IChatService {
/**
* 兜底 MCP 工具装配（无智能体时使用，保留原有 3 个硬编码客户端逻辑）
*
* @param sessionId 会话ID，用于 MCP 工具事件按会话推送 SSE
*/
private ToolProvider buildDefaultMcpToolProvider(Long userId) {
private ToolProvider buildDefaultMcpToolProvider(String sessionId) {
String npxCommand = resolveNpxCommand();
McpTransport playwrightTransport = new StdioMcpTransport.Builder()
.command(List.of(npxCommand, "-y", "@playwright/mcp@latest"))
@@ -470,7 +466,7 @@ public class ChatServiceFacade implements IChatService {
.build();
McpClient playwrightMcpClient = new DefaultMcpClient.Builder()
.transport(playwrightTransport)
.listener(new MyMcpClientListener(userId))
.listener(new MyMcpClientListener(sessionId))
.build();
String userDir = System.getProperty("user.dir");
@@ -481,7 +477,7 @@ public class ChatServiceFacade implements IChatService {
.build();
McpClient filesystemMcpClient = new DefaultMcpClient.Builder()
.transport(filesystemTransport)
.listener(new MyMcpClientListener(userId))
.listener(new MyMcpClientListener(sessionId))
.build();
return McpToolProvider.builder()
@@ -566,16 +562,15 @@ public class ChatServiceFacade implements IChatService {
// 4. 获取用户信息
Long userId = LoginHelper.getUserId();
String tokenValue = StpUtil.getTokenValue();
// 5. 建立 SSE 连接（用于前端监听）
sseEmitterManager.connect(userId, tokenValue);
// 5. 建立 SSE 连接（用于前端监听，按会话隔离）
sseEmitterManager.connect(String.valueOf(chatRequest.getSessionId()));
// 保存用户消息
chatMessageService.saveChatMessage(userId, chatRequest.getSessionId(), chatRequest.getContent(), RoleType.USER.getName(), chatRequest.getModel());
// 6. 创建组合 handler：同时发送到 SSE 和外部 handler
StreamingChatResponseHandler combinedHandler = createCombinedHandler(userId, tokenValue, externalHandler);
StreamingChatResponseHandler combinedHandler = createCombinedHandler(String.valueOf(chatRequest.getSessionId()), externalHandler);
// 7. 发起对话
StreamingChatModel streamingChatModel = chatService.buildStreamingChatModel(chatModelVo, chatRequest);
@@ -799,12 +794,11 @@ public class ChatServiceFacade implements IChatService {
/**
* 创建组合响应处理器 - 同时发送到 SSE 和外部 handler
*
* @param userId 用户ID
* @param tokenValue 会话令牌
* @param sessionId 会话ID（SSE 按会话隔离推送）
* @param externalHandler 外部响应处理器（可为 null）
* @return 组合的流式响应处理器
*/
protected StreamingChatResponseHandler createCombinedHandler(Long userId, String tokenValue,
protected StreamingChatResponseHandler createCombinedHandler(String sessionId,
StreamingChatResponseHandler externalHandler) {
return new StreamingChatResponseHandler() {
@@ -817,7 +811,7 @@ public class ChatServiceFacade implements IChatService {
messageBuffer.append(partialResponse);
// 2. 发送内容事件到 SSE（前端可通过 SSE 监听）
SseMessageUtils.sendContent(userId, partialResponse);
SseMessageUtils.sendContent(sessionId, partialResponse);
// 3. 转发给外部 handler（Workflow 等模块可处理）
if (externalHandler != null) {
@@ -828,7 +822,7 @@ public class ChatServiceFacade implements IChatService {
@Override
public void onPartialThinking(PartialThinking partialThinking) {
// 发送推理内容到 SSE（前端通过 reasoning 事件监听）
SseMessageUtils.sendReasoning(userId, partialThinking.text());
SseMessageUtils.sendReasoning(sessionId, partialThinking.text());
// 转发给外部 handler
if (externalHandler != null) {
@@ -840,10 +834,10 @@ public class ChatServiceFacade implements IChatService {
public void onCompleteResponse(ChatResponse completeResponse) {
try {
// 1. 发送完成事件
SseMessageUtils.sendDone(userId);
SseMessageUtils.sendDone(sessionId);
// 2. 关闭 SSE 连接
SseMessageUtils.completeConnection(userId, tokenValue);
SseMessageUtils.completeConnection(sessionId);
// 3. 转发给外部 handler
if (externalHandler != null) {
@@ -857,7 +851,7 @@ public class ChatServiceFacade implements IChatService {
@Override
public void onError(Throwable error) {
// 发送错误事件
SseMessageUtils.sendError(userId, error.getMessage());
SseMessageUtils.sendError(sessionId, error.getMessage());
log.error("流式响应错误: {}", error.getMessage(), error);
// 转发给外部 handler

1
ruoyi-modules/ruoyi-chat/src/main/java/org/ruoyi/trace/RagTraceNodeTypes.java
View File

@@ -14,5 +14,4 @@ public final class RagTraceNodeTypes {
public static final String NODE_RETRIEVAL = "RETRIEVAL";
public static final String NODE_RERANK = "RERANK";
public static final String NODE_LLM_CALL = "LLM_CALL";
public static final String NODE_STREAM = "STREAM";
}