fix(core): Handle each Chat Hub segment once in queue mode (#40791)

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

packages/@n8n/decorators/src/execution-lifecycle/lifecycle-metadata.ts

@@ -56,6 +56,11 @@ export type WorkflowExecuteAfterContext = {
 	retryOf?: string;
 	/** Who initiated the run. Unset means a regular user-initiated run. */
 	source?: WorkflowExecutionSource;
+	/**
+	 * False when this process only enqueued the run and waited for its job. In queue mode,
+	 * the process that ran the job fires its own `workflowExecuteAfter` for the same run.
+	 */
+	executedInThisProcess: boolean;
 };
 
 export type WorkflowExecuteResumeContext = {

packages/cli/src/execution-lifecycle/__tests__/execution-lifecycle-hooks.test.ts

@@ -6,6 +6,7 @@ import { ExecutionsConfig } from '@n8n/config';
 import type { Project, User } from '@n8n/db';
 import { ExecutionRepository, UserRepository } from '@n8n/db';
 import { LifecycleMetadata } from '@n8n/decorators';
+import type { WorkflowExecuteAfterContext } from '@n8n/decorators';
 import { Container } from '@n8n/di';
 import { stringify } from 'flatted';
 import {
@@ -536,6 +537,67 @@ describe('Execution Lifecycle Hooks', () => {
 				Container.set(LifecycleMetadata, original);
 			}
 		});
+
+		describe('executedInThisProcess', () => {
+			class RecordingHandler {
+				contexts: WorkflowExecuteAfterContext[] = [];
+
+				async onWorkflowExecuteAfter(ctx: WorkflowExecuteAfterContext) {
+					this.contexts.push(ctx);
+				}
+			}
+
+			const recorder = new RecordingHandler();
+
+			const runModuleHandler = async (hooks: ExecutionLifecycleHooks) => {
+				await hooks.runHook('workflowExecuteAfter', [successfulRun, {}]);
+				return recorder.contexts.at(-1);
+			};
+
+			let original: LifecycleMetadata;
+
+			beforeEach(() => {
+				recorder.contexts = [];
+				const recording = new LifecycleMetadata();
+				recording.register({
+					handlerClass: RecordingHandler as unknown as HandlerClass,
+					methodName: 'onWorkflowExecuteAfter',
+					eventName: 'workflowExecuteAfter',
+				});
+				Container.set(RecordingHandler, recorder);
+				original = Container.get(LifecycleMetadata);
+				Container.set(LifecycleMetadata, recording);
+			});
+
+			afterEach(() => {
+				Container.set(LifecycleMetadata, original);
+			});
+
+			it('should be false for the hooks of a process that only enqueued the run', async () => {
+				const hooks = getLifecycleHooksForScalingMain(
+					{ executionMode: 'manual', workflowData },
+					executionId,
+				);
+
+				expect(await runModuleHandler(hooks)).toMatchObject({ executedInThisProcess: false });
+			});
+
+			it('should be true for the hooks of a process that runs the execution', async () => {
+				const workerHooks = getLifecycleHooksForScalingWorker(
+					{ executionMode: 'manual', workflowData },
+					executionId,
+				);
+				const regularHooks = getLifecycleHooksForRegularMain(
+					{ executionMode: 'manual', workflowData },
+					executionId,
+				);
+
+				expect(await runModuleHandler(workerHooks)).toMatchObject({ executedInThisProcess: true });
+				expect(await runModuleHandler(regularHooks)).toMatchObject({
+					executedInThisProcess: true,
+				});
+			});
+		});
 	});
 
 	describe('getLifecycleHooksForRegularMain', () => {

packages/cli/src/execution-lifecycle/execution-lifecycle-hooks.ts

@@ -53,8 +53,13 @@ class ModulesHooksRegistry {
 	/**
 	 * `source` is not carried on `ExecutionLifecycleHooks`, so it is passed in
 	 * here to let module handlers tell agent-initiated runs from user ones.
+	 * `executedInThisProcess` is false for the hooks of a process that only enqueued the run.
 	 */
-	addHooks(hooks: ExecutionLifecycleHooks, source?: IWorkflowExecutionDataProcess['source']) {
+	addHooks(
+		hooks: ExecutionLifecycleHooks,
+		source?: IWorkflowExecutionDataProcess['source'],
+		{ executedInThisProcess = true }: { executedInThisProcess?: boolean } = {},
+	) {
 		const handlers = Container.get(LifecycleMetadata).getHandlers();
 
 		for (const { handlerClass, methodName, eventName } of handlers) {
@@ -72,6 +77,7 @@ class ModulesHooksRegistry {
 							executionId: this.executionId,
 							retryOf: this.retryOf,
 							source,
+							executedInThisProcess,
 						};
 
 						return await instance[methodName].call(instance, context);
@@ -977,7 +983,7 @@ export function getLifecycleHooksForScalingMain(
 	hooks.handlers.nodeExecuteBefore = [];
 	hooks.handlers.nodeExecuteAfter = [];
 
-	Container.get(ModulesHooksRegistry).addHooks(hooks, source);
+	Container.get(ModulesHooksRegistry).addHooks(hooks, source, { executedInThisProcess: false });
 
 	return hooks;
 }

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

@@ -171,8 +171,12 @@ describe('ChatHubExecutionWatcherService', () => {
 				...overrides,
 			}) as IRun;
 
-		const createAfterContext = (executionId: string, runData: IRun): WorkflowExecuteAfterContext =>
-			({ executionId, runData }) as WorkflowExecuteAfterContext;
+		const createAfterContext = (
+			executionId: string,
+			runData: IRun,
+			{ executedInThisProcess = true }: { executedInThisProcess?: boolean } = {},
+		): WorkflowExecuteAfterContext =>
+			({ executionId, runData, executedInThisProcess }) as WorkflowExecuteAfterContext;
 
 		it('should skip if context not found', async () => {
 			executionStore.get.mockResolvedValue(null);
@@ -457,9 +461,50 @@ describe('ChatHubExecutionWatcherService', () => {
 						createMessageOnResume: true,
 					});
 				});
+
+				it('should handle a segment once when the enqueuing process also reports it', async () => {
+					let stored = createContext({ responseMode: 'responseNodes' });
+					executionStore.get.mockImplementation(async () => stored);
+					executionStore.update.mockImplementation(async (_executionId, updates) => {
+						stored = { ...stored, ...updates };
+					});
+					chatHubExecutionService.extractMessage.mockReturnValue('Found the incident.');
+					executionPersistence.findSingleExecution.mockResolvedValue(
+						createExecution('@n8n/n8n-nodes-langchain.chat', { operation: 'send' }),
+					);
+					const runData = createRunData({ status: 'waiting' });
+
+					await service.handleWorkflowExecuteAfter(createAfterContext(EXECUTION_ID, runData));
+					await service.handleWorkflowExecuteAfter(
+						createAfterContext(EXECUTION_ID, runData, { executedInThisProcess: false }),
+					);
+
+					expect(messageRepository.createChatMessage).toHaveBeenCalledTimes(1);
+					expect(chatStreamService.endExecution).toHaveBeenCalledTimes(1);
+					expect(executionManager.runWorkflow).toHaveBeenCalledTimes(1);
+					expect(messageRepository.updateChatMessage).not.toHaveBeenCalledWith(
+						MOCK_NEW_MESSAGE_ID,
+						expect.anything(),
+					);
+				});
 			});
 		});
 
+		it('should ignore a call from a process that only enqueued the run', async () => {
+			executionStore.get.mockResolvedValue(createContext({ responseMode: 'responseNodes' }));
+
+			await service.handleWorkflowExecuteAfter(
+				createAfterContext(EXECUTION_ID, createRunData({ status: 'waiting' }), {
+					executedInThisProcess: false,
+				}),
+			);
+
+			expect(executionStore.get).not.toHaveBeenCalled();
+			expect(messageRepository.updateChatMessage).not.toHaveBeenCalled();
+			expect(chatStreamService.endExecution).not.toHaveBeenCalled();
+			expect(executionManager.runWorkflow).not.toHaveBeenCalled();
+		});
+
 		it('should not remove context when finished is false', async () => {
 			const context = createContext();
 			executionStore.get.mockResolvedValue(context);

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

@@ -1483,6 +1483,7 @@ describe('chatHub', () => {
 								runData,
 								newStaticData: {},
 								executionId,
+								executedInThisProcess: true,
 							});
 						});
 
@@ -1601,6 +1602,7 @@ describe('chatHub', () => {
 								runData,
 								newStaticData: {},
 								executionId,
+								executedInThisProcess: true,
 							});
 						});
 
