feat(core): Move n8n Chat production messages onto the agent queue (no-changelog) (#40102)

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);
 	});
 });

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');

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' }),

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<UserRepository>();
 	const chatExecutions = mock<AgentChatExecutionService>();
 	const integrations = mock<ChatIntegrationService>();
+	const orchestrator = mock<AgentExecutionOrchestratorService>();
 	const sender = { send: vi.fn(), close: vi.fn(async () => {}) };
 	let consumer: AgentMessageQueueConsumer;
 
@@ -112,6 +114,7 @@ describe('AgentMessageQueueConsumer', () => {
 			mock<CredentialsService>(),
 			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',

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),

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<void> {
 		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<Parameters<AgentMessageQueueService['enqueue']>[0] | undefined>,
+	): Promise<void> {
+		const delivery = initSseStream(res);
+		const { send, abortSignal } = delivery;
+		let accepted = false;
+		let subscription: ReturnType<AgentQueuedPreviewStreamService['subscribe']> | 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<AgentQueuedPreviewStreamService['subscribe']> | 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<AgentChatQueueResponse> {
+		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<void> {
+		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<AgentChatQueueResponse> {
 		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<void> {
-		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<void> {
-		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<void> {
-		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({

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;
 		}

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<boolean> {
 		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<boolean> {
-		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(

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<User> {
+	private async getThreadOwner(thread: AgentExecutionThread): Promise<User> {
 		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<void> {
+		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,

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<QueuedPreviewMessage, 'userId' | 'messageId'> | QueuedIntegrationMessage;
+	payload: Omit<QueuedUserChatMessage, 'userId' | 'messageId'> | 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<AgentMessageQueue> {
 		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<AgentChatQueueResponse> {
+	async listPending(input: PendingMessageScope): Promise<AgentChatQueueResponse> {
+		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<void> {
+	async removePending(input: PendingMessageScope & { queueId: string }): Promise<void> {
 		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<void> {
+	async updatePending(
+		input: PendingMessageScope & { queueId: string; message: string },
+	): Promise<void> {
 		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<PendingMessageScope, 'threadId'>,
 		ctx: OperationContext = {},
 	): Promise<void> {
 		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)

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<void> {
+		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,

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<AgentExecution> {
 			)
 			.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(

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'

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

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,

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);