diff --git a/packages/@n8n/decorators/src/execution-lifecycle/lifecycle-metadata.ts b/packages/@n8n/decorators/src/execution-lifecycle/lifecycle-metadata.ts index d344041cc1e..5fd3c3700d0 100644 --- a/packages/@n8n/decorators/src/execution-lifecycle/lifecycle-metadata.ts +++ b/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 = { diff --git a/packages/cli/src/execution-lifecycle/__tests__/execution-lifecycle-hooks.test.ts b/packages/cli/src/execution-lifecycle/__tests__/execution-lifecycle-hooks.test.ts index 4c69981bdf5..2f4e294d290 100644 --- a/packages/cli/src/execution-lifecycle/__tests__/execution-lifecycle-hooks.test.ts +++ b/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', () => { diff --git a/packages/cli/src/execution-lifecycle/execution-lifecycle-hooks.ts b/packages/cli/src/execution-lifecycle/execution-lifecycle-hooks.ts index 3408c5c152b..82426d2bbd4 100644 --- a/packages/cli/src/execution-lifecycle/execution-lifecycle-hooks.ts +++ b/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; } diff --git a/packages/cli/src/modules/chat-hub/__tests__/chat-hub-execution-watcher.service.test.ts b/packages/cli/src/modules/chat-hub/__tests__/chat-hub-execution-watcher.service.test.ts index 18c643b7e74..5617406c17e 100644 --- a/packages/cli/src/modules/chat-hub/__tests__/chat-hub-execution-watcher.service.test.ts +++ b/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); diff --git a/packages/cli/src/modules/chat-hub/__tests__/chat-hub.service.integration.test.ts b/packages/cli/src/modules/chat-hub/__tests__/chat-hub.service.integration.test.ts index a1b2eb44086..e98afa85c24 100644 --- a/packages/cli/src/modules/chat-hub/__tests__/chat-hub.service.integration.test.ts +++ b/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, }); }); diff --git a/packages/cli/src/modules/chat-hub/chat-hub-execution-watcher.service.ts b/packages/cli/src/modules/chat-hub/chat-hub-execution-watcher.service.ts index 4cbf74a1c7d..3a93a8b73b9 100644 --- a/packages/cli/src/modules/chat-hub/chat-hub-execution-watcher.service.ts +++ b/packages/cli/src/modules/chat-hub/chat-hub-execution-watcher.service.ts @@ -117,6 +117,11 @@ export class ChatHubExecutionWatcherService { async handleWorkflowExecuteAfter(ctx: WorkflowExecuteAfterContext): Promise { 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);