Merge e7440ab1f0 into ce826c2612

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

@@ -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,136 @@ 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();
+		});
+
+		/** A refused system turn fails the item, streams an error and never runs. */
+		function expectRefused(item: ReturnType<typeof systemClaim>) {
+			expect(queue.recordFailure).toHaveBeenCalledWith(
+				item,
+				expect.objectContaining({ message: 'You can no longer execute this agent' }),
+				expect.any(AbortSignal),
+			);
+			expect(chatExecutions.register).not.toHaveBeenCalled();
+			expect(systemAgentExecution.consume).not.toHaveBeenCalled();
+			expect(sender.send).toHaveBeenCalledWith(expect.objectContaining({ type: 'error' }));
+		}
+
+		it('records a failure when the provider no longer authorizes the owner', async () => {
+			registerSystemAgent(false);
+			const item = systemClaim('session');
+
+			await consumeOne(item);
+
+			expectRefused(item);
+		});
+
+		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);
+
+			expectRefused(item);
+			expect(provider.authorize).not.toHaveBeenCalled();
+		});
+
+		it('records a failure for a system message without a registered provider', async () => {
+			const item = systemClaim('session');
+
+			await consumeOne(item);
+
+			expectRefused(item);
+		});
+
+		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();
+		});
+	});
 });

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

@@ -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(),
 		);
 	});
 

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

@@ -20,6 +20,10 @@ 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 { chatSurfaceForKind, type AgentChatSurface } from './utils/agent-thread-access';
 
 @Service()
 export class AgentMessageQueueConsumer {
@@ -40,6 +44,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 {
@@ -133,11 +139,11 @@ export class AgentMessageQueueConsumer {
 			controller.signal,
 			this.executionService.getAbortSignal(admission.executionId),
 		]);
-		const sender =
-			payload.kind === 'integration' ? undefined : this.previewStreams.createSender(item.id);
+		const surface = payload.kind === 'integration' ? undefined : chatSurfaceForKind(payload.kind);
+		const sender = surface && this.previewStreams.createSender(item.id);
 		try {
-			if (sender) {
-				const user = await this.getThreadOwner(thread);
+			if (surface && sender) {
+				const user = await this.getThreadOwner(thread, surface);
 				this.chatExecutionService.register(
 					{
 						projectId: thread.projectId,
@@ -145,7 +151,7 @@ export class AgentMessageQueueConsumer {
 						threadId: thread.id,
 						userId: user.id,
 						executionId: admission.executionId,
-						surface: payload.kind === 'n8n_chat' ? 'n8n-chat' : 'preview',
+						surface,
 					},
 					controller,
 				);
@@ -156,11 +162,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 +188,35 @@ 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,
+		surface: AgentChatSurface,
+	): 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, surface))) {
 			throw new UserError('You can no longer execute this agent');
 		}
 		return user;
 	}
 
+	private async canExecute(
+		user: User,
+		thread: AgentExecutionThread,
+		surface: AgentChatSurface,
+	): Promise<boolean> {
+		if (surface !== 'system') {
+			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,

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

@@ -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,8 +34,8 @@ import {
 	type QueuedUserChatMessage,
 } from './types/agent-queued-message';
 import {
-	canContinueThreadInN8nChat,
-	canContinueThreadInPreview,
+	canContinueThreadOnSurface,
+	chatSurfaceForKind,
 	type AgentSessionMode,
 } from './utils/agent-thread-access';
 import { buildInboundUserMessage, readInboundUserMessage } from './utils/inbound-attachments';
@@ -76,6 +77,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 +92,20 @@ 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.
+		// System agents run even when the admin turns off custom agents.
+		const isSystem = payload.kind === 'system';
+		if (!isSystem) await this.settingsService.assertEnabled();
+		const agent = isSystem
+			? this.systemAgents.get(input.agentId)
+			: 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 (!isSystem) await this.settingsService.assertEnabled(ctx);
 				await this.executionService.prepareThread(
 					{
 						...input,
@@ -185,6 +192,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 +343,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))
+			!canContinueThreadOnSurface(chatSurfaceForKind(input.kind), thread, input.userId, source)
 		) {
 			throw new NotFoundError('Session not found');
 		}
@@ -353,9 +363,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);

