From 4ddd17b803a8bc05c445ef80124aa309e92f162c Mon Sep 17 00:00:00 2001 From: Jaakko Husso Date: Thu, 15 Oct 2026 23:55:50 +0300 Subject: [PATCH] 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 --- .../agents/agent-chat-attachment.service.ts | 8 + .../cli/src/modules/agents/agents.module.ts | 1 + .../agent-chat-attachment.repository.ts | 5 + .../system-agent-chat.controller.test.ts | 252 +++++++++++++++ .../system-agent-chat.controller.ts | 301 ++++++++++++++++++ 5 files changed, 567 insertions(+) create mode 100644 packages/cli/src/modules/agents/system-agents/__tests__/system-agent-chat.controller.test.ts create mode 100644 packages/cli/src/modules/agents/system-agents/system-agent-chat.controller.ts diff --git a/packages/cli/src/modules/agents/agent-chat-attachment.service.ts b/packages/cli/src/modules/agents/agent-chat-attachment.service.ts index e2fbb0fbb90..dfb66b094a9 100644 --- a/packages/cli/src/modules/agents/agent-chat-attachment.service.ts +++ b/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 { + return await this.repository.findByIdForSystemAgent(attachmentId, agentId); + } + /** Return metadata only when the caller can read the session. */ async getForAgent( attachmentId: string, diff --git a/packages/cli/src/modules/agents/agents.module.ts b/packages/cli/src/modules/agents/agents.module.ts index 8af0ff74e85..8d033fd97f0 100644 --- a/packages/cli/src/modules/agents/agents.module.ts +++ b/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'); diff --git a/packages/cli/src/modules/agents/repositories/agent-chat-attachment.repository.ts b/packages/cli/src/modules/agents/repositories/agent-chat-attachment.repository.ts index e5839fc0562..1392639fef6 100644 --- a/packages/cli/src/modules/agents/repositories/agent-chat-attachment.repository.ts +++ b/packages/cli/src/modules/agents/repositories/agent-chat-attachment.repository.ts @@ -25,6 +25,11 @@ export class AgentChatAttachmentRepository extends Repository { + 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 diff --git a/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-chat.controller.test.ts b/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-chat.controller.test.ts new file mode 100644 index 00000000000..17996fbe0ae --- /dev/null +++ b/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({ id: 'user-1' }); +const thread = { id: 'thread-1', projectId: 'project-1' } as AgentExecutionThread; + +function setup() { + const systemAgents = mock(); + const chatRelay = mock(); + const messageQueue = mock(); + const chatExecutionService = mock(); + const orchestrator = mock(); + const attachments = mock(); + const checkpointStorage = mock(); + 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

(params: P, query: Record = {}) { + return { user, params, query } as unknown as AuthenticatedRequest

; +} + +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(), + 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(), + { 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(), + 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({ threadId: 'thread-1', resourceId: 'system-chat:user-2' }), + ); + + await expect( + controller.getChatAttachment( + request({ agentId: AGENT_ID, attachmentId: 'att-1' }), + mock(), + ), + ).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({ + threadId: 'thread-1', + resourceId: 'system-chat:user-1', + }); + attachments.getForSystemAgent.mockResolvedValue(attachment); + + await controller.getChatAttachment( + request({ agentId: AGENT_ID, attachmentId: 'att-1' }), + mock(), + ); + + 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(), 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(), 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(), 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 }); + }); +}); diff --git a/packages/cli/src/modules/agents/system-agents/system-agent-chat.controller.ts b/packages/cli/src/modules/agents/system-agents/system-agent-chat.controller.ts new file mode 100644 index 00000000000..704927b738a --- /dev/null +++ b/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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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, + }; + } +}