@@ -1733,6 +1735,7 @@ describe('chatHub', () => {
 								runData,
 								newStaticData: {},
 								executionId,
+								executedInThisProcess: true,
 							});
 						});
 
@@ -1874,6 +1877,7 @@ describe('chatHub', () => {
 								runData,
 								newStaticData: {},
 								executionId,
+								executedInThisProcess: true,
 							});
 						});
 
@@ -1930,6 +1934,7 @@ describe('chatHub', () => {
 								runData,
 								newStaticData: {},
 								executionId: capturedExecutionId,
+								executedInThisProcess: true,
 							});
 						});
 					});
@@ -2064,6 +2069,7 @@ describe('chatHub', () => {
 								runData,
 								newStaticData: {},
 								executionId,
+								executedInThisProcess: true,
 							});
 						});
 
@@ -2205,6 +2211,7 @@ describe('chatHub', () => {
 								runData,
 								newStaticData: {},
 								executionId,
+								executedInThisProcess: true,
 							});
 						});
 
@@ -2298,6 +2305,7 @@ describe('chatHub', () => {
 									runData,
 									newStaticData: {},
 									executionId: capturedExecutionId,
+									executedInThisProcess: true,
 								});
 							});
 						});
@@ -2449,6 +2457,7 @@ describe('chatHub', () => {
 								runData,
 								newStaticData: {},
 								executionId,
+								executedInThisProcess: true,
 							});
 						});
 