packages/cli/src/modules/agents/agent-turn-execution.service.ts

@@ -58,6 +58,12 @@ interface ExecuteTurnConfig {
 	backgroundJobSignal?: AgentBackgroundJobSignal;
 	/** The chat surface this turn runs under. Undefined for integrations and tasks. */
 	chatSurface?: AgentChatSurface;
+	/**
+	 * The controller of the chat turn. Defaults to a new controller. A cancel
+	 * aborts it, and the caller can abort it to stop the turn at any time. The
+	 * request signal stops only a turn that is not admitted yet.
+	 */
+	chatController?: AbortController;
 	automaticPreviewContinuation?: boolean;
 	/** The abort signal is a wake lease, not a chat request. */
 	isWakeRun?: boolean;
@@ -132,7 +138,11 @@ export class AgentTurnExecutionService {
 			turn = await config.prepare();
 			const preparedTurn = turn;
 			if (config.chatSurface !== undefined) {
-				chatControl = this.createChatExecutionControl(preparedTurn, config.chatSurface);
+				chatControl = this.createChatExecutionControl(
+					preparedTurn,
+					config.chatSurface,
+					config.chatController,
+				);
 				preparedTurn.options.abortSignal = chatControl.controller.signal;
 			}
 			const stream = await this.admitTurn(preparedTurn, config, recorder, state, chatControl);
@@ -158,13 +168,13 @@ export class AgentTurnExecutionService {
 	private createChatExecutionControl(
 		turn: AgentTurnRequest,
 		surface: AgentChatSurface,
+		controller = new AbortController(),
 	): ChatExecutionControl {
 		const userId = turn.recording.access.ownerId;
 		if (turn.recording.access.accessScope !== 'user' || !userId) {
 			throw new UnexpectedError('A preview execution must have an owning user.');
 		}
 		const requestSignal = turn.options.abortSignal;
-		const controller = new AbortController();
 		const abort = () => controller.abort(requestSignal?.reason);
 		if (requestSignal?.aborted) abort();
 		else requestSignal?.addEventListener('abort', abort, { once: true });

packages/cli/src/modules/agents/system-agents/system-agent-access.ts

@@ -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);
+}

packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts

@@ -0,0 +1,665 @@
+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';
+
+type SystemQueueInput = Parameters<AgentMessageQueueService['enqueue']>[0];
+
+interface SystemMessageInput {
+	message: string;
+	attachments?: StoredAttachmentRef[];
+	options?: SystemAgentTurnOptions;
+	messageId?: string;
+	/** A machine turn: it stays out of the transcript. */
+	hidden?: boolean;
+}
+
+interface ResumeParams {
+	agentId: string;
+	user: User;
+	threadId: string;
+	resumeData: unknown;
+	runId?: string;
+	toolCallId?: string;
+	send?: (event: AgentSseEvent) => void;
+}
+
+interface ResumeHandle {
+	runId: string;
+	toolCallId: string;
+	done: Promise<void>;
+}
+
+/** The SDK run options of a handle, without its abort signal. `runTurn` links that signal. */
+function runOptionsOf(handle: SystemAgentTurnHandle) {
+	const { abortSignal: _abortSignal, ...runOptions } = handle.runOptions ?? {};
+	return runOptions;
+}
+
+/**
+ * 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 ||
+				// The access check above ran for `params.projectId`, not for the project of this thread.
+				existing.projectId !== params.projectId
+			) {
+				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');
+	}
+
+	/** Whether the thread is not idle. It uses the same rule as `getStatus`. */
+	async isBusy(thread: AgentExecutionThread): Promise<boolean> {
+		return (await this.resolveStatus(thread)).status !== 'idle';
+	}
+
+	async getStatus(thread: AgentExecutionThread): Promise<{
+		status: SystemAgentThreadStatus;
+		latestExecutionId: string | null;
+		checkpoint: SerializableAgentState | null;
+	}> {
+		const { status, checkpoint } = await this.resolveStatus(thread);
+		const latest = await this.executionRepository.findLatestByThreadId(thread.id);
+		return { status, latestExecutionId: latest?.id ?? null, checkpoint };
+	}
+
+	/**
+	 * A queued message (for example a hidden follow-up turn) counts as running,
+	 * so the thread is not idle between a turn and its follow-up. An open
+	 * suspension still wins: the queue waits for the user to answer it.
+	 */
+	private async resolveStatus(thread: AgentExecutionThread): Promise<{
+		status: SystemAgentThreadStatus;
+		checkpoint: SerializableAgentState | null;
+	}> {
+		if (await this.executionRepository.existsRunningByThread(thread.id)) {
+			return { status: 'running', checkpoint: null };
+		}
+		const checkpoint = await this.checkpointStorage.findSuspendedForThread(
+			thread.agentId,
+			thread.id,
+		);
+		if (checkpoint) return { status: 'suspended', checkpoint };
+		const queued = await this.messageQueue.hasQueuedMessages(thread.id);
+		return { status: queued ? 'running' : 'idle', checkpoint: null };
+	}
+
+	// ── 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: SystemMessageInput & { agentId: string; user: User; threadId: string },
+	) {
+		const thread = await this.getUsableThread(params.agentId, params.user, params.threadId);
+		return await this.messageQueue.enqueue(this.queueInput(thread, params.user, params));
+	}
+
+	/**
+	 * 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 controller = new AbortController();
+		const handle = await provider.prepareTurn({
+			type: 'start',
+			user,
+			thread,
+			resourceId,
+			abortSignal: controller.signal,
+			executionId: admission.executionId,
+			message: payload.message,
+			attachments,
+			options: payload.options ?? {},
+		});
+		const text = handle.input ?? payload.message;
+		this.attachFileStore(handle, thread);
+		const input =
+			attachments.length > 0 && typeof text === 'string'
+				? buildInboundUserMessage(text, attachments)
+				: text;
+		await this.runTurn({
+			thread,
+			handle,
+			send,
+			controller,
+			admission,
+			prepare: async () => ({
+				type: 'start',
+				input,
+				options: {
+					...runOptionsOf(handle),
+					persistence: { threadId: thread.id, resourceId, hostMetadata: handle.hostMetadata },
+					// The claim signal stops the turn only until the runtime admits it.
+					abortSignal: signal,
+				},
+				recording: {
+					...claim.recording,
+					...(handle.hideUserMessage ? { hideUserMessageFromTranscript: true } : {}),
+				},
+			}),
+		});
+	}
+
+	/**
+	 * Resume a suspended tool call. Runs the continuation in the background:
+	 * `done` settles when the turn ends and logs a failure instead of rejecting.
+	 */
+	async resume(params: ResumeParams): Promise<ResumeHandle> {
+		const resumed = await this.startResume(params);
+		const done = resumed.done.catch((error: unknown) => {
+			this.logger.warn('System agent resume failed', { threadId: params.threadId, error });
+		});
+		return { ...resumed, done };
+	}
+
+	/** Start a resume. `done` rejects when the turn fails. */
+	private async startResume(params: ResumeParams): Promise<ResumeHandle> {
+		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),
+		);
+		// A client run id must name the open suspension, not an earlier run.
+		if (!checkpoint || !pending || (params.runId !== undefined && params.runId !== pending.runId)) {
+			throw new UserError('This action is no longer waiting for input');
+		}
+		const { runId } = pending;
+		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,
+		});
+		this.attachFileStore(handle, thread);
+		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 done = this.runTurn({
+			thread,
+			handle,
+			send: params.send ?? (() => {}),
+			controller,
+			prepare: async () => ({
+				type: 'resume',
+				resumeData: params.resumeData,
+				options: {
+					...runOptionsOf(handle),
+					runId,
+					toolCallId: pending.toolCallId,
+					hostMetadata: handle.hostMetadata,
+				},
+				recording,
+			}),
+		});
+		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<SystemQueueInput> {
+		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 this.queueInput(thread, params.user, {
+			message: params.message,
+			messageId: params.messageId,
+			attachments,
+			options,
+		});
+	}
+
+	/** The queue input of a system message in an existing thread. */
+	private queueInput(
+		thread: AgentExecutionThread,
+		user: User,
+		input: SystemMessageInput,
+	): SystemQueueInput {
+		return {
+			agentId: thread.agentId,
+			projectId: thread.projectId,
+			threadId: thread.id,
+			sessionMode: 'existing',
+			source: SYSTEM_CHAT_SOURCE,
+			payload: {
+				kind: 'system',
+				userId: user.id,
+				message: input.message,
+				resourceId: this.resourceIdFor(user),
+				...(input.attachments?.length ? { attachments: input.attachments } : {}),
+				...(input.messageId ? { messageId: input.messageId } : {}),
+				...(input.options ? { options: input.options } : {}),
+				...(input.hidden ? { hidden: true } : {}),
+			},
+		};
+	}
+
+	/** 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');
+		// A failure rejects, so the route can send an error event.
+		const { done } = await this.startResume({
+			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. It checks thread
+	 * ownership only, not the floor: a user who lost access can still stop a turn.
+	 */
+	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',
+		});
+	}
+
+	/**
+	 * Attach the attachment store on every turn. Memory keeps file references,
+	 * and without the store a later turn cannot load the files of earlier turns.
+	 */
+	private attachFileStore(handle: SystemAgentTurnHandle, thread: AgentExecutionThread): void {
+		handle.agent.fileStore(
+			this.attachmentService.getFileStore(
+				{ agentId: thread.agentId, projectId: thread.projectId },
+				handle.agent.snapshot.model.provider ?? '',
+			),
+		);
+	}
+
+	/**
+	 * Run a prepared turn under `controller`, the controller the provider got
+	 * the signal of. A cancel aborts it, and the abort signal in the run
+	 * options of the handle stops the turn at any time.
+	 */
+	private async runTurn({
+		thread,
+		handle,
+		send,
+		controller,
+		admission,
+		prepare,
+	}: {
+		thread: AgentExecutionThread;
+		handle: SystemAgentTurnHandle;
+		send: (event: AgentSseEvent) => void;
+		controller: AbortController;
+		/** Set when the queue admitted the turn. A resume admits its own turn. */
+		admission?: AgentExecutionAdmission;
+		prepare: () => Promise<AgentTurnRequest>;
+	}): Promise<void> {
+		let status: SystemAgentTurnStatus = 'completed';
+		let executionId = admission?.executionId;
+		let error: unknown;
+		const providerSignal = handle.runOptions?.abortSignal;
+		const stopTurn = () => controller.abort(providerSignal?.reason);
+		if (providerSignal?.aborted) stopTurn();
+		else providerSignal?.addEventListener('abort', stopTurn, { once: true });
+		try {
+			const stream = this.turnExecutionService.execute({
+				admittedExecution: admission,
+				agentInstance: handle.agent,
+				toolRegistry: handle.toolRegistry ?? new Map(),
+				mcpServerAttributions: new Map(),
+				context: { projectId: thread.projectId, agentId: thread.agentId, threadId: thread.id },
+				chatSurface: 'system',
+				chatController: controller,
+				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 = controller.signal.aborted ? 'cancelled' : 'errored';
+			throw caught;
+		} finally {
+			providerSignal?.removeEventListener('abort', stopTurn);
+			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,
+				});
+			}
+			// Like Preview, only a completed turn sends `done`. An error has its own event.
+			if (status === 'completed') {
+				send({ type: 'done', sessionId: thread.id, executionId: executionId ?? '' });
+			}
+		}
+	}
+}

