From 447fc866bb0a841decfa5e8a29fba06057bfc5bc Mon Sep 17 00:00:00 2001 From: Jaakko Husso Date: Thu, 15 Oct 2026 19:25:36 +0000 Subject: [PATCH] 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 --- .../agents/agent-message-queue.service.ts | 22 ++- .../agent-message-queue.repository.ts | 5 +- .../agents/types/agent-queued-message.ts | 24 ++- .../agent-execution.repository.test.ts | 149 ++++++++++++++++++ 4 files changed, 188 insertions(+), 12 deletions(-) diff --git a/packages/cli/src/modules/agents/agent-message-queue.service.ts b/packages/cli/src/modules/agents/agent-message-queue.service.ts index 39982a40591..19225e50712 100644 --- a/packages/cli/src/modules/agents/agent-message-queue.service.ts +++ b/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; +} diff --git a/packages/cli/src/modules/agents/repositories/agent-message-queue.repository.ts b/packages/cli/src/modules/agents/repositories/agent-message-queue.repository.ts index 090030f411b..bf35dd3f6c7 100644 --- a/packages/cli/src/modules/agents/repositories/agent-message-queue.repository.ts +++ b/packages/cli/src/modules/agents/repositories/agent-message-queue.repository.ts @@ -58,10 +58,13 @@ export class AgentMessageQueueRepository extends BaseRepository { 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]) diff --git a/packages/cli/src/modules/agents/types/agent-queued-message.ts b/packages/cli/src/modules/agents/types/agent-queued-message.ts index 2c00121196c..2c0fa264976 100644 --- a/packages/cli/src/modules/agents/types/agent-queued-message.ts +++ b/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; + /** + * 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 } | (Omit< QueuedIntegrationMessage, keyof QueuedMessageInput | 'modelMessage' | 'author' | 'platformThreadId' | 'messageContext' diff --git a/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts b/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts index 02af9ba9316..9b2ab244d22 100644 --- a/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts +++ b/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'],