feat(core): Add chat routes for system agents (no-changelog)

Add the /agents/system/:agentId routes: chat, resume, cancel, history,
attachments, background tasks and the queue. The routes have no project
in the URL, so SystemAgentExecutionService checks access.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

packages/cli/src/modules/agents/agent-chat-attachment.service.ts

@@ -112,6 +112,14 @@ export class AgentChatAttachmentService {
 		}
 	}
 
+	/** An attachment of a system agent. The caller checks access to its thread. */
+	async getForSystemAgent(
+		attachmentId: string,
+		agentId: string,
+	): Promise<AgentChatAttachment | null> {
+		return await this.repository.findByIdForSystemAgent(attachmentId, agentId);
+	}
+
 	/** Return metadata only when the caller can read the session. */
 	async getForAgent(
 		attachmentId: string,

packages/cli/src/modules/agents/agents.module.ts

@@ -32,6 +32,7 @@ export class AgentsModule implements ModuleInterface {
 		await import('./agent-knowledge.controller.js');
 		await import('./agent-publish.controller.js');
 		await import('./agent-chat.controller.js');
+		await import('./system-agents/system-agent-chat.controller.js');
 		await import('./agent-integrations.controller.js');
 		await import('./agent-slack-integrations.controller.js');
 		await import('./agent-teams-integrations.controller.js');

packages/cli/src/modules/agents/repositories/agent-chat-attachment.repository.ts

@@ -25,6 +25,11 @@ export class AgentChatAttachmentRepository extends Repository<AgentChatAttachmen
 		return await this.findOneBy({ id, agentId: scope.agentId, projectId: scope.projectId });
 	}
 
+	/** An attachment of a system agent. The caller checks access to its thread. */
+	async findByIdForSystemAgent(id: string, agentId: string): Promise<AgentChatAttachment | null> {
+		return await this.findOneBy({ id, agentId });
+	}
+
 	/**
 	 * Thread lookup scoped to a project or agent — the scope carries the
 	 * caller's authorization and lets the matching composite index

packages/cli/src/modules/agents/system-agents/__tests__/system-agent-chat.controller.test.ts

@@ -0,0 +1,252 @@
+import type { AuthenticatedRequest, User } from '@n8n/db';
+import { BadRequestError, NotFoundError } from '@n8n/errors';
+import type { Response } from 'express';
+import { mock } from 'vitest-mock-extended';
+
+import type { AgentChatAttachmentService } from '../../agent-chat-attachment.service';
+import type { AgentChatExecutionService } from '../../agent-chat-execution.service';
+import type { AgentChatRelayService } from '../../agent-chat-relay.service';
+import type { AgentExecutionOrchestratorService } from '../../agent-execution-orchestrator.service';
+import type { AgentMessageQueueService } from '../../agent-message-queue.service';
+import type { FlushableResponse } from '../../agent-sse-stream';
+import type { AgentChatAttachment } from '../../entities/agent-chat-attachment.entity';
+import type { AgentExecutionThread } from '../../entities/agent-execution-thread.entity';
+import type { N8NCheckpointStorage } from '../../integrations/n8n-checkpoint-storage';
+import { SystemAgentChatController } from '../system-agent-chat.controller';
+import type { SystemAgentExecutionService } from '../system-agent-execution.service';
+
+const AGENT_ID = 'test-system-agent';
+const user = mock<User>({ id: 'user-1' });
+const thread = { id: 'thread-1', projectId: 'project-1' } as AgentExecutionThread;
+
+function setup() {
+	const systemAgents = mock<SystemAgentExecutionService>();
+	const chatRelay = mock<AgentChatRelayService>();
+	const messageQueue = mock<AgentMessageQueueService>();
+	const chatExecutionService = mock<AgentChatExecutionService>();
+	const orchestrator = mock<AgentExecutionOrchestratorService>();
+	const attachments = mock<AgentChatAttachmentService>();
+	const checkpointStorage = mock<N8NCheckpointStorage>();
+	systemAgents.getUsableThread.mockResolvedValue(thread);
+	systemAgents.resourceIdFor.mockImplementation(({ id }) => `system-chat:${id}`);
+	systemAgents.getThread.mockResolvedValue(thread);
+	const controller = new SystemAgentChatController(
+		systemAgents,
+		chatRelay,
+		messageQueue,
+		chatExecutionService,
+		orchestrator,
+		checkpointStorage,
+		attachments,
+	);
+	return {
+		controller,
+		systemAgents,
+		chatRelay,
+		messageQueue,
+		chatExecutionService,
+		orchestrator,
+		checkpointStorage,
+		attachments,
+	};
+}
+
+function request<P extends object>(params: P, query: Record<string, string> = {}) {
+	return { user, params, query } as unknown as AuthenticatedRequest<P>;
+}
+
+describe('SystemAgentChatController', () => {
+	it('prepares a chat message with the working project from the query', async () => {
+		const { controller, systemAgents, chatRelay } = setup();
+		chatRelay.relayQueuedMessage.mockImplementation(async (_res, accept) => {
+			await accept(vi.fn(), new AbortController().signal);
+		});
+
+		await controller.chat(
+			request({}, { projectId: 'project-1' }),
+			mock<FlushableResponse>(),
+			AGENT_ID,
+			{ message: 'hi', sessionId: 'thread-1' } as never,
+		);
+
+		expect(systemAgents.prepareChatMessage).toHaveBeenCalledWith(
+			expect.objectContaining({
+				agentId: AGENT_ID,
+				user,
+				projectId: 'project-1',
+				sessionId: 'thread-1',
+			}),
+		);
+	});
+
+	it('lists the queue of a usable thread as system messages', async () => {
+		const { controller, systemAgents, messageQueue } = setup();
+
+		await controller.getQueuedMessages(request({ agentId: AGENT_ID, threadId: 'thread-1' }));
+
+		expect(systemAgents.getUsableThread).toHaveBeenCalledWith(AGENT_ID, user, 'thread-1');
+		expect(messageQueue.listPending).toHaveBeenCalledWith({
+			projectId: 'project-1',
+			agentId: AGENT_ID,
+			threadId: 'thread-1',
+			userId: 'user-1',
+			kind: 'system',
+		});
+	});
+
+	it('steers a queued system message after the floor check', async () => {
+		const { controller, systemAgents, messageQueue } = setup();
+
+		await controller.steerQueuedMessage(
+			request({ agentId: AGENT_ID, threadId: 'thread-1', queueId: '7' }),
+			mock<Response>(),
+			{ executionId: 'exec-1' } as never,
+		);
+
+		expect(systemAgents.getUsableThread).toHaveBeenCalled();
+		expect(messageQueue.steer).toHaveBeenCalledWith(
+			expect.objectContaining({ queueId: '7', executionId: 'exec-1', kind: 'system' }),
+		);
+	});
+
+	it('rejects an invalid queue id before it loads the thread', async () => {
+		const { controller, systemAgents } = setup();
+
+		await expect(
+			controller.removeQueuedMessage(
+				request({ agentId: AGENT_ID, threadId: 'thread-1', queueId: 'x' }),
+			),
+		).rejects.toThrow(BadRequestError);
+		expect(systemAgents.getUsableThread).not.toHaveBeenCalled();
+	});
+
+	it('lets the owner stop a turn with the thread working project', async () => {
+		const { controller, systemAgents, chatExecutionService } = setup();
+
+		await controller.cancelChatExecution(
+			request({}),
+			mock<Response>(),
+			AGENT_ID,
+			'thread-1',
+			'exec-1',
+		);
+
+		expect(systemAgents.getThread).toHaveBeenCalledWith(AGENT_ID, user, 'thread-1');
+		expect(chatExecutionService.requestCancel).toHaveBeenCalledWith({
+			projectId: 'project-1',
+			agentId: AGENT_ID,
+			threadId: 'thread-1',
+			executionId: 'exec-1',
+			userId: 'user-1',
+			surface: 'system',
+		});
+	});
+
+	describe('attachments', () => {
+		it('refuses an attachment of another user', async () => {
+			const { controller, attachments, chatRelay } = setup();
+			attachments.getForSystemAgent.mockResolvedValue(
+				mock<AgentChatAttachment>({ threadId: 'thread-1', resourceId: 'system-chat:user-2' }),
+			);
+
+			await expect(
+				controller.getChatAttachment(
+					request({ agentId: AGENT_ID, attachmentId: 'att-1' }),
+					mock<Response>(),
+				),
+			).rejects.toThrow(NotFoundError);
+			expect(chatRelay.streamAttachment).not.toHaveBeenCalled();
+		});
+
+		it('streams an attachment of a usable thread of the user', async () => {
+			const { controller, attachments, chatRelay, systemAgents } = setup();
+			const attachment = mock<AgentChatAttachment>({
+				threadId: 'thread-1',
+				resourceId: 'system-chat:user-1',
+			});
+			attachments.getForSystemAgent.mockResolvedValue(attachment);
+
+			await controller.getChatAttachment(
+				request({ agentId: AGENT_ID, attachmentId: 'att-1' }),
+				mock<Response>(),
+			);
+
+			expect(systemAgents.getUsableThread).toHaveBeenCalledWith(AGENT_ID, user, 'thread-1');
+			expect(chatRelay.streamAttachment).toHaveBeenCalledWith(attachment, expect.anything());
+		});
+	});
+	it('resumes a run through the execution service and closes the stream', async () => {
+		const { controller, systemAgents, chatRelay } = setup();
+		const execution = {
+			send: vi.fn(),
+			abortSignal: new AbortController().signal,
+			onExecutionStarted: vi.fn(),
+			onChunk: vi.fn(),
+			close: vi.fn(),
+		};
+		chatRelay.createChatExecution.mockReturnValue(execution);
+
+		await controller.chatResume(request({}), mock<FlushableResponse>(), AGENT_ID, {
+			runId: 'run-1',
+			toolCallId: 'tc-1',
+			resumeData: { approved: true },
+		} as never);
+
+		expect(systemAgents.resumeRun).toHaveBeenCalledWith({
+			agentId: AGENT_ID,
+			user,
+			runId: 'run-1',
+			toolCallId: 'tc-1',
+			resumeData: { approved: true },
+			send: execution.send,
+		});
+		expect(execution.close).toHaveBeenCalled();
+	});
+
+	it('sends a refused resume as an error event', async () => {
+		const { controller, systemAgents, chatRelay } = setup();
+		const execution = {
+			send: vi.fn(),
+			abortSignal: new AbortController().signal,
+			onExecutionStarted: vi.fn(),
+			onChunk: vi.fn(),
+			close: vi.fn(),
+		};
+		chatRelay.createChatExecution.mockReturnValue(execution);
+		systemAgents.resumeRun.mockRejectedValue(new NotFoundError('Session not found'));
+
+		await controller.chatResume(request({}), mock<FlushableResponse>(), AGENT_ID, {
+			runId: 'run-1',
+			toolCallId: 'tc-1',
+			resumeData: {},
+		} as never);
+
+		expect(execution.send).toHaveBeenCalledWith({ type: 'error', message: 'Session not found' });
+		expect(execution.close).toHaveBeenCalled();
+	});
+
+	it('reads the history of a usable thread only', async () => {
+		const { controller, systemAgents, orchestrator } = setup();
+		systemAgents.getUsableThread.mockRejectedValue(new NotFoundError('Session not found'));
+
+		await expect(
+			controller.getChatMessages(request({ agentId: AGENT_ID, threadId: 'thread-1' })),
+		).rejects.toThrow(NotFoundError);
+		expect(orchestrator.getConversationHistory).not.toHaveBeenCalled();
+	});
+
+	it('cancels a suspended run in the memory scope of the user', async () => {
+		const { controller, systemAgents, orchestrator } = setup();
+		orchestrator.cancelChatRun.mockResolvedValue(true);
+
+		const result = await controller.cancelChatRun(request({}), mock<Response>(), AGENT_ID, 'run-1');
+
+		expect(systemAgents.assertRegistered).toHaveBeenCalledWith(AGENT_ID);
+		expect(orchestrator.cancelChatRun).toHaveBeenCalledWith({
+			agentId: AGENT_ID,
+			runId: 'run-1',
+			resourceId: 'system-chat:user-1',
+		});
+		expect(result).toEqual({ cancelled: true });
+	});
+});

packages/cli/src/modules/agents/system-agents/system-agent-chat.controller.ts

@@ -0,0 +1,301 @@
+import {
+	type AgentBackgroundJobsResponse,
+	AgentChatMessageDto,
+	type AgentChatMessagesResponse,
+	AgentChatQueueReorderDto,
+	type AgentChatQueueResponse,
+	AgentChatQueueSteerDto,
+	AgentChatQueueUpdateDto,
+	AgentChatResumeDto,
+} from '@n8n/api-types';
+import type { AuthenticatedRequest } from '@n8n/db';
+import { Body, Delete, Get, Param, Patch, Post, RestController } from '@n8n/decorators';
+import { BadRequestError, NotFoundError } from '@n8n/errors';
+import type { Response } from 'express';
+
+import { AgentChatAttachmentService } from '../agent-chat-attachment.service';
+import { AgentChatExecutionService } from '../agent-chat-execution.service';
+import { AgentChatRelayService } from '../agent-chat-relay.service';
+import { AgentExecutionOrchestratorService } from '../agent-execution-orchestrator.service';
+import { AgentMessageQueueService } from '../agent-message-queue.service';
+import type { FlushableResponse } from '../agent-sse-stream';
+import { N8NCheckpointStorage } from '../integrations/n8n-checkpoint-storage';
+import { withOpenSuspensions } from '../utils/messages-envelope';
+import { SystemAgentExecutionService } from './system-agent-execution.service';
+
+/**
+ * Chat routes of code-defined system agents. They are not project routes: the
+ * working project comes from the thread, or from the `projectId` query for a
+ * new session. So the routes take no `@ProjectScope` decorator, and the
+ * project-agent routes keep their `agent:execute` guards unchanged.
+ *
+ * `SystemAgentExecutionService` checks access on every route: thread
+ * ownership, and for routes that read or continue a conversation also the
+ * runtime floor (`project:read` on the working project) and the provider's
+ * own checks. Stopping a turn needs thread ownership only.
+ */
+@RestController('/agents/system')
+export class SystemAgentChatController {
+	constructor(
+		private readonly systemAgents: SystemAgentExecutionService,
+		private readonly chatRelay: AgentChatRelayService,
+		private readonly messageQueue: AgentMessageQueueService,
+		private readonly chatExecutionService: AgentChatExecutionService,
+		private readonly orchestrator: AgentExecutionOrchestratorService,
+		private readonly checkpointStorage: N8NCheckpointStorage,
+		private readonly attachments: AgentChatAttachmentService,
+	) {}
+
+	/**
+	 * Queue a message. An existing session keeps its working project. A new
+	 * session needs `projectId` in the query.
+	 */
+	@Post('/:agentId/chat', { usesTemplates: true })
+	async chat(
+		req: AuthenticatedRequest<{}, {}, {}, { projectId?: string }>,
+		res: FlushableResponse,
+		@Param('agentId') agentId: string,
+		@Body payload: AgentChatMessageDto,
+	) {
+		const projectId = typeof req.query.projectId === 'string' ? req.query.projectId : undefined;
+		const { message, sessionId, messageId, newSession, attachments, clientContext } = payload;
+		await this.chatRelay.relayQueuedMessage(
+			res,
+			async () =>
+				await this.systemAgents.prepareChatMessage({
+					agentId,
+					user: req.user,
+					projectId,
+					sessionId: newSession ? undefined : sessionId,
+					message,
+					messageId,
+					clientContext,
+					storeAttachments: async (threadId, projectId) =>
+						await this.chatRelay.storeChatAttachments({
+							attachments,
+							agentId,
+							projectId,
+							threadId,
+							resourceId: this.systemAgents.resourceIdFor(req.user),
+						}),
+				}),
+		);
+	}
+
+	@Post('/:agentId/chat/resume', { usesTemplates: true })
+	async chatResume(
+		req: AuthenticatedRequest,
+		res: FlushableResponse,
+		@Param('agentId') agentId: string,
+		@Body payload: AgentChatResumeDto,
+	) {
+		const execution = this.chatRelay.createChatExecution(res);
+		try {
+			execution.abortSignal.throwIfAborted();
+			await this.systemAgents.resumeRun({
+				agentId,
+				user: req.user,
+				runId: payload.runId,
+				toolCallId: payload.toolCallId,
+				resumeData: payload.resumeData,
+				send: execution.send,
+			});
+		} catch (error) {
+			execution.send({
+				type: 'error',
+				message: error instanceof Error ? error.message : 'Resume failed',
+			});
+		} finally {
+			execution.close();
+		}
+	}
+
+	/** Stop a running turn. Only the thread owner can stop it. */
+	@Delete('/:agentId/chat/:threadId/executions/:executionId')
+	async cancelChatExecution(
+		req: AuthenticatedRequest,
+		_res: Response,
+		@Param('agentId') agentId: string,
+		@Param('threadId') threadId: string,
+		@Param('executionId') executionId: string,
+	) {
+		const thread = await this.systemAgents.getThread(agentId, req.user, threadId);
+		const cancelRequested = await this.chatExecutionService.requestCancel({
+			projectId: thread.projectId,
+			agentId,
+			threadId: thread.id,
+			executionId,
+			userId: req.user.id,
+			surface: 'system',
+		});
+		return { cancelRequested };
+	}
+
+	/** Cancel a suspended run of the user. */
+	@Delete('/:agentId/chat/runs/:runId')
+	async cancelChatRun(
+		req: AuthenticatedRequest,
+		_res: Response,
+		@Param('agentId') agentId: string,
+		@Param('runId') runId: string,
+	) {
+		this.systemAgents.assertRegistered(agentId);
+		const cancelled = await this.orchestrator.cancelChatRun({
+			agentId,
+			runId,
+			resourceId: this.systemAgents.resourceIdFor(req.user),
+		});
+		return { cancelled };
+	}
+
+	@Get('/:agentId/chat/:threadId/messages')
+	async getChatMessages(
+		req: AuthenticatedRequest<{ agentId: string; threadId: string }>,
+	): Promise<AgentChatMessagesResponse> {
+		const { agentId, threadId } = req.params;
+		const thread = await this.systemAgents.getUsableThread(agentId, req.user, threadId);
+		const history = await this.orchestrator.getConversationHistory({
+			threadId: thread.id,
+			projectId: thread.projectId,
+			agentId,
+			userId: req.user.id,
+		});
+		const checkpoint = await this.checkpointStorage.findSuspendedForThread(agentId, thread.id);
+		const ownCheckpoint =
+			checkpoint?.persistence?.resourceId === this.systemAgents.resourceIdFor(req.user)
+				? checkpoint
+				: null;
+		if (!history) {
+			return { ...withOpenSuspensions([], ownCheckpoint), activeExecutionId: null };
+		}
+		return {
+			...withOpenSuspensions(history.messages, ownCheckpoint, {
+				appendInactiveCheckpointMessages: false,
+			}),
+			activeExecutionId: history.activeExecutionId,
+		};
+	}
+
+	@Get('/:agentId/chat/attachments/:attachmentId')
+	async getChatAttachment(
+		req: AuthenticatedRequest<{ agentId: string; attachmentId: string }>,
+		res: Response,
+	) {
+		const { agentId, attachmentId } = req.params;
+		this.systemAgents.assertRegistered(agentId);
+		const attachment = await this.attachments.getForSystemAgent(attachmentId, agentId);
+		if (attachment?.resourceId !== this.systemAgents.resourceIdFor(req.user)) {
+			throw new NotFoundError(`Attachment "${attachmentId}" not found`);
+		}
+		await this.systemAgents.getUsableThread(agentId, req.user, attachment.threadId);
+		await this.chatRelay.streamAttachment(attachment, res);
+	}
+
+	/** System agents do not run Agents background jobs. */
+	@Get('/:agentId/chat/:threadId/background-tasks')
+	async getBackgroundJobs(
+		req: AuthenticatedRequest<{ agentId: string; threadId: string }>,
+	): Promise<AgentBackgroundJobsResponse> {
+		await this.systemAgents.getUsableThread(req.params.agentId, req.user, req.params.threadId);
+		return { tasks: [] };
+	}
+
+	// ── Queue ────────────────────────────────────────────────────────────────
+
+	@Get('/:agentId/chat/:threadId/queue')
+	async getQueuedMessages(
+		req: AuthenticatedRequest<{ agentId: string; threadId: string }>,
+	): Promise<AgentChatQueueResponse> {
+		const thread = await this.systemAgents.getUsableThread(
+			req.params.agentId,
+			req.user,
+			req.params.threadId,
+		);
+		return await this.messageQueue.listPending({
+			...this.queueScope(req, thread.projectId),
+			kind: 'system',
+		});
+	}
+
+	@Patch('/:agentId/chat/:threadId/queue/:queueId')
+	async updateQueuedMessage(
+		req: AuthenticatedRequest<{ agentId: string; threadId: string; queueId: string }>,
+		_res: Response,
+		@Body payload: AgentChatQueueUpdateDto,
+	): Promise<void> {
+		const thread = await this.usableQueueThread(req);
+		await this.messageQueue.updatePending({
+			...this.queueScope(req, thread.projectId),
+			queueId: req.params.queueId,
+			message: payload.message,
+			kind: 'system',
+		});
+	}
+
+	@Post('/:agentId/chat/:threadId/queue/:queueId/reorder')
+	async reorderQueuedMessage(
+		req: AuthenticatedRequest<{ agentId: string; threadId: string; queueId: string }>,
+		_res: Response,
+		@Body payload: AgentChatQueueReorderDto,
+	): Promise<void> {
+		const thread = await this.usableQueueThread(req);
+		await this.messageQueue.reorderPending({
+			...this.queueScope(req, thread.projectId),
+			queueId: req.params.queueId,
+			targetQueueId: payload.targetQueueId,
+			expectedQueueIds: payload.expectedQueueIds,
+			kind: 'system',
+		});
+	}
+
+	@Delete('/:agentId/chat/:threadId/queue/:queueId')
+	async removeQueuedMessage(
+		req: AuthenticatedRequest<{ agentId: string; threadId: string; queueId: string }>,
+	) {
+		const thread = await this.usableQueueThread(req);
+		await this.messageQueue.removePending({
+			...this.queueScope(req, thread.projectId),
+			queueId: req.params.queueId,
+			kind: 'system',
+		});
+		return { removed: true };
+	}
+
+	@Post('/:agentId/chat/:threadId/queue/:queueId/steer')
+	async steerQueuedMessage(
+		req: AuthenticatedRequest<{ agentId: string; threadId: string; queueId: string }>,
+		_res: Response,
+		@Body payload: AgentChatQueueSteerDto,
+	): Promise<void> {
+		const thread = await this.usableQueueThread(req);
+		await this.messageQueue.steer({
+			...this.queueScope(req, thread.projectId),
+			queueId: req.params.queueId,
+			executionId: payload.executionId,
+			kind: 'system',
+		});
+	}
+
+	private async usableQueueThread(
+		req: AuthenticatedRequest<{ agentId: string; threadId: string; queueId: string }>,
+	) {
+		if (!/^[1-9]\d*$/.test(req.params.queueId)) throw new BadRequestError('Invalid queue ID');
+		return await this.systemAgents.getUsableThread(
+			req.params.agentId,
+			req.user,
+			req.params.threadId,
+		);
+	}
+
+	private queueScope(
+		req: AuthenticatedRequest<{ agentId: string; threadId: string }>,
+		projectId: string,
+	) {
+		return {
+			projectId,
+			agentId: req.params.agentId,
+			threadId: req.params.threadId,
+			userId: req.user.id,
+		};
+	}
+}