fix(console): isolate workflow SSE request callbacks

console/frontend/_tests_/sse-request.test.js

@@ -1,16 +1,8 @@
 import assert from 'node:assert/strict';
-import { readFileSync } from 'node:fs';
-import { dirname, resolve } from 'node:path';
 import test from 'node:test';
-import { fileURLToPath } from 'node:url';
 import { fetchEventSource } from '@microsoft/fetch-event-source';
 import { fetchSseWithContext } from '../src/utils/sse-request.ts';
-
-const frontendRoot = resolve(dirname(fileURLToPath(import.meta.url)), '..');
-const workflowChatSource = readFileSync(
-  resolve(frontendRoot, 'src/components/workflow/store/flow-chat-function.ts'),
-  'utf8'
-);
+import { createWorkflowSseLifecycle } from '../src/components/workflow/store/workflow-sse.ts';
 
 const originalDocument = globalThis.document;
 const originalWindow = globalThis.window;
@@ -153,37 +145,52 @@ test('throwing from an SSE error callback prevents POST request replay', async (
   assert.equal(requestCount, 1);
 });
 
-test('workflow POST streams stop retries and finalize active UI state', () => {
-  assert.doesNotMatch(
-    workflowChatSource,
-    /onerror\(\)\s*\{\s*get\(\)\.controllerRef\?\.abort\(\);\s*\}/,
-    'aborting and returning from onerror still schedules the default retry'
-  );
-  assert.equal(
-    workflowChatSource.match(/handleWorkflowSseError\(error, get/g)?.length,
-    3,
-    'chat, resume, and interrupt-abort POST streams must share no-retry handling'
-  );
-  assert.match(
-    workflowChatSource,
-    /get\(\)\.wsMessageStatus\s*!==\s*'end'[\s\S]*handleFlowStop\([\s\S]*throw requestError/,
-    'transport failures must finalize an active workflow before rejecting'
-  );
-  assert.equal(
-    workflowChatSource.match(/void fetchEventSource\(/g)?.length,
-    3,
-    'all workflow POST stream promises must be handled explicitly'
-  );
-  assert.equal(
-    workflowChatSource.match(
-      /onclose\(\) \{\s*handleWorkflowSseClose\(get\);\s*\}/g
-    )?.length,
-    2,
-    'chat and resume streams must finalize when the connection closes early'
-  );
-  assert.equal(
-    workflowChatSource.match(/\.catch\(\(\) => undefined\);/g)?.length,
-    3,
-    'expected workflow SSE rejections must not become unhandled promises'
-  );
+test('late callbacks from a completed request cannot finalize its successor', () => {
+  let activeRequest = 'request-a';
+  const finalized = [];
+  const handledMessages = [];
+  const requestA = createWorkflowSseLifecycle({
+    isCurrent: () => activeRequest === 'request-a',
+    finalize: () => finalized.push('request-a'),
+    handleMessage: message => {
+      handledMessages.push(`request-a:${message}`);
+      return message === 'terminal frame';
+    },
+  });
+
+  requestA.onMessage('terminal frame');
+  activeRequest = 'request-b';
+  const requestB = createWorkflowSseLifecycle({
+    isCurrent: () => activeRequest === 'request-b',
+    finalize: () => finalized.push('request-b'),
+    handleMessage: message => {
+      handledMessages.push(`request-b:${message}`);
+      return false;
+    },
+  });
+
+  requestA.onMessage('late frame');
+  requestA.onClose();
+  assert.throws(() => requestA.onError(new Error('late socket error')));
+  requestB.onMessage('current frame');
+  assert.deepEqual(finalized, []);
+  assert.deepEqual(handledMessages, [
+    'request-a:terminal frame',
+    'request-b:current frame',
+  ]);
+});
+
+test('an active stream that closes early is finalized only once', () => {
+  let finalizeCount = 0;
+  const lifecycle = createWorkflowSseLifecycle({
+    isCurrent: () => true,
+    finalize: () => {
+      finalizeCount += 1;
+    },
+  });
+
+  lifecycle.onClose();
+  assert.equal(finalizeCount, 1);
+  assert.throws(() => lifecycle.onError(new Error('late socket error')));
+  assert.equal(finalizeCount, 1);
 });

console/frontend/src/components/workflow/store/flow-chat-function.ts

@@ -31,6 +31,11 @@ import i18n from 'i18next';
 import { cloneDeep } from 'lodash';
 import { getFixedUrl, getAuthorization } from '@/components/workflow/utils';
 import { fetchEventSource } from '@microsoft/fetch-event-source';
+import {
+  createWorkflowSseLifecycle,
+  throwFatalWorkflowSseError,
+  type WorkflowSseLifecycle,
+} from './workflow-sse';
 
 const initInterruptChat: InterruptChatType = {
   eventId: '',
@@ -520,48 +525,37 @@ const handleMessageEnd = (data: WebSocketMessageData, get): void => {
   get().setInterruptChat({ ...initInterruptChat });
   handleRunningNodeStatus();
 };
-const handleWorkflowSseError = (
-  error: unknown,
+const createActiveWorkflowSseLifecycle = (
+  controller: AbortController,
   get,
-  finalizeExecution = true
-): never => {
-  const requestError =
-    error instanceof Error ? error : new Error(String(error));
-  console.error('[Workflow SSE] Request failed', requestError);
-  if (finalizeExecution && get().wsMessageStatus !== 'end') {
-    handleFlowStop(
-      {
-        code: -1,
-        message: i18n.t(
-          'workflow.nodes.chatDebugger.workflowConnectionInterrupted'
-        ),
-      },
-      get
-    );
-  }
-  throw requestError;
-};
-const handleWorkflowSseClose = (get): void => {
-  if (get().wsMessageStatus === 'end') return;
-
-  console.error('[Workflow SSE] Connection closed before completion');
-  handleFlowStop(
-    {
-      code: -1,
-      message: i18n.t(
-        'workflow.nodes.chatDebugger.workflowConnectionInterrupted'
+  handleSseMessage: (event: MessageEvent) => boolean
+): WorkflowSseLifecycle<MessageEvent> =>
+  createWorkflowSseLifecycle({
+    isCurrent: () => get().controllerRef === controller,
+    finalize: () =>
+      handleFlowStop(
+        {
+          code: -1,
+          message: i18n.t(
+            'workflow.nodes.chatDebugger.workflowConnectionInterrupted'
+          ),
+        },
+        get
       ),
-    },
-    get
-  );
-};
+    handleMessage: handleSseMessage,
+    onTransportError: error =>
+      console.error('[Workflow SSE] Request failed', error),
+  });
 const handleResumeChat = (content, get, set): void => {
   const currentFlow = useFlowsManager.getState().currentFlow;
   const nodes = useFlowStore.getState().nodes;
   const edges = useFlowStore.getState().edges;
+  get().controllerRef?.abort();
+  const controller = new AbortController();
   set({
     wsMessageStatus: 'start',
     debuggering: true,
+    controllerRef: controller,
   });
   pushAskToChatList(
     [
@@ -595,23 +589,31 @@ const handleResumeChat = (content, get, set): void => {
     params.version = get().versionId;
     params.promptDebugger = true;
   }
+  const sseLifecycle = createActiveWorkflowSseLifecycle(
+    controller,
+    get,
+    event => {
+      handleMessage(nodes, edges, event, get, set);
+      return get().wsMessageStatus === 'end';
+    }
+  );
   void fetchEventSource(url, {
     method: 'POST',
     headers: {
       'Content-Type': 'application/json',
       Authorization: getAuthorization(),
     },
     body: JSON.stringify(params),
-    signal: get().controllerRef?.signal,
+    signal: controller.signal,
     openWhenHidden: true,
     onerror(error) {
-      handleWorkflowSseError(error, get);
+      sseLifecycle.onError(error);
     },
     onclose() {
-      handleWorkflowSseClose(get);
+      sseLifecycle.onClose();
     },
     onmessage(e) {
-      handleMessage(nodes, edges, e, get, set);
+      sseLifecycle.onMessage(e);
     },
   }).catch(() => undefined);
   clearNodeStatus(get);
@@ -621,8 +623,10 @@ const runDebugger = (obj: unknown): void => {
   const currentFlow = useFlowsManager.getState().currentFlow;
   const historyVersion = useFlowsManager.getState().historyVersion;
   const url = getFixedUrl('/workflow/chat');
+  get().controllerRef?.abort();
+  const controller = new AbortController();
   set({
-    controllerRef: new AbortController(),
+    controllerRef: controller,
   });
   const inputs = {};
   const enterlist = enters ?? get().startNodeParams;
@@ -652,23 +656,31 @@ const runDebugger = (obj: unknown): void => {
     params.version = get().versionId;
     params.promptDebugger = true;
   }
+  const sseLifecycle = createActiveWorkflowSseLifecycle(
+    controller,
+    get,
+    event => {
+      handleMessage(nodes, edges, event, get, set);
+      return get().wsMessageStatus === 'end';
+    }
+  );
   void fetchEventSource(url, {
     method: 'POST',
     headers: {
       'Content-Type': 'application/json',
       Authorization: getAuthorization(),
     },
     body: JSON.stringify(params),
-    signal: get().controllerRef?.signal,
+    signal: controller.signal,
     openWhenHidden: true,
     onerror(error) {
-      handleWorkflowSseError(error, get);
+      sseLifecycle.onError(error);
     },
     onclose() {
-      handleWorkflowSseClose(get);
+      sseLifecycle.onClose();
     },
     onmessage(e) {
-      handleMessage(nodes, edges, e, get, set);
+      sseLifecycle.onMessage(e);
     },
   }).catch(() => undefined);
   clearNodeStatus(get);
@@ -949,9 +961,11 @@ const handleStopConversation = (get): void => {
       body: JSON.stringify(params),
       openWhenHidden: true,
       onerror(error) {
-        handleWorkflowSseError(error, get, false);
+        throwFatalWorkflowSseError(error);
       },
-    }).catch(() => undefined);
+    }).catch(error =>
+      console.error('[Workflow SSE] Abort request failed', error)
+    );
   }
   get().setChatList(chatList => [
     ...chatList,

console/frontend/src/components/workflow/store/workflow-sse.ts

@@ -0,0 +1,55 @@
+interface WorkflowSseLifecycleOptions<TMessage> {
+  isCurrent: () => boolean;
+  finalize: () => void;
+  handleMessage?: (message: TMessage) => boolean;
+  onTransportError?: (error: Error) => void;
+}
+
+export interface WorkflowSseLifecycle<TMessage> {
+  onClose: () => void;
+  onError: (error: unknown) => never;
+  onMessage: (message: TMessage) => void;
+}
+
+const normalizeError = (error: unknown): Error =>
+  error instanceof Error ? error : new Error(String(error));
+
+export const throwFatalWorkflowSseError = (error: unknown): never => {
+  throw normalizeError(error);
+};
+
+export const createWorkflowSseLifecycle = <TMessage = unknown>({
+  isCurrent,
+  finalize,
+  handleMessage,
+  onTransportError,
+}: WorkflowSseLifecycleOptions<TMessage>): WorkflowSseLifecycle<TMessage> => {
+  let completed = false;
+
+  const finalizeIfActive = (error: Error): void => {
+    if (completed || !isCurrent()) return;
+
+    completed = true;
+    onTransportError?.(error);
+    finalize();
+  };
+
+  return {
+    onClose(): void {
+      finalizeIfActive(
+        new Error('Workflow SSE connection closed before completion')
+      );
+    },
+    onError(error: unknown): never {
+      const requestError = normalizeError(error);
+      finalizeIfActive(requestError);
+      throw requestError;
+    },
+    onMessage(message: TMessage): void {
+      if (completed || !isCurrent()) return;
+      if (handleMessage?.(message)) {
+        completed = true;
+      }
+    },
+  };
+};