fix(core): Let the streaming heartbeat stop on Chat Hub responses (#40792)

Co-authored-by: Claude Opus 5.5 <[email protected]>

packages/cli/src/modules/chat-hub/__tests__/chat-hub-execution.service.test.ts

@@ -1,7 +1,8 @@
 import type { Logger } from '@n8n/backend-common';
-import type { ExecutionRepository } from '@n8n/db';
+import type { ExecutionRepository, User } from '@n8n/db';
+import type { Response } from 'express';
 import type { InstanceSettings } from 'n8n-core';
-import type { IRun, IRunExecutionData } from 'n8n-workflow';
+import type { IRun, IRunExecutionData, IWorkflowBase } from 'n8n-workflow';
 import { NodeConnectionTypes } from 'n8n-workflow';
 import { mock } from 'vitest-mock-extended';
 
@@ -9,6 +10,7 @@ import type { ActiveExecutions } from '@/active-executions';
 import type { ChatExecutionManager } from '@/chat/chat-execution-manager';
 import type { ExecutionPersistence } from '@/executions/execution-persistence';
 import type { ExecutionService } from '@/executions/execution.service';
+import { StreamingWebhookResponseHeartbeat } from '@/webhooks/streaming-webhook-response-heartbeat';
 import type { WorkflowExecutionService } from '@/workflows/workflow-execution.service';
 
 import type { ChatHubExecutionStore } from '../chat-hub-execution-store.service';
@@ -27,7 +29,7 @@ describe('ChatHubExecutionService', () => {
 	const executionPersistence = mock<ExecutionPersistence>();
 	const executionManager = mock<ChatExecutionManager>();
 	const activeExecutions = mock<ActiveExecutions>();
-	const instanceSettings = mock<InstanceSettings>();
+	const instanceSettings = mock<InstanceSettings>({ isMultiMain: false });
 	const chatStreamService = mock<ChatStreamService>();
 	const chatHubWorkflowService = mock<ChatHubWorkflowService>();
 	const chatHubExecutionStore = mock<ChatHubExecutionStore>();
@@ -336,4 +338,32 @@ describe('ChatHubExecutionService', () => {
 			expect(service.extractErrorMessage(runData)).toBeUndefined();
 		});
 	});
+
+	describe('streaming response adapter', () => {
+		it('should let the streaming heartbeat stop without an error', async () => {
+			let adapter: Response | undefined;
+			workflowExecutionService.executeChatWorkflow.mockImplementation(
+				async (_user, _workflowData, _executionData, httpResponse) => {
+					adapter = httpResponse;
+					return { executionId: '1' };
+				},
+			);
+			activeExecutions.getPostExecutePromise.mockResolvedValue({ status: 'success' } as IRun);
+
+			await service.executeChatWorkflowWithCleanup(
+				mock<User>({ id: 'user-1' }),
+				{ provider: 'anthropic', model: 'claude-sonnet-5-5' },
+				mock<IWorkflowBase>({ id: 'workflow-1', name: 'Chat' }),
+				mock<IRunExecutionData>(),
+				'session-1',
+				'previous-message-1',
+				null,
+				'streaming',
+			);
+
+			expect(adapter).toBeDefined();
+			const heartbeat = new StreamingWebhookResponseHeartbeat(adapter as Response);
+			expect(() => heartbeat.stop()).not.toThrow();
+		});
+	});
 });

packages/cli/src/modules/chat-hub/chat-hub-execution.service.ts

@@ -659,6 +659,7 @@ export class ChatHubExecutionService {
 
 			on: (_event: string, _handler: (...args: unknown[]) => void) => adapter,
 			once: (_event: string, _handler: (...args: unknown[]) => void) => adapter,
+			off: (_event: string, _handler: (...args: unknown[]) => void) => adapter,
 			emit: (_event: string, ..._args: unknown[]) => true,
 		};