@@ -2658,6 +2667,7 @@ describe('chatHub', () => {
 								runData,
 								newStaticData: {},
 								executionId,
+								executedInThisProcess: true,
 							});
 						});
 
@@ -2793,6 +2803,7 @@ describe('chatHub', () => {
 								runData,
 								newStaticData: {},
 								executionId,
+								executedInThisProcess: true,
 							});
 						});
 
@@ -2909,6 +2920,7 @@ describe('chatHub', () => {
 								runData,
 								newStaticData: {},
 								executionId,
+								executedInThisProcess: true,
 							});
 						});
 
@@ -3025,6 +3037,7 @@ describe('chatHub', () => {
 								runData,
 								newStaticData: {},
 								executionId,
+								executedInThisProcess: true,
 							});
 						});
 

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

@@ -117,6 +117,11 @@ export class ChatHubExecutionWatcherService {
 	async handleWorkflowExecuteAfter(ctx: WorkflowExecuteAfterContext): Promise<void> {
 		const { runData, executionId } = ctx;
 
+		// An auto-resume enqueues the next segment from the worker, so that worker also waits
+		// for the job and gets a second call for the same segment. The process that ran the
+		// segment handles it.
+		if (!ctx.executedInThisProcess) return;
+
 		const context = await this.executionStore.get(executionId);
 		if (!context) return; // Not a tracked chat hub execution
 
@@ -144,10 +149,7 @@ export class ChatHubExecutionWatcherService {
 			return;
 		}
 
-		// NOTE: This check is required because on multi-main/queue mode, the resumed execution's
-		// 'workflowExecuteAfter' hook fires one more time after it should have stopped in 'waiting'
-		// state on a Chat response node. On this final hook call the runData.status is 'success' even though
-		// the execution isn't finished. On single-main mode this does not happen.
+		// A `success` run that has not finished must not end the conversation.
 		// oxlint-disable-next-line typescript/no-deprecated
 		if (runData.finished) {
 			await this.pushFinalResults(context, message);