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

feat(core): Add a system-agent layer to the Agents runtime (no-changelog)

Browse Source 
- A module can register a code-defined instance agent through a
  SystemAgentProvider. SystemAgentRegistry keeps the providers.
- SystemAgentExecutionService manages the threads and turns of a system
  agent. It checks the access floor (owned thread, project:read on the
  working project) and then the provider.
- The queue accepts the system kind only for a registered provider. The
  consumer sends system turns to the execution service.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Jaakko Husso 2026-10-15 23:55:47 +03:00 committed by GitHub
parent 60b56086e1
commit ab2ff827a5
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 1866 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

135
packages/cli/src/modules/agents/__tests__/agent-message-queue-consumer.service.test.ts
View File

@ -21,6 +21,9 @@ import type { AgentMessageEntity } from '../entities/agent-message.entity';
import type { AgentChatBridge } from '../integrations/agent-chat-bridge';
import type { ChatIntegrationService } from '../integrations/chat-integration.service';
import type { AgentMessageQueueRepository } from '../repositories/agent-message-queue.repository';
import type { SystemAgentExecutionService } from '../system-agents/system-agent-execution.service';
import { SystemAgentRegistry } from '../system-agents/system-agent-registry';
import type { SystemAgentProvider } from '../system-agents/system-agent.types';
import type { AgentQueueDispatch, QueuedIntegrationMessage } from '../types/agent-queued-message';
vi.mock('@/permissions.ee/check-access', () => ({ userHasScopes: vi.fn() }));
@ -36,6 +39,8 @@ describe('AgentMessageQueueConsumer', () => {
const integrations = mock<ChatIntegrationService>();
const orchestrator = mock<AgentExecutionOrchestratorService>();
const sender = { send: vi.fn(), close: vi.fn(async () => {}) };
const systemAgentExecution = mock<SystemAgentExecutionService>();
let systemAgents: SystemAgentRegistry;
let consumer: AgentMessageQueueConsumer;
function claim(threadId: string, integration = false): ClaimedAgentMessage {
@ -90,6 +95,7 @@ describe('AgentMessageQueueConsumer', () => {
beforeEach(() => {
vi.resetAllMocks();
vi.mocked(userHasScopes).mockResolvedValue(true);
systemAgents = new SystemAgentRegistry();
users.findByIdWithRole.mockResolvedValue(mock<User>({ id: 'user', disabled: false }));
repository.findThreadIds.mockResolvedValue([]);
executions.getAbortSignal.mockImplementation(() => new AbortController().signal);
@ -117,6 +123,8 @@ describe('AgentMessageQueueConsumer', () => {
integrations,
orchestrator,
mockLogger(),
systemAgents,
systemAgentExecution,
);
});
@ -440,4 +448,131 @@ describe('AgentMessageQueueConsumer', () => {
expect.any(AbortSignal),
);
});
describe('system agents', () => {
function registerSystemAgent(authorized = true) {
const provider = {
agentId: 'agent',
name: 'System Agent',
authorize: vi.fn(async () => authorized),
prepareTurn: vi.fn(),
} satisfies SystemAgentProvider;
systemAgents.register(provider);
return provider;
}
function systemClaim(threadId: string): ClaimedAgentMessage {
const item = claim(threadId);
return {
...item,
payload: { kind: 'system', message: 'input', resourceId: 'system-chat:user' },
item: mock<AgentMessageQueue>({ ...item.item, payload: { kind: 'system' } }),
};
}
async function consumeOne(item: ClaimedAgentMessage) {
repository.findThreadIds.mockResolvedValue([item.thread.id]);
queue.claimNext.mockResolvedValueOnce(item).mockResolvedValue(null);
consumer.start();
await vi.waitFor(() =>
expect(queue.settle).toHaveBeenCalledWith(item.thread.id, item.admission.executionId),
);
}
it('routes a system message to the system-agent runtime after the floor check', async () => {
const provider = registerSystemAgent();
const item = systemClaim('session');
systemAgentExecution.consume.mockImplementation(async (_claim, _user, _signal, send) => {
send({ type: 'done', sessionId: 'session', executionId: item.admission.executionId });
});
await consumeOne(item);
expect(systemAgentExecution.consume).toHaveBeenCalledWith(
item,
expect.objectContaining({ id: 'user' }),
expect.any(AbortSignal),
sender.send,
);
expect(chatExecutions.register).toHaveBeenCalledWith(
expect.objectContaining({ userId: 'user', surface: 'system' }),
expect.any(AbortController),
);
// The runtime floor (`project:read`) and the provider decide access,
// not the project `agent:execute` scope.
expect(userHasScopes).toHaveBeenCalledWith(
expect.objectContaining({ id: 'user' }),
['project:read'],
false,
{ projectId: 'project' },
);
expect(userHasScopes).not.toHaveBeenCalledWith(
expect.anything(),
['agent:execute'],
expect.anything(),
expect.anything(),
);
expect(provider.authorize).toHaveBeenCalledWith(
expect.objectContaining({ id: 'user' }),
'project',
);
expect(testRuns.prepareDraftRun).not.toHaveBeenCalled();
expect(chatExecutions.register).toHaveBeenCalledWith(
expect.objectContaining({ userId: 'user', threadId: 'session' }),
expect.any(AbortController),
);
expect(sender.send).toHaveBeenCalledWith(expect.objectContaining({ type: 'done' }));
expect(queue.recordFailure).not.toHaveBeenCalled();
});
it('records a failure when the provider no longer authorizes the owner', async () => {
registerSystemAgent(false);
const item = systemClaim('session');
await consumeOne(item);
expect(queue.recordFailure).toHaveBeenCalledWith(
item,
expect.objectContaining({ message: 'You can no longer execute this agent' }),
expect.any(AbortSignal),
);
expect(systemAgentExecution.consume).not.toHaveBeenCalled();
});
it('records a failure when the owner can no longer read the working project', async () => {
const provider = registerSystemAgent();
vi.mocked(userHasScopes).mockResolvedValue(false);
const item = systemClaim('session');
await consumeOne(item);
expect(queue.recordFailure).toHaveBeenCalled();
expect(provider.authorize).not.toHaveBeenCalled();
expect(systemAgentExecution.consume).not.toHaveBeenCalled();
});
it('records a failure for a system message without a registered provider', async () => {
const item = systemClaim('session');
await consumeOne(item);
expect(queue.recordFailure).toHaveBeenCalled();
expect(systemAgentExecution.consume).not.toHaveBeenCalled();
});
it('keeps the agent:execute check for Preview messages of a system agent id', async () => {
registerSystemAgent();
const item = claim('session');
await consumeOne(item);
expect(userHasScopes).toHaveBeenCalledWith(
expect.objectContaining({ id: 'user' }),
['agent:execute'],
false,
{ projectId: 'project' },
);
expect(systemAgentExecution.consume).not.toHaveBeenCalled();
expect(testRuns.prepareDraftRun).toHaveBeenCalled();
});
});
});

2
packages/cli/src/modules/agents/__tests__/agent-message-queue.service.test.ts
View File

@ -19,6 +19,7 @@ import type { AgentExecutionThreadRepository } from '../repositories/agent-execu
import type { AgentMessageQueueRepository } from '../repositories/agent-message-queue.repository';
import type { AgentMessageRepository } from '../repositories/agent-message.repository';
import type { AgentRepository } from '../repositories/agent.repository';
import { SystemAgentRegistry } from '../system-agents/system-agent-registry';
function makeThread(overrides: Partial<AgentExecutionThread> = {}): AgentExecutionThread {
return mock<AgentExecutionThread>({
@ -68,6 +69,7 @@ describe('AgentMessageQueueService', () => {
messages,
steering,
settingsService,
new SystemAgentRegistry(),
);
});

55
packages/cli/src/modules/agents/agent-message-queue-consumer.service.ts
View File

@ -20,6 +20,11 @@ import type { AgentExecutionThread } from './entities/agent-execution-thread.ent
import type { AgentChatBridge } from './integrations/agent-chat-bridge';
import { ChatIntegrationService } from './integrations/chat-integration.service';
import { AgentMessageQueueRepository } from './repositories/agent-message-queue.repository';
import { canUseSystemAgent } from './system-agents/system-agent-access';
import { SystemAgentExecutionService } from './system-agents/system-agent-execution.service';
import { SystemAgentRegistry } from './system-agents/system-agent-registry';
import type { AgentQueuedMessage } from './types/agent-queued-message';
import type { AgentChatSurface } from './utils/agent-thread-access';
@Service()
export class AgentMessageQueueConsumer {
@ -40,6 +45,8 @@ export class AgentMessageQueueConsumer {
private readonly integrations: ChatIntegrationService,
private readonly orchestrator: AgentExecutionOrchestratorService,
private readonly logger: Logger,
private readonly systemAgents: SystemAgentRegistry,
private readonly systemAgentExecution: SystemAgentExecutionService,
) {}
start(): void {
@ -137,7 +144,7 @@ export class AgentMessageQueueConsumer {
payload.kind === 'integration' ? undefined : this.previewStreams.createSender(item.id);
try {
if (sender) {
const user = await this.getThreadOwner(thread);
const user = await this.getThreadOwner(thread, payload.kind === 'system');
this.chatExecutionService.register(
{
projectId: thread.projectId,
@ -145,7 +152,7 @@ export class AgentMessageQueueConsumer {
threadId: thread.id,
userId: user.id,
executionId: admission.executionId,
surface: payload.kind === 'n8n_chat' ? 'n8n-chat' : 'preview',
surface: chatSurfaceFor(payload.kind),
},
controller,
);
@ -156,11 +163,15 @@ export class AgentMessageQueueConsumer {
inputMessageIds: admission.inputMessageIds,
message: payload.message,
});
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),
);
await this.chatExecutionService.settle(admission.executionId, async () => {
if (payload.kind === 'system') {
await this.systemAgentExecution.consume(claim, user, signal, sender.send);
} else if (payload.kind === 'n8n_chat') {
await this.consumeN8nChat(claim, user, signal, sender.send);
} else {
await this.consumePreview(claim, user, signal, sender.send);
}
});
} else {
await this.consumeIntegration(claim, signal, bridge);
}
@ -178,19 +189,32 @@ export class AgentMessageQueueConsumer {
}
}
private async getThreadOwner(thread: AgentExecutionThread): Promise<User> {
/**
* Load the thread owner and check that they can still run the agent. A
* system agent uses the runtime floor and its provider checks instead of
* the project `agent:execute` scope.
*/
private async getThreadOwner(thread: AgentExecutionThread, systemKind: boolean): Promise<User> {
if (!thread.ownerId) throw new UserError('You can no longer execute this agent');
const user = await this.userRepository.findByIdWithRole(thread.ownerId);
if (
!user ||
user.disabled ||
!(await userHasScopes(user, ['agent:execute'], false, { projectId: thread.projectId }))
) {
if (!user || user.disabled || !(await this.canExecute(user, thread, systemKind))) {
throw new UserError('You can no longer execute this agent');
}
return user;
}
private async canExecute(
user: User,
thread: AgentExecutionThread,
systemKind: boolean,
): Promise<boolean> {
if (!systemKind) {
return await userHasScopes(user, ['agent:execute'], false, { projectId: thread.projectId });
}
const provider = this.systemAgents.get(thread.agentId);
return provider !== undefined && (await canUseSystemAgent(provider, user, thread.projectId));
}
private async consumePreview(
claim: ClaimedAgentMessage,
user: User,
@ -293,3 +317,8 @@ export class AgentMessageQueueConsumer {
await bridge.consumeQueuedMessage(payload, thread.id, claim.admission, signal, connection);
}
}
function chatSurfaceFor(kind: AgentQueuedMessage['kind']): AgentChatSurface {
if (kind === 'n8n_chat') return 'n8n-chat';
return kind === 'system' ? 'system' : 'preview';
}

47
packages/cli/src/modules/agents/agent-message-queue.service.ts
View File

@ -24,6 +24,7 @@ import {
AgentMessageRepository,
} from './repositories/agent-message.repository';
import { AgentRepository } from './repositories/agent.repository';
import { SystemAgentRegistry } from './system-agents/system-agent-registry';
import {
acceptsSteering,
type AgentExecutionAdmission,
@ -33,6 +34,7 @@ import {
type QueuedUserChatMessage,
} from './types/agent-queued-message';
import {
canContinueSystemThread,
canContinueThreadInN8nChat,
canContinueThreadInPreview,
type AgentSessionMode,
@ -76,6 +78,7 @@ export class AgentMessageQueueService {
private readonly messages: AgentMessageRepository,
private readonly steering: AgentMessageSteeringService,
private readonly settingsService: AgentsSettingsService,
private readonly systemAgents: SystemAgentRegistry,
) {}
/** Save a pending message. It is durably accepted when the transaction commits. */
@ -90,15 +93,22 @@ export class AgentMessageQueueService {
},
onInserted?: (queueId: string) => void,
): Promise<{ status: 'accepted'; item: AgentMessageQueue } | { status: 'duplicate' }> {
await this.settingsService.assertEnabled();
const agent = await this.agentRepository.findByIdAndProjectId(input.agentId, input.projectId);
if (!agent) throw new UserError('Agent not found');
const { payload } = input;
// Only the system kind reaches a system agent. Project lookups never return instance agents.
const systemAgent =
payload.kind === 'system' ? this.systemAgents.get(input.agentId) : undefined;
// System agents run even when the admin turns off custom agents.
if (!systemAgent) await this.settingsService.assertEnabled();
const agent =
payload.kind === 'system'
? systemAgent
: await this.agentRepository.findByIdAndProjectId(input.agentId, input.projectId);
if (!agent) throw new UserError('Agent not found');
const messageId = queuedMessageId(input.agentId, input.threadId, payload);
let item: AgentMessageQueue;
try {
item = await this.txRunner.run({}, async (ctx) => {
await this.settingsService.assertEnabled(ctx);
if (!systemAgent) await this.settingsService.assertEnabled(ctx);
await this.executionService.prepareThread(
{
...input,
@ -185,6 +195,11 @@ export class AgentMessageQueueService {
return await this.repository.enqueue(threadId, input.id, queueDispatch, ctx);
}
/** Whether a message waits in the thread queue, hidden machine turns included. */
async hasQueuedMessages(threadId: string): Promise<boolean> {
return await this.repository.hasItems(threadId);
}
async listPending(input: PendingMessageScope): Promise<AgentChatQueueResponse> {
const { kind } = input;
const thread = await this.threadRepository.findOneBy({ id: input.threadId });
@ -331,9 +346,7 @@ export class AgentMessageQueueService {
if (
thread.projectId !== input.projectId ||
thread.agentId !== input.agentId ||
!(input.kind === 'n8n_chat'
? canContinueThreadInN8nChat(thread, input.userId, source)
: canContinueThreadInPreview(thread, input.userId, source))
!canContinueThread(input.kind, thread, input.userId, source)
) {
throw new NotFoundError('Session not found');
}
@ -353,9 +366,11 @@ export class AgentMessageQueueService {
): Promise<ClaimedAgentMessage | null> {
let steeringChanged = false;
const claimed = await this.txRunner.run({}, async (ctx) => {
if (!(await this.settingsService.getEnabled(ctx))) return null;
const thread = await this.threadRepository.lockById(threadId, ctx);
if (!thread) return null;
// The admin setting for custom agents does not apply to system agents.
if (!this.systemAgents.has(thread.agentId) && !(await this.settingsService.getEnabled(ctx)))
return null;
steeringChanged = await this.steering.releaseInactive(thread, ctx);
if (await this.isBlocked(thread, ctx)) return null;
const active = await this.repository.findActive(threadId, ctx);
@ -493,3 +508,19 @@ export class AgentMessageQueueService {
function isHidden(item: AgentMessageQueue): boolean {
return item.message.origin?.hidden === true;
}
function canContinueThread(
kind: QueuedUserChatMessage['kind'],
thread: AgentExecutionThread,
userId: string,
source: string | null | undefined,
): boolean {
switch (kind) {
case 'n8n_chat':
return canContinueThreadInN8nChat(thread, userId, source);
case 'system':
return canContinueSystemThread(thread, userId, source);
case 'preview':
return canContinueThreadInPreview(thread, userId, source);
}
}

791
packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts Normal file
View File

@ -0,0 +1,791 @@
import type { Logger } from '@n8n/backend-common';
import type { TransactionRunner, User } from '@n8n/db';
import { NotFoundError } from '@n8n/errors';
import { mock } from 'vitest-mock-extended';
import { userHasScopes } from '@/permissions.ee/check-access';
import type { AgentChatExecutionService } from '../../agent-chat-execution.service';
import type { AgentExecutionService } from '../../agent-execution.service';
import type {
AgentMessageQueueService,
ClaimedAgentMessage,
} from '../../agent-message-queue.service';
import type {
AgentTurnExecutionService,
AgentTurnRequest,
} from '../../agent-turn-execution.service';
import type { AgentExecutionThread } from '../../entities/agent-execution-thread.entity';
import type { N8NCheckpointStorage } from '../../integrations/n8n-checkpoint-storage';
import type { AgentExecutionRepository } from '../../repositories/agent-execution.repository';
import type { AgentExecutionThreadRepository } from '../../repositories/agent-execution-thread.repository';
import type { AgentRepository } from '../../repositories/agent.repository';
import type { AgentExecutionStreamChunk } from '../../types/agent-steering';
import { SystemAgentExecutionService } from '../system-agent-execution.service';
import { SystemAgentRegistry } from '../system-agent-registry';
import type {
SystemAgentProvider,
SystemAgentTurn,
SystemAgentTurnHandle,
} from '../system-agent.types';
vi.mock('@/permissions.ee/check-access', () => ({ userHasScopes: vi.fn() }));
const AGENT_ID = 'test-assistant';
const user = mock<User>({ id: 'user-1' });
const thread = {
id: 'thread-1',
agentId: AGENT_ID,
agentName: 'Test Assistant',
projectId: 'project-1',
accessScope: 'user',
ownerId: 'user-1',
} as AgentExecutionThread;
/** The checkpoint of a turn that waits for the answer to one tool call. */
function suspendedCheckpoint(resourceId = 'system-chat:user-1') {
return {
runId: 'run-1',
pendingToolCalls: {
'tc-1': {
toolCallId: 'tc-1',
toolName: 'ask-user',
input: {},
suspended: true,
suspendPayload: { requestId: 'req-1' },
resumeSchema: {},
runId: 'run-1',
},
},
persistence: {
threadId: 'thread-1',
resourceId,
hostMetadata: { turnOptions: { runId: 'host-run-1' } },
},
};
}
function setup(chunks: AgentExecutionStreamChunk[] = []) {
const registry = new SystemAgentRegistry();
const threadRepository = mock<AgentExecutionThreadRepository>();
const executionRepository = mock<AgentExecutionRepository>();
const turnExecutionService = mock<AgentTurnExecutionService>();
const messageQueue = mock<AgentMessageQueueService>();
const checkpointStorage = mock<N8NCheckpointStorage>();
const executionService = mock<AgentExecutionService>();
const chatExecutionService = mock<AgentChatExecutionService>();
const agentRepository = mock<AgentRepository>();
const onChunk = vi.fn();
const onSettled = vi.fn(async () => {});
const handle: SystemAgentTurnHandle = {
agent: { name: 'runtime-agent' } as unknown as SystemAgentTurnHandle['agent'],
input: 'model input',
hostMetadata: { turnOptions: { runId: 'host-run-1' } },
onChunk,
onSettled,
};
const provider = {
agentId: AGENT_ID,
name: 'Test Assistant',
authorize: vi.fn(async (_user: User, _projectId: string) => true),
prepareTurn: vi.fn(async (_turn: SystemAgentTurn) => handle),
normalizeResumeData: vi.fn((data: unknown) => ({ normalized: data })),
} satisfies SystemAgentProvider;
registry.register(provider);
threadRepository.findOwnedById.mockResolvedValue(thread);
executionRepository.findExecution.mockResolvedValue(null);
// Collect the prepared request, then stream the scripted chunks.
const prepared: AgentTurnRequest[] = [];
turnExecutionService.execute.mockImplementation(async function* (config) {
prepared.push(await config.prepare());
yield* chunks;
});
const txRunner = mock<TransactionRunner>();
const service = new SystemAgentExecutionService(
mock<Logger>(),
registry,
agentRepository,
threadRepository,
executionRepository,
executionService,
turnExecutionService,
messageQueue,
chatExecutionService,
checkpointStorage,
txRunner,
mock(),
);
return {
service,
provider,
handle,
onChunk,
onSettled,
prepared,
threadRepository,
executionRepository,
turnExecutionService,
messageQueue,
checkpointStorage,
txRunner,
executionService,
chatExecutionService,
agentRepository,
registry,
};
}
const suspendedChunk = {
type: 'tool-call-suspended',
runId: 'run-1',
toolCallId: 'tc-1',
toolName: 'ask-user',
input: {},
suspendPayload: { requestId: 'req-1' },
} as unknown as AgentExecutionStreamChunk;
function claimFor(options: Record<string, unknown> = {}): ClaimedAgentMessage {
return {
item: { id: 'queue-1' },
thread,
payload: {
kind: 'system',
message: 'Build me a workflow',
resourceId: 'system-chat:user-1',
options,
},
admission: { executionId: 'exec-1', startedAt: new Date(), inputMessageIds: ['m-1'] },
recording: { threadId: 'thread-1' },
} as unknown as ClaimedAgentMessage;
}
const textChunk = { type: 'text-delta', id: 't-1', delta: 'Hello' } as AgentExecutionStreamChunk;
describe('SystemAgentExecutionService', () => {
beforeEach(() => {
vi.mocked(userHasScopes).mockReset().mockResolvedValue(true);
});
describe('sendMessage', () => {
it('enqueues a system message with the provider turn options', async () => {
const { service, messageQueue, provider } = setup();
await service.sendMessage({
agentId: AGENT_ID,
user,
threadId: 'thread-1',
message: 'hello',
options: { runId: 'run-1', timeZone: 'Europe/Helsinki' },
hidden: true,
});
expect(provider.authorize).toHaveBeenCalledWith(user, 'project-1');
expect(messageQueue.enqueue).toHaveBeenCalledWith({
agentId: AGENT_ID,
projectId: 'project-1',
threadId: 'thread-1',
sessionMode: 'existing',
source: 'system_chat',
payload: {
kind: 'system',
userId: 'user-1',
message: 'hello',
resourceId: 'system-chat:user-1',
options: { runId: 'run-1', timeZone: 'Europe/Helsinki' },
hidden: true,
},
});
});
it('refuses a user the provider does not authorize', async () => {
const { service, messageQueue, provider } = setup();
provider.authorize.mockResolvedValue(false);
await expect(
service.sendMessage({ agentId: AGENT_ID, user, threadId: 'thread-1', message: 'hello' }),
).rejects.toThrow(NotFoundError);
expect(messageQueue.enqueue).not.toHaveBeenCalled();
});
it('refuses a thread the user does not own', async () => {
const { service, threadRepository, messageQueue } = setup();
threadRepository.findOwnedById.mockResolvedValue(null);
await expect(
service.sendMessage({ agentId: AGENT_ID, user, threadId: 'thread-1', message: 'hello' }),
).rejects.toThrow(NotFoundError);
expect(messageQueue.enqueue).not.toHaveBeenCalled();
});
});
describe('consume', () => {
it('prepares the turn and runs the provider handle on the Agents runtime', async () => {
const { service, provider, handle, turnExecutionService, prepared } = setup([textChunk]);
const signal = new AbortController().signal;
await service.consume(claimFor({ runId: 'run-1' }), user, signal, vi.fn());
expect(provider.prepareTurn).toHaveBeenCalledWith(
expect.objectContaining({
type: 'start',
user,
thread,
resourceId: 'system-chat:user-1',
executionId: 'exec-1',
message: 'Build me a workflow',
attachments: [],
options: { runId: 'run-1' },
}),
);
expect(turnExecutionService.execute).toHaveBeenCalledWith(
expect.objectContaining({
agentInstance: handle.agent,
admittedExecution: expect.objectContaining({ executionId: 'exec-1' }),
context: { projectId: 'project-1', agentId: AGENT_ID, threadId: 'thread-1' },
}),
);
expect(prepared[0]).toMatchObject({
type: 'start',
input: 'model input',
options: {
persistence: {
threadId: 'thread-1',
resourceId: 'system-chat:user-1',
hostMetadata: { turnOptions: { runId: 'host-run-1' } },
},
},
});
});
it('hides the user message from the transcript when the handle asks for it', async () => {
const { service, handle, prepared } = setup();
handle.hideUserMessage = true;
await service.consume(claimFor(), user, new AbortController().signal, vi.fn());
expect(prepared[0].recording).toMatchObject({ hideUserMessageFromTranscript: true });
});
it('forwards chunks to the handle and the client, and settles as completed', async () => {
const { service, onChunk, onSettled } = setup([textChunk]);
const send = vi.fn();
await service.consume(claimFor(), user, new AbortController().signal, send);
expect(onChunk).toHaveBeenCalledWith(textChunk);
expect(send).toHaveBeenCalledWith(expect.objectContaining({ type: 'text-delta' }));
expect(onSettled).toHaveBeenCalledWith({
status: 'completed',
executionId: 'exec-1',
error: undefined,
});
expect(send).toHaveBeenLastCalledWith({
type: 'done',
sessionId: 'thread-1',
executionId: 'exec-1',
});
});
it('settles as suspended after a tool suspends, without a done event', async () => {
const { service, onSettled } = setup([suspendedChunk]);
const send = vi.fn();
await service.consume(claimFor(), user, new AbortController().signal, send);
expect(onSettled).toHaveBeenCalledWith(expect.objectContaining({ status: 'suspended' }));
expect(send).not.toHaveBeenCalledWith(expect.objectContaining({ type: 'done' }));
});
it('settles as errored with the error of an error chunk', async () => {
const failure = new Error('model failed');
const { service, onSettled } = setup([
{ type: 'error', error: failure } as AgentExecutionStreamChunk,
]);
await service.consume(claimFor(), user, new AbortController().signal, vi.fn());
expect(onSettled).toHaveBeenCalledWith({
status: 'errored',
executionId: 'exec-1',
error: failure,
});
});
it('settles as cancelled when the recorded execution was cancelled', async () => {
const { service, onSettled, executionRepository } = setup([textChunk]);
executionRepository.findExecution.mockResolvedValue({ status: 'cancelled' } as never);
await service.consume(claimFor(), user, new AbortController().signal, vi.fn());
expect(onSettled).toHaveBeenCalledWith(expect.objectContaining({ status: 'cancelled' }));
});
it('settles as errored and rethrows when the turn throws', async () => {
const { service, onSettled, turnExecutionService } = setup();
const failure = new Error('stream broke');
// eslint-disable-next-line require-yield
turnExecutionService.execute.mockImplementation(async function* () {
throw failure;
});
await expect(
service.consume(claimFor(), user, new AbortController().signal, vi.fn()),
).rejects.toThrow('stream broke');
expect(onSettled).toHaveBeenCalledWith(
expect.objectContaining({ status: 'errored', error: failure }),
);
});
it('does not prepare a turn when the claim is already aborted', async () => {
const { service, provider } = setup();
const controller = new AbortController();
controller.abort();
await expect(service.consume(claimFor(), user, controller.signal, vi.fn())).rejects.toThrow();
expect(provider.prepareTurn).not.toHaveBeenCalled();
});
});
describe('resume', () => {
it('finds the suspended tool call in the checkpoint and prepares a resume turn', async () => {
const { service, provider, checkpointStorage, prepared } = setup([textChunk]);
checkpointStorage.findSuspendedForThread.mockResolvedValue(suspendedCheckpoint() as never);
const result = await service.resume({
agentId: AGENT_ID,
user,
threadId: 'thread-1',
resumeData: { approved: true },
});
await result.done;
expect(result).toMatchObject({ runId: 'run-1', toolCallId: 'tc-1' });
expect(checkpointStorage.findSuspendedForThread).toHaveBeenCalledWith(AGENT_ID, 'thread-1');
expect(provider.prepareTurn).toHaveBeenCalledWith(
expect.objectContaining({
type: 'resume',
thread,
resourceId: 'system-chat:user-1',
runId: 'run-1',
toolCallId: 'tc-1',
checkpointHostMetadata: { turnOptions: { runId: 'host-run-1' } },
resumeData: { approved: true },
}),
);
expect(prepared[0]).toMatchObject({
type: 'resume',
resumeData: { approved: true },
options: { runId: 'run-1', toolCallId: 'tc-1' },
});
});
it('refuses when no tool call waits for input', async () => {
const { service, provider, checkpointStorage } = setup();
checkpointStorage.findSuspendedForThread.mockResolvedValue(null);
await expect(
service.resume({ agentId: AGENT_ID, user, threadId: 'thread-1', resumeData: {} }),
).rejects.toThrow('This action is no longer waiting for input');
expect(provider.prepareTurn).not.toHaveBeenCalled();
});
it('refuses a checkpoint stored for another memory resource', async () => {
const { service, provider, checkpointStorage } = setup();
checkpointStorage.findSuspendedForThread.mockResolvedValue(
suspendedCheckpoint('system-chat:user-2') as never,
);
await expect(
service.resume({ agentId: AGENT_ID, user, threadId: 'thread-1', resumeData: {} }),
).rejects.toThrow(NotFoundError);
expect(provider.prepareTurn).not.toHaveBeenCalled();
});
it('throws NotFoundError when the provider does not authorize the user', async () => {
const { service, provider, checkpointStorage } = setup();
provider.authorize.mockResolvedValue(false);
checkpointStorage.findSuspendedForThread.mockResolvedValue(suspendedCheckpoint() as never);
await expect(
service.resume({ agentId: AGENT_ID, user, threadId: 'thread-1', resumeData: {} }),
).rejects.toThrow(NotFoundError);
expect(provider.prepareTurn).not.toHaveBeenCalled();
});
});
describe('resumeRun', () => {
it('normalizes the client resume data with the provider before it resumes', async () => {
const { service, provider, checkpointStorage } = setup([textChunk]);
checkpointStorage.load.mockResolvedValue({
persistence: { threadId: 'thread-1' },
} as never);
checkpointStorage.findSuspendedForThread.mockResolvedValue(suspendedCheckpoint() as never);
await service.resumeRun({
agentId: AGENT_ID,
user,
runId: 'run-1',
toolCallId: 'tc-1',
resumeData: { kind: 'approval', approved: true },
send: vi.fn(),
});
expect(checkpointStorage.load).toHaveBeenCalledWith('run-1', AGENT_ID);
expect(provider.normalizeResumeData).toHaveBeenCalledWith({
kind: 'approval',
approved: true,
});
expect(provider.prepareTurn).toHaveBeenCalledWith(
expect.objectContaining({
type: 'resume',
resumeData: { normalized: { kind: 'approval', approved: true } },
}),
);
});
it('refuses a run without a stored checkpoint', async () => {
const { service, checkpointStorage } = setup();
checkpointStorage.load.mockResolvedValue(undefined);
await expect(
service.resumeRun({
agentId: AGENT_ID,
user,
runId: 'run-1',
toolCallId: 'tc-1',
resumeData: {},
send: vi.fn(),
}),
).rejects.toThrow('This action is no longer waiting for input');
});
});
describe('getStatus', () => {
it('reports a thread with a queued message as running', async () => {
const { service, executionRepository, checkpointStorage, messageQueue } = setup();
executionRepository.existsRunningByThread.mockResolvedValue(false);
checkpointStorage.findSuspendedForThread.mockResolvedValue(null);
messageQueue.hasQueuedMessages.mockResolvedValue(true);
expect((await service.getStatus(thread)).status).toBe('running');
});
it('keeps a suspended thread suspended while a message waits', async () => {
const { service, executionRepository, checkpointStorage, messageQueue } = setup();
executionRepository.existsRunningByThread.mockResolvedValue(false);
checkpointStorage.findSuspendedForThread.mockResolvedValue(suspendedCheckpoint() as never);
messageQueue.hasQueuedMessages.mockResolvedValue(true);
expect((await service.getStatus(thread)).status).toBe('suspended');
});
it('reports an idle thread with an empty queue', async () => {
const { service, executionRepository, checkpointStorage, messageQueue } = setup();
executionRepository.existsRunningByThread.mockResolvedValue(false);
checkpointStorage.findSuspendedForThread.mockResolvedValue(null);
messageQueue.hasQueuedMessages.mockResolvedValue(false);
expect((await service.getStatus(thread)).status).toBe('idle');
});
});
describe('access floor', () => {
it('refuses a user who cannot read the working project, before the provider checks', async () => {
const { service, provider } = setup();
vi.mocked(userHasScopes).mockResolvedValue(false);
await expect(service.assertCanUse(AGENT_ID, user, 'project-1')).rejects.toThrow(
NotFoundError,
);
expect(userHasScopes).toHaveBeenCalledWith(user, ['project:read'], false, {
projectId: 'project-1',
});
expect(provider.authorize).not.toHaveBeenCalled();
});
it('loads a usable thread with the floor on its working project', async () => {
const { service, provider } = setup();
expect(await service.getUsableThread(AGENT_ID, user, 'thread-1')).toBe(thread);
expect(userHasScopes).toHaveBeenCalledWith(user, ['project:read'], false, {
projectId: 'project-1',
});
expect(provider.authorize).toHaveBeenCalledWith(user, 'project-1');
});
it('refuses a thread the user does not own', async () => {
const { service, threadRepository } = setup();
threadRepository.findOwnedById.mockResolvedValue(null);
await expect(service.getUsableThread(AGENT_ID, user, 'thread-1')).rejects.toThrow(
NotFoundError,
);
});
});
describe('prepareChatMessage', () => {
it('keeps the working project of an existing session', async () => {
const { service, threadRepository } = setup();
threadRepository.findOwnedById.mockResolvedValue(thread);
const input = await service.prepareChatMessage({
agentId: AGENT_ID,
user,
sessionId: 'thread-1',
message: 'hi',
});
expect(input).toMatchObject({
projectId: 'project-1',
threadId: 'thread-1',
payload: { kind: 'system' },
});
});
it('refuses a new session without a working project', async () => {
const { service, threadRepository } = setup();
threadRepository.findOwnedById.mockResolvedValue(null);
await expect(
service.prepareChatMessage({ agentId: AGENT_ID, user, message: 'hi' }),
).rejects.toThrow('A new session needs a working project');
});
it('refuses a project that does not match the existing session', async () => {
const { service, threadRepository } = setup();
threadRepository.findOwnedById.mockResolvedValue(thread);
await expect(
service.prepareChatMessage({
agentId: AGENT_ID,
user,
projectId: 'other-project',
sessionId: 'thread-1',
message: 'hi',
}),
).rejects.toThrow(NotFoundError);
});
});
describe('assertCanUse', () => {
it('throws NotFoundError for an agent id without a registered provider', async () => {
const { service } = setup();
await expect(service.assertCanUse('unknown-agent', user, 'project-1')).rejects.toThrow(
NotFoundError,
);
});
});
describe('register', () => {
it('seeds the instance agent row before it registers the provider', async () => {
const { service, agentRepository, registry } = setup();
const provider = {
agentId: 'other-agent',
name: 'Other Agent',
authorize: vi.fn(async () => true),
prepareTurn: vi.fn(),
} satisfies SystemAgentProvider;
await service.register(provider);
expect(agentRepository.ensureInstanceAgent).toHaveBeenCalledWith(
'other-agent',
'Other Agent',
);
expect(registry.get('other-agent')).toBe(provider);
});
it('does not register a provider whose agent row cannot be seeded', async () => {
const { service, agentRepository, registry } = setup();
agentRepository.ensureInstanceAgent.mockRejectedValue(new Error('project agent'));
await expect(
service.register({
agentId: 'other-agent',
name: 'Other Agent',
authorize: vi.fn(async () => true),
prepareTurn: vi.fn(),
}),
).rejects.toThrow('project agent');
expect(registry.has('other-agent')).toBe(false);
});
});
describe('hidden turns', () => {
it('runs a hidden queued message like any other system turn', async () => {
const { service, provider, prepared } = setup([textChunk]);
const claim = claimFor({ followUp: true });
claim.payload = {
kind: 'system',
message: 'Build me a workflow',
resourceId: 'system-chat:user-1',
options: { followUp: true },
hidden: true,
};
await service.consume(claim, user, new AbortController().signal, vi.fn());
expect(provider.prepareTurn).toHaveBeenCalledWith(
expect.objectContaining({ type: 'start', options: { followUp: true } }),
);
expect(prepared[0]).toMatchObject({ type: 'start', input: 'model input' });
});
});
describe('steerIntoRunningTurn', () => {
it('steers a queued system message into the running turn', async () => {
const { service, executionRepository, messageQueue } = setup();
executionRepository.findSteerable.mockResolvedValue({ id: 'exec-1' } as never);
expect(await service.steerIntoRunningTurn(thread, user, '7')).toBe(true);
expect(messageQueue.steer).toHaveBeenCalledWith({
projectId: 'project-1',
agentId: AGENT_ID,
threadId: 'thread-1',
userId: 'user-1',
queueId: '7',
executionId: 'exec-1',
kind: 'system',
});
});
it('leaves the message queued when no turn accepts steering', async () => {
const { service, executionRepository, messageQueue } = setup();
executionRepository.findSteerable.mockResolvedValue(null);
expect(await service.steerIntoRunningTurn(thread, user, '7')).toBe(false);
expect(messageQueue.steer).not.toHaveBeenCalled();
});
it('leaves the message queued when the queue refuses the steer', async () => {
const { service, executionRepository, messageQueue } = setup();
executionRepository.findSteerable.mockResolvedValue({ id: 'exec-1' } as never);
messageQueue.steer.mockRejectedValue(new Error('This message is no longer available'));
expect(await service.steerIntoRunningTurn(thread, user, '7')).toBe(false);
});
});
describe('cancel', () => {
it('requests a cancel of the latest execution of an owned thread', async () => {
const { service, executionRepository, chatExecutionService } = setup();
executionRepository.findLatestByThreadId.mockResolvedValue({ id: 'exec-2' } as never);
chatExecutionService.requestCancel.mockResolvedValue(true);
expect(await service.cancel(AGENT_ID, user, 'thread-1')).toBe(true);
expect(chatExecutionService.requestCancel).toHaveBeenCalledWith({
projectId: 'project-1',
agentId: AGENT_ID,
threadId: 'thread-1',
executionId: 'exec-2',
userId: 'user-1',
surface: 'system',
});
});
it('returns false for a thread without executions', async () => {
const { service, executionRepository, chatExecutionService } = setup();
executionRepository.findLatestByThreadId.mockResolvedValue(null);
expect(await service.cancel(AGENT_ID, user, 'thread-1')).toBe(false);
expect(chatExecutionService.requestCancel).not.toHaveBeenCalled();
});
it('refuses a thread the user does not own', async () => {
const { service, threadRepository, chatExecutionService } = setup();
threadRepository.findOwnedById.mockResolvedValue(null);
await expect(service.cancel(AGENT_ID, user, 'thread-1')).rejects.toThrow(NotFoundError);
expect(chatExecutionService.requestCancel).not.toHaveBeenCalled();
});
});
describe('threads', () => {
it('deletes an owned thread through the Agents execution service', async () => {
const { service, executionService } = setup();
executionService.deleteThread.mockResolvedValue(true);
await service.deleteThread(AGENT_ID, user, 'thread-1');
expect(executionService.deleteThread).toHaveBeenCalledWith(
'project-1',
AGENT_ID,
'thread-1',
'user-1',
);
});
it('refuses to move a thread into a project the user cannot use', async () => {
const { service, provider, threadRepository } = setup();
provider.authorize.mockImplementation(
async (_user: User, projectId: string) => projectId === 'project-1',
);
await expect(
service.updateThread(AGENT_ID, user, 'thread-1', { projectId: 'project-2' }),
).rejects.toThrow(NotFoundError);
expect(threadRepository.moveOwned).not.toHaveBeenCalled();
});
it('renames and moves a thread in one transaction', async () => {
const { service, threadRepository, txRunner, executionRepository, checkpointStorage } =
setup();
const ctx = {};
executionRepository.existsRunningByThread.mockResolvedValue(false);
checkpointStorage.findSuspendedForThread.mockResolvedValue(null);
txRunner.run.mockImplementation(async (_ctx, fn) => await fn(ctx));
threadRepository.renameOwned.mockResolvedValue(true);
threadRepository.moveOwned.mockResolvedValue(true);
const updated = await service.updateThread(AGENT_ID, user, 'thread-1', {
title: 'Renamed',
projectId: 'project-2',
});
expect(threadRepository.renameOwned).toHaveBeenCalledWith(
AGENT_ID,
user.id,
'thread-1',
'Renamed',
ctx,
);
expect(threadRepository.moveOwned).toHaveBeenCalledWith(
AGENT_ID,
user.id,
'thread-1',
'project-2',
ctx,
);
expect(updated).toMatchObject({ title: 'Renamed', projectId: 'project-2' });
});
it('reports a thread that was deleted during the update as not found', async () => {
const { service, threadRepository, txRunner } = setup();
txRunner.run.mockImplementation(async (ctx, fn) => await fn(ctx));
threadRepository.renameOwned.mockResolvedValue(false);
await expect(
service.updateThread(AGENT_ID, user, 'thread-1', { title: 'Renamed' }),
).rejects.toThrow(NotFoundError);
});
it('refuses to create a session with the id of a thread of another user', async () => {
const { service, threadRepository, txRunner } = setup();
threadRepository.findOneBy.mockResolvedValue({ ...thread, ownerId: 'user-2' } as never);
await expect(
service.createThread({
agentId: AGENT_ID,
user,
projectId: 'project-1',
threadId: 'thread-1',
}),
).rejects.toThrow(NotFoundError);
expect(txRunner.run).not.toHaveBeenCalled();
});
});
});

22
packages/cli/src/modules/agents/system-agents/system-agent-access.ts Normal file
View File

@ -0,0 +1,22 @@
import type { User } from '@n8n/db';
import { userHasScopes } from '@/permissions.ee/check-access';
import type { SystemAgentProvider } from './system-agent.types';
/**
* Whether the user can use a system agent in a working project. The runtime
* floor applies to every system agent: the user must be able to read the
* working project, because the agent acts in it with the user's permissions.
* The floor uses `project:read`, not `agent:execute`: `agent:execute` gates
* the custom agents of a project. The provider then adds its own checks.
* Thread ownership is checked where the thread is loaded.
*/
export async function canUseSystemAgent(
provider: SystemAgentProvider,
user: User,
projectId: string,
): Promise<boolean> {
if (!(await userHasScopes(user, ['project:read'], false, { projectId }))) return false;
return await provider.authorize(user, projectId);
}

599
packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts Normal file
View File

@ -0,0 +1,599 @@
import type { SerializableAgentState } from '@n8n/agents';
import type { AgentSseEvent } from '@n8n/api-types';
import { Logger } from '@n8n/backend-common';
import { TransactionRunner, type User } from '@n8n/db';
import { Service } from '@n8n/di';
import { NotFoundError } from '@n8n/errors';
import { UserError } from 'n8n-workflow';
import { randomUUID } from 'node:crypto';
import { AgentChatAttachmentService } from '../agent-chat-attachment.service';
import { AgentChatExecutionService } from '../agent-chat-execution.service';
import { AgentExecutionService, type StartExecutionParams } from '../agent-execution.service';
import type { ClaimedAgentMessage } from '../agent-message-queue.service';
import { AgentMessageQueueService } from '../agent-message-queue.service';
import { emitChunkEvents } from '../agent-sse-stream';
import { AgentTurnExecutionService, type AgentTurnRequest } from '../agent-turn-execution.service';
import type { AgentExecutionThread } from '../entities/agent-execution-thread.entity';
import { N8NCheckpointStorage } from '../integrations/n8n-checkpoint-storage';
import { AgentExecutionRepository } from '../repositories/agent-execution.repository';
import { AgentExecutionThreadRepository } from '../repositories/agent-execution-thread.repository';
import { AgentRepository } from '../repositories/agent.repository';
import type { StoredAttachmentRef } from '../types/agent-chat-attachment';
import type { AgentExecutionAdmission } from '../types/agent-queued-message';
import { systemChatMemoryResourceId } from '../utils/agent-memory-scope';
import { buildInboundUserMessage } from '../utils/inbound-attachments';
import { SYSTEM_CHAT_SOURCE } from '../utils/agent-thread-access';
import { canUseSystemAgent } from './system-agent-access';
import { SystemAgentRegistry } from './system-agent-registry';
import type {
SystemAgentProvider,
SystemAgentTurnHandle,
SystemAgentTurnOptions,
SystemAgentTurnStatus,
} from './system-agent.types';
type SuspendedToolCall = Extract<
SerializableAgentState['pendingToolCalls'][string],
{ suspended: true }
>;
export type SystemAgentThreadStatus = 'running' | 'suspended' | 'idle';
function combineSignals(signal: AbortSignal, extra: AbortSignal | undefined): AbortSignal {
return extra ? AbortSignal.any([signal, extra]) : signal;
}
/**
* Runs code-defined instance agents on the Agents runtime. A conversation is
* a private session (`accessScope = 'user'`). Its `projectId` is the working
* project. The queue, steering, checkpoints, cancel and recording are shared
* with project-agent chats, but the conversation has its own `system` chat
* surface: its own memory resource scope, transcript source and access rule.
*
* Every entry point that starts or continues a turn checks the runtime floor:
* the user owns the thread, can read its working project, and passes the
* provider's own checks.
*/
@Service()
export class SystemAgentExecutionService {
constructor(
private readonly logger: Logger,
private readonly registry: SystemAgentRegistry,
private readonly agentRepository: AgentRepository,
private readonly threadRepository: AgentExecutionThreadRepository,
private readonly executionRepository: AgentExecutionRepository,
private readonly executionService: AgentExecutionService,
private readonly turnExecutionService: AgentTurnExecutionService,
private readonly messageQueue: AgentMessageQueueService,
private readonly chatExecutionService: AgentChatExecutionService,
private readonly checkpointStorage: N8NCheckpointStorage,
private readonly txRunner: TransactionRunner,
private readonly attachmentService: AgentChatAttachmentService,
) {}
/** Register a provider and make sure its instance agent row exists. */
async register(provider: SystemAgentProvider): Promise<void> {
await this.agentRepository.ensureInstanceAgent(provider.agentId, provider.name);
this.registry.register(provider);
}
resourceIdFor(user: Pick<User, 'id'>): string {
return systemChatMemoryResourceId(user.id);
}
getProvider(agentId: string): SystemAgentProvider {
const provider = this.registry.get(agentId);
if (!provider) throw new NotFoundError(`Agent "${agentId}" not found`);
return provider;
}
/** Throws NotFoundError when no provider is registered for the id. */
assertRegistered(agentId: string): void {
this.getProvider(agentId);
}
/** The runtime floor plus the provider checks, for a working project. */
async assertCanUse(agentId: string, user: User, projectId: string): Promise<SystemAgentProvider> {
const provider = this.getProvider(agentId);
// Refuse with "not found", so a user without access learns nothing about the agent.
if (!(await canUseSystemAgent(provider, user, projectId))) {
throw new NotFoundError(`Agent "${agentId}" not found`);
}
return provider;
}
/** Load a thread the user owns and can still use: thread ownership, the floor and the provider. */
async getUsableThread(
agentId: string,
user: User,
threadId: string,
): Promise<AgentExecutionThread> {
const thread = await this.getThread(agentId, user, threadId);
await this.assertCanUse(agentId, user, thread.projectId);
return thread;
}
// ── Threads ──────────────────────────────────────────────────────────────
async createThread(params: {
agentId: string;
user: User;
projectId: string;
threadId?: string;
title?: string;
}): Promise<AgentExecutionThread> {
const provider = await this.assertCanUse(params.agentId, params.user, params.projectId);
const threadId = params.threadId ?? randomUUID();
const existing = await this.threadRepository.findOneBy({ id: threadId });
if (existing) {
if (
existing.agentId !== params.agentId ||
existing.accessScope !== 'user' ||
existing.ownerId !== params.user.id
) {
throw new NotFoundError('Session not found');
}
return existing;
}
return await this.txRunner.run({}, async (ctx) => {
const { thread } = await this.threadRepository.findOrCreate(
threadId,
provider.agentId,
provider.name,
params.projectId,
{ accessScope: 'user', ownerId: params.user.id },
ctx,
);
if (params.title) {
await this.threadRepository.renameOwned(
provider.agentId,
params.user.id,
thread.id,
params.title,
ctx,
);
thread.title = params.title;
}
return thread;
});
}
async listThreads(agentId: string, user: User, limit?: number) {
this.assertRegistered(agentId);
return await this.threadRepository.findOwnedByAgent(agentId, user.id, { limit });
}
/** Load a thread the user owns. It checks ownership only, not the floor. */
async getThread(agentId: string, user: User, threadId: string): Promise<AgentExecutionThread> {
this.assertRegistered(agentId);
const thread = await this.threadRepository.findOwnedById(agentId, user.id, threadId);
if (!thread) throw new NotFoundError('Session not found');
return thread;
}
async updateThread(
agentId: string,
user: User,
threadId: string,
changes: { title?: string; projectId?: string },
): Promise<AgentExecutionThread> {
const thread = await this.getThread(agentId, user, threadId);
if (changes.projectId && changes.projectId !== thread.projectId) {
await this.assertCanUse(agentId, user, changes.projectId);
if (await this.isBusy(thread)) {
throw new UserError('Wait for the current turn to finish before you change the project');
}
}
await this.txRunner.run({}, async (ctx) => {
const { title, projectId } = changes;
const renamed =
title === undefined ||
(await this.threadRepository.renameOwned(agentId, user.id, threadId, title, ctx));
const moved =
projectId === undefined ||
(await this.threadRepository.moveOwned(agentId, user.id, threadId, projectId, ctx));
if (!renamed || !moved) throw new NotFoundError('Session not found');
});
if (changes.title !== undefined) thread.title = changes.title;
if (changes.projectId !== undefined) thread.projectId = changes.projectId;
return thread;
}
/** Delete a thread with its executions, memory and stored files. */
async deleteThread(agentId: string, user: User, threadId: string): Promise<void> {
const thread = await this.getThread(agentId, user, threadId);
const deleted = await this.executionService.deleteThread(
thread.projectId,
agentId,
thread.id,
user.id,
);
if (!deleted) throw new NotFoundError('Session not found');
}
async isBusy(thread: AgentExecutionThread): Promise<boolean> {
if (await this.executionRepository.existsRunningByThread(thread.id)) return true;
return (
(await this.checkpointStorage.findSuspendedForThread(thread.agentId, thread.id)) !== null
);
}
/**
* Status of the thread. A queued message (for example a hidden follow-up
* turn) counts as running, so a client does not see the thread as idle
* between a turn and its follow-up. An open suspension still wins: the
* queue waits for the user to answer it.
*/
async getStatus(thread: AgentExecutionThread): Promise<{
status: SystemAgentThreadStatus;
latestExecutionId: string | null;
checkpoint: SerializableAgentState | null;
}> {
const running = await this.executionRepository.existsRunningByThread(thread.id);
const checkpoint = running
? null
: await this.checkpointStorage.findSuspendedForThread(thread.agentId, thread.id);
const latest = await this.executionRepository.findLatestByThreadId(thread.id);
const queued =
!running && !checkpoint && (await this.messageQueue.hasQueuedMessages(thread.id));
let status: SystemAgentThreadStatus = 'idle';
if (running || queued) status = 'running';
else if (checkpoint) status = 'suspended';
return { status, latestExecutionId: latest?.id ?? null, checkpoint };
}
// ── Messages ─────────────────────────────────────────────────────────────
/**
* Queue a message. The queue starts it when the conversation is idle.
* While a turn runs, the caller can steer the queued message into it.
* A `hidden` message is a machine turn: it stays out of the transcript.
*/
async sendMessage(params: {
agentId: string;
user: User;
threadId: string;
message: string;
attachments?: StoredAttachmentRef[];
options?: SystemAgentTurnOptions;
messageId?: string;
hidden?: boolean;
}) {
const thread = await this.getUsableThread(params.agentId, params.user, params.threadId);
return await this.messageQueue.enqueue({
agentId: params.agentId,
projectId: thread.projectId,
threadId: thread.id,
sessionMode: 'existing',
source: SYSTEM_CHAT_SOURCE,
payload: {
kind: 'system',
userId: params.user.id,
message: params.message,
resourceId: this.resourceIdFor(params.user),
...(params.attachments?.length ? { attachments: params.attachments } : {}),
...(params.messageId ? { messageId: params.messageId } : {}),
...(params.options ? { options: params.options } : {}),
...(params.hidden ? { hidden: true } : {}),
},
});
}
/**
* Steer a queued message into the running turn. Returns false when no turn
* accepts steering; the message then waits in the queue.
*/
async steerIntoRunningTurn(
thread: AgentExecutionThread,
user: User,
queueId: string,
): Promise<boolean> {
const execution = await this.executionRepository.findSteerable(thread.id);
if (!execution) return false;
try {
await this.messageQueue.steer({
projectId: thread.projectId,
agentId: thread.agentId,
threadId: thread.id,
userId: user.id,
queueId,
executionId: execution.id,
kind: 'system',
});
return true;
} catch (error) {
this.logger.debug('Could not steer system agent message', { threadId: thread.id, error });
return false;
}
}
/**
* Run a claimed system-agent message. The queue consumer calls this after
* it checked the runtime floor for the thread owner.
*/
async consume(
claim: ClaimedAgentMessage,
user: User,
signal: AbortSignal,
send: (event: AgentSseEvent) => void,
): Promise<void> {
const { thread, admission, payload } = claim;
if (payload.kind !== 'system') return;
const provider = this.getProvider(thread.agentId);
signal.throwIfAborted();
const { resourceId } = payload;
const attachments = payload.attachments ?? [];
const handle = await provider.prepareTurn({
type: 'start',
user,
thread,
resourceId,
abortSignal: signal,
executionId: admission.executionId,
message: payload.message,
attachments,
options: payload.options ?? {},
});
const text = handle.input ?? payload.message;
if (attachments.length > 0) {
// Messages keep file references. The runtime loads bytes from binary data per call.
handle.agent.fileStore(
this.attachmentService.getFileStore(
{ agentId: thread.agentId, projectId: thread.projectId },
handle.agent.snapshot.model.provider ?? '',
),
);
}
const input =
attachments.length > 0 && typeof text === 'string'
? buildInboundUserMessage(text, attachments)
: text;
const abortSignal = combineSignals(signal, handle.runOptions?.abortSignal);
await this.runTurn(provider, thread, handle, send, abortSignal, admission, async () => ({
type: 'start',
input,
options: {
persistence: { threadId: thread.id, resourceId, hostMetadata: handle.hostMetadata },
...handle.runOptions,
abortSignal,
},
recording: {
...claim.recording,
...(handle.hideUserMessage ? { hideUserMessageFromTranscript: true } : {}),
},
}));
}
/** Resume a suspended tool call. Runs the continuation in the background. */
async resume(params: {
agentId: string;
user: User;
threadId: string;
resumeData: unknown;
runId?: string;
toolCallId?: string;
send?: (event: AgentSseEvent) => void;
}): Promise<{ runId: string; toolCallId: string; done: Promise<void> }> {
const thread = await this.getUsableThread(params.agentId, params.user, params.threadId);
const provider = this.getProvider(params.agentId);
const checkpoint = await this.checkpointStorage.findSuspendedForThread(
thread.agentId,
thread.id,
);
const pending = Object.values(checkpoint?.pendingToolCalls ?? {}).find(
(toolCall): toolCall is SuspendedToolCall =>
toolCall.suspended &&
(params.toolCallId === undefined || toolCall.toolCallId === params.toolCallId),
);
if (!checkpoint || !pending) throw new UserError('This action is no longer waiting for input');
const runId = params.runId ?? pending.runId;
const resourceId = this.resourceIdFor(params.user);
if (checkpoint.persistence?.resourceId !== resourceId) {
throw new NotFoundError('Session not found');
}
const controller = new AbortController();
const handle = await provider.prepareTurn({
type: 'resume',
user: params.user,
thread,
resourceId,
abortSignal: controller.signal,
runId,
toolCallId: pending.toolCallId,
checkpointHostMetadata: checkpoint.persistence.hostMetadata ?? {},
resumeData: params.resumeData,
});
const recording: StartExecutionParams = {
threadId: thread.id,
agentId: thread.agentId,
agentName: thread.agentName,
projectId: thread.projectId,
resourceId,
access: { accessScope: 'user', ownerId: params.user.id },
userMessage: null,
sessionMode: 'existing',
source: SYSTEM_CHAT_SOURCE,
};
const abortSignal = combineSignals(controller.signal, handle.runOptions?.abortSignal);
const done = this.runTurn(
provider,
thread,
handle,
params.send ?? (() => {}),
abortSignal,
undefined,
async () => ({
type: 'resume',
resumeData: params.resumeData,
options: {
...handle.runOptions,
runId,
toolCallId: pending.toolCallId,
hostMetadata: handle.hostMetadata,
abortSignal,
},
recording,
}),
).catch((error: unknown) => {
this.logger.warn('System agent resume failed', { threadId: thread.id, error });
});
return { runId, toolCallId: pending.toolCallId, done };
}
/**
* Build the queue input for a message sent through the system-agent chat
* route. Creates the session when the client starts a new one.
*/
async prepareChatMessage(params: {
agentId: string;
user: User;
/** The working project of a new session. An existing session keeps its own. */
projectId?: string;
sessionId?: string;
message: string;
messageId?: string;
clientContext?: Record<string, unknown>;
storeAttachments?: (
threadId: string,
projectId: string,
) => Promise<StoredAttachmentRef[] | undefined>;
}): Promise<Parameters<AgentMessageQueueService['enqueue']>[0]> {
this.assertRegistered(params.agentId);
const existing = params.sessionId
? await this.threadRepository.findOwnedById(params.agentId, params.user.id, params.sessionId)
: null;
if (existing && params.projectId && existing.projectId !== params.projectId) {
throw new NotFoundError('Session not found');
}
const projectId = existing?.projectId ?? params.projectId;
if (!projectId) throw new UserError('A new session needs a working project');
const provider = await this.assertCanUse(params.agentId, params.user, projectId);
const thread =
existing ??
(await this.createThread({
agentId: params.agentId,
user: params.user,
projectId,
...(params.sessionId ? { threadId: params.sessionId } : {}),
}));
const options =
(await provider.chatTurnOptions?.(params.user, thread, params.clientContext)) ?? {};
const attachments = await params.storeAttachments?.(thread.id, thread.projectId);
return {
agentId: params.agentId,
projectId: thread.projectId,
threadId: thread.id,
sessionMode: 'existing',
source: SYSTEM_CHAT_SOURCE,
payload: {
kind: 'system',
userId: params.user.id,
message: params.message,
resourceId: this.resourceIdFor(params.user),
...(params.messageId ? { messageId: params.messageId } : {}),
...(attachments?.length ? { attachments } : {}),
options,
},
};
}
/** Resume by run id, for the system-agent chat resume route. Streams to `send`. */
async resumeRun(params: {
agentId: string;
user: User;
runId: string;
toolCallId: string;
resumeData: unknown;
send: (event: AgentSseEvent) => void;
}): Promise<void> {
const provider = this.getProvider(params.agentId);
const state = await this.checkpointStorage.load(params.runId, params.agentId);
const threadId = state?.persistence?.threadId;
if (!threadId) throw new UserError('This action is no longer waiting for input');
const { done } = await this.resume({
agentId: params.agentId,
user: params.user,
threadId,
runId: params.runId,
toolCallId: params.toolCallId,
resumeData: provider.normalizeResumeData
? provider.normalizeResumeData(params.resumeData)
: params.resumeData,
send: params.send,
});
await done;
}
/** Stop the running turn, or cancel the suspended one. */
async cancel(agentId: string, user: User, threadId: string): Promise<boolean> {
const thread = await this.getThread(agentId, user, threadId);
const latest = await this.executionRepository.findLatestByThreadId(thread.id);
if (!latest) return false;
return await this.chatExecutionService.requestCancel({
projectId: thread.projectId,
agentId,
threadId: thread.id,
executionId: latest.id,
userId: user.id,
surface: 'system',
});
}
private async runTurn(
provider: SystemAgentProvider,
thread: AgentExecutionThread,
handle: SystemAgentTurnHandle,
send: (event: AgentSseEvent) => void,
abortSignal: AbortSignal,
admission: AgentExecutionAdmission | undefined,
prepare: () => Promise<AgentTurnRequest>,
): Promise<void> {
let status: SystemAgentTurnStatus = 'completed';
let executionId = admission?.executionId;
let error: unknown;
try {
const stream = this.turnExecutionService.execute({
admittedExecution: admission,
agentInstance: handle.agent,
toolRegistry: handle.toolRegistry ?? new Map(),
mcpServerAttributions: new Map(),
context: { projectId: thread.projectId, agentId: provider.agentId, threadId: thread.id },
chatSurface: 'system',
onExecutionStarted: (id) => {
executionId = id;
},
prepare,
});
for await (const chunk of stream) {
emitChunkEvents(chunk, send);
if (chunk.type === 'tool-call-suspended') status = 'suspended';
if (chunk.type === 'error' && status !== 'suspended') {
status = 'errored';
error = chunk.error;
}
handle.onChunk?.(chunk);
}
} catch (caught) {
error = caught;
status = abortSignal.aborted ? 'cancelled' : 'errored';
throw caught;
} finally {
const execution = executionId
? await this.executionRepository.findExecution(executionId)
: null;
if (execution?.status === 'cancelled') status = 'cancelled';
try {
await handle.onSettled?.({ status, executionId, error });
} catch (settleError) {
this.logger.warn('System agent turn settle hook failed', {
threadId: thread.id,
error: settleError,
});
}
if (status !== 'suspended' && status !== 'cancelled') {
send({ type: 'done', sessionId: thread.id, executionId: executionId ?? '' });
}
}
}
}

29
packages/cli/src/modules/agents/system-agents/system-agent-registry.ts Normal file
View File

@ -0,0 +1,29 @@
import { Service } from '@n8n/di';
import { UnexpectedError } from 'n8n-workflow';
import type { SystemAgentProvider } from './system-agent.types';
/**
* Holds the code-defined instance agents. A module registers its provider at
* init through `SystemAgentExecutionService.register`. With no provider, every
* system-agent hook in the Agents runtime is inert.
*/
@Service()
export class SystemAgentRegistry {
private readonly providers = new Map<string, SystemAgentProvider>();
register(provider: SystemAgentProvider): void {
if (this.providers.has(provider.agentId)) {
throw new UnexpectedError(`System agent "${provider.agentId}" is already registered`);
}
this.providers.set(provider.agentId, provider);
}
get(agentId: string): SystemAgentProvider | undefined {
return this.providers.get(agentId);
}
has(agentId: string): boolean {
return this.providers.has(agentId);
}
}

94
packages/cli/src/modules/agents/system-agents/system-agent.types.ts Normal file
View File

@ -0,0 +1,94 @@
import type {
Agent as RuntimeAgent,
ExecutionOptions,
JSONObject,
Message,
ResumeOptions,
RunOptions,
} from '@n8n/agents';
import type { User } from '@n8n/db';
import type { AgentExecutionThread } from '../entities/agent-execution-thread.entity';
import type { ToolRegistry } from '../tool-registry';
import type { StoredAttachmentRef } from '../types/agent-chat-attachment';
import type { AgentExecutionStreamChunk } from '../types/agent-steering';
/** Provider-defined, JSON-safe turn options. The queue stores them with the message. */
export type SystemAgentTurnOptions = Record<string, unknown>;
interface SystemAgentTurnBase {
user: User;
thread: AgentExecutionThread;
resourceId: string;
abortSignal: AbortSignal;
}
export interface SystemAgentStartTurn extends SystemAgentTurnBase {
type: 'start';
/** Execution id when the queue admitted the turn. */
executionId?: string;
/** The message as the user wrote it. It is the stored transcript text. */
message: string;
attachments: StoredAttachmentRef[];
options: SystemAgentTurnOptions;
}
export interface SystemAgentResumeTurn extends SystemAgentTurnBase {
type: 'resume';
runId: string;
toolCallId: string;
/** Host metadata the suspended run stored in its checkpoint. */
checkpointHostMetadata: JSONObject;
resumeData: unknown;
}
export type SystemAgentTurn = SystemAgentStartTurn | SystemAgentResumeTurn;
export type SystemAgentTurnStatus = 'completed' | 'suspended' | 'errored' | 'cancelled';
export interface SystemAgentTurnOutcome {
status: SystemAgentTurnStatus;
executionId?: string;
error?: unknown;
}
/** A built runtime for one turn, plus the hooks the host calls while the turn runs. */
export interface SystemAgentTurnHandle {
agent: RuntimeAgent;
toolRegistry?: ToolRegistry;
/** Model input for a start turn. Defaults to the stored message. */
input?: string | Message[];
/** Stored on the checkpoint. A resume receives it back. */
hostMetadata?: JSONObject;
/** Extra SDK run options, such as telemetry. */
runOptions?: Partial<RunOptions & ExecutionOptions> & Partial<ResumeOptions>;
/** Hide the user message from the transcript (machine turns). */
hideUserMessage?: boolean;
onChunk?: (chunk: AgentExecutionStreamChunk) => void;
onSettled?: (outcome: SystemAgentTurnOutcome) => Promise<void>;
}
/**
* A code-defined, instance-level agent. The Agents runtime owns the queue,
* steering, checkpoints, HITL resume and recording. The provider builds a
* fresh runtime for each turn and adds its own access checks on top of the
* runtime floor (see `canUseSystemAgent`).
*/
export interface SystemAgentProvider {
readonly agentId: string;
readonly name: string;
/**
* Agent-specific access checks for the user in the working project. The
* runtime calls this only after its own floor check passed.
*/
authorize(user: User, projectId: string): Promise<boolean>;
prepareTurn(turn: SystemAgentTurn): Promise<SystemAgentTurnHandle>;
/** Turn options for a message sent through the system-agent chat route. */
chatTurnOptions?(
user: User,
thread: AgentExecutionThread,
clientContext?: Record<string, unknown>,
): Promise<SystemAgentTurnOptions>;
/** Convert a resume payload from a client into the tool's resume data. */
normalizeResumeData?(resumeData: unknown): unknown;
}

123
packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts
View File

@ -95,6 +95,7 @@ import { AgentChatAttachmentRepository } from '@/modules/agents/repositories/age
import { AgentExecutionThreadRepository } from '@/modules/agents/repositories/agent-execution-thread.repository';
import { AgentExecutionRepository } from '@/modules/agents/repositories/agent-execution.repository';
import { AgentRepository } from '@/modules/agents/repositories/agent.repository';
import { SystemAgentRegistry } from '@/modules/agents/system-agents/system-agent-registry';
import { N8N_CHAT_PRODUCTION_SOURCE } from '@/modules/agents/utils/agent-thread-access';
import { createMember, createAdmin } from '../../shared/db/users';
@ -163,6 +164,7 @@ describe('AgentExecutionRepository', () => {
function recordingServices(
memoryBackend: ReturnType<N8nMemory['getImplementation']> = mock(),
connection?: DataSource,
systemAgents = new SystemAgentRegistry(),
) {
const txRunner = new TypeOrmTransactionRunner(
connection ?? repository.manager.connection,
@ -243,6 +245,7 @@ describe('AgentExecutionRepository', () => {
messageRepository,
steering,
settingsService,
systemAgents,
);
const chatExecutionService = mock<AgentChatExecutionService>();
chatExecutionService.settle.mockImplementation(async (_executionId, finalize) => {
@ -2627,6 +2630,88 @@ describe('AgentExecutionRepository', () => {
);
});
describe('system agents', () => {
const systemAgentId = 'test-system-agent';
async function systemServices() {
const registry = new SystemAgentRegistry();
registry.register({
agentId: systemAgentId,
name: 'Test System Agent',
authorize: async () => true,
prepareTurn: vi.fn(),
});
await agentRepo.ensureInstanceAgent(systemAgentId, 'Test System Agent');
return recordingServices(undefined, undefined, registry);
}
function systemInput(
threadId: string,
message: string,
sessionMode: 'new' | 'existing' = 'existing',
kind: 'system' | 'preview' = 'system',
): Parameters<AgentMessageQueueService['enqueue']>[0] {
return {
agentId: systemAgentId,
projectId,
threadId,
sessionMode,
source: 'system_chat',
payload: { kind, message, userId: owner.id, resourceId: `system-chat:${owner.id}` },
};
}
it('queues and claims system messages while custom agents are turned off', async () => {
const services = await systemServices();
await services.settingsService.setEnabled(false);
const threadId = uuid();
const item = await enqueue(services, systemInput(threadId, 'Hello', 'new'));
expect(await services.queue.hasQueuedMessages(threadId)).toBe(true);
expect(await threadRepo.findOneByOrFail({ id: threadId })).toMatchObject({
agentId: systemAgentId,
projectId,
accessScope: 'user',
ownerId: owner.id,
});
const claimed = await claim(services, threadId);
expect(claimed.item.id).toBe(item.id);
expect(claimed.payload).toMatchObject({ kind: 'system', message: 'Hello' });
// A system thread has its own chat surface, so its recording is not a Preview chat.
expect(claimed.recording.previewChat).toBe(false);
await finish(services, claimed);
expect(await services.queue.hasQueuedMessages(threadId)).toBe(false);
});
it('refuses a system message for an agent without a registered provider', async () => {
const services = recordingServices();
await agentRepo.ensureInstanceAgent(systemAgentId, 'Test System Agent');
await expect(services.queue.enqueue(systemInput(uuid(), 'Hello', 'new'))).rejects.toThrow(
'Agent not found',
);
});
it('refuses a Preview message for a system agent', async () => {
const services = await systemServices();
await expect(
services.queue.enqueue(systemInput(uuid(), 'Hello', 'new', 'preview')),
).rejects.toThrow('Agent not found');
});
it('refuses a system message for a project agent', async () => {
const services = await systemServices();
await expect(
services.queue.enqueue({
...systemInput(uuid(), 'Hello', 'new'),
agentId,
}),
).rejects.toThrow('Agent not found');
});
});
it('consumes a hidden turn without listing it and passes turn options through unchanged', async () => {
const services = recordingServices();
const threadId = uuid();
@ -2709,17 +2794,35 @@ describe('AgentExecutionRepository', () => {
expect((await claim(services, threadId)).item.id).toBe(reordered.id);
});
it('lets a user steer and reorder system agent input like Preview input', async () => {
const services = recordingServices();
it('lets a user steer and reorder system agent input on its own surface', async () => {
// The system kind needs a registered provider and its instance agent row.
const systemAgentId = 'test-system-agent';
const registry = new SystemAgentRegistry();
registry.register({
agentId: systemAgentId,
name: 'Test System Agent',
authorize: async () => true,
prepareTurn: vi.fn(),
});
await agentRepo.ensureInstanceAgent(systemAgentId, 'Test System Agent');
const services = recordingServices(undefined, undefined, registry);
const threadId = uuid();
const target = { projectId, agentId, threadId, userId: owner.id, kind: 'system' as const };
const target = {
projectId,
agentId: systemAgentId,
threadId,
userId: owner.id,
kind: 'system' as const,
};
const systemInput = (message: string, sessionMode: 'new' | 'existing' = 'existing') => ({
...input(threadId, message, sessionMode),
agentId: systemAgentId,
source: 'system_chat',
payload: {
kind: 'system' as const,
message,
userId: owner.id,
resourceId: `draft-chat:${owner.id}`,
resourceId: `system-chat:${owner.id}`,
},
});
await enqueue(services, systemInput('Active', 'new'));
@ -2729,10 +2832,10 @@ describe('AgentExecutionRepository', () => {
const c = await enqueue(services, systemInput('C'));
const steered = await enqueue(services, systemInput('Steer'));
expect(await services.queue.listPending({ ...target, kind: 'preview' })).toEqual({
items: [],
steerableExecutionId: active.admission.executionId,
});
// A system thread has its own chat surface, so the Preview access rule refuses it.
await expect(services.queue.listPending({ ...target, kind: 'preview' })).rejects.toThrow(
'Session not found',
);
const pending = await services.queue.listPending(target);
expect(pending.steerableExecutionId).toBe(active.admission.executionId);
expect(pending.items.map(({ id }) => id)).toEqual([b.id, c.id, steered.id]);
@ -2743,7 +2846,7 @@ describe('AgentExecutionRepository', () => {
queueId: steered.id,
executionId: active.admission.executionId,
}),
).rejects.toThrow('no longer available');
).rejects.toThrow('Session not found');
await services.queue.steer({
...target,
queueId: steered.id,
@ -2765,7 +2868,7 @@ describe('AgentExecutionRepository', () => {
{
...target,
executionId: active.admission.executionId,
resourceId: `draft-chat:${owner.id}`,
resourceId: `system-chat:${owner.id}`,
},
{ messages: [], lastCreatedAt: 0, completing: false, canContinue: true },
recorder,