packages/cli/src/modules/agents/system-agents/system-agent-registry.ts

@@ -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);
+	}
+}

packages/cli/src/modules/agents/system-agents/system-agent.types.ts

@@ -0,0 +1,98 @@
+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;
+	/** Aborts when the turn stops: a user cancel, the queue, or `runOptions.abortSignal`. */
+	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. `abortSignal` stops the turn at
+	 * any time. The host sets `persistence`, so a provider cannot change it.
+	 */
+	runOptions?: Omit<Partial<RunOptions & ExecutionOptions> & Partial<ResumeOptions>, 'persistence'>;
+	/** 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;
+}

packages/cli/src/modules/agents/utils/__tests__/agent-thread-access.test.ts

@@ -0,0 +1,46 @@
+import type { AgentExecutionThread } from '../../entities/agent-execution-thread.entity';
+import {
+	canContinueThreadOnSurface,
+	chatSurfaceForKind,
+	N8N_CHAT_PRODUCTION_SOURCE,
+	SYSTEM_CHAT_SOURCE,
+} from '../agent-thread-access';
+
+const thread = {
+	accessScope: 'user',
+	ownerId: 'user-1',
+	parentThreadId: null,
+	taskId: null,
+} as AgentExecutionThread;
+
+describe('chatSurfaceForKind', () => {
+	it.each([
+		['preview', 'preview'],
+		['n8n_chat', 'n8n-chat'],
+		['system', 'system'],
+	] as const)('maps the %s kind to the %s surface', (kind, surface) => {
+		expect(chatSurfaceForKind(kind)).toBe(surface);
+	});
+});
+
+describe('canContinueThreadOnSurface', () => {
+	it.each([
+		['preview', 'chat'],
+		['n8n-chat', N8N_CHAT_PRODUCTION_SOURCE],
+		['system', SYSTEM_CHAT_SOURCE],
+	] as const)('lets the owner continue a %s thread with its own source', (surface, source) => {
+		expect(canContinueThreadOnSurface(surface, thread, 'user-1', source)).toBe(true);
+	});
+
+	it.each([
+		['preview', SYSTEM_CHAT_SOURCE],
+		['n8n-chat', SYSTEM_CHAT_SOURCE],
+		['system', 'chat'],
+	] as const)('refuses a %s thread with a %s source', (surface, source) => {
+		expect(canContinueThreadOnSurface(surface, thread, 'user-1', source)).toBe(false);
+	});
+
+	it('refuses a user who does not own the thread', () => {
+		expect(canContinueThreadOnSurface('system', thread, 'user-2', SYSTEM_CHAT_SOURCE)).toBe(false);
+	});
+});

packages/cli/src/modules/agents/utils/agent-thread-access.ts

@@ -1,4 +1,5 @@
 import type { AgentExecutionThread } from '../entities/agent-execution-thread.entity';
+import type { QueuedUserChatMessage } from '../types/agent-queued-message';
 import {
 	draftChatMemoryResourceId,
 	productionChatMemoryResourceId,
@@ -18,6 +19,18 @@ export const SYSTEM_CHAT_SOURCE = 'system_chat' as const;
  */
 export type AgentChatSurface = 'preview' | 'n8n-chat' | 'system';
 
