fix(web): clean up pipeline debug websocket

web/src/app/home/pipelines/components/debug-dialog/DebugDialog.tsx

@@ -150,6 +150,11 @@ export default function DebugDialog({
   const fileInputRef = useRef<HTMLInputElement>(null);
   const wsClientRef = useRef<WebSocketClient | null>(null);
   const isInitializingRef = useRef<boolean>(false);
+  const historyRequestGenerationRef = useRef(0);
+
+  const invalidateHistoryRequests = useCallback(() => {
+    historyRequestGenerationRef.current++;
+  }, []);
 
   const scrollToBottom = useCallback(() => {
     setTimeout(() => {
@@ -165,13 +170,16 @@ export default function DebugDialog({
 
   const loadMessages = useCallback(
     async (pipelineId: string) => {
+      const generation = ++historyRequestGenerationRef.current;
       try {
         const response = await httpClient.getWebSocketHistoryMessages(
           pipelineId,
           sessionType,
         );
+        if (generation !== historyRequestGenerationRef.current) return;
         setMessages(response.messages);
       } catch (error) {
+        if (generation !== historyRequestGenerationRef.current) return;
         console.error('Failed to load messages:', error);
       }
     },
@@ -186,24 +194,30 @@ export default function DebugDialog({
         return;
       }
 
+      let wsClient: WebSocketClient | null = null;
+      let errorReported = false;
       try {
         isInitializingRef.current = true;
 
         // Disconnect old connection
-        if (wsClientRef.current) {
+        const previousClient = wsClientRef.current;
-          wsClientRef.current.disconnect();
         wsClientRef.current = null;
-        }
+        previousClient?.disconnect();
 
         // Create new connection
-        const wsClient = new WebSocketClient(pipelineId, sessionType);
+        wsClient = new WebSocketClient(pipelineId, sessionType);
+        // Store the client before awaiting connect so effect cleanup can also
+        // cancel sockets that are still authenticating.
+        wsClientRef.current = wsClient;
 
         wsClient
           .onConnected(() => {
+            if (wsClientRef.current !== wsClient) return;
             setIsConnected(true);
             isInitializingRef.current = false;
           })
           .onMessage((wsMessage) => {
+            if (wsClientRef.current !== wsClient) return;
             // Convert WebSocketMessage to Message type
             const message: Message = {
               ...wsMessage,
@@ -228,27 +242,33 @@ export default function DebugDialog({
             });
           })
           .onError((error) => {
+            if (wsClientRef.current !== wsClient) return;
+            errorReported = true;
             console.error('WebSocket error:', error);
             setIsConnected(false);
             isInitializingRef.current = false;
             toast.error(t('pipelines.debugDialog.connectionError'));
           })
           .onClose(() => {
+            if (wsClientRef.current !== wsClient) return;
             setIsConnected(false);
             isInitializingRef.current = false;
           })
           .onBroadcast((message) => {
+            if (wsClientRef.current !== wsClient) return;
             toast.info(message);
           });
 
         await wsClient.connect();
-        wsClientRef.current = wsClient;
       } catch (error) {
+        if (!wsClient || wsClientRef.current !== wsClient) return;
         console.error('WebSocket connection failed:', error);
         setIsConnected(false);
         isInitializingRef.current = false;
+        if (!errorReported) {
           toast.error(t('pipelines.debugDialog.connectionFailed'));
         }
+      }
     },
     [sessionType, t],
   );
@@ -263,24 +283,28 @@ export default function DebugDialog({
     if (open) {
       setSelectedPipelineId(pipelineId);
     } else {
+      invalidateHistoryRequests();
       // Disconnect WebSocket immediately when dialog closes
       if (wsClientRef.current) {
-        wsClientRef.current.disconnect();
+        const wsClient = wsClientRef.current;
         wsClientRef.current = null;
+        wsClient.disconnect();
         setIsConnected(false);
         isInitializingRef.current = false;
       }
     }
 
     return () => {
+      invalidateHistoryRequests();
       // Disconnect WebSocket on component unmount
       if (wsClientRef.current) {
-        wsClientRef.current.disconnect();
+        const wsClient = wsClientRef.current;
         wsClientRef.current = null;
+        wsClient.disconnect();
         isInitializingRef.current = false;
       }
     };
-  }, [open, pipelineId]);
+  }, [open, pipelineId, invalidateHistoryRequests]);
 
   // Reload messages and reconnect when sessionType or selectedPipelineId changes
   useEffect(() => {

web/src/app/infra/websocket/WebSocketClient.ts

@@ -31,9 +31,12 @@ export class WebSocketClient {
   private reconnectAttempts = 0;
   private maxReconnectAttempts = 5;
   private reconnectDelay = 3000; // 3秒重连间隔
+  private reconnectTimeout: ReturnType<typeof setTimeout> | null = null;
   private heartbeatInterval: NodeJS.Timeout | null = null;
   private heartbeatIntervalMs = 30000; // 30秒
   private isConnecting = false; // 防止重复连接
+  private shouldReconnect = true;
+  private disconnectedByUser = false;
 
   // 事件回调
   private onConnectedCallback?: (data: WebSocketResponse) => void;
@@ -54,6 +57,13 @@ export class WebSocketClient {
   public connect(): Promise<string> {
     return new Promise((resolve, reject) => {
       try {
+        this.disconnectedByUser = false;
+        this.shouldReconnect = true;
+        if (this.reconnectTimeout) {
+          clearTimeout(this.reconnectTimeout);
+          this.reconnectTimeout = null;
+        }
+
         // 防止重复连接
         if (
           this.isConnecting ||
@@ -82,21 +92,27 @@ export class WebSocketClient {
           window.location.host;
         const url = `${protocol}//${host}/api/v1/pipelines/${this.pipelineId}/ws/connect?session_type=${this.sessionType}`;
 
-        this.ws = new WebSocket(url);
+        const socket = new WebSocket(url);
+        this.ws = socket;
 
         // 连接打开
-        this.ws.onopen = () => {
+        socket.onopen = () => {
+          if (this.disconnectedByUser || this.ws !== socket) {
+            socket.close();
+            return;
+          }
           this.isConnecting = false;
           const token = this.token || localStorage.getItem('token');
           const workspaceUuid = getActiveWorkspaceUuid();
           if (!token || !workspaceUuid) {
             const error = new Error('WebSocket认证信息缺失');
+            this.shouldReconnect = false;
             this.onErrorCallback?.(error);
-            this.ws?.close();
+            socket.close();
             reject(error);
             return;
           }
-          this.ws?.send(
+          socket.send(
             JSON.stringify({
               type: 'authenticate',
               token,
@@ -106,11 +122,17 @@ export class WebSocketClient {
         };
 
         // 接收消息
-        this.ws.onmessage = (event) => {
+        socket.onmessage = (event) => {
+          if (this.disconnectedByUser || this.ws !== socket) return;
           try {
             const data: WebSocketResponse = JSON.parse(event.data);
             this.handleMessage(data);
 
+            if (data.type === 'error' && !this.connectionId) {
+              reject(new Error(data.message || 'WebSocket连接失败'));
+              return;
+            }
+
             // 第一次连接成功
             if (data.type === 'connected' && data.connection_id) {
               // Only a fully authenticated runtime connection should reset
@@ -128,22 +150,36 @@ export class WebSocketClient {
         };
 
         // 连接关闭
-        this.ws.onclose = () => {
+        socket.onclose = () => {
+          if (this.ws === socket) {
+            this.ws = null;
+            this.connectionId = null;
+          }
           this.isConnecting = false;
           this.stopHeartbeat();
+          if (this.disconnectedByUser) return;
           this.onCloseCallback?.();
 
           // 自动重连
-          if (this.reconnectAttempts < this.maxReconnectAttempts) {
+          if (
+            this.shouldReconnect &&
+            this.reconnectAttempts < this.maxReconnectAttempts
+          ) {
             this.reconnectAttempts++;
-            setTimeout(() => {
+            this.reconnectTimeout = setTimeout(() => {
+              this.reconnectTimeout = null;
+              if (!this.shouldReconnect || this.disconnectedByUser) return;
               this.connect().catch(console.error);
             }, this.reconnectDelay * this.reconnectAttempts);
           }
         };
 
         // 连接错误
-        this.ws.onerror = (event) => {
+        socket.onerror = (event) => {
+          if (this.disconnectedByUser || this.ws !== socket) {
+            reject(new Error('WebSocket连接已取消'));
+            return;
+          }
           console.error('WebSocket错误:', event);
           this.isConnecting = false;
           const error = new Error('WebSocket连接失败');
@@ -208,6 +244,13 @@ export class WebSocketClient {
       case 'error':
         const error = new Error(data.message || '未知错误');
         this.onErrorCallback?.(error);
+        // Authentication/resource errors happen before the `connected`
+        // handshake. Retrying them cannot recover and would leak error toasts
+        // after the user leaves the Pipeline page.
+        if (!this.connectionId) {
+          this.shouldReconnect = false;
+          this.ws?.close();
+        }
         break;
 
       default:
@@ -270,21 +313,27 @@ export class WebSocketClient {
    * 断开连接
    */
   public disconnect() {
-    if (this.ws) {
+    this.disconnectedByUser = true;
-      this.stopHeartbeat();
+    this.shouldReconnect = false;
-
-      // 停止自动重连
     this.reconnectAttempts = this.maxReconnectAttempts;
-
+    if (this.reconnectTimeout) {
-      // 发送断开消息
+      clearTimeout(this.reconnectTimeout);
-      if (this.ws.readyState === WebSocket.OPEN) {
+      this.reconnectTimeout = null;
-        this.ws.send(JSON.stringify({ type: 'disconnect' }));
+    }
+
+    if (this.ws) {
+      this.stopHeartbeat();
+      const socket = this.ws;
+
+      // 发送断开消息
+      if (socket.readyState === WebSocket.OPEN) {
+        socket.send(JSON.stringify({ type: 'disconnect' }));
       }
 
-      this.ws.close();
       this.ws = null;
       this.connectionId = null;
       this.isConnecting = false;
+      socket.close();
     }
   }
 

web/tests/unit/processor-detail-workbench.test.mjs

@@ -45,6 +45,20 @@ test('agent and pipeline details share the split processor workbench', () => {
     websocketClient,
     /data\.type === 'connected'[\s\S]*this\.reconnectAttempts = 0/,
   );
+  assert.match(websocketClient, /private reconnectTimeout:/);
+  assert.match(websocketClient, /private disconnectedByUser = false/);
+  assert.match(
+    websocketClient,
+    /if \(!this\.connectionId\)[\s\S]*this\.shouldReconnect = false/,
+  );
+  assert.match(
+    pipelineDebug,
+    /wsClientRef\.current = wsClient;[\s\S]*await wsClient\.connect\(\)/,
+  );
+  assert.match(
+    pipelineDebug,
+    /if \(wsClientRef\.current !== wsClient\) return;/,
+  );
   assert.match(pipelineDebug, /data-slot="scroll-area-viewport"/);
   assert.doesNotMatch(pipelineDebug, /scrollIntoView/);
 });