github-mirror/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2026-10-17 13:39:44 +02:00
Code Issues Actions 14 Packages Projects Releases Wiki Activity Graph

Merge bc2de3b222 into 06f01713b6

Browse Source
This commit is contained in:
Jaakko Husso 2026-10-16 19:08:54 +00:00 committed by GitHub
commit 343fc68cd6
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 211 additions and 187 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
packages/cli/src/modules/agents/__tests__/agent-chat.controller.test.ts
View File

@ -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 {

198
packages/cli/src/modules/agents/agent-chat-relay.service.ts Normal file
View File

@ -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<StoredAttachmentRef[] | undefined> {
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<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(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<ReturnType<AgentChatAttachmentService['getStream']>>;
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;
}
}
}

197
packages/cli/src/modules/agents/agent-chat.controller.ts
View File

@ -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<StoredAttachmentRef[] | undefined> {
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<void> {
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<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(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<ReturnType<AgentChatAttachmentService['getStream']>>;
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')