From bc2de3b2227876aa1880d7a9027a34a3d42aa04a Mon Sep 17 00:00:00 2001 From: Jaakko Husso Date: Thu, 15 Oct 2026 23:55:42 +0300 Subject: [PATCH] refactor(core): Move the Agents chat relay into its own service (no-changelog) Move the SSE relay and attachment helpers from AgentChatController into AgentChatRelayService, so another chat controller can share them. The code does not change. Co-Authored-By: Claude Opus 5.5 --- .../__tests__/agent-chat.controller.test.ts | 3 +- .../agents/agent-chat-relay.service.ts | 198 ++++++++++++++++++ .../modules/agents/agent-chat.controller.ts | 197 +---------------- 3 files changed, 211 insertions(+), 187 deletions(-) create mode 100644 packages/cli/src/modules/agents/agent-chat-relay.service.ts 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 5a7d480ba0a..2981fd2ffd9 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 @@ -33,6 +33,7 @@ import type { AgentChatAttachment } from '../entities/agent-chat-attachment.enti import type { FlushableResponse } from '../agent-sse-stream'; import type { AgentTestChatService } from '../agent-test-chat.service'; import { AgentTestRunService } from '../agent-test-run.service'; +import { AgentChatRelayService } from '../agent-chat-relay.service'; import type { AgentsService } from '../agents.service'; import type { AgentsBuilderService } from '../builder/agents-builder.service'; import type { N8NCheckpointStorage } from '../integrations/n8n-checkpoint-storage'; @@ -104,8 +105,8 @@ function makeController() { backgroundJobService, chatExecutionService, messageQueue, - previewStreams, agentsConfig, + new AgentChatRelayService(messageQueue, previewStreams, agentChatAttachmentService), ); return { diff --git a/packages/cli/src/modules/agents/agent-chat-relay.service.ts b/packages/cli/src/modules/agents/agent-chat-relay.service.ts new file mode 100644 index 00000000000..4946501f338 --- /dev/null +++ b/packages/cli/src/modules/agents/agent-chat-relay.service.ts @@ -0,0 +1,198 @@ +import { + type AgentChatAttachmentPayload, + type AgentSseEvent, + MAX_AGENT_CHAT_ATTACHMENT_SIZE_BYTES, + MAX_AGENT_CHAT_ATTACHMENT_SIZE_MB, + ViewableMimeTypes, +} from '@n8n/api-types'; +import { Service } from '@n8n/di'; +import { BadRequestError, NotFoundError } from '@n8n/errors'; +import { sanitizeFilename } from '@n8n/utils/files/sanitize-filename'; +import type { Response } from 'express'; +import { FileNotFoundError, getHtmlSandboxCSP } from 'n8n-core'; +import { pipeline } from 'node:stream/promises'; + +import { AgentChatAttachmentService } from './agent-chat-attachment.service'; +import { AgentMessageQueueService } from './agent-message-queue.service'; +import { AgentQueuedPreviewStreamService } from './agent-queued-preview-stream.service'; +import { type FlushableResponse, initSseStream, toChatErrorEvent } from './agent-sse-stream'; +import type { AgentChatAttachment } from './entities/agent-chat-attachment.entity'; +import type { StoredAttachmentRef } from './types/agent-chat-attachment'; +import { resolveInboundMimeType } from './utils/inbound-attachments'; + +/** + * Request plumbing shared by the Agents chat controllers: SSE delivery of a + * queued message, inbound attachment storage and attachment downloads. + */ +@Service() +export class AgentChatRelayService { + constructor( + private readonly messageQueue: AgentMessageQueueService, + private readonly queuedPreviewStreams: AgentQueuedPreviewStreamService, + private readonly agentChatAttachmentService: AgentChatAttachmentService, + ) {} + + createChatExecution(res: FlushableResponse) { + const delivery = initSseStream(res); + const requestController = new AbortController(); + const abandon = () => requestController.abort(); + delivery.abortSignal.addEventListener('abort', abandon, { once: true }); + if (delivery.abortSignal.aborted) abandon(); + return { + send: delivery.send, + abortSignal: requestController.signal, + onExecutionStarted: (id: string, sessionId: string, inputMessageIds: string[]) => { + delivery.abortSignal.removeEventListener('abort', abandon); + delivery.send({ type: 'execution-started', executionId: id, sessionId, inputMessageIds }); + }, + onChunk: delivery.onChunk, + close: () => { + delivery.abortSignal.removeEventListener('abort', abandon); + delivery.close(); + }, + }; + } + + /** Decode, sniff, and persist inbound chat attachments; returns refs for the user turn. */ + async storeChatAttachments(params: { + attachments: AgentChatAttachmentPayload[] | undefined; + agentId: string; + projectId: string; + threadId: string; + resourceId: string; + source?: string; + }): Promise { + const { attachments, agentId, projectId, threadId, resourceId, source = 'chat' } = params; + if (!attachments?.length) return undefined; + + const stored: StoredAttachmentRef[] = []; + try { + for (const attachment of attachments) { + const data = Buffer.from(attachment.data, 'base64'); + if (data.byteLength === 0) { + throw new BadRequestError(`Attachment "${attachment.fileName}" is empty`); + } + if (data.byteLength > MAX_AGENT_CHAT_ATTACHMENT_SIZE_BYTES) { + throw new BadRequestError( + `Attachment "${attachment.fileName}" exceeds the ${MAX_AGENT_CHAT_ATTACHMENT_SIZE_MB} MB limit`, + ); + } + + const mimeType = await resolveInboundMimeType(attachment.mimeType, data); + const row = await this.agentChatAttachmentService.storeInbound({ + agentId, + projectId, + threadId, + resourceId, + source, + fileName: attachment.fileName, + mimeType, + data, + }); + stored.push({ + id: row.id, + fileName: row.fileName, + mimeType: row.mimeType, + sizeBytes: row.fileSizeBytes, + }); + } + } catch (error) { + // Nothing references the already-stored attachments of a rejected message. + await this.agentChatAttachmentService.deleteByIds(stored.map((ref) => ref.id)); + throw error; + } + return stored; + } + + /** + * 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. + */ + 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(toChatErrorEvent(error, 'Chat failed')); + } 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(); + } + } + + async streamAttachment(attachment: AgentChatAttachment, res: Response) { + const attachmentId = attachment.id; + // Open the stream before writing headers: bytes can be gone while the row + // remains (out-of-band storage cleanup), and that must surface as a clean + // 404 rather than a half-written response. + let stream: Awaited>; + try { + stream = await this.agentChatAttachmentService.getStream(attachment); + } catch (error) { + if (error instanceof FileNotFoundError) { + throw new NotFoundError(`Attachment "${attachmentId}" is no longer available`); + } + throw error; + } + + res.setHeader('Content-Type', attachment.mimeType); + res.setHeader('Content-Length', attachment.fileSizeBytes); + res.setHeader('X-Content-Type-Options', 'nosniff'); + // Sandbox anything rendered inline: attachments are user-supplied content + // served same-origin, so active content in them must never script against + // the n8n session (same posture as the binary-data controller). + res.setHeader('Content-Security-Policy', getHtmlSandboxCSP()); + // Non-viewable types must not render inline in the browser. + if (!ViewableMimeTypes.includes(attachment.mimeType.toLowerCase())) { + res.setHeader( + 'Content-Disposition', + `attachment; filename="${sanitizeFilename(attachment.fileName)}"`, + ); + } + + // pipeline destroys the source when the client disconnects mid-transfer, + // so aborted downloads don't leak file descriptors or object-store sockets. + try { + await pipeline(stream, res); + } catch (error) { + if ( + error instanceof Error && + 'code' in error && + error.code === 'ERR_STREAM_PREMATURE_CLOSE' + ) { + return; + } + throw error; + } + } +} diff --git a/packages/cli/src/modules/agents/agent-chat.controller.ts b/packages/cli/src/modules/agents/agent-chat.controller.ts index a7b032a5951..09af75ff96d 100644 --- a/packages/cli/src/modules/agents/agent-chat.controller.ts +++ b/packages/cli/src/modules/agents/agent-chat.controller.ts @@ -1,18 +1,13 @@ import { type AgentBackgroundJobsResponse, - type AgentChatAttachmentPayload, AgentChatMessageDto, AgentChatQueueUpdateDto, AgentChatQueueSteerDto, AgentChatQueueReorderDto, type AgentChatMessagesResponse, type AgentChatQueueResponse, - type AgentSseEvent, AgentChatResumeDto, - MAX_AGENT_CHAT_ATTACHMENT_SIZE_BYTES, - MAX_AGENT_CHAT_ATTACHMENT_SIZE_MB, N8N_CHAT_INTEGRATION_TYPE, - ViewableMimeTypes, } from '@n8n/api-types'; import { AgentsConfig } from '@n8n/config'; import type { AuthenticatedRequest } from '@n8n/db'; @@ -28,10 +23,7 @@ import { } from '@n8n/decorators'; import { scrubSecretsInText } from '@n8n/utils/scrub-secrets'; import { redactDeep } from '@n8n/utils/redaction/redact-text'; -import { sanitizeFilename } from '@n8n/utils/files/sanitize-filename'; import type { Response } from 'express'; -import { FileNotFoundError, getHtmlSandboxCSP } from 'n8n-core'; -import { pipeline } from 'node:stream/promises'; import { randomUUID } from 'node:crypto'; import { CredentialsService } from '@/credentials/credentials.service'; @@ -40,11 +32,9 @@ import { BadRequestError, NotFoundError } from '@n8n/errors'; import { AgentsCredentialProvider } from './adapters/agents-credential-provider'; import { AgentChatAttachmentService } from './agent-chat-attachment.service'; import { AgentN8nChatUnavailableError } from './agent-n8n-chat-unavailable.error'; -import type { AgentChatAttachment } from './entities/agent-chat-attachment.entity'; -import type { StoredAttachmentRef } from './types/agent-chat-attachment'; +import { AgentChatRelayService } from './agent-chat-relay.service'; import { AgentExecutionOrchestratorService } from './agent-execution-orchestrator.service'; import { AgentMessageQueueService } from './agent-message-queue.service'; -import { AgentQueuedPreviewStreamService } from './agent-queued-preview-stream.service'; import { AgentChatExecutionService } from './agent-chat-execution.service'; import { AgentExecutionService } from './agent-execution.service'; import { @@ -53,7 +43,7 @@ import { threadBelongsTo, } from './utils/agent-thread-access'; import { messagesToDto } from './agent-message-mapper'; -import { type FlushableResponse, initSseStream, toChatErrorEvent } from './agent-sse-stream'; +import { type FlushableResponse, toChatErrorEvent } from './agent-sse-stream'; import { AgentTestChatService, chatThreadId } from './agent-test-chat.service'; import { AgentTestRunService } from './agent-test-run.service'; import { AgentsService } from './agents.service'; @@ -64,7 +54,6 @@ import { productionChatMemoryResourceId, userIdFromDraftChatMemoryResourceId, } from './utils/agent-memory-scope'; -import { resolveInboundMimeType } from './utils/inbound-attachments'; import { withOpenSuspensions } from './utils/messages-envelope'; @RestController('/projects/:projectId/agents/v2') @@ -81,82 +70,10 @@ export class AgentChatController { private readonly backgroundJobService: AgentBackgroundJobService, private readonly chatExecutionService: AgentChatExecutionService, private readonly messageQueue: AgentMessageQueueService, - private readonly queuedPreviewStreams: AgentQueuedPreviewStreamService, private readonly agentsConfig: AgentsConfig, + private readonly chatRelay: AgentChatRelayService, ) {} - private createChatExecution(res: FlushableResponse) { - const delivery = initSseStream(res); - const requestController = new AbortController(); - const abandon = () => requestController.abort(); - delivery.abortSignal.addEventListener('abort', abandon, { once: true }); - if (delivery.abortSignal.aborted) abandon(); - return { - send: delivery.send, - abortSignal: requestController.signal, - onExecutionStarted: (id: string, sessionId: string, inputMessageIds: string[]) => { - delivery.abortSignal.removeEventListener('abort', abandon); - delivery.send({ type: 'execution-started', executionId: id, sessionId, inputMessageIds }); - }, - onChunk: delivery.onChunk, - close: () => { - delivery.abortSignal.removeEventListener('abort', abandon); - delivery.close(); - }, - }; - } - - /** Decode, sniff, and persist inbound chat attachments; returns refs for the user turn. */ - private async storeChatAttachments(params: { - attachments: AgentChatAttachmentPayload[] | undefined; - agentId: string; - projectId: string; - threadId: string; - resourceId: string; - source?: string; - }): Promise { - const { attachments, agentId, projectId, threadId, resourceId, source = 'chat' } = params; - if (!attachments?.length) return undefined; - - const stored: StoredAttachmentRef[] = []; - try { - for (const attachment of attachments) { - const data = Buffer.from(attachment.data, 'base64'); - if (data.byteLength === 0) { - throw new BadRequestError(`Attachment "${attachment.fileName}" is empty`); - } - if (data.byteLength > MAX_AGENT_CHAT_ATTACHMENT_SIZE_BYTES) { - throw new BadRequestError( - `Attachment "${attachment.fileName}" exceeds the ${MAX_AGENT_CHAT_ATTACHMENT_SIZE_MB} MB limit`, - ); - } - - const mimeType = await resolveInboundMimeType(attachment.mimeType, data); - const row = await this.agentChatAttachmentService.storeInbound({ - agentId, - projectId, - threadId, - resourceId, - source, - fileName: attachment.fileName, - mimeType, - data, - }); - stored.push({ - id: row.id, - fileName: row.fileName, - mimeType: row.mimeType, - sizeBytes: row.fileSizeBytes, - }); - } - } catch (error) { - // Nothing references the already-stored attachments of a rejected message. - await this.agentChatAttachmentService.deleteByIds(stored.map((ref) => ref.id)); - throw error; - } - return stored; - } - private async requireProductionChat(agentId: string, projectId: string): Promise { if (!(await this.agentsService.isN8nChatPublished(agentId, projectId))) { throw new NotFoundError('Agent is not available in n8n Chat'); @@ -198,52 +115,6 @@ export class AgentChatController { throw new NotFoundError(`Thread "${threadId}" 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(toChatErrorEvent(error, 'Chat failed')); - } 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'); } @@ -258,7 +129,7 @@ export class AgentChatController { ) { const { projectId } = req.params; const resourceId = productionChatMemoryResourceId(req.user.id); - await this.relayQueuedMessage(res, async () => { + await this.chatRelay.relayQueuedMessage(res, async () => { if (!(await this.agentsService.isN8nChatPublished(agentId, projectId))) { throw new AgentN8nChatUnavailableError(); } @@ -266,7 +137,7 @@ export class AgentChatController { // 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({ + const attachments = await this.chatRelay.storeChatAttachments({ attachments: payload.attachments, agentId, projectId, @@ -300,7 +171,7 @@ export class AgentChatController { @Param('agentId') agentId: string, @Body payload: AgentChatResumeDto, ) { - const execution = this.createChatExecution(res); + const execution = this.chatRelay.createChatExecution(res); const { send, onChunk, abortSignal, onExecutionStarted } = execution; let executionId: string | undefined; try { @@ -358,7 +229,7 @@ export class AgentChatController { agentId, ); - await this.relayQueuedMessage(res, async (send, abortSignal) => { + await this.chatRelay.relayQueuedMessage(res, async (send, abortSignal) => { const prepared = await this.agentTestRunService.prepareDraftRun({ agentId, projectId, @@ -394,7 +265,7 @@ export class AgentChatController { kind: 'preview', message, messageId, - attachments: await this.storeChatAttachments({ + attachments: await this.chatRelay.storeChatAttachments({ attachments, agentId, projectId, @@ -418,7 +289,7 @@ export class AgentChatController { ) { const { projectId } = req.params; const { runId, toolCallId, resumeData } = payload; - const execution = this.createChatExecution(res); + const execution = this.chatRelay.createChatExecution(res); const { send, onChunk, abortSignal, onExecutionStarted } = execution; try { abortSignal.throwIfAborted(); @@ -591,7 +462,7 @@ export class AgentChatController { throw new NotFoundError(`Attachment "${attachmentId}" not found`); } await this.requireProductionThread(attachment.threadId, projectId, agentId, req.user.id); - await this.streamAttachment(attachment, res); + await this.chatRelay.streamAttachment(attachment, res); } // Same reasoning as `getProductionChatMessages`: `messageQueue.listPending` @@ -1079,53 +950,7 @@ export class AgentChatController { if (attachment.source === N8N_CHAT_PRODUCTION_SOURCE) { throw new NotFoundError(`Attachment "${attachmentId}" not found`); } - await this.streamAttachment(attachment, res); - } - - private async streamAttachment(attachment: AgentChatAttachment, res: Response) { - const attachmentId = attachment.id; - // Open the stream before writing headers: bytes can be gone while the row - // remains (out-of-band storage cleanup), and that must surface as a clean - // 404 rather than a half-written response. - let stream: Awaited>; - try { - stream = await this.agentChatAttachmentService.getStream(attachment); - } catch (error) { - if (error instanceof FileNotFoundError) { - throw new NotFoundError(`Attachment "${attachmentId}" is no longer available`); - } - throw error; - } - - res.setHeader('Content-Type', attachment.mimeType); - res.setHeader('Content-Length', attachment.fileSizeBytes); - res.setHeader('X-Content-Type-Options', 'nosniff'); - // Sandbox anything rendered inline: attachments are user-supplied content - // served same-origin, so active content in them must never script against - // the n8n session (same posture as the binary-data controller). - res.setHeader('Content-Security-Policy', getHtmlSandboxCSP()); - // Non-viewable types must not render inline in the browser. - if (!ViewableMimeTypes.includes(attachment.mimeType.toLowerCase())) { - res.setHeader( - 'Content-Disposition', - `attachment; filename="${sanitizeFilename(attachment.fileName)}"`, - ); - } - - // pipeline destroys the source when the client disconnects mid-transfer, - // so aborted downloads don't leak file descriptors or object-store sockets. - try { - await pipeline(stream, res); - } catch (error) { - if ( - error instanceof Error && - 'code' in error && - error.code === 'ERR_STREAM_PREMATURE_CLOSE' - ) { - return; - } - throw error; - } + await this.chatRelay.streamAttachment(attachment, res); } @Delete('/:agentId/chat/messages')