+/** The chat surface that a queued user chat message runs on. */
+export function chatSurfaceForKind(kind: QueuedUserChatMessage['kind']): AgentChatSurface {
+	switch (kind) {
+		case 'n8n_chat':
+			return 'n8n-chat';
+		case 'system':
+			return 'system';
+		case 'preview':
+			return 'preview';
+	}
+}
+
 /** The memory resource ID that scopes a thread to its owning user on the given chat surface. */
 export function chatSurfaceMemoryResourceId(surface: AgentChatSurface, userId: string): string {
 	switch (surface) {
@@ -82,6 +95,23 @@ export function canContinueSystemThread(
 	);
 }
 
+/** Whether the user can continue the thread with a message on the chat surface. */
+export function canContinueThreadOnSurface(
+	surface: AgentChatSurface,
+	thread: AgentExecutionThread,
+	userId: string,
+	source?: string | null,
+): boolean {
+	switch (surface) {
+		case 'n8n-chat':
+			return canContinueThreadInN8nChat(thread, userId, source);
+		case 'system':
+			return canContinueSystemThread(thread, userId, source);
+		case 'preview':
+			return canContinueThreadInPreview(thread, userId, source);
+	}
+}
+
 export function canUseTopLevelDraftThread(thread: AgentExecutionThread, userId: string): boolean {
 	return (
 		thread.accessScope === 'user' && thread.ownerId === userId && thread.parentThreadId === null

packages/cli/test/integration/agents/system-agent-turn-abort.test.ts

@@ -0,0 +1,183 @@
+import { Agent as RuntimeAgent, Tool } from '@n8n/agents';
+import { testDb, testModules } from '@n8n/backend-test-utils';
+import { ProjectRepository, type User } from '@n8n/db';
+import { Container } from '@n8n/di';
+import { convertArrayToReadableStream, MockLanguageModelV3 } from 'ai/test';
+import { z } from 'zod';
+
+import { AgentMessageQueueConsumer } from '@/modules/agents/agent-message-queue-consumer.service';
+import { N8NCheckpointStorage } from '@/modules/agents/integrations/n8n-checkpoint-storage';
+import { N8nMemory } from '@/modules/agents/integrations/n8n-memory';
+import { AgentExecutionRepository } from '@/modules/agents/repositories/agent-execution.repository';
+import { SystemAgentExecutionService } from '@/modules/agents/system-agents/system-agent-execution.service';
+import type {
+	SystemAgentProvider,
+	SystemAgentTurn,
+	SystemAgentTurnOutcome,
+} from '@/modules/agents/system-agents/system-agent.types';
+
+import { createOwner } from '../shared/db/users';
+
+type ModelStreamPart = Awaited<
+	ReturnType<MockLanguageModelV3['doStream']>
+>['stream'] extends ReadableStream<infer Part>
+	? Part
+	: never;
+
+const AGENT_ID = 'test-system-agent';
+const usage = {
+	inputTokens: { total: 1, noCache: 1, cacheRead: 0, cacheWrite: 0 },
+	outputTokens: { total: 1, text: 1, reasoning: 0 },
+};
+
+/** The first model call asks for the `wait` tool. The next call answers. */
+function scriptedModel() {
+	let calls = 0;
+	return new MockLanguageModelV3({
+		provider: 'mock',
+		modelId: 'system-turn',
+		doStream: async () => {
+			const first = calls++ === 0;
+			return {
+				stream: convertArrayToReadableStream<ModelStreamPart>([
+					{ type: 'stream-start', warnings: [] },
+					...(first
+						? [{ type: 'tool-call' as const, toolCallId: 'wait-1', toolName: 'wait', input: '{}' }]
+						: [
+								{ type: 'text-start' as const, id: 'answer' },
+								{ type: 'text-delta' as const, id: 'answer', delta: 'Done' },
+								{ type: 'text-end' as const, id: 'answer' },
+							]),
+					{
+						type: 'finish',
+						finishReason: { unified: first ? 'tool-calls' : 'stop', raw: undefined },
+						usage,
+					},
+				]),
+			};
+		},
+	});
+}
+
+async function until(check: () => boolean, timeoutMs = 10_000) {
+	const start = Date.now();
+	while (!check()) {
+		if (Date.now() - start > timeoutMs) throw new Error('Timed out');
+		await new Promise((resolve) => setTimeout(resolve, 20));
+	}
+}
+
+interface TurnProbe {
+	turnSignal?: AbortSignal;
+	providerAbort?: AbortController;
+	toolStarted?: boolean;
+	toolEnd?: 'aborted' | 'timeout';
+	outcome?: SystemAgentTurnOutcome;
+}
+
+describe('System agent turn abort', () => {
+	let probe: TurnProbe;
+	let service: SystemAgentExecutionService;
+	let owner: User;
+	let projectId: string;
+
+	const provider: SystemAgentProvider = {
+		agentId: AGENT_ID,
+		name: 'Test System Agent',
+		authorize: async () => true,
+		async prepareTurn(turn: SystemAgentTurn) {
+			probe.turnSignal = turn.abortSignal;
+			probe.providerAbort = new AbortController();
+			// The tool waits for the run signal, so the turn stays active until it aborts.
+			const wait = new Tool('wait')
+				.description('Wait')
+				.input(z.object({}))
+				.handler(async (_input, ctx) => {
+					probe.toolStarted = true;
+					probe.toolEnd = await new Promise((resolve) => {
+						const timer = setTimeout(() => resolve('timeout'), 5_000);
+						ctx.abortSignal?.addEventListener('abort', () => {
+							clearTimeout(timer);
+							resolve('aborted');
+						});
+					});
+					return 'waited';
+				});
+			const agent = new RuntimeAgent('Test System Agent')
+				.instructions('Use the wait tool.')
+				.model(scriptedModel())
+				.tool(wait)
+				.memory(Container.get(N8nMemory).getImplementation(AGENT_ID))
+				.checkpoint(Container.get(N8NCheckpointStorage).getStorage(AGENT_ID));
+			return {
+				agent,
+				runOptions: { abortSignal: probe.providerAbort.signal },
+				onSettled: async (outcome) => {
+					probe.outcome = outcome;
+				},
+			};
+		},
+	};
+
+	beforeAll(async () => {
+		await testModules.loadModules(['agents']);
+		await testDb.init();
+		service = Container.get(SystemAgentExecutionService);
+		await service.register(provider);
+		Container.get(AgentMessageQueueConsumer).start();
+		owner = await createOwner();
+		projectId = (await Container.get(ProjectRepository).getPersonalProjectForUserOrFail(owner.id))
+			.id;
+	});
+
+	afterAll(async () => {
+		Container.get(AgentMessageQueueConsumer).stop();
+		await testDb.terminate();
+	});
+
+	beforeEach(() => {
+		probe = {};
+	});
+
+	/** Queue a message and wait until the turn runs the `wait` tool. */
+	async function startTurn() {
+		const thread = await service.createThread({ agentId: AGENT_ID, user: owner, projectId });
+		await service.sendMessage({
+			agentId: AGENT_ID,
+			user: owner,
+			threadId: thread.id,
+			message: 'Go',
+		});
+		await until(() => probe.toolStarted === true);
+		return thread;
+	}
+
+	async function latestStatus(threadId: string) {
+		return (await Container.get(AgentExecutionRepository).findLatestByThreadId(threadId))?.status;
+	}
+
+	it('aborts the turn signal of the provider when the user cancels the turn', async () => {
+		const thread = await startTurn();
+		expect(probe.turnSignal?.aborted).toBe(false);
+
+		expect(await service.cancel(AGENT_ID, owner, thread.id)).toBe(true);
+		await until(() => probe.outcome !== undefined);
+
+		expect(probe.turnSignal?.aborted).toBe(true);
+		expect(probe.toolEnd).toBe('aborted');
+		expect(probe.outcome?.status).toBe('cancelled');
+		expect(await latestStatus(thread.id)).toBe('cancelled');
+	});
+
+	it('stops an admitted turn when the provider aborts its run signal', async () => {
+		const thread = await startTurn();
+
+		probe.providerAbort?.abort();
+		await until(() => probe.outcome !== undefined);
+
+		expect(probe.toolEnd).toBe('aborted');
+		expect(probe.turnSignal?.aborted).toBe(true);
+		expect(probe.outcome?.status).toBe('cancelled');
+		expect(await latestStatus(thread.id)).toBe('cancelled');
+	});
+});

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

@@ -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,98 @@ 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 system message without a provider while custom agents are turned off', async () => {
+				const services = recordingServices();
+				await services.settingsService.setEnabled(false);
+				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 +2804,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 +2842,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 +2856,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 +2878,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,