Merge branch 'fix/sse-session-trace-cleanup'

ruoyi-admin/src/main/resources/application.yml

@@ -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
 

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

@@ -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:{}",
+            log.info("SSE发送主题订阅消息topic:{} session:{} session keys:{} message:{}",
-                SSE_TOPIC, sseMessageDto.getUserIds(), sseMessageDto.getMessage());
+                SSE_TOPIC, sseMessageDto.getSessionId(), sseMessageDto.getUserIds(), sseMessageDto.getMessage());
         });
     }
 

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

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

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

@@ -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());
+            log.info("SSE主题订阅收到消息session:{} session keys={} message={}",
-            // 如果key不为空就按照key发消息 如果为空就群发
+                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());

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

@@ -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));
+    }
+
     /**
      * 是否开启
      */

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

@@ -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 "";
-}

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

@@ -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);
-        }
-    }
-}

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

@@ -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);
-    }
 }

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

@@ -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;
-
         /**
          * 错误信息最大长度。
          */

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

@@ -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";
 }

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

@@ -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();

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

@@ -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 调用" */

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

@@ -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<>());
 

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

@@ -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);
-    }
 }

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

@@ -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";
-        }
-    }
-}

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

@@ -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());
-    }
 }

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

@@ -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));
-    }
 }

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

@@ -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对象
+        // 根据会话ID连接SSE对象（每会话一个连接，避免同用户多会话串台）
-        SseEmitter sseEmitter = sseEmitterManager.connect(userId, tokenValue);
+        SseEmitter sseEmitter = sseEmitterManager.connect(String.valueOf(sessionId));
         if (!sseEmitterHelper.checkOrComplete(user, sseEmitter)) {
             return sseEmitter;
         }

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

@@ -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) {
+    public MyMcpClientListener(String sessionId) {
-        this.userId = userId;
+        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) {
+        if (sessionId == null) {
-            log.warn("userId 为空，无法推送 MCP 事件");
+            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());

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

@@ -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 字段
+        // 智能体解析：传入 agentId 时按智能体绑定的模型覆盖 model 字段
-        //    （前端默认走智能体；enableThinking 不再作为对话模式开关，Supervisor 多 Agent 编排成为默认智能体路径）
         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.sendContent(sessionId, result);
-                SseMessageUtils.sendDone(userId);
+                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 连接（用于前端监听）
+        // 5. 建立 SSE 连接（用于前端监听，按会话隔离）
-        sseEmitterManager.connect(userId, tokenValue);
+        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 sessionId       会话ID（SSE 按会话隔离推送）
-     * @param tokenValue      会话令牌
      * @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

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

@@ -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";
 }