From 56aa3d8857ebb3304f28bc590e7eba36eab1219a Mon Sep 17 00:00:00 2001 From: Benjamin Schroth <68321970+schrothbn@users.noreply.github.com> Date: Thu, 1 Oct 2026 16:38:31 +0000 Subject: [PATCH] feat(core): Move n8n Chat production messages onto the agent queue (no-changelog) (#40102) --- .../__tests__/agent-chat.controller.test.ts | 124 ++++--- ...ent-execution-orchestrator.service.test.ts | 40 +++ .../__tests__/agent-execution.service.test.ts | 14 + ...ent-message-queue-consumer.service.test.ts | 62 ++++ .../__tests__/queued-message-id.test.ts | 13 +- .../modules/agents/agent-chat.controller.ts | 316 ++++++++++-------- .../agent-execution-orchestrator.service.ts | 36 +- .../modules/agents/agent-execution.service.ts | 30 +- .../agent-message-queue-consumer.service.ts | 46 ++- .../agents/agent-message-queue.service.ts | 90 ++--- .../agents/agent-turn-execution.service.ts | 23 ++ .../agent-execution.repository.ts | 16 +- .../agents/types/agent-queued-message.ts | 9 +- .../agents/utils/agent-thread-access.ts | 12 + .../modules/agents/utils/queued-message-id.ts | 4 +- .../agent-execution.repository.test.ts | 72 +++- 16 files changed, 630 insertions(+), 277 deletions(-) diff --git a/packages/cli/src/modules/agents/__tests__/agent-chat.controller.test.ts b/packages/cli/src/modules/agents/__tests__/agent-chat.controller.test.ts index c580661ac07..d9b604a6ab8 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-chat.controller.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-chat.controller.test.ts @@ -166,6 +166,9 @@ describe('AgentChatController route access scopes', () => { ['cancelProductionChatRun', 'agent:execute'], ['getProductionChatMessages', 'agent:read'], ['getProductionChatAttachment', 'agent:read'], + ['getProductionQueuedMessages', 'agent:read'], + ['updateProductionQueuedMessage', 'agent:execute'], + ['removeProductionQueuedMessage', 'agent:execute'], ['chat', 'agent:execute'], ['chatResume', 'agent:execute'], ['cancelChatRun', 'agent:execute'], @@ -215,6 +218,7 @@ describe('AgentChatController queue mutations', () => { ...params, userId: 'user-1', ...payload, + ...(operation === 'updatePending' ? { kind: 'preview' } : {}), }); }, ); @@ -1024,36 +1028,68 @@ describe('AgentChatController production n8n Chat', () => { expect(agentExecutionOrchestratorService.executeForN8nChatPublished).not.toHaveBeenCalled(); }); - it('records a published turn with a user-owned memory resource', async () => { - const { controller, agentsService, agentExecutionOrchestratorService } = makeController(); + it('queues a published turn with a user-owned memory resource and relays its stream', async () => { + const { controller, agentsService, messageQueue, previewStreams } = makeController(); agentsService.isN8nChatPublished.mockResolvedValue(true); - agentExecutionOrchestratorService.executeForN8nChatPublished.mockImplementation( - async function* (config) { - config.onExecutionStarted?.('exec-99', 'thread-1', ['message-1']); - yield { type: 'text-delta', id: 'text-1', delta: 'Hi' }; - }, + const writes: string[] = []; + const pending = controller.productionChat( + request as never, + makeSseResponse(writes), + 'agent-1', + { message: 'hello', messageId: 'c4b02d7b-2088-41ce-9c6b-faf8c7b83d8a' } as never, ); + await vi.waitFor(() => expect(messageQueue.enqueue).toHaveBeenCalled()); + expect(messageQueue.enqueue.mock.lastCall?.[0]).toMatchObject({ + agentId: 'agent-1', + projectId: 'project-1', + sessionMode: 'new', + source: 'n8n_chat_production', + payload: { + kind: 'n8n_chat', + message: 'hello', + messageId: 'c4b02d7b-2088-41ce-9c6b-faf8c7b83d8a', + userId: 'user-1', + resourceId: 'n8n-chat-production:user-1', + }, + }); + const threadId = messageQueue.enqueue.mock.lastCall![0].threadId; + const started = { + type: 'execution-started' as const, + executionId: 'exec-99', + sessionId: threadId, + inputMessageIds: ['message-1'], + }; + await vi.waitFor(() => expect(writes.join('')).toContain('message-queued')); + previewStreams.handleRelay({ queueId: 'queue-1', sequence: 1, event: started }); + previewStreams.handleRelay({ queueId: 'queue-1', sequence: 2, event: null }); + await pending; + expect( + writes.filter((line) => line.startsWith('data:')).map((line) => JSON.parse(line.slice(6))), + ).toEqual([{ type: 'message-queued', queueId: 'queue-1', sessionId: threadId }, started]); + }); + + it('keeps the client session ID for a new session so a retry is a duplicate', async () => { + const { controller, agentsService, agentExecutionService, messageQueue } = makeController(); + agentsService.isN8nChatPublished.mockResolvedValue(true); + messageQueue.enqueue.mockResolvedValue({ status: 'duplicate' }); const writes: string[] = []; await controller.productionChat(request as never, makeSseResponse(writes), 'agent-1', { message: 'hello', - } as never); - expect(agentExecutionOrchestratorService.executeForN8nChatPublished).toHaveBeenCalledWith( - expect.objectContaining({ - agentId: 'agent-1', - message: 'hello', - sessionMode: 'new', - memory: expect.objectContaining({ resourceId: 'n8n-chat-production:user-1' }), - }), - ); - expect(writes.some((line) => line.includes('"delta":"Hi"'))).toBe(true); - expect( - writes.filter((line) => line.startsWith('data:')).map((line) => JSON.parse(line.slice(6))), - ).toContainEqual({ - type: 'execution-started', - executionId: 'exec-99', sessionId: 'thread-1', - inputMessageIds: ['message-1'], + newSession: true, + } as never); + expect(agentExecutionService.canUseProductionChatThread).toHaveBeenCalledWith( + 'thread-1', + 'project-1', + 'agent-1', + 'user-1', + 'new', + ); + expect(messageQueue.enqueue.mock.lastCall?.[0]).toMatchObject({ + threadId: 'thread-1', + sessionMode: 'new', }); + expect(writes).toContain('data: {"type":"done"}\n\n'); }); it('rejects a foreign session before saving an attachment', async () => { @@ -1121,25 +1157,29 @@ describe('AgentChatController production n8n Chat', () => { expect(writes.some((line) => line.includes('agent_unavailable'))).toBe(true); }); - it('does not finish a turn that suspended for HITL', async () => { - const { controller, agentsService, agentExecutionOrchestratorService } = makeController(); + it('scopes the pending-message API to n8n Chat items', async () => { + const { controller, agentsService, messageQueue } = makeController(); agentsService.isN8nChatPublished.mockResolvedValue(true); - agentExecutionOrchestratorService.executeForN8nChatPublished.mockImplementation( - async function* () { - yield { - type: 'tool-call-suspended', - toolCallId: 'call-1', - toolName: 'ask_questions', - runId: 'run-1', - suspendPayload: { type: 'questions', questions: [] }, - }; - }, - ); - const writes: string[] = []; - await controller.productionChat(request as never, makeSseResponse(writes), 'agent-1', { - message: 'Ask me first', - } as never); - expect(writes.some((line) => line.includes('"type":"tool-call-suspended"'))).toBe(true); - expect(writes.some((line) => line.includes('"type":"done"'))).toBe(false); + const params = { + projectId: 'project-1', + agentId: 'agent-1', + threadId: 'thread-1', + queueId: '1', + }; + const req = { params, user: { id: 'user-1' } } as never; + await controller.getProductionQueuedMessages(req); + await controller.updateProductionQueuedMessage(req, makeSseResponse([]), { message: 'Edited' }); + await controller.removeProductionQueuedMessage(req); + const scope = { ...params, userId: 'user-1', kind: 'n8n_chat' }; + expect(messageQueue.listPending).toHaveBeenCalledWith(scope); + expect(messageQueue.updatePending).toHaveBeenCalledWith({ ...scope, message: 'Edited' }); + expect(messageQueue.removePending).toHaveBeenCalledWith(scope); + + agentsService.isN8nChatPublished.mockResolvedValue(false); + await expect(controller.getProductionQueuedMessages(req)).rejects.toThrow(NotFoundError); + await expect( + controller.updateProductionQueuedMessage(req, makeSseResponse([]), { message: 'x' }), + ).rejects.toThrow(NotFoundError); + await expect(controller.removeProductionQueuedMessage(req)).rejects.toThrow(NotFoundError); }); }); diff --git a/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts index dcae37f6375..11d749d71ec 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts @@ -2054,6 +2054,46 @@ describe('AgentExecutionOrchestratorService', () => { }, ); + it('finalizes an admitted execution with its telemetry when the runtime build fails', async () => { + const { service, runtimeCacheService, executionService, agentRepository } = makeService(); + const buildError = new UserError('Credential "OpenAI" not found'); + runtimeCacheService.getRuntime.mockRejectedValue(buildError); + agentRepository.findByIdAndProjectId.mockResolvedValue({ + id: agentId, + name: 'Support Agent', + schema, + activeVersion: { schema }, + integrations: [], + } as unknown as Agent); + + await expect( + collect( + service.executeForChatPublished({ + agentId, + projectId, + message: 'from slack', + memory: { threadId: 'thread-1', resourceId: 'platform-user-1' }, + integrationType: 'slack', + sandboxPrincipalHash: integrationPrincipalHash, + admittedExecution: { + executionId: 'admitted-1', + startedAt: new Date(), + inputMessageIds: ['message-1'], + }, + }), + ), + ).rejects.toBe(buildError); + + expect(executionService.startExecutionRecording).not.toHaveBeenCalled(); + expect(executionService.finalizeExecution).toHaveBeenCalledWith( + 'admitted-1', + expect.objectContaining({ + telemetry: expect.objectContaining({ runType: 'production' }), + record: expect.objectContaining({ finishReason: 'error' }), + }), + ); + }); + it('rethrows the build error without recording when the agent no longer exists', async () => { const { service, runtimeCacheService, executionService, agentRepository } = makeService(); const buildError = new Error('boom'); diff --git a/packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts index b9233676081..4eade5f0611 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts @@ -1684,6 +1684,20 @@ describe('AgentExecutionService', () => { ).toBe(allowed); }); + it('rejects a new session ID that another memory scope already uses', async () => { + agentExecutionThreadRepository.findOneBy.mockResolvedValue(null); + memoryBackend.getThread.mockResolvedValue(mock({ resourceId: 'draft-chat:user-1' })); + expect( + await service.canUseProductionChatThread( + 'thread-1', + 'project-1', + 'agent-1', + 'user-1', + 'new', + ), + ).toBe(false); + }); + it('rejects another owner and an unknown existing session', async () => { agentExecutionThreadRepository.findOneBy.mockResolvedValue( makeThread({ ownerId: 'other-user' }), diff --git a/packages/cli/src/modules/agents/__tests__/agent-message-queue-consumer.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-message-queue-consumer.service.test.ts index dd5302382b7..1ddaec34c83 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-message-queue-consumer.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-message-queue-consumer.service.test.ts @@ -8,6 +8,7 @@ import type { CredentialsService } from '@/credentials/credentials.service'; import { userHasScopes } from '@/permissions.ee/check-access'; import type { AgentChatExecutionService } from '../agent-chat-execution.service'; +import type { AgentExecutionOrchestratorService } from '../agent-execution-orchestrator.service'; import type { AgentExecutionService } from '../agent-execution.service'; import { AgentMessageQueueConsumer } from '../agent-message-queue-consumer.service'; import type { AgentMessageQueueService, ClaimedAgentMessage } from '../agent-message-queue.service'; @@ -32,6 +33,7 @@ describe('AgentMessageQueueConsumer', () => { const users = mock(); const chatExecutions = mock(); const integrations = mock(); + const orchestrator = mock(); const sender = { send: vi.fn(), close: vi.fn(async () => {}) }; let consumer: AgentMessageQueueConsumer; @@ -112,6 +114,7 @@ describe('AgentMessageQueueConsumer', () => { mock(), chatExecutions, integrations, + orchestrator, mockLogger(), ); }); @@ -177,6 +180,65 @@ describe('AgentMessageQueueConsumer', () => { ); }); + it('runs n8n Chat messages against the published agent on the claimed execution', async () => { + const item = claim('session'); + item.payload = { kind: 'n8n_chat', message: 'input', resourceId: 'n8n-chat:user' }; + item.item.payload = { kind: 'n8n_chat' }; + repository.findThreadIds.mockResolvedValue(['session']); + queue.claimNext.mockResolvedValueOnce(item).mockResolvedValue(null); + orchestrator.executeForN8nChatPublished.mockImplementation(async function* () { + yield { type: 'text-delta', id: 'text', delta: 'hi' }; + }); + consumer.start(); + await vi.waitFor(() => + expect(queue.settle).toHaveBeenCalledWith('session', item.admission.executionId), + ); + expect(chatExecutions.register).toHaveBeenCalledWith( + expect.objectContaining({ userId: 'user', productionN8nChat: true }), + expect.any(AbortController), + ); + expect(orchestrator.executeForN8nChatPublished).toHaveBeenCalledWith( + expect.objectContaining({ + user: expect.objectContaining({ id: 'user' }), + memory: { threadId: 'session', resourceId: 'n8n-chat:user' }, + sessionMode: 'existing', + admittedExecution: item.admission, + }), + ); + expect(testRuns.prepareDraftRun).not.toHaveBeenCalled(); + expect(sender.send).toHaveBeenCalledWith({ + type: 'done', + sessionId: 'session', + executionId: item.admission.executionId, + }); + expect(queue.recordFailure).not.toHaveBeenCalled(); + }); + + it('does not finish an n8n Chat turn that suspended for HITL', async () => { + const item = claim('session'); + item.payload = { kind: 'n8n_chat', message: 'input', resourceId: 'n8n-chat:user' }; + item.item.payload = { kind: 'n8n_chat' }; + repository.findThreadIds.mockResolvedValue(['session']); + queue.claimNext.mockResolvedValueOnce(item).mockResolvedValue(null); + orchestrator.executeForN8nChatPublished.mockImplementation(async function* () { + yield { + type: 'tool-call-suspended', + toolCallId: 'call-1', + toolName: 'ask_questions', + runId: 'run-1', + suspendPayload: { type: 'questions', questions: [] }, + }; + }); + consumer.start(); + await vi.waitFor(() => + expect(queue.settle).toHaveBeenCalledWith('session', item.admission.executionId), + ); + expect(sender.send).toHaveBeenCalledWith( + expect.objectContaining({ type: 'tool-call-suspended' }), + ); + expect(sender.send).not.toHaveBeenCalledWith(expect.objectContaining({ type: 'done' })); + }); + it.each([ 'missing owner', 'deleted owner', diff --git a/packages/cli/src/modules/agents/__tests__/queued-message-id.test.ts b/packages/cli/src/modules/agents/__tests__/queued-message-id.test.ts index e017fd184d5..1f5ccb05828 100644 --- a/packages/cli/src/modules/agents/__tests__/queued-message-id.test.ts +++ b/packages/cli/src/modules/agents/__tests__/queued-message-id.test.ts @@ -1,10 +1,13 @@ import { mock } from 'vitest-mock-extended'; -import type { QueuedIntegrationMessage, QueuedPreviewMessage } from '../types/agent-queued-message'; +import type { + QueuedIntegrationMessage, + QueuedUserChatMessage, +} from '../types/agent-queued-message'; import { queuedMessageId } from '../utils/queued-message-id'; describe('queuedMessageId', () => { - const preview: QueuedPreviewMessage = { + const preview: QueuedUserChatMessage = { kind: 'preview', userId: 'user-1', messageId: 'C4B02D7B-2088-41CE-9C6B-FAF8C7B83D8A', @@ -36,6 +39,12 @@ describe('queuedMessageId', () => { ).toBe('14926758-36d7-5db0-a865-962afe3ad861'); }); + it('separates n8n Chat submissions from Preview submissions', () => { + expect(queuedMessageId('agent-1', 'session-1', { ...preview, kind: 'n8n_chat' })).not.toBe( + queuedMessageId('agent-1', 'session-1', preview), + ); + }); + it('separates Preview agents, users, sessions, and submissions', () => { const ids = [ queuedMessageId('agent-1', 'session-1', preview), diff --git a/packages/cli/src/modules/agents/agent-chat.controller.ts b/packages/cli/src/modules/agents/agent-chat.controller.ts index 31c8df2e5b9..1db20875dc9 100644 --- a/packages/cli/src/modules/agents/agent-chat.controller.ts +++ b/packages/cli/src/modules/agents/agent-chat.controller.ts @@ -7,6 +7,7 @@ import { AgentChatQueueReorderDto, type AgentChatMessagesResponse, type AgentChatQueueResponse, + type AgentSseEvent, AgentChatResumeDto, MAX_AGENT_CHAT_ATTACHMENT_SIZE_BYTES, MAX_AGENT_CHAT_ATTACHMENT_SIZE_MB, @@ -39,7 +40,6 @@ import { AgentChatAttachmentService } from './agent-chat-attachment.service'; import type { AgentChatAttachment } from './entities/agent-chat-attachment.entity'; import type { StoredAttachmentRef } from './types/agent-chat-attachment'; import { AgentExecutionOrchestratorService } from './agent-execution-orchestrator.service'; -import { AgentExecutionRecordingError } from './agent-execution-recording.error'; import { AgentMessageQueueService } from './agent-message-queue.service'; import { AgentQueuedPreviewStreamService } from './agent-queued-preview-stream.service'; import { @@ -47,7 +47,11 @@ import { AgentTurnAlreadyRunningError, } from './agent-chat-execution.service'; import { AgentExecutionService } from './agent-execution.service'; -import { N8N_CHAT_PRODUCTION_SOURCE, threadBelongsTo } from './utils/agent-thread-access'; +import { + type AgentSessionMode, + N8N_CHAT_PRODUCTION_SOURCE, + threadBelongsTo, +} from './utils/agent-thread-access'; import { messagesToDto } from './agent-message-mapper'; import { type FlushableResponse, initSseStream } from './agent-sse-stream'; import { AgentTestChatService, chatThreadId } from './agent-test-chat.service'; @@ -163,6 +167,7 @@ export class AgentChatController { projectId: string, agentId: string, userId: string, + sessionMode: AgentSessionMode = 'existing', ): Promise { if ( !(await this.agentExecutionService.canUseProductionChatThread( @@ -170,12 +175,68 @@ export class AgentChatController { projectId, agentId, userId, - 'existing', + sessionMode, )) ) throw new NotFoundError('Session not found'); } + /** + * Enqueue a chat message and relay its execution to the browser stream. + * `accept` validates the request and stores attachments. It returns nothing after it sends its own error. + */ + private async relayQueuedMessage( + res: FlushableResponse, + accept: ( + send: (event: AgentSseEvent) => void, + abortSignal: AbortSignal, + ) => Promise[0] | undefined>, + ): Promise { + const delivery = initSseStream(res); + const { send, abortSignal } = delivery; + let accepted = false; + let subscription: ReturnType | undefined; + let attachments: StoredAttachmentRef[] | undefined; + try { + const input = await accept(send, abortSignal); + if (!input) return; + attachments = input.payload.attachments; + abortSignal.throwIfAborted(); + const result = await this.messageQueue.enqueue(input, (queueId) => { + subscription = this.queuedPreviewStreams.subscribe(queueId, delivery); + }); + if (result.status === 'duplicate') { + send({ type: 'done' }); + return; + } + accepted = true; + subscription?.accepted(); + send({ type: 'message-queued', queueId: result.item.id, sessionId: input.threadId }); + await subscription?.done; + } catch (error) { + send({ + type: 'error', + message: error instanceof Error ? error.message : 'Chat failed', + ...(error instanceof AgentTurnAlreadyRunningError + ? { errorCode: 'turn_already_running' } + : {}), + }); + } finally { + // Committed messages own their attachments, including after a disconnect. + if (!accepted && attachments?.length) { + await this.agentChatAttachmentService + .deleteByIds(attachments.map((ref) => ref.id)) + .catch(() => {}); + } + subscription?.close(); + delivery.close(); + } + } + + private assertQueueId(queueId: string): void { + if (!/^[1-9]\d*$/.test(queueId)) throw new BadRequestError('Invalid queue ID'); + } + @Post('/:agentId/n8n-chat', { usesTemplates: true }) @ProjectScope('agent:execute') async productionChat( @@ -185,84 +246,44 @@ export class AgentChatController { @Body payload: AgentChatMessageDto, ) { const { projectId } = req.params; - const execution = this.createChatExecution(res); - const { send, onChunk, abortSignal } = execution; - let executionId: string | undefined; - // `onExecutionRecorded` only fires once the turn finalizes. A disconnect or an - // error before that leaves the started id as the only way to name the execution. - let startedExecutionId: string | undefined; - const onExecutionStarted = (id: string, sessionId: string, inputMessageIds: string[]) => { - startedExecutionId = id; - execution.onExecutionStarted(id, sessionId, inputMessageIds); - }; - let storedAttachments: StoredAttachmentRef[] | undefined; - try { + const resourceId = productionChatMemoryResourceId(req.user.id); + await this.relayQueuedMessage(res, async (send) => { if (!(await this.agentsService.isN8nChatPublished(agentId, projectId))) { send({ type: 'error', message: 'This agent is not available in n8n Chat.', errorCode: 'agent_unavailable', }); - return; + return undefined; } const sessionMode = payload.sessionId && !payload.newSession ? 'existing' : 'new'; - const threadId = sessionMode === 'existing' ? payload.sessionId : randomUUID(); - if (!threadId) throw new NotFoundError('Session not found'); - if (sessionMode === 'existing') { - await this.requireProductionThread(threadId, projectId, agentId, req.user.id); - } - storedAttachments = await this.storeChatAttachments({ + // Keep a client session ID for a new session, so a retry stays a duplicate. + const threadId = payload.sessionId ?? randomUUID(); + await this.requireProductionThread(threadId, projectId, agentId, req.user.id, sessionMode); + const attachments = await this.storeChatAttachments({ attachments: payload.attachments, agentId, projectId, threadId, - resourceId: productionChatMemoryResourceId(req.user.id), + resourceId, source: N8N_CHAT_PRODUCTION_SOURCE, }); - abortSignal.throwIfAborted(); - const stream = this.agentExecutionOrchestratorService.executeForN8nChatPublished({ + return { agentId, projectId, - user: req.user, - message: payload.message, - memory: { threadId, resourceId: productionChatMemoryResourceId(req.user.id) }, - attachments: storedAttachments, + threadId, sessionMode, - onExecutionStarted, - onExecutionRecorded: (id) => { - executionId = id; + source: N8N_CHAT_PRODUCTION_SOURCE, + payload: { + kind: 'n8n_chat', + message: payload.message, + messageId: payload.messageId, + attachments, + userId: req.user.id, + resourceId, }, - abortSignal, - }); - let suspended = false; - let failed = false; - for await (const chunk of stream) { - onChunk(chunk); - if (chunk.type === 'tool-call-suspended') suspended = true; - if (chunk.type === 'error') failed = true; - } - executionId ??= startedExecutionId; - if (!suspended && !failed) { - send({ type: 'done', sessionId: threadId, ...(executionId ? { executionId } : {}) }); - } - } catch (error) { - executionId ??= startedExecutionId; - if (error instanceof AgentExecutionRecordingError) executionId ??= error.executionId; - if (!executionId && storedAttachments?.length) { - await this.agentChatAttachmentService - .deleteByIds(storedAttachments.map((ref) => ref.id)) - .catch(() => {}); - } - send({ - type: 'error', - message: error instanceof Error ? error.message : 'Chat failed', - ...(error instanceof AgentTurnAlreadyRunningError - ? { errorCode: 'turn_already_running' } - : {}), - }); - } finally { - execution.close(); - } + }; + }); } @Post('/:agentId/n8n-chat/resume', { usesTemplates: true }) @@ -274,12 +295,8 @@ export class AgentChatController { @Body payload: AgentChatResumeDto, ) { const execution = this.createChatExecution(res); - const { send, onChunk, abortSignal } = execution; - let startedExecutionId: string | undefined; - const onExecutionStarted = (id: string, sessionId: string, inputMessageIds: string[]) => { - startedExecutionId = id; - execution.onExecutionStarted(id, sessionId, inputMessageIds); - }; + const { send, onChunk, abortSignal, onExecutionStarted } = execution; + let executionId: string | undefined; try { if (!(await this.agentsService.isN8nChatPublished(agentId, req.params.projectId))) { send({ @@ -302,6 +319,9 @@ export class AgentChatController { source: N8N_CHAT_PRODUCTION_SOURCE, expectedMemory: { resourceId: productionChatMemoryResourceId(req.user.id) }, onExecutionStarted, + onExecutionRecorded: (id) => { + executionId = id; + }, abortSignal, }); let suspended = false; @@ -312,10 +332,7 @@ export class AgentChatController { if (chunk.type === 'error') failed = true; } if (!suspended && !failed) { - send({ - type: 'done', - ...(startedExecutionId ? { executionId: startedExecutionId } : {}), - }); + send({ type: 'done', ...(executionId ? { executionId } : {}) }); } } catch (error) { send({ @@ -349,12 +366,7 @@ export class AgentChatController { agentId, ); - const delivery = initSseStream(res); - const { send, abortSignal } = delivery; - let accepted = false; - let subscription: ReturnType | undefined; - let storedAttachments: StoredAttachmentRef[] | undefined; - try { + await this.relayQueuedMessage(res, async (send, abortSignal) => { const prepared = await this.agentTestRunService.prepareDraftRun({ agentId, projectId, @@ -364,10 +376,10 @@ export class AgentChatController { newSession, credentialProvider, }); - if (abortSignal.aborted) return; + if (abortSignal.aborted) return undefined; if (prepared.status === 'session_not_found') { send({ type: 'error', message: 'Session not found' }); - return; + return undefined; } if (prepared.status === 'agent_misconfigured') { send({ @@ -376,66 +388,32 @@ export class AgentChatController { errorCode: 'agent_misconfigured', missing: prepared.missing, }); - return; + return undefined; } - const threadId = prepared.sessionId; - storedAttachments = await this.storeChatAttachments({ - attachments, + const resourceId = draftChatMemoryResourceId(req.user.id); + return { agentId, projectId, threadId, - resourceId: draftChatMemoryResourceId(req.user.id), - }); - abortSignal.throwIfAborted(); - - const result = await this.messageQueue.enqueue( - { - agentId, - projectId, - threadId, - sessionMode: prepared.sessionMode, - source: 'chat', - payload: { - kind: 'preview', - message, - messageId, - attachments: storedAttachments, - userId: req.user.id, - resourceId: draftChatMemoryResourceId(req.user.id), - }, + sessionMode: prepared.sessionMode, + source: 'chat', + payload: { + kind: 'preview', + message, + messageId, + attachments: await this.storeChatAttachments({ + attachments, + agentId, + projectId, + threadId, + resourceId, + }), + userId: req.user.id, + resourceId, }, - (queueId) => { - subscription = this.queuedPreviewStreams.subscribe(queueId, delivery); - }, - ); - if (result.status === 'duplicate') { - send({ type: 'done' }); - return; - } - accepted = true; - subscription?.accepted(); - send({ type: 'message-queued', queueId: result.item.id, sessionId: threadId }); - await subscription?.done; - } catch (error) { - const errorMessage = error instanceof Error ? error.message : 'Chat failed'; - send({ - type: 'error', - message: errorMessage, - ...(error instanceof AgentTurnAlreadyRunningError - ? { errorCode: 'turn_already_running' } - : {}), - }); - } finally { - // Committed messages own their attachments, including after a disconnect. - if (!accepted && storedAttachments?.length) { - await this.agentChatAttachmentService - .deleteByIds(storedAttachments.map((ref) => ref.id)) - .catch(() => {}); - } - subscription?.close(); - delivery.close(); - } + }; + }); } @Post('/:agentId/chat/resume', { usesTemplates: true }) @@ -620,6 +598,57 @@ export class AgentChatController { await this.streamAttachment(attachment, res); } + @Get('/:agentId/n8n-chat/:threadId/queue') + @ProjectScope('agent:read') + async getProductionQueuedMessages( + req: AuthenticatedRequest<{ projectId: string; agentId: string; threadId: string }>, + ): Promise { + await this.requireProductionChat(req.params.agentId, req.params.projectId); + return await this.messageQueue.listPending({ + ...req.params, + userId: req.user.id, + kind: 'n8n_chat', + }); + } + + @Patch('/:agentId/n8n-chat/:threadId/queue/:queueId') + @ProjectScope('agent:execute') + async updateProductionQueuedMessage( + req: AuthenticatedRequest<{ + projectId: string; + agentId: string; + threadId: string; + queueId: string; + }>, + _res: Response, + @Body payload: AgentChatQueueUpdateDto, + ): Promise { + this.assertQueueId(req.params.queueId); + await this.requireProductionChat(req.params.agentId, req.params.projectId); + await this.messageQueue.updatePending({ + ...req.params, + userId: req.user.id, + message: payload.message, + kind: 'n8n_chat', + }); + } + + @Delete('/:agentId/n8n-chat/:threadId/queue/:queueId') + @ProjectScope('agent:execute') + async removeProductionQueuedMessage( + req: AuthenticatedRequest<{ + projectId: string; + agentId: string; + threadId: string; + queueId: string; + }>, + ) { + this.assertQueueId(req.params.queueId); + await this.requireProductionChat(req.params.agentId, req.params.projectId); + await this.messageQueue.removePending({ ...req.params, userId: req.user.id, kind: 'n8n_chat' }); + return { removed: true }; + } + @Get('/:agentId/chat/:threadId/queue') @ProjectScope('agent:read') async getQueuedMessages( @@ -627,7 +656,11 @@ export class AgentChatController { ): Promise { const agent = await this.agentsService.findById(req.params.agentId, req.params.projectId); if (!agent) throw new NotFoundError('Agent not found'); - return await this.messageQueue.listPending({ ...req.params, userId: req.user.id }); + return await this.messageQueue.listPending({ + ...req.params, + userId: req.user.id, + kind: 'preview', + }); } @Patch('/:agentId/chat/:threadId/queue/:queueId') @@ -642,13 +675,14 @@ export class AgentChatController { _res: Response, @Body payload: AgentChatQueueUpdateDto, ): Promise { - if (!/^[1-9]\d*$/.test(req.params.queueId)) throw new BadRequestError('Invalid queue ID'); + this.assertQueueId(req.params.queueId); const agent = await this.agentsService.findById(req.params.agentId, req.params.projectId); if (!agent) throw new NotFoundError('Agent not found'); await this.messageQueue.updatePending({ ...req.params, userId: req.user.id, message: payload.message, + kind: 'preview', }); } @@ -664,7 +698,7 @@ export class AgentChatController { _res: Response, @Body payload: AgentChatQueueReorderDto, ): Promise { - if (!/^[1-9]\d*$/.test(req.params.queueId)) throw new BadRequestError('Invalid queue ID'); + this.assertQueueId(req.params.queueId); const agent = await this.agentsService.findById(req.params.agentId, req.params.projectId); if (!agent) throw new NotFoundError('Agent not found'); await this.messageQueue.reorderPending({ @@ -685,10 +719,10 @@ export class AgentChatController { queueId: string; }>, ) { - if (!/^[1-9]\d*$/.test(req.params.queueId)) throw new BadRequestError('Invalid queue ID'); + this.assertQueueId(req.params.queueId); const agent = await this.agentsService.findById(req.params.agentId, req.params.projectId); if (!agent) throw new NotFoundError('Agent not found'); - await this.messageQueue.removePending({ ...req.params, userId: req.user.id }); + await this.messageQueue.removePending({ ...req.params, userId: req.user.id, kind: 'preview' }); return { removed: true }; } @@ -704,7 +738,7 @@ export class AgentChatController { _res: Response, @Body payload: AgentChatQueueSteerDto, ): Promise { - if (!/^[1-9]\d*$/.test(req.params.queueId)) throw new BadRequestError('Invalid queue ID'); + this.assertQueueId(req.params.queueId); const agent = await this.agentsService.findById(req.params.agentId, req.params.projectId); if (!agent) throw new NotFoundError('Agent not found'); await this.messageQueue.steer({ diff --git a/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts b/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts index c60b6ee43bd..09ac792b78d 100644 --- a/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts +++ b/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts @@ -724,11 +724,13 @@ export class AgentExecutionOrchestratorService { access: { accessScope: 'user', ownerId: user.id }, sessionMode, abortSignal, + admittedExecution: config.admittedExecution, }, ), async (runtime) => { const messageContext = this.createN8nChatMessageContext(memory, user.id); return this.streamChatResponse({ + admittedExecution: config.admittedExecution, access: { accessScope: 'user', ownerId: user.id }, messageContext, onAdmitted: async () => @@ -1076,7 +1078,6 @@ export class AgentExecutionOrchestratorService { try { return await this.runtimeCacheService.getRuntime(params); } catch (error) { - if (admittedExecution) throw error; abortSignal?.throwIfAborted(); const { agentId, projectId } = params; let agent; @@ -1091,22 +1092,25 @@ export class AgentExecutionOrchestratorService { params.usePublishedVersion && agent.activeVersion?.schema ? getPublishedAgentSnapshot(agent) : agent; - await this.turnExecutionService.recordFailedStart( - { - ...recording, - agentId, - agentName: selected.schema?.name ?? agent.name, - projectId, - telemetry: { - userId: params.attributionUserId ?? params.user?.id, - runType: params.usePublishedVersion ? 'production' : 'test', - configuration: buildAgentConfigurationTelemetry(selected), - }, + const failed = { + ...recording, + agentId, + agentName: selected.schema?.name ?? agent.name, + projectId, + telemetry: { + userId: params.attributionUserId ?? params.user?.id, + runType: params.usePublishedVersion ? ('production' as const) : ('test' as const), + configuration: buildAgentConfigurationTelemetry(selected), }, - error, - onExecutionRecorded, - { previewChat: params.previewChat, automaticContinuationRunId }, - ); + }; + if (admittedExecution) { + await this.turnExecutionService.recordFailedAdmission(admittedExecution, failed, error); + } else { + await this.turnExecutionService.recordFailedStart(failed, error, onExecutionRecorded, { + previewChat: params.previewChat, + automaticContinuationRunId, + }); + } } throw error; } diff --git a/packages/cli/src/modules/agents/agent-execution.service.ts b/packages/cli/src/modules/agents/agent-execution.service.ts index 69200a2d0f9..1f8eb6a21dc 100644 --- a/packages/cli/src/modules/agents/agent-execution.service.ts +++ b/packages/cli/src/modules/agents/agent-execution.service.ts @@ -44,11 +44,14 @@ import { } from './execution-log/agent-execution-log-store'; import { N8nMemory } from './integrations/n8n-memory'; import { N8NCheckpointStorage } from './integrations/n8n-checkpoint-storage'; -import { draftChatMemoryResourceId } from './utils/agent-memory-scope'; import { + draftChatMemoryResourceId, + productionChatMemoryResourceId, +} from './utils/agent-memory-scope'; +import { + canContinueThreadInN8nChat, canContinueThreadInPreview, canUseTopLevelDraftThread, - N8N_CHAT_PRODUCTION_SOURCE, threadBelongsTo, type AgentSessionMode, } from './utils/agent-thread-access'; @@ -901,7 +904,7 @@ export class AgentExecutionService { return canContinueThreadInPreview(thread, userId, sources.get(threadId)); } if (options.sessionMode === 'existing') return false; - return await this.canUseUnrecordedDraftThread(threadId, agentId, userId); + return await this.canUseUnrecordedThread(threadId, agentId, draftChatMemoryResourceId(userId)); } async canUseProductionChatThread( @@ -912,24 +915,31 @@ export class AgentExecutionService { sessionMode: AgentSessionMode, ): Promise { const thread = await this.findThreadById(threadId); - if (!thread) return sessionMode === 'new'; + if (!thread) + return ( + sessionMode === 'new' && + (await this.canUseUnrecordedThread( + threadId, + agentId, + productionChatMemoryResourceId(userId), + )) + ); if ( thread.projectId !== projectId || thread.agentId !== agentId || - !canUseTopLevelDraftThread(thread, userId) || - thread.taskId !== null + !canUseTopLevelDraftThread(thread, userId) ) return false; const sources = await this.agentExecutionRepository.findFirstSourceByThreadIds([threadId]); - return sources.get(threadId) === N8N_CHAT_PRODUCTION_SOURCE; + return canContinueThreadInN8nChat(thread, userId, sources.get(threadId)); } - private async canUseUnrecordedDraftThread( + /** A session ID without a thread is free only when no other memory scope uses it. */ + private async canUseUnrecordedThread( threadId: string, agentId: string, - userId: string, + resourceId: string, ): Promise { - const resourceId = draftChatMemoryResourceId(userId); const memory = await this.n8nMemory.getImplementation(agentId).getThread(threadId); if (memory && memory.resourceId !== resourceId) return false; return await this.checkpointStorage.hasNoConflictingThreadResource( diff --git a/packages/cli/src/modules/agents/agent-message-queue-consumer.service.ts b/packages/cli/src/modules/agents/agent-message-queue-consumer.service.ts index 0c55a2292c9..02ddc7bab63 100644 --- a/packages/cli/src/modules/agents/agent-message-queue-consumer.service.ts +++ b/packages/cli/src/modules/agents/agent-message-queue-consumer.service.ts @@ -10,6 +10,7 @@ import { userHasScopes } from '@/permissions.ee/check-access'; import { AgentsCredentialProvider } from './adapters/agents-credential-provider'; import { AgentChatExecutionService } from './agent-chat-execution.service'; +import { AgentExecutionOrchestratorService } from './agent-execution-orchestrator.service'; import { AgentExecutionService } from './agent-execution.service'; import { AgentMessageQueueService, type ClaimedAgentMessage } from './agent-message-queue.service'; import { AgentQueuedPreviewStreamService } from './agent-queued-preview-stream.service'; @@ -37,6 +38,7 @@ export class AgentMessageQueueConsumer { private readonly credentialsService: CredentialsService, private readonly chatExecutionService: AgentChatExecutionService, private readonly integrations: ChatIntegrationService, + private readonly orchestrator: AgentExecutionOrchestratorService, private readonly logger: Logger, ) {} @@ -95,7 +97,7 @@ export class AgentMessageQueueConsumer { try { const claim = await this.queue.claimNext(threadId, async (item, thread, ctx) => { if (this.stopped) return false; - if (item.payload.kind === 'preview') return true; + if (item.payload.kind !== 'integration') return true; const source = item.message.origin?.source; if (!source) throw new UnexpectedError('Queued integration input has no source'); const connection = await this.repository.findPublishedConnection( @@ -132,10 +134,10 @@ export class AgentMessageQueueConsumer { this.executionService.getAbortSignal(admission.executionId), ]); const sender = - payload.kind === 'preview' ? this.previewStreams.createSender(item.id) : undefined; + payload.kind === 'integration' ? undefined : this.previewStreams.createSender(item.id); try { - if (payload.kind === 'preview' && sender) { - const user = await this.getPreviewOwner(thread); + if (sender) { + const user = await this.getThreadOwner(thread); this.chatExecutionService.register( { projectId: thread.projectId, @@ -143,6 +145,7 @@ export class AgentMessageQueueConsumer { threadId: thread.id, userId: user.id, executionId: admission.executionId, + productionN8nChat: payload.kind === 'n8n_chat', }, controller, ); @@ -153,9 +156,10 @@ export class AgentMessageQueueConsumer { inputMessageIds: admission.inputMessageIds, message: payload.message, }); - await this.chatExecutionService.settle( - admission.executionId, - async () => await this.consumePreview(claim, user, signal, sender.send), + await this.chatExecutionService.settle(admission.executionId, async () => + payload.kind === 'n8n_chat' + ? await this.consumeN8nChat(claim, user, signal, sender.send) + : await this.consumePreview(claim, user, signal, sender.send), ); } else { await this.consumeIntegration(claim, signal, bridge); @@ -177,7 +181,7 @@ export class AgentMessageQueueConsumer { } } - private async getPreviewOwner(thread: AgentExecutionThread): Promise { + private async getThreadOwner(thread: AgentExecutionThread): Promise { if (!thread.ownerId) throw new UserError('You can no longer execute this agent'); const user = await this.userRepository.findByIdWithRole(thread.ownerId); if ( @@ -247,6 +251,32 @@ export class AgentMessageQueueConsumer { send({ type: 'done', sessionId: thread.id, executionId: admission.executionId }); } + private async consumeN8nChat( + claim: ClaimedAgentMessage, + user: User, + signal: AbortSignal, + send: (event: AgentSseEvent) => void, + ): Promise { + const { thread, admission, payload } = claim; + const stream = this.orchestrator.executeForN8nChatPublished({ + agentId: thread.agentId, + projectId: thread.projectId, + user, + message: payload.message, + memory: { threadId: thread.id, resourceId: payload.resourceId }, + attachments: payload.attachments, + sessionMode: 'existing', + admittedExecution: admission, + abortSignal: signal, + }); + let completed = true; + for await (const chunk of stream) { + emitChunkEvents(chunk, send); + if (chunk.type === 'tool-call-suspended' || chunk.type === 'error') completed = false; + } + if (completed) send({ type: 'done', sessionId: thread.id, executionId: admission.executionId }); + } + private async consumeIntegration( claim: ClaimedAgentMessage, signal: AbortSignal, diff --git a/packages/cli/src/modules/agents/agent-message-queue.service.ts b/packages/cli/src/modules/agents/agent-message-queue.service.ts index 670f913582c..34b8dfc951a 100644 --- a/packages/cli/src/modules/agents/agent-message-queue.service.ts +++ b/packages/cli/src/modules/agents/agent-message-queue.service.ts @@ -28,21 +28,34 @@ import type { AgentQueuedMessage, AgentQueueDispatch, QueuedIntegrationMessage, - QueuedPreviewMessage, + QueuedUserChatMessage, } from './types/agent-queued-message'; -import { canContinueThreadInPreview, type AgentSessionMode } from './utils/agent-thread-access'; +import { + canContinueThreadInN8nChat, + canContinueThreadInPreview, + type AgentSessionMode, +} from './utils/agent-thread-access'; import { buildInboundUserMessage, readInboundUserMessage } from './utils/inbound-attachments'; import { queuedMessageId } from './utils/queued-message-id'; export interface ClaimedAgentMessage { item: AgentMessageQueue; - payload: Omit | QueuedIntegrationMessage; + payload: Omit | QueuedIntegrationMessage; thread: AgentExecutionThread; /** Admission identifies the committed execution that the turn pipeline must reuse. */ admission: AgentExecutionAdmission; recording: StartExecutionParams; } +interface PendingMessageScope { + projectId: string; + agentId: string; + threadId: string; + userId: string; + /** The chat surface that owns the session. */ + kind: QueuedUserChatMessage['kind']; +} + @Service() export class AgentMessageQueueService { /** The main consumer installs this callback. Webhook processes only enqueue. */ @@ -88,9 +101,9 @@ export class AgentMessageQueueService { userMessage: payload.message, resourceId: payload.resourceId, access: - payload.kind === 'preview' - ? { accessScope: 'user', ownerId: payload.userId } - : { accessScope: 'project', ownerId: null }, + payload.kind === 'integration' + ? { accessScope: 'project', ownerId: null } + : { accessScope: 'user', ownerId: payload.userId }, }, ctx, ); @@ -112,7 +125,7 @@ export class AgentMessageQueueService { } throw error; } - if (payload.kind === 'preview') this.updates.notifyQueueUpdated(item.threadId); + if (payload.kind !== 'integration') this.updates.notifyQueueUpdated(item.threadId); this.onAvailable?.(item.threadId); return { status: 'accepted', item }; } @@ -126,7 +139,7 @@ export class AgentMessageQueueService { ): Promise { const { message, resourceId, attachments = [], ...dispatch } = payload; const [content] = buildInboundUserMessage(message, attachments); - let queueDispatch: AgentQueueDispatch = { kind: 'preview' }; + let queueDispatch: AgentQueueDispatch; let modelContent: Message | undefined; let author: AgentMessageAuthor | undefined; let origin: AgentMessageOrigin = { source }; @@ -153,6 +166,8 @@ export class AgentMessageQueueService { platformMessageId, platformThreadId, }; + } else { + queueDispatch = { kind: dispatch.kind }; } const input = await this.messages.createInput( { id: messageId, threadId, resourceId, content, modelContent, author, origin }, @@ -161,22 +176,19 @@ export class AgentMessageQueueService { return await this.repository.enqueue(threadId, input.id, queueDispatch, ctx); } - async listPending(input: { - projectId: string; - agentId: string; - threadId: string; - userId: string; - }): Promise { + async listPending(input: PendingMessageScope): Promise { + const { kind } = input; const thread = await this.threadRepository.findOneBy({ id: input.threadId }); - // A client-created Preview session can have no accepted messages yet. + // A client-created session can have no accepted messages yet. if (!thread) return { items: [], steerableExecutionId: null }; - await this.assertPreviewAccess(thread, input); + await this.assertUserChatAccess(thread, input); const items = await this.repository.listPending(thread.id); - const steerable = await this.steering.findEligible(thread); + // Only Preview executions accept steering. + const steerable = kind === 'preview' ? await this.steering.findEligible(thread) : null; return { steerableExecutionId: steerable?.id ?? null, items: items - .filter((item) => item.payload.kind === 'preview') + .filter((item) => item.payload.kind === kind) // Show accepted steers first. Keep future turns in their saved queue order. .sort( (a, b) => @@ -192,19 +204,13 @@ export class AgentMessageQueueService { }; } - async removePending(input: { - projectId: string; - agentId: string; - threadId: string; - userId: string; - queueId: string; - }): Promise { + async removePending(input: PendingMessageScope & { queueId: string }): Promise { const removed = await this.txRunner.run({}, async (ctx) => { const thread = await this.threadRepository.lockById(input.threadId, ctx); if (!thread) throw new NotFoundError('Session not found'); - await this.assertPreviewAccess(thread, input, ctx); + await this.assertUserChatAccess(thread, input, ctx); const item = await this.repository.findItem(thread.id, input.queueId, ctx); - if (!item || item.payload.kind !== 'preview') + if (!item || item.payload.kind !== input.kind) throw new NotFoundError('Queued message not found'); if ( item.executionId !== null || @@ -223,20 +229,15 @@ export class AgentMessageQueueService { await this.attachments.deleteByIds(removed.attachmentIds); } - async updatePending(input: { - projectId: string; - agentId: string; - threadId: string; - userId: string; - queueId: string; - message: string; - }): Promise { + async updatePending( + input: PendingMessageScope & { queueId: string; message: string }, + ): Promise { await this.txRunner.run({}, async (ctx) => { const thread = await this.threadRepository.lockById(input.threadId, ctx); if (!thread) throw new NotFoundError('Session not found'); - await this.assertPreviewAccess(thread, input, ctx); + await this.assertUserChatAccess(thread, input, ctx); const item = await this.repository.findItem(thread.id, input.queueId, ctx); - if (!item || item.payload.kind !== 'preview') + if (!item || item.payload.kind !== input.kind) throw new NotFoundError('Queued message not found'); if (item.executionId !== null) throw new ConflictError('This message has already started'); if (item.steeringExecutionId !== null) @@ -263,7 +264,7 @@ export class AgentMessageQueueService { await this.txRunner.run({}, async (ctx) => { const thread = await this.threadRepository.lockById(input.threadId, ctx); if (!thread) throw new NotFoundError('Session not found'); - await this.assertPreviewAccess(thread, input, ctx); + await this.assertUserChatAccess(thread, { ...input, kind: 'preview' }, ctx); const moved = await this.repository.movePending( thread.id, input.queueId, @@ -289,7 +290,7 @@ export class AgentMessageQueueService { await this.txRunner.run({}, async (ctx) => { const thread = await this.threadRepository.lockById(input.threadId, ctx); if (!thread) throw new NotFoundError('Session not found'); - await this.assertPreviewAccess(thread, input, ctx); + await this.assertUserChatAccess(thread, { ...input, kind: 'preview' }, ctx); const item = await this.repository.findItem(thread.id, input.queueId, ctx); if (!item || item.payload.kind !== 'preview') throw new ConflictError('This message is no longer available'); @@ -308,16 +309,19 @@ export class AgentMessageQueueService { this.updates.notifyQueueUpdated(input.threadId); } - private async assertPreviewAccess( + private async assertUserChatAccess( thread: AgentExecutionThread, - input: { projectId: string; agentId: string; userId: string }, + input: Omit, ctx: OperationContext = {}, ): Promise { const sources = await this.executionRepository.findFirstSourceByThreadIds([thread.id], ctx); + const source = sources.get(thread.id); if ( thread.projectId !== input.projectId || thread.agentId !== input.agentId || - !canContinueThreadInPreview(thread, input.userId, sources.get(thread.id)) + !(input.kind === 'n8n_chat' + ? canContinueThreadInN8nChat(thread, input.userId, source) + : canContinueThreadInPreview(thread, input.userId, source)) ) { throw new NotFoundError('Session not found'); } @@ -449,7 +453,7 @@ export class AgentMessageQueueService { ...readInboundUserMessage(item.message.content), resourceId: item.message.resourceId, }; - if (item.payload.kind === 'preview') return { ...item.payload, ...input }; + if (item.payload.kind !== 'integration') return { ...item.payload, ...input }; if (!item.message.author) throw new UnexpectedError('Queued integration input has no author'); const { origin } = item.message; if (!origin?.source || !origin.integrationConnectionId || !origin.platformThreadId) diff --git a/packages/cli/src/modules/agents/agent-turn-execution.service.ts b/packages/cli/src/modules/agents/agent-turn-execution.service.ts index 46921a0165d..73ea021939b 100644 --- a/packages/cli/src/modules/agents/agent-turn-execution.service.ts +++ b/packages/cli/src/modules/agents/agent-turn-execution.service.ts @@ -342,6 +342,29 @@ export class AgentTurnExecutionService { onExecutionRecorded?.(recordedId); } + /** Finalize an admitted execution whose runtime could not start. */ + async recordFailedAdmission( + admission: AgentExecutionAdmission, + params: StartExecutionParams, + executionError: unknown, + ): Promise { + const recorder = this.createRecorder( + undefined, + undefined, + undefined, + undefined, + admission.startedAt, + ); + recorder.record({ type: 'error', error: executionError }); + recorder.record({ type: 'finish', finishReason: 'error' }); + await this.finalizeExecution({ + executionId: admission.executionId, + executionStarted: false, + executionError, + params: { ...params, record: recorder.getMessageRecord() }, + }); + } + async recordFailedStart( params: StartExecutionParams, executionError: unknown, diff --git a/packages/cli/src/modules/agents/repositories/agent-execution.repository.ts b/packages/cli/src/modules/agents/repositories/agent-execution.repository.ts index 310bf36cb00..e7c439afc20 100644 --- a/packages/cli/src/modules/agents/repositories/agent-execution.repository.ts +++ b/packages/cli/src/modules/agents/repositories/agent-execution.repository.ts @@ -1,7 +1,7 @@ import type { AgentExecutionStatus } from '@n8n/api-types'; import { BaseRepository, TransactionRunner, type OperationContext } from '@n8n/db'; import { Service } from '@n8n/di'; -import { DataSource, IsNull, Not } from '@n8n/typeorm'; +import { DataSource, In, IsNull, Not } from '@n8n/typeorm'; import type { QueryDeepPartialEntity } from '@n8n/typeorm/query-builder/QueryPartialEntity'; import { AgentExecution } from '../entities/agent-execution.entity'; @@ -242,7 +242,19 @@ export class AgentExecutionRepository extends BaseRepository { ) .getRawMany<{ threadId: string; source: string }>(); - return new Map(rows.map((r) => [r.threadId, r.source])); + const sources = new Map(rows.map((r) => [r.threadId, r.source])); + const unrecorded = threadIds.filter((id) => !sources.has(id)); + if (unrecorded.length === 0) return sources; + // Accepted input keeps its origin after removal from the queue. It names the surface before any execution. + const inputs = await this.managerFor(ctx).find(AgentMessageEntity, { + where: { threadId: In(unrecorded), origin: Not(IsNull()) }, + order: { createdAt: 'ASC', id: 'ASC' }, + }); + for (const input of inputs) { + const source = input.origin?.source; + if (source && !sources.has(input.threadId)) sources.set(input.threadId, source); + } + return sources; } async findLatestStatusesByThreadIds( diff --git a/packages/cli/src/modules/agents/types/agent-queued-message.ts b/packages/cli/src/modules/agents/types/agent-queued-message.ts index f439012ce04..c5c1ce9930a 100644 --- a/packages/cli/src/modules/agents/types/agent-queued-message.ts +++ b/packages/cli/src/modules/agents/types/agent-queued-message.ts @@ -15,8 +15,9 @@ interface QueuedMessageInput { attachments?: StoredAttachmentRef[]; } -export interface QueuedPreviewMessage extends QueuedMessageInput { - kind: 'preview'; +/** A message from an n8n user. Preview runs the draft agent. n8n Chat runs the published agent. */ +export interface QueuedUserChatMessage extends QueuedMessageInput { + kind: 'preview' | 'n8n_chat'; userId: string; messageId?: string; } @@ -34,11 +35,11 @@ export interface QueuedIntegrationMessage extends QueuedMessageInput { slackThreadContext?: BridgeExecutionContext['slackThreadContext']; } -export type AgentQueuedMessage = QueuedPreviewMessage | QueuedIntegrationMessage; +export type AgentQueuedMessage = QueuedUserChatMessage | QueuedIntegrationMessage; /** Queue storage keeps dispatch data. Conversation input belongs to the referenced message. */ export type AgentQueueDispatch = - | { kind: 'preview' } + | { kind: QueuedUserChatMessage['kind'] } | (Omit< QueuedIntegrationMessage, keyof QueuedMessageInput | 'modelMessage' | 'author' | 'platformThreadId' | 'messageContext' diff --git a/packages/cli/src/modules/agents/utils/agent-thread-access.ts b/packages/cli/src/modules/agents/utils/agent-thread-access.ts index 187f8f3293f..f97c15b9d03 100644 --- a/packages/cli/src/modules/agents/utils/agent-thread-access.ts +++ b/packages/cli/src/modules/agents/utils/agent-thread-access.ts @@ -33,6 +33,18 @@ export function canContinueThreadInPreview( ); } +export function canContinueThreadInN8nChat( + thread: AgentExecutionThread, + userId: string, + source?: string | null, +): boolean { + return ( + canUseTopLevelDraftThread(thread, userId) && + thread.taskId === null && + source === N8N_CHAT_PRODUCTION_SOURCE + ); +} + export function canUseTopLevelDraftThread(thread: AgentExecutionThread, userId: string): boolean { return ( thread.accessScope === 'user' && thread.ownerId === userId && thread.parentThreadId === null diff --git a/packages/cli/src/modules/agents/utils/queued-message-id.ts b/packages/cli/src/modules/agents/utils/queued-message-id.ts index ac514999e23..645c44b66bf 100644 --- a/packages/cli/src/modules/agents/utils/queued-message-id.ts +++ b/packages/cli/src/modules/agents/utils/queued-message-id.ts @@ -10,11 +10,11 @@ export function queuedMessageId( threadId: string, payload: AgentQueuedMessage, ): string | undefined { - if (payload.kind === 'preview') { + if (payload.kind !== 'integration') { if (!payload.messageId) return undefined; return uuidv5( JSON.stringify([ - 'preview', + payload.kind, agentId, payload.userId, threadId, diff --git a/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts b/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts index b2521ab1c4f..0dd3ea5e9f8 100644 --- a/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts +++ b/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts @@ -1621,7 +1621,7 @@ describe('AgentExecutionRepository', () => { const initialInputs = await services.messageRepository.findExecutionInputs([ predecessor.id, ]); - const target = { projectId, agentId, threadId, userId: owner.id }; + const target = { projectId, agentId, threadId, userId: owner.id, kind: 'preview' as const }; const b = await enqueue(services, input(threadId, 'B')); const c = await enqueue(services, input(threadId, 'C')); const runtime = fixture.makeAgent(services.checkpointStorage.getStorage(agentId)); @@ -1701,7 +1701,7 @@ describe('AgentExecutionRepository', () => { const local = recordingServices(); const remote = recordingServices(undefined, peer); const threadId = uuid(); - const target = { projectId, agentId, threadId, userId: owner.id }; + const target = { projectId, agentId, threadId, userId: owner.id, kind: 'preview' as const }; await enqueue(local, input(threadId, 'A', 'new')); const active = await claim(local, threadId); const data = input(threadId, 'B'); @@ -1762,7 +1762,7 @@ describe('AgentExecutionRepository', () => { const local = recordingServices(); const remote = recordingServices(undefined, peer); const threadId = uuid(); - const target = { projectId, agentId, threadId, userId: owner.id }; + const target = { projectId, agentId, threadId, userId: owner.id, kind: 'preview' as const }; await enqueue(local, input(threadId, 'A', 'new')); const active = await claim(local, threadId); const executionId = active.admission.executionId; @@ -1897,7 +1897,7 @@ describe('AgentExecutionRepository', () => { it('rolls back messages and timeline markers if consumption fails', async () => { const services = recordingServices(); const threadId = uuid(); - const target = { projectId, agentId, threadId, userId: owner.id }; + const target = { projectId, agentId, threadId, userId: owner.id, kind: 'preview' as const }; await enqueue(services, input(threadId, 'A', 'new')); const active = await claim(services, threadId); const executionId = active.admission.executionId; @@ -1951,7 +1951,7 @@ describe('AgentExecutionRepository', () => { const local = recordingServices(); const remote = recordingServices(undefined, peer); const threadId = uuid(); - const target = { projectId, agentId, threadId, userId: owner.id }; + const target = { projectId, agentId, threadId, userId: owner.id, kind: 'preview' as const }; await enqueue(local, input(threadId, 'A', 'new')); const active = await claim(local, threadId); const executionId = active.admission.executionId; @@ -2048,6 +2048,7 @@ describe('AgentExecutionRepository', () => { agentId, threadId, userId: owner.id, + kind: 'preview' as const, queueId: c.id, executionId, }; @@ -2242,10 +2243,59 @@ describe('AgentExecutionRepository', () => { }, ); + it('identifies a session from its queued first message before any execution exists', async () => { + const services = recordingServices(Container.get(N8nMemory).getImplementation(agentId)); + const threadId = uuid(); + const pending = await enqueue(services, { + ...input(threadId, 'hello', 'new'), + source: 'n8n_chat_production', + payload: { + kind: 'n8n_chat', + message: 'hello', + userId: owner.id, + resourceId: `n8n-chat-production:${owner.id}`, + }, + }); + expect((await repository.findFirstSourceByThreadIds([threadId])).get(threadId)).toBe( + 'n8n_chat_production', + ); + const target = { projectId, agentId, threadId, userId: owner.id, kind: 'preview' as const }; + expect( + (await services.queue.listPending({ ...target, kind: 'n8n_chat' })).items.map( + ({ id }) => id, + ), + ).toEqual([pending.id]); + await expect(services.queue.listPending({ ...target, kind: 'preview' })).rejects.toThrow( + 'Session not found', + ); + + // Removing the only input leaves a session without a source. Its owner keeps access. + await services.queue.removePending({ ...target, kind: 'n8n_chat', queueId: pending.id }); + expect( + await services.executionService.canUseProductionChatThread( + threadId, + projectId, + agentId, + owner.id, + 'existing', + ), + ).toBe(true); + expect(await services.queue.listPending({ ...target, kind: 'n8n_chat' })).toEqual({ + items: [], + steerableExecutionId: null, + }); + expect( + await services.executionService.canUseDraftThread(threadId, projectId, agentId, owner.id, { + previewChat: true, + sessionMode: 'existing', + }), + ).toBe(false); + }); + it('lists only pending Preview input and removes its attachments without affecting the active run', async () => { const services = recordingServices(Container.get(N8nMemory).getImplementation(agentId)); const threadId = uuid(); - const target = { projectId, agentId, threadId, userId: owner.id }; + const target = { projectId, agentId, threadId, userId: owner.id, kind: 'preview' as const }; await enqueue(services, input(threadId, 'active', 'new')); const active = await claim(services, threadId); const pendingInput = input(threadId, 'pending', 'existing', uuid()); @@ -2317,6 +2367,7 @@ describe('AgentExecutionRepository', () => { agentId, threadId, userId: owner.id, + kind: 'preview' as const, queueId: first.id, message: 'updated', }; @@ -2383,7 +2434,14 @@ describe('AgentExecutionRepository', () => { const remote = recordingServices(undefined, peer); const threadId = uuid(); const item = await enqueue(local, input(threadId, 'first', 'new')); - const target = { projectId, agentId, threadId, userId: owner.id, queueId: item.id }; + const target = { + projectId, + agentId, + threadId, + userId: owner.id, + queueId: item.id, + kind: 'preview' as const, + }; const acquired = createDeferredPromise(); const release = createDeferredPromise(); const lock = local.threads.lockById.bind(local.threads);