feat(core): Add hidden turns, turn options and a system kind to the Agents queue (no-changelog) (#40614)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

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

@@ -172,7 +172,11 @@ export class AgentMessageQueueService {
 				platformThreadId,
 			};
 		} else {
-			queueDispatch = { kind: dispatch.kind };
+			queueDispatch = {
+				kind: dispatch.kind,
+				...(dispatch.options ? { options: dispatch.options } : {}),
+			};
+			if (dispatch.hidden) origin = { source, hidden: true };
 		}
 		const input = await this.messages.createInput(
 			{ id: messageId, threadId, resourceId, content, modelContent, author, origin },
@@ -192,7 +196,8 @@ export class AgentMessageQueueService {
 		return {
 			steerableExecutionId: steerable?.id ?? null,
 			items: items
-				.filter((item) => item.payload.kind === kind)
+				// Hidden machine turns are not user messages.
+				.filter((item) => item.payload.kind === kind && !isHidden(item))
 				// Show accepted steers first. Keep future turns in their saved queue order.
 				.sort(
 					(a, b) =>
@@ -214,7 +219,7 @@ export class AgentMessageQueueService {
 			if (!thread) throw new NotFoundError('Session not found');
 			await this.assertUserChatAccess(thread, input, ctx);
 			const item = await this.repository.findItem(thread.id, input.queueId, ctx);
-			if (!item || item.payload.kind !== input.kind)
+			if (!item || item.payload.kind !== input.kind || isHidden(item))
 				throw new NotFoundError('Queued message not found');
 			if (
 				item.executionId !== null ||
@@ -241,7 +246,7 @@ export class AgentMessageQueueService {
 			if (!thread) throw new NotFoundError('Session not found');
 			await this.assertUserChatAccess(thread, input, ctx);
 			const item = await this.repository.findItem(thread.id, input.queueId, ctx);
-			if (!item || item.payload.kind !== input.kind)
+			if (!item || item.payload.kind !== input.kind || isHidden(item))
 				throw new NotFoundError('Queued message not found');
 			if (item.executionId !== null) throw new ConflictError('This message has already started');
 			if (item.steeringExecutionId !== null)
@@ -299,7 +304,7 @@ export class AgentMessageQueueService {
 			if (!thread) throw new NotFoundError('Session not found');
 			await this.assertUserChatAccess(thread, input, ctx);
 			const item = await this.repository.findItem(thread.id, input.queueId, ctx);
-			if (!item || item.payload.kind !== input.kind)
+			if (!item || item.payload.kind !== input.kind || isHidden(item))
 				throw new ConflictError('This message is no longer available');
 			const execution = await this.steering.findEligible(thread, ctx);
 			if (
@@ -462,7 +467,8 @@ export class AgentMessageQueueService {
 			...readInboundUserMessage(item.message.content),
 			resourceId: item.message.resourceId,
 		};
-		if (item.payload.kind !== 'integration') return { ...item.payload, ...input };
+		if (item.payload.kind !== 'integration')
+			return { ...item.payload, ...input, ...(isHidden(item) ? { hidden: true } : {}) };
 		if (!item.message.author) throw new UnexpectedError('Queued integration input has no author');
 		const { origin } = item.message;
 		if (!origin?.source || !origin.integrationConnectionId || !origin.platformThreadId)
@@ -483,3 +489,7 @@ export class AgentMessageQueueService {
 		};
 	}
 }
+
+function isHidden(item: AgentMessageQueue): boolean {
+	return item.message.origin?.hidden === true;
+}

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

@@ -58,10 +58,13 @@ export class AgentMessageQueueRepository extends BaseRepository<AgentMessageQueu
 		ctx: OperationContext,
 	): Promise<boolean> {
 		const manager = this.managerFor(ctx);
-		const items = await manager.find(AgentMessageQueue, {
+		const pending = await manager.find(AgentMessageQueue, {
 			where: { threadId, executionId: IsNull(), steeringExecutionId: IsNull() },
+			relations: { message: true },
 			order: { position: 'ASC', id: 'ASC' },
 		});
+		// Clients do not see hidden machine turns, so they keep their positions.
+		const items = pending.filter((entry) => entry.message.origin?.hidden !== true);
 		if (
 			items.length !== expectedIds.length ||
 			items.some((item, index) => item.id !== expectedIds[index])

packages/cli/src/modules/agents/types/agent-queued-message.ts

@@ -15,16 +15,29 @@ interface QueuedMessageInput {
 	attachments?: StoredAttachmentRef[];
 }
 
-/** A message from an n8n user. Preview runs the draft agent. n8n Chat runs the published agent. */
+/**
+ * A message from an n8n user. Preview runs the draft agent. n8n Chat runs the
+ * published agent. System runs a code-defined system agent.
+ */
 export interface QueuedUserChatMessage extends QueuedMessageInput {
-	kind: 'preview' | 'n8n_chat';
+	kind: 'preview' | 'n8n_chat' | 'system';
 	userId: string;
 	messageId?: string;
+	/** Turn options for a system agent. The queue keeps them opaque; the provider defines their shape. */
+	options?: Record<string, unknown>;
+	/**
+	 * A machine turn. The message is model input only: it is stored with
+	 * `origin.hidden`, stays out of the transcript and the pending list.
+	 */
+	hidden?: boolean;
 }
 
-/** One rule for steering eligibility: the allow-list of kinds. Integrations never steer. */
+/**
+ * One rule for steering eligibility: the allow-list of kinds. Integrations never steer.
+ * A running hidden `system` turn accepts steering too: the user sees its output.
+ */
 export const acceptsSteering = (kind: AgentQueuedMessage['kind']): boolean =>
-	kind === 'preview' || kind === 'n8n_chat';
+	kind === 'preview' || kind === 'n8n_chat' || kind === 'system';
 
 export interface QueuedIntegrationMessage extends QueuedMessageInput {
 	kind: 'integration';
@@ -43,7 +56,8 @@ export type AgentQueuedMessage = QueuedUserChatMessage | QueuedIntegrationMessag
 
 /** Queue storage keeps dispatch data. Conversation input belongs to the referenced message. */
 export type AgentQueueDispatch =
-	| { kind: QueuedUserChatMessage['kind'] }
+	// `hidden` is stored on the message origin, not in the dispatch.
+	| { kind: QueuedUserChatMessage['kind']; options?: Record<string, unknown> }
 	| (Omit<
 			QueuedIntegrationMessage,
 			keyof QueuedMessageInput | 'modelMessage' | 'author' | 'platformThreadId' | 'messageContext'

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

@@ -2627,6 +2627,155 @@ describe('AgentExecutionRepository', () => {
 			);
 		});
 
+		it('consumes a hidden turn without listing it and passes turn options through unchanged', async () => {
+			const services = recordingServices();
+			const threadId = uuid();
+			const target = { projectId, agentId, threadId, userId: owner.id, kind: 'preview' as const };
+			await enqueue(services, input(threadId, 'Visible', 'new'));
+			const active = await claim(services, threadId);
+			const options = { mode: 'plan', nested: { ids: ['a', 'b'] } };
+			const hiddenInput = {
+				...input(threadId, 'Machine follow-up'),
+				payload: {
+					kind: 'preview' as const,
+					message: 'Machine follow-up',
+					userId: owner.id,
+					resourceId: `draft-chat:${owner.id}`,
+					hidden: true,
+					options,
+				},
+			};
+			const hidden = await enqueue(services, hiddenInput);
+			const pending = await enqueue(services, input(threadId, 'Pending'));
+
+			expect((await services.queue.listPending(target)).items.map(({ id }) => id)).toEqual([
+				pending.id,
+			]);
+			expect(await services.queueRepository.findOneByOrFail({ id: hidden.id })).toMatchObject({
+				payload: { kind: 'preview', options },
+			});
+			expect(
+				await services.messageRepository.findOneByOrFail({ id: hidden.messageId }),
+			).toMatchObject({ origin: { source: 'chat', hidden: true } });
+			await expect(services.queue.removePending({ ...target, queueId: hidden.id })).rejects.toThrow(
+				'Queued message not found',
+			);
+			await expect(
+				services.queue.updatePending({ ...target, queueId: hidden.id, message: 'Edited' }),
+			).rejects.toThrow('Queued message not found');
+			await expect(
+				services.queue.steer({
+					...target,
+					queueId: hidden.id,
+					executionId: active.admission.executionId,
+				}),
+			).rejects.toThrow('no longer available');
+			// The client only knows the visible items. Reordering them leaves the hidden turn in place.
+			const reordered = await enqueue(services, input(threadId, 'Reordered'));
+			await services.queue.reorderPending({
+				...target,
+				queueId: reordered.id,
+				targetQueueId: pending.id,
+				expectedQueueIds: [pending.id, reordered.id],
+			});
+			expect((await services.queue.listPending(target)).items.map(({ id }) => id)).toEqual([
+				reordered.id,
+				pending.id,
+			]);
+
+			await finish(services, active);
+			const consumed = await claim(services, threadId);
+			expect(consumed.item.id).toBe(hidden.id);
+			expect(consumed.payload).toEqual({
+				kind: 'preview',
+				options,
+				hidden: true,
+				message: 'Machine follow-up',
+				resourceId: hiddenInput.payload.resourceId,
+				attachments: [],
+			});
+			await finish(services, consumed);
+			const detail = await services.executionService.getThreadDetail(
+				threadId,
+				projectId,
+				agentId,
+				owner.id,
+			);
+			expect(
+				executionsToMessagesDto(detail!.executions)
+					.filter(({ role }) => role === 'user')
+					.map(({ content }) => content),
+			).toEqual([[{ type: 'text', text: 'Visible' }]]);
+			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();
+			const threadId = uuid();
+			const target = { projectId, agentId, threadId, userId: owner.id, kind: 'system' as const };
+			const systemInput = (message: string, sessionMode: 'new' | 'existing' = 'existing') => ({
+				...input(threadId, message, sessionMode),
+				payload: {
+					kind: 'system' as const,
+					message,
+					userId: owner.id,
+					resourceId: `draft-chat:${owner.id}`,
+				},
+			});
+			await enqueue(services, systemInput('Active', 'new'));
+			const active = await claim(services, threadId);
+			expect(active.recording.previewChat).toBe(false);
+			const b = await enqueue(services, systemInput('B'));
+			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,
+			});
+			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]);
+			await expect(
+				services.queue.steer({
+					...target,
+					kind: 'preview',
+					queueId: steered.id,
+					executionId: active.admission.executionId,
+				}),
+			).rejects.toThrow('no longer available');
+			await services.queue.steer({
+				...target,
+				queueId: steered.id,
+				executionId: active.admission.executionId,
+			});
+			await services.queue.reorderPending({
+				...target,
+				queueId: c.id,
+				targetQueueId: b.id,
+				expectedQueueIds: [b.id, c.id],
+			});
+			expect((await services.queue.listPending(target)).items.map(({ id }) => id)).toEqual([
+				steered.id,
+				c.id,
+				b.id,
+			]);
+			const recorder = new ExecutionRecorder();
+			const consumed = await services.steering.consume(
+				{
+					...target,
+					executionId: active.admission.executionId,
+					resourceId: `draft-chat:${owner.id}`,
+				},
+				{ messages: [], lastCreatedAt: 0, completing: false, canContinue: true },
+				recorder,
+				new AbortController().signal,
+			);
+			expect(consumed.messages.map(({ id }) => id)).toEqual([steered.messageId]);
+			await finish(services, active, 'stop', recorder);
+			expect((await claim(services, threadId)).item.id).toBe(c.id);
+		});
+
 		it.each([
 			['claim', 'remove'],
 			['remove', 'remove'],