fix(trace): 用户取消后将 trace run 收尾为 CANCELLED

bootstrap/src/main/java/com/nageoffer/ai/ragent/admin/service/impl/DashboardServiceImpl.java

@@ -58,6 +58,7 @@ public class DashboardServiceImpl implements DashboardService {
 
     private static final String STATUS_SUCCESS = "SUCCESS";
     private static final String STATUS_ERROR = "ERROR";
+    private static final String STATUS_CANCELLED = "CANCELLED";
     private static final String ROLE_ASSISTANT = "assistant";
     private static final String NO_DOC_REPLY = "未检索到与问题相关的文档内容。";
     private static final String GRANULARITY_DAY = "day";
@@ -114,7 +115,9 @@ public DashboardPerformanceVO loadPerformance(String window) {
 
         long success = countTraceRuns(range.start, range.end, STATUS_SUCCESS);
         long error = countTraceRuns(range.start, range.end, STATUS_ERROR);
-        long total = success + error;
+        // 被用户取消的会话同样是一次真实请求，必须计入分母，否则「用户等不及点了停止」这一负面信号会被系统性剔除
+        long cancelled = countTraceRuns(range.start, range.end, STATUS_CANCELLED);
+        long total = success + error + cancelled;
         long assistantCount = countAssistantMessages(range.start, range.end);
         long noDocCount = countNoDocMessages(range.start, range.end);
         long slowCount = durations.stream().filter(duration -> duration > SLOW_LATENCY_THRESHOLD_MS).count();

bootstrap/src/main/java/com/nageoffer/ai/ragent/rag/service/RagTraceRecordService.java

@@ -31,6 +31,19 @@ public interface RagTraceRecordService {
 
     void finishRun(String traceId, String status, String errorMessage, Date endTime, long durationMs);
 
+    /**
+     * 将指定任务仍处于 RUNNING 的 trace run 收尾为 CANCELLED
+     * <p>
+     * 取消信号在 provider client 层被拦截（{@code ForwardingStreamCallback} 对外部取消不透传 delegate，
+     * 否则流式 failover 切换候选时会终止用户 SSE），run 级终态无法由回调链驱动，只能按 taskId 单独上报
+     * </p>
+     *
+     * @param taskId  流式任务 ID
+     * @param endTime 取消发生的时间
+     * @return 是否确实有一行由 RUNNING 翻转为 CANCELLED
+     */
+    boolean cancelRunByTaskId(String taskId, Date endTime);
+
     void startNode(RagTraceNodeDO node);
 
     void finishNode(String traceId, String nodeId, String status, String errorMessage, Date endTime, long durationMs);

bootstrap/src/main/java/com/nageoffer/ai/ragent/rag/service/handler/StreamTaskManager.java

@@ -24,6 +24,7 @@
 import com.nageoffer.ai.ragent.rag.dto.CompletionPayload;
 import com.nageoffer.ai.ragent.framework.web.SseEmitterSender;
 import com.nageoffer.ai.ragent.infra.chat.StreamCancellationHandle;
+import com.nageoffer.ai.ragent.rag.service.RagTraceRecordService;
 import jakarta.annotation.PostConstruct;
 import jakarta.annotation.PreDestroy;
 import lombok.SneakyThrows;
@@ -34,6 +35,7 @@
 import org.springframework.stereotype.Component;
 
 import java.time.Duration;
+import java.util.Date;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Supplier;
 
@@ -51,10 +53,12 @@ public class StreamTaskManager {
             .build();
 
     private final RedissonClient redissonClient;
+    private final RagTraceRecordService traceRecordService;
     private int listenerId = -1;
 
-    public StreamTaskManager(RedissonClient redissonClient) {
+    public StreamTaskManager(RedissonClient redissonClient, RagTraceRecordService traceRecordService) {
         this.redissonClient = redissonClient;
+        this.traceRecordService = traceRecordService;
     }
 
     @PostConstruct
@@ -149,6 +153,26 @@ private void cancelLocal(String taskId) {
             sendCancelAndDone(taskInfo.sender, payload);
             taskInfo.sender.complete();
         }
+
+        reportTraceRunCancelled(taskId);
+    }
+
+    /**
+     * 上报 run 级取消终态
+     * <p>
+     * 取消信号在 provider client 层就被 {@code ForwardingStreamCallback} 拦截了（对外部取消不透传 delegate，
+     * 否则流式 failover 切换候选时会终止用户 SSE），run 级终态无法由回调链驱动。
+     * 而 {@code cancelLocal} 的 CAS 成功分支是全链路唯一能确定「这是用户主动取消」而非「failover 内部取消」的位置，
+     * 因此在这里上报
+     * </p>
+     */
+    private void reportTraceRunCancelled(String taskId) {
+        try {
+            traceRecordService.cancelRunByTaskId(taskId, new Date());
+        } catch (Exception e) {
+            // trace 是旁路观测，失败不能影响取消本身
+            log.warn("上报 trace run 取消状态失败，任务ID：{}", taskId, e);
+        }
     }
 
     public void unregister(String taskId) {

bootstrap/src/main/java/com/nageoffer/ai/ragent/rag/service/impl/RagTraceRecordServiceImpl.java

@@ -35,6 +35,9 @@
 @RequiredArgsConstructor
 public class RagTraceRecordServiceImpl implements RagTraceRecordService {
 
+    private static final String STATUS_RUNNING = "RUNNING";
+    private static final String STATUS_CANCELLED = "CANCELLED";
+
     private final RagTraceRunMapper runMapper;
     private final RagTraceNodeMapper nodeMapper;
 
@@ -55,6 +58,28 @@ public void finishRun(String traceId, String status, String errorMessage, Date e
                 .eq(RagTraceRunDO::getTraceId, traceId));
     }
 
+    @Override
+    public boolean cancelRunByTaskId(String taskId, Date endTime) {
+        RagTraceRunDO running = runMapper.selectOne(Wrappers.lambdaQuery(RagTraceRunDO.class)
+                .eq(RagTraceRunDO::getTaskId, taskId)
+                .eq(RagTraceRunDO::getStatus, STATUS_RUNNING)
+                .orderByDesc(RagTraceRunDO::getStartTime)
+                .last("LIMIT 1"));
+        if (running == null) {
+            return false;
+        }
+
+        RagTraceRunDO update = RagTraceRunDO.builder()
+                .status(STATUS_CANCELLED)
+                .endTime(endTime)
+                .durationMs(Math.max(0, endTime.getTime() - running.getStartTime().getTime()))
+                .build();
+        // 带 status 的条件更新：与正常终态（onComplete / onError）竞争时只有一方能生效
+        return runMapper.update(update, Wrappers.lambdaUpdate(RagTraceRunDO.class)
+                .eq(RagTraceRunDO::getTraceId, running.getTraceId())
+                .eq(RagTraceRunDO::getStatus, STATUS_RUNNING)) > 0;
+    }
+
     @Override
     public void startNode(RagTraceNodeDO node) {
         nodeMapper.insert(node);

bootstrap/src/test/java/com/nageoffer/ai/ragent/admin/service/impl/DashboardPerformanceCancelledTest.java

@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.nageoffer.ai.ragent.admin.service.impl;
+
+import com.baomidou.mybatisplus.core.conditions.Wrapper;
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.nageoffer.ai.ragent.admin.controller.vo.DashboardPerformanceVO;
+import com.nageoffer.ai.ragent.rag.dao.entity.RagTraceRunDO;
+import com.nageoffer.ai.ragent.rag.dao.mapper.ConversationMapper;
+import com.nageoffer.ai.ragent.rag.dao.mapper.ConversationMessageMapper;
+import com.nageoffer.ai.ragent.rag.dao.mapper.RagTraceRunMapper;
+import com.nageoffer.ai.ragent.user.dao.mapper.UserMapper;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentMatcher;
+
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.argThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * 被用户取消的会话必须计入成功率分母
+ * <p>
+ * 否则「用户等不及点了停止」这一最该被捕捉的负面信号会被系统性剔除，看板呈现幸存者偏差
+ * </p>
+ */
+class DashboardPerformanceCancelledTest {
+
+    private RagTraceRunMapper traceRunMapper;
+    private ConversationMessageMapper messageMapper;
+    private DashboardServiceImpl dashboardService;
+
+    @BeforeEach
+    void setUp() {
+        traceRunMapper = mock(RagTraceRunMapper.class);
+        messageMapper = mock(ConversationMessageMapper.class);
+        dashboardService = new DashboardServiceImpl(
+                mock(UserMapper.class),
+                mock(ConversationMapper.class),
+                messageMapper,
+                traceRunMapper);
+    }
+
+    @Test
+    void countsCancelledRunsInSuccessRateDenominator() {
+        stubTraceRunCount("SUCCESS", 6L);
+        stubTraceRunCount("ERROR", 2L);
+        stubTraceRunCount("CANCELLED", 2L);
+        when(traceRunMapper.selectObjs(any())).thenReturn(Collections.emptyList());
+        when(messageMapper.selectCount(any())).thenReturn(0L);
+
+        DashboardPerformanceVO performance = dashboardService.loadPerformance("24h");
+
+        // 分母 = 6 + 2 + 2 = 10，而非只算成功与失败的 8
+        assertEquals(60.0, performance.getSuccessRate());
+        assertEquals(20.0, performance.getErrorRate());
+    }
+
+    private void stubTraceRunCount(String status, long count) {
+        when(traceRunMapper.selectCount(argThat(wrapperMatching(status)))).thenReturn(count);
+    }
+
+    private static ArgumentMatcher<Wrapper<RagTraceRunDO>> wrapperMatching(String status) {
+        return wrapper -> {
+            if (!(wrapper instanceof QueryWrapper<?> queryWrapper)) {
+                return false;
+            }
+            // MyBatis-Plus 的条件值是惰性写入 paramNameValuePairs 的，先触发 SQL 片段生成
+            queryWrapper.getTargetSql();
+            return queryWrapper.getParamNameValuePairs().containsValue(status);
+        };
+    }
+}

bootstrap/src/test/java/com/nageoffer/ai/ragent/rag/service/handler/StreamTaskManagerCancelTraceTest.java

@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.nageoffer.ai.ragent.rag.service.handler;
+
+import com.nageoffer.ai.ragent.infra.chat.StreamCancellationHandle;
+import com.nageoffer.ai.ragent.rag.service.RagTraceRecordService;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.redisson.api.RTopic;
+import org.redisson.api.RedissonClient;
+import org.redisson.api.listener.MessageListener;
+
+import java.util.Date;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * 用户主动取消时，对应的 trace run 必须收尾为 CANCELLED
+ * <p>
+ * 取消信号在 provider client 层被 {@code ForwardingStreamCallback.finishExternally} 刻意拦截（不透传 delegate，
+ * 否则流式 failover 切换候选时会终止用户 SSE），因此 run 级终态只能由本类上报 ——
+ * 全链路只有这里能区分「用户主动取消」与「failover 内部取消」
+ * </p>
+ */
+class StreamTaskManagerCancelTraceTest {
+
+    private static final String TASK_ID = "task-1";
+
+    private RedissonClient redissonClient;
+    private RTopic topic;
+    private RagTraceRecordService traceRecordService;
+    private StreamTaskManager taskManager;
+    private StreamCancellationHandle handle;
+
+    @BeforeEach
+    @SuppressWarnings("unchecked")
+    void setUp() {
+        redissonClient = mock(RedissonClient.class);
+        topic = mock(RTopic.class);
+        traceRecordService = mock(RagTraceRecordService.class);
+        handle = mock(StreamCancellationHandle.class);
+        when(redissonClient.getTopic(any(String.class))).thenReturn(topic);
+
+        taskManager = new StreamTaskManager(redissonClient, traceRecordService);
+    }
+
+    @Test
+    void reportsCancelledTraceRunOnUserCancel() {
+        MessageListener<String> listener = subscribeAndCaptureListener();
+        taskManager.bindHandle(TASK_ID, handle);
+
+        listener.onMessage("channel", TASK_ID);
+
+        verify(handle).cancel();
+        verify(traceRecordService).cancelRunByTaskId(eq(TASK_ID), any(Date.class));
+    }
+
+    @Test
+    void reportsCancelledTraceRunOnlyOnce() {
+        MessageListener<String> listener = subscribeAndCaptureListener();
+        taskManager.bindHandle(TASK_ID, handle);
+
+        listener.onMessage("channel", TASK_ID);
+        listener.onMessage("channel", TASK_ID);
+
+        verify(traceRecordService, times(1)).cancelRunByTaskId(eq(TASK_ID), any(Date.class));
+    }
+
+    @Test
+    void skipsTraceReportForUnknownTask() {
+        MessageListener<String> listener = subscribeAndCaptureListener();
+
+        listener.onMessage("channel", "not-registered");
+
+        verify(traceRecordService, never()).cancelRunByTaskId(any(), any());
+    }
+
+    @Test
+    void stillCancelsStreamWhenTraceReportFails() {
+        MessageListener<String> listener = subscribeAndCaptureListener();
+        taskManager.bindHandle(TASK_ID, handle);
+        doThrow(new RuntimeException("trace 库不可用"))
+                .when(traceRecordService).cancelRunByTaskId(any(), any());
+
+        listener.onMessage("channel", TASK_ID);
+
+        verify(handle).cancel();
+    }
+
+    @SuppressWarnings("unchecked")
+    private MessageListener<String> subscribeAndCaptureListener() {
+        taskManager.subscribe();
+        ArgumentCaptor<MessageListener<String>> captor = ArgumentCaptor.forClass(MessageListener.class);
+        verify(topic).addListener(eq(String.class), captor.capture());
+        return captor.getValue();
+    }
+}