diff --git a/packages/cli/src/modules/chat-hub/__tests__/chat-hub-execution.service.test.ts b/packages/cli/src/modules/chat-hub/__tests__/chat-hub-execution.service.test.ts index a8c9e82be49..d8facef304e 100644 --- a/packages/cli/src/modules/chat-hub/__tests__/chat-hub-execution.service.test.ts +++ b/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(); const executionManager = mock(); const activeExecutions = mock(); - const instanceSettings = mock(); + const instanceSettings = mock({ isMultiMain: false }); const chatStreamService = mock(); const chatHubWorkflowService = mock(); const chatHubExecutionStore = mock(); @@ -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({ id: 'user-1' }), + { provider: 'anthropic', model: 'claude-sonnet-5-5' }, + mock({ id: 'workflow-1', name: 'Chat' }), + mock(), + 'session-1', + 'previous-message-1', + null, + 'streaming', + ); + + expect(adapter).toBeDefined(); + const heartbeat = new StreamingWebhookResponseHeartbeat(adapter as Response); + expect(() => heartbeat.stop()).not.toThrow(); + }); + }); }); diff --git a/packages/cli/src/modules/chat-hub/chat-hub-execution.service.ts b/packages/cli/src/modules/chat-hub/chat-hub-execution.service.ts index 149a58ff6e5..2ce8338413c 100644 --- a/packages/cli/src/modules/chat-hub/chat-hub-execution.service.ts +++ b/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, };