diff --git a/packages/@n8n/api-types/src/agent-sse.ts b/packages/@n8n/api-types/src/agent-sse.ts index c0a3cdc6bc5..cce9c69affd 100644 --- a/packages/@n8n/api-types/src/agent-sse.ts +++ b/packages/@n8n/api-types/src/agent-sse.ts @@ -73,7 +73,8 @@ export type ForwardedChildChunkWire = }; export type AgentSseEvent = - | { type: 'execution-started'; executionId: string; sessionId: string } + | { type: 'message-queued'; queueId: string; sessionId: string } + | { type: 'execution-started'; executionId: string; sessionId: string; message?: string } | { type: 'start-step' } | { type: 'finish-step' } | { type: 'text-start'; id: string } diff --git a/packages/@n8n/api-types/src/agents/dto.ts b/packages/@n8n/api-types/src/agents/dto.ts index 6c557dfcba0..32a050bd6c3 100644 --- a/packages/@n8n/api-types/src/agents/dto.ts +++ b/packages/@n8n/api-types/src/agents/dto.ts @@ -260,6 +260,10 @@ export class AgentChatMessageDto extends Z.class(agentChatMessageShape) { } } +export class AgentChatQueueUpdateDto extends Z.class({ + message: z.string(), +}) {} + export class AgentChatResumeDto extends Z.class({ runId: z.string().min(1), toolCallId: z.string().min(1), diff --git a/packages/@n8n/api-types/src/agents/types.ts b/packages/@n8n/api-types/src/agents/types.ts index 4ac8e3f0e1f..3353563a078 100644 --- a/packages/@n8n/api-types/src/agents/types.ts +++ b/packages/@n8n/api-types/src/agents/types.ts @@ -380,6 +380,17 @@ export interface AgentBuilderOpenSuspension { suspendPayload?: unknown; } +export interface AgentChatQueueItem { + id: string; + message: string; + attachments?: Array<{ id: string; fileName: string; mimeType: string; sizeBytes: number }>; + createdAt: string; +} + +export interface AgentChatQueueResponse { + items: AgentChatQueueItem[]; +} + /** Chat history envelope returned by the agent chat messages endpoints. */ export interface AgentChatMessagesResponse { messages: AgentPersistedMessageDto[]; diff --git a/packages/@n8n/api-types/src/push/agents.ts b/packages/@n8n/api-types/src/push/agents.ts index 8c213412052..d2ad92acb89 100644 --- a/packages/@n8n/api-types/src/push/agents.ts +++ b/packages/@n8n/api-types/src/push/agents.ts @@ -29,4 +29,13 @@ export type AgentUpdated = { }; }; -export type AgentsPushMessage = AgentExecutionUpdated | AgentUpdated | AgentBackgroundJobsUpdated; +export type AgentMessageQueueUpdated = { + type: 'agentMessageQueueUpdated'; + data: { projectId: string; agentId: string; threadId: string }; +}; + +export type AgentsPushMessage = + | AgentExecutionUpdated + | AgentUpdated + | AgentBackgroundJobsUpdated + | AgentMessageQueueUpdated; diff --git a/packages/@n8n/decorators/src/pubsub/pubsub-metadata.ts b/packages/@n8n/decorators/src/pubsub/pubsub-metadata.ts index 93ae7a026b2..1c538eb2bfd 100644 --- a/packages/@n8n/decorators/src/pubsub/pubsub-metadata.ts +++ b/packages/@n8n/decorators/src/pubsub/pubsub-metadata.ts @@ -28,6 +28,7 @@ export type PubSubEventName = | 'relay-agent-execution-update' | 'relay-agent-queued-chat' | 'cancel-agent-chat-execution' + | 'relay-agent-message-queue-update' | 'relay-agent-background-tasks-update' | 'relay-agent-update' | 'resume-agent-workflow-tool' diff --git a/packages/cli/src/modules/agents/PREVIEW_CHAT_RECOVERY.md b/packages/cli/src/modules/agents/PREVIEW_CHAT_RECOVERY.md index 892a94efa57..d165a3fbc70 100644 --- a/packages/cli/src/modules/agents/PREVIEW_CHAT_RECOVERY.md +++ b/packages/cli/src/modules/agents/PREVIEW_CHAT_RECOVERY.md @@ -1,15 +1,19 @@ # Preview chat recovery -An accepted preview turn continues when its browser connection closes. This applies to new messages and resumed interactions. The server records the execution and registers its cancellation handle before it sends `execution-started`. The event contains `executionId` and `sessionId`. +An accepted preview turn continues when its browser connection closes. This applies to new messages and resumed interactions. For a new message, the server commits the session and queue item before it sends `message-queued`. This event contains `queueId` and `sessionId`. The composer clears the submitted draft on acceptance. When the consumer claims the item, it records the execution and registers its cancellation handle before it sends `execution-started`. The event contains `executionId` and `sessionId`. The original connection receives live SSE events. After a reload or disconnect, the client reads the recorded transcript. It follows later `agentExecutionUpdated` notifications with fresh history reads. Each successful read replaces the recovered transcript. A failed read keeps the last known state. The client does not submit the message or resume request again. `GET /projects/:projectId/agents/v2/:agentId/chat/:threadId/messages` returns `messages`, `openSuspensions`, and `activeExecutionId`. The active ID is present even before the first output. A `null` ID means that the recorded thread has no running execution. -All tabs of the owning user can follow the execution and request Stop. Other users cannot read or cancel a private preview turn. New-message submission stays blocked while a turn runs. The draft remains editable. A competing request receives the `turn_already_running` error and keeps its draft. +All tabs of the owning user can follow the execution and request Stop. Other users cannot read or cancel a private preview turn. Users can submit messages while a turn runs or waits for approval. Pending messages appear only in the queue panel. Each message enters the conversation when its execution starts. Typed answers to question cards still resume the suspended interaction. Waiting SSE requests do not block history recovery or approval actions. -Stop uses `DELETE /projects/:projectId/agents/v2/:agentId/chat/:threadId/executions/:executionId`. It requires `agent:execute` and the preview owner. The server cancels the local execution or relays the request to the main that owns it. The response `{ cancelRequested }` reports the request. Recorded state determines completion. A late request cannot cancel a later execution. Suspended turns use the existing checkpoint cancellation path. +Stop uses `DELETE /projects/:projectId/agents/v2/:agentId/chat/:threadId/executions/:executionId`. It requires `agent:execute` and the preview owner. The server cancels the local execution or relays the request to the main that owns it. The response `{ cancelRequested }` reports the request. Recorded state determines completion. A late request cannot cancel a later execution. Suspended turns use the existing checkpoint cancellation path. Stop leaves pending messages in place. The next message starts after cancellation settles. Recovery guarantees the latest retained timeline, followed by newer snapshots. Updates can arrive in batches. Snapshots do not contain original text-stream IDs or every transient live event. Child text and reasoning share the existing 4,000-character limit for each delegation. Text beyond that limit is not part of recovery. -This contract does not include token replay, queued messages, execution fencing, or survival of backend process failure. The existing Agents enablement controls the feature. +`GET /projects/:projectId/agents/v2/:agentId/chat/:threadId/queue` restores pending messages in FIFO order. It requires `agent:read` and the Preview session owner. `DELETE` on the same path with `/:queueId` requires `agent:execute`. Removal locks the session and rejects an item that has started. Successful removal discards its attachments. Queue changes use `agentMessageQueueUpdated` notifications across mains. Execution updates also refresh the queue. Neither notification replays stream events. + +Accepted input survives backend process failure through the durable queue. Interrupted executions are not replayed. Valid approval checkpoints retain their place. The existing Agents enablement controls the feature. This contract does not include token replay, sender deduplication, or queue reordering. + +Users can edit the text of a pending message. Edits retain its attachments and queue position. The update uses the session lock and fails if the consumer has claimed the item. Queue notifications refresh the edited text across mains. The execution-start event carries the accepted text so waiting streams do not show the original draft. If a message starts during editing, the editor keeps the unsaved text available to copy until the user dismisses it. diff --git a/packages/cli/src/modules/agents/__tests__/agent-chat.controller.test.ts b/packages/cli/src/modules/agents/__tests__/agent-chat.controller.test.ts index 86e555755cf..d3271379d5a 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-chat.controller.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-chat.controller.test.ts @@ -158,6 +158,9 @@ describe('AgentChatController route access scopes', () => { ['cancelChatRun', 'agent:execute'], ['cancelChatExecution', 'agent:execute'], ['getChatMessages', 'agent:read'], + ['getQueuedMessages', 'agent:read'], + ['removeQueuedMessage', 'agent:execute'], + ['updateQueuedMessage', 'agent:execute'], ['getBackgroundJobs', 'agent:read'], ['getTestChatMessages', 'agent:read'], ['clearTestChatMessages', 'agent:update'], @@ -166,6 +169,31 @@ describe('AgentChatController route access scopes', () => { }); }); +describe('AgentChatController queue editing', () => { + it('reads the message after the request and response arguments supplied by the registry', async () => { + const { controller, agentsService, messageQueue } = makeController(); + agentsService.findById.mockResolvedValue({ id: 'agent-1' } as never); + const params = { + projectId: 'project-1', + agentId: 'agent-1', + threadId: 'thread-1', + queueId: '1', + }; + + await Reflect.apply(controller.updateQueuedMessage, controller, [ + { params, user: { id: 'user-1' } }, + makeSseResponse([]), + { message: 'Edited message' }, + ]); + + expect(messageQueue.updatePending).toHaveBeenCalledWith({ + ...params, + userId: 'user-1', + message: 'Edited message', + }); + }); +}); + describe('AgentChatController background tasks', () => { const thread = mock({ id: 'thread-1', @@ -737,8 +765,14 @@ describe('AgentChatController attachment cleanup on failed turns', () => { expect(agentChatAttachmentService.deleteByIds).not.toHaveBeenCalled(); }); - it('relays existing SSE events from the main that claims the message', async () => { + it('acknowledges committed acceptance and relays events from the claiming main', async () => { const { controller, messageQueue, previewStreams } = makeController(); + const commit = createDeferredPromise(); + messageQueue.enqueue.mockImplementationOnce(async (_input, onInserted) => { + onInserted?.('queue-1'); + await commit.promise; + return mock({ id: 'queue-1' }); + }); const { res, events } = makeCleanupSseResponse(); const request = controller.chat( { params: { projectId: 'project-1' }, user: { id: 'user-1' } } as never, @@ -747,6 +781,15 @@ describe('AgentChatController attachment cleanup on failed turns', () => { { message: 'hi', sessionId: 'thread-1', newSession: true } as never, ); await vi.waitFor(() => expect(messageQueue.enqueue).toHaveBeenCalled()); + expect(events()).toEqual([]); + commit.resolve(); + await vi.waitFor(() => + expect(events()).toContainEqual({ + type: 'message-queued', + queueId: 'queue-1', + sessionId: 'thread-1', + }), + ); const started = { type: 'execution-started' as const, executionId: 'exec-1', @@ -758,7 +801,11 @@ describe('AgentChatController attachment cleanup on failed turns', () => { previewStreams.handleRelay({ queueId: 'queue-1', sequence: 2, event: done }); previewStreams.handleRelay({ queueId: 'queue-1', sequence: 3, event: null }); await request; - expect(events()).toEqual([started, done]); + expect(events()).toEqual([ + { type: 'message-queued', queueId: 'queue-1', sessionId: 'thread-1' }, + started, + done, + ]); expect(res.end).toHaveBeenCalledOnce(); }); diff --git a/packages/cli/src/modules/agents/__tests__/agent-execution-update-broadcaster.test.ts b/packages/cli/src/modules/agents/__tests__/agent-execution-update-broadcaster.test.ts index eaa8ba58ce9..18ecd56f907 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-execution-update-broadcaster.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-execution-update-broadcaster.test.ts @@ -81,7 +81,7 @@ describe('AgentExecutionUpdateBroadcaster', () => { expect(publisher.publishCommand).not.toHaveBeenCalled(); }); - it('sends private execution and task updates only to the owner through the relay', async () => { + it('sends private execution, task, and queue updates only to the owner through the relay', async () => { Object.defineProperty(instanceSettings, 'isWorker', { value: true, configurable: true }); threadRepository.findOneBy.mockResolvedValue({ ...thread, @@ -90,7 +90,8 @@ describe('AgentExecutionUpdateBroadcaster', () => { }); broadcaster.notify(update); broadcaster.notifyBackgroundJobsUpdated('agent-1', 'thread-1'); - await vi.waitFor(() => expect(publisher.publishCommand).toHaveBeenCalledTimes(2)); + broadcaster.notifyQueueUpdated('thread-1'); + await vi.waitFor(() => expect(publisher.publishCommand).toHaveBeenCalledTimes(3)); for (const [command] of publisher.publishCommand.mock.calls) { expect(command).toMatchObject({ payload: { userIds: ['user-2'] } }); } @@ -134,12 +135,13 @@ describe('AgentExecutionUpdateBroadcaster', () => { expect(publisher.publishCommand).not.toHaveBeenCalled(); }); - it('delivers relayed task updates without another relay', () => { + it.each([ + ['handleBackgroundJobsRelay', 'agentBackgroundTasksUpdated'], + ['handleQueueRelay', 'agentMessageQueueUpdated'], + ] as const)('delivers %s updates without another relay', (handler, type) => { const data = { projectId: 'project-1', agentId: 'agent-1', threadId: 'thread-1' }; - broadcaster.handleBackgroundJobsRelay({ data, userIds: ['user-2'] }); - expect(push.sendToUsers).toHaveBeenCalledWith({ type: 'agentBackgroundTasksUpdated', data }, [ - 'user-2', - ]); + broadcaster[handler]({ data, userIds: ['user-2'] }); + expect(push.sendToUsers).toHaveBeenCalledWith({ type, data }, ['user-2']); expect(publisher.publishCommand).not.toHaveBeenCalled(); }); diff --git a/packages/cli/src/modules/agents/__tests__/agent-message-queue-consumer.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-message-queue-consumer.service.test.ts index 5bd376b892c..8322b7e122e 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-message-queue-consumer.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-message-queue-consumer.service.test.ts @@ -98,6 +98,8 @@ describe('AgentMessageQueueConsumer', () => { it('runs independent sessions without waiting and reuses each claimed execution', async () => { const first = claim('first'); + first.item.payload.message = 'edited first message'; + first.recording.userMessage = first.item.payload.message; const second = claim('second'); const waiting = createDeferredPromise(); repository.findThreadIds.mockResolvedValue(['first', 'second']); @@ -119,6 +121,12 @@ describe('AgentMessageQueueConsumer', () => { expect(queue.settle).toHaveBeenCalledWith('second', second.admission.executionId), ); expect(queue.settle).not.toHaveBeenCalledWith('first', first.admission.executionId); + expect(sender.send).toHaveBeenCalledWith({ + type: 'execution-started', + executionId: first.admission.executionId, + sessionId: 'first', + message: 'edited first message', + }); expect(testRuns.prepareDraftRun).toHaveBeenCalledWith( expect.objectContaining({ newSession: false, diff --git a/packages/cli/src/modules/agents/agent-chat.controller.ts b/packages/cli/src/modules/agents/agent-chat.controller.ts index 18ac9a53bd7..cc3369dce78 100644 --- a/packages/cli/src/modules/agents/agent-chat.controller.ts +++ b/packages/cli/src/modules/agents/agent-chat.controller.ts @@ -2,14 +2,25 @@ import { type AgentBackgroundJobsResponse, type AgentChatAttachmentPayload, AgentChatMessageDto, + AgentChatQueueUpdateDto, type AgentChatMessagesResponse, + type AgentChatQueueResponse, AgentChatResumeDto, MAX_AGENT_CHAT_ATTACHMENT_SIZE_BYTES, MAX_AGENT_CHAT_ATTACHMENT_SIZE_MB, ViewableMimeTypes, } from '@n8n/api-types'; import type { AuthenticatedRequest } from '@n8n/db'; -import { Body, Delete, Get, Param, Post, ProjectScope, RestController } from '@n8n/decorators'; +import { + Body, + Delete, + Get, + Param, + Patch, + Post, + ProjectScope, + RestController, +} from '@n8n/decorators'; import { scrubSecretsInText } from '@n8n/utils/scrub-secrets'; import { sanitizeFilename } from '@n8n/utils/files/sanitize-filename'; import type { Response } from 'express'; @@ -193,7 +204,7 @@ export class AgentChatController { }); abortSignal.throwIfAborted(); - await this.messageQueue.enqueue( + const item = await this.messageQueue.enqueue( { agentId, projectId, @@ -214,6 +225,7 @@ export class AgentChatController { ); accepted = true; subscription?.accepted(); + send({ type: 'message-queued', queueId: item.id, sessionId: threadId }); await subscription?.done; } catch (error) { // Committed messages own their attachments, including after a disconnect. @@ -322,6 +334,55 @@ export class AgentChatController { return { cancelled }; } + @Get('/:agentId/chat/:threadId/queue') + @ProjectScope('agent:read') + async getQueuedMessages( + req: AuthenticatedRequest<{ projectId: string; agentId: string; threadId: string }>, + ): Promise { + const agent = await this.agentsService.findById(req.params.agentId, req.params.projectId); + if (!agent) throw new NotFoundError('Agent not found'); + return await this.messageQueue.listPending({ ...req.params, userId: req.user.id }); + } + + @Patch('/:agentId/chat/:threadId/queue/:queueId') + @ProjectScope('agent:execute') + async updateQueuedMessage( + req: AuthenticatedRequest<{ + projectId: string; + agentId: string; + threadId: string; + queueId: string; + }>, + _res: Response, + @Body payload: AgentChatQueueUpdateDto, + ): Promise { + if (!/^[1-9]\d*$/.test(req.params.queueId)) throw new BadRequestError('Invalid queue ID'); + const agent = await this.agentsService.findById(req.params.agentId, req.params.projectId); + if (!agent) throw new NotFoundError('Agent not found'); + await this.messageQueue.updatePending({ + ...req.params, + userId: req.user.id, + message: payload.message, + }); + } + + @Delete('/:agentId/chat/:threadId/queue/:queueId') + @ProjectScope('agent:execute') + async removeQueuedMessage( + req: AuthenticatedRequest<{ + projectId: string; + agentId: string; + threadId: string; + queueId: string; + }>, + ) { + if (!/^[1-9]\d*$/.test(req.params.queueId)) throw new BadRequestError('Invalid queue ID'); + const agent = await this.agentsService.findById(req.params.agentId, req.params.projectId); + if (!agent) throw new NotFoundError('Agent not found'); + await this.messageQueue.removePending({ ...req.params, userId: req.user.id }); + return { removed: true }; + } + @Get('/:agentId/chat/:threadId/background-tasks') @ProjectScope('agent:read') async getBackgroundJobs( diff --git a/packages/cli/src/modules/agents/agent-execution-update-broadcaster.ts b/packages/cli/src/modules/agents/agent-execution-update-broadcaster.ts index b17db6e3dd4..e528233ce28 100644 --- a/packages/cli/src/modules/agents/agent-execution-update-broadcaster.ts +++ b/packages/cli/src/modules/agents/agent-execution-update-broadcaster.ts @@ -77,6 +77,32 @@ export class AgentExecutionUpdateBroadcaster { } } + notifyQueueUpdated(threadId: string): void { + void this.broadcastQueueUpdated(threadId).catch((error: unknown) => { + this.logger.warn('Failed to broadcast agent queue update', { threadId, error }); + }); + } + + private async broadcastQueueUpdated(threadId: string): Promise { + const thread = await this.threadRepository.findOneBy({ id: threadId }); + if (!thread || thread.accessScope !== 'user' || !thread.ownerId) return; + const data = { projectId: thread.projectId, agentId: thread.agentId, threadId }; + const userIds = await this.getRecipients(thread); + if (userIds.length === 0) return; + this.push.sendToUsers({ type: 'agentMessageQueueUpdated', data }, userIds); + if (this.instanceSettings.isWorker || this.instanceSettings.isMultiMain) { + await this.publisher.publishCommand({ + command: 'relay-agent-message-queue-update', + payload: { data, userIds }, + }); + } + } + + @OnPubSubEvent('relay-agent-message-queue-update', { instanceType: 'main' }) + handleQueueRelay({ data, userIds }: PubSubCommandMap['relay-agent-message-queue-update']): void { + this.push.sendToUsers({ type: 'agentMessageQueueUpdated', data }, userIds); + } + private async getRecipients(thread: AgentExecutionThread): Promise { const userIds = await this.recipients.getProjectReaders(thread.projectId); return userIds.filter((id) => threadBelongsTo(thread, thread.projectId, thread.agentId, id)); diff --git a/packages/cli/src/modules/agents/agent-message-queue-consumer.service.ts b/packages/cli/src/modules/agents/agent-message-queue-consumer.service.ts index b716a73da04..3acfafe99af 100644 --- a/packages/cli/src/modules/agents/agent-message-queue-consumer.service.ts +++ b/packages/cli/src/modules/agents/agent-message-queue-consumer.service.ts @@ -142,6 +142,12 @@ export class AgentMessageQueueConsumer { }, controller, ); + sender.send({ + type: 'execution-started', + executionId: admission.executionId, + sessionId: thread.id, + message: item.payload.message, + }); await this.chatExecutionService.settle( admission.executionId, async () => await this.consumePreview(claim, signal, sender.send), @@ -224,8 +230,6 @@ export class AgentMessageQueueConsumer { admittedExecution: admission, abortSignal: signal, errorMode: 'forward', - onExecutionStarted: (executionId, sessionId) => - send({ type: 'execution-started', executionId, sessionId }), onChunk: (chunk) => emitChunkEvents(chunk, send), }); if (result.status === 'completed') 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 37f8f8da913..d4d9989fc8d 100644 --- a/packages/cli/src/modules/agents/agent-message-queue.service.ts +++ b/packages/cli/src/modules/agents/agent-message-queue.service.ts @@ -1,7 +1,14 @@ +import type { AgentChatQueueResponse } from '@n8n/api-types'; import { TransactionRunner, type OperationContext } from '@n8n/db'; import { Service } from '@n8n/di'; import { OperationalError, UserError } from 'n8n-workflow'; +import { ConflictError } from '@/errors/response-errors/conflict.error'; +import { BadRequestError } from '@/errors/response-errors/bad-request.error'; +import { NotFoundError } from '@/errors/response-errors/not-found.error'; + +import { AgentChatAttachmentService } from './agent-chat-attachment.service'; +import { AgentExecutionUpdateBroadcaster } from './agent-execution-update-broadcaster'; import { AgentExecutionService, type StartExecutionParams } from './agent-execution.service'; import type { AgentExecutionThread } from './entities/agent-execution-thread.entity'; import type { AgentMessageQueue } from './entities/agent-message-queue.entity'; @@ -12,7 +19,7 @@ import { AgentExecutionThreadRepository } from './repositories/agent-execution-t import { AgentMessageQueueRepository } from './repositories/agent-message-queue.repository'; import { AgentRepository } from './repositories/agent.repository'; import type { AgentExecutionAdmission, AgentQueuedMessage } from './types/agent-queued-message'; -import type { AgentSessionMode } from './utils/agent-thread-access'; +import { canContinueThreadInPreview, type AgentSessionMode } from './utils/agent-thread-access'; export interface ClaimedAgentMessage { item: AgentMessageQueue; @@ -35,6 +42,8 @@ export class AgentMessageQueueService { private readonly executionService: AgentExecutionService, private readonly checkpointStorage: N8NCheckpointStorage, private readonly agentRepository: AgentRepository, + private readonly attachments: AgentChatAttachmentService, + private readonly updates: AgentExecutionUpdateBroadcaster, ) {} /** Save a pending message. It is durably accepted when the transaction commits. */ @@ -70,10 +79,105 @@ export class AgentMessageQueueService { onInserted?.(inserted.id); return inserted; }); + if (payload.kind === 'preview') this.updates.notifyQueueUpdated(item.threadId); this.onAvailable?.(item.threadId); return item; } + async listPending(input: { + projectId: string; + agentId: string; + threadId: string; + userId: string; + }): Promise { + const thread = await this.threadRepository.findOneBy({ id: input.threadId }); + // A client-created Preview session can have no accepted messages yet. + if (!thread) return { items: [] }; + await this.assertPreviewAccess(thread, input); + const items = await this.repository.listPending(thread.id); + return { + items: items + .filter((item) => item.payload.kind === 'preview') + .map((item) => ({ + id: item.id, + message: item.payload.message, + attachments: item.payload.attachments, + createdAt: item.createdAt.toISOString(), + })), + }; + } + + async removePending(input: { + projectId: string; + agentId: string; + threadId: string; + userId: string; + queueId: string; + }): Promise { + const item = await this.txRunner.run({}, async (ctx) => { + const thread = await this.threadRepository.lockById(input.threadId, ctx); + if (!thread) throw new NotFoundError('Session not found'); + await this.assertPreviewAccess(thread, input, ctx); + const item = await this.repository.findItem(thread.id, input.queueId, ctx); + if (!item || item.payload.kind !== 'preview') + throw new NotFoundError('Queued message not found'); + if ( + item.executionId !== null || + !(await this.repository.removePending(thread.id, item.id, ctx)) + ) { + throw new ConflictError('This message has already started'); + } + return item; + }); + this.updates.notifyQueueUpdated(item.threadId); + await this.attachments.deleteByIds(item.payload.attachments?.map(({ id }) => id) ?? []); + } + + async updatePending(input: { + projectId: string; + agentId: string; + threadId: string; + userId: string; + queueId: string; + message: string; + }): Promise { + await this.txRunner.run({}, async (ctx) => { + const thread = await this.threadRepository.lockById(input.threadId, ctx); + if (!thread) throw new NotFoundError('Session not found'); + await this.assertPreviewAccess(thread, input, ctx); + const item = await this.repository.findItem(thread.id, input.queueId, ctx); + if (!item || item.payload.kind !== 'preview') + throw new NotFoundError('Queued message not found'); + if (item.executionId !== null) throw new ConflictError('This message has already started'); + const message = input.message.trim(); + if (!message && !item.payload.attachments?.length) + throw new BadRequestError('A message or attachment is required'); + const updated = await this.repository.updatePendingPayload( + thread.id, + item.id, + { ...item.payload, message }, + ctx, + ); + if (!updated) throw new ConflictError('This message has already started'); + }); + this.updates.notifyQueueUpdated(input.threadId); + } + + private async assertPreviewAccess( + thread: AgentExecutionThread, + input: { projectId: string; agentId: string; userId: string }, + ctx: OperationContext = {}, + ): Promise { + const sources = await this.executionRepository.findFirstSourceByThreadIds([thread.id], ctx); + if ( + thread.projectId !== input.projectId || + thread.agentId !== input.agentId || + !canContinueThreadInPreview(thread, input.userId, sources.get(thread.id)) + ) { + throw new NotFoundError('Session not found'); + } + } + /** * Give the oldest pending message exclusive use of the session for its execution. * Running work and valid suspended checkpoints block a claim. diff --git a/packages/cli/src/modules/agents/repositories/agent-execution.repository.ts b/packages/cli/src/modules/agents/repositories/agent-execution.repository.ts index 2a410cb76b9..a1f5ea985af 100644 --- a/packages/cli/src/modules/agents/repositories/agent-execution.repository.ts +++ b/packages/cli/src/modules/agents/repositories/agent-execution.repository.ts @@ -152,7 +152,10 @@ export class AgentExecutionRepository extends BaseRepository { * * Returns one row per thread from that thread's earliest matching run. */ - async findFirstSourceByThreadIds(threadIds: string[]): Promise> { + async findFirstSourceByThreadIds( + threadIds: string[], + ctx: OperationContext = {}, + ): Promise> { if (threadIds.length === 0) return new Map(); // Correlated subquery: for each thread, pick the row with the smallest @@ -160,7 +163,9 @@ export class AgentExecutionRepository extends BaseRepository { // so Postgres preserves their camelCase (it lowercases unquoted names), // and the table name is read from metadata so DB_TABLE_PREFIX is respected. const tableName = this.metadata.tablePath; - const rows = await this.createQueryBuilder('e') + const rows = await this.managerFor(ctx) + .getRepository(AgentExecution) + .createQueryBuilder('e') .select(['e."threadId" AS "threadId"', 'e."source" AS "source"']) .where('e."threadId" IN (:...threadIds)', { threadIds }) .andWhere('e."source" IS NOT NULL') 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 0d9d2fbfec2..e2e7e22f83a 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 @@ -25,6 +25,37 @@ export class AgentMessageQueueRepository extends BaseRepository([ 'relay-agent-execution-update', 'relay-agent-queued-chat', 'cancel-agent-chat-execution', + 'relay-agent-message-queue-update', 'relay-agent-background-tasks-update', 'relay-agent-update', 'resume-agent-workflow-tool', diff --git a/packages/cli/src/scaling/pubsub/__tests__/publisher.service.test.ts b/packages/cli/src/scaling/pubsub/__tests__/publisher.service.test.ts index 944728f468c..747c8a8d787 100644 --- a/packages/cli/src/scaling/pubsub/__tests__/publisher.service.test.ts +++ b/packages/cli/src/scaling/pubsub/__tests__/publisher.service.test.ts @@ -173,6 +173,7 @@ describe('Publisher', () => { 'relay-agent-execution-update', 'cancel-agent-chat-execution', 'relay-agent-background-tasks-update', + 'relay-agent-message-queue-update', 'relay-agent-update', 'display-workflow-activation', 'display-workflow-deactivation', diff --git a/packages/cli/src/scaling/pubsub/pubsub.event-map.ts b/packages/cli/src/scaling/pubsub/pubsub.event-map.ts index 6144a93fb73..e77eb7645a5 100644 --- a/packages/cli/src/scaling/pubsub/pubsub.event-map.ts +++ b/packages/cli/src/scaling/pubsub/pubsub.event-map.ts @@ -147,6 +147,11 @@ export type PubSubCommandMap = { userId: string; }; + 'relay-agent-message-queue-update': { + data: PushPayload<'agentMessageQueueUpdated'>; + userIds: string[]; + }; + 'relay-agent-background-tasks-update': { data: PushPayload<'agentBackgroundTasksUpdated'>; userIds: string[]; diff --git a/packages/cli/src/scaling/pubsub/pubsub.types.ts b/packages/cli/src/scaling/pubsub/pubsub.types.ts index dca30af5ea0..ec93562382e 100644 --- a/packages/cli/src/scaling/pubsub/pubsub.types.ts +++ b/packages/cli/src/scaling/pubsub/pubsub.types.ts @@ -62,6 +62,7 @@ export namespace PubSub { export type RelayAgentExecutionUpdate = ToCommand<'relay-agent-execution-update'>; export type RelayAgentQueuedChat = ToCommand<'relay-agent-queued-chat'>; export type CancelAgentChatExecution = ToCommand<'cancel-agent-chat-execution'>; + export type RelayAgentMessageQueueUpdate = ToCommand<'relay-agent-message-queue-update'>; export type RelayAgentBackgroundJobsUpdate = ToCommand<'relay-agent-background-tasks-update'>; export type RelayAgentUpdate = ToCommand<'relay-agent-update'>; export type ResumeAgentWorkflowTool = ToCommand<'resume-agent-workflow-tool'>; @@ -112,6 +113,7 @@ export namespace PubSub { | Commands.RelayAgentExecutionUpdate | Commands.RelayAgentQueuedChat | Commands.CancelAgentChatExecution + | Commands.RelayAgentMessageQueueUpdate | Commands.RelayAgentBackgroundJobsUpdate | Commands.RelayAgentUpdate | Commands.ResumeAgentWorkflowTool 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 802d439b32c..76380ff8b66 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 @@ -178,6 +178,8 @@ describe('AgentExecutionRepository', () => { executionService, checkpointStorage, connection ? new AgentRepository(connection) : agentRepo, + attachmentService, + mock(), ); return { txRunner, @@ -1058,6 +1060,157 @@ describe('AgentExecutionRepository', () => { await services.queue.settle(item.thread.id, item.admission.executionId); } + it('lists only pending Preview input and removes its attachments without affecting the active run', async () => { + const services = recordingServices(); + const threadId = uuid(); + const target = { projectId, agentId, threadId, userId: owner.id }; + await services.queue.enqueue(input(threadId, 'active', 'new')); + const active = await claim(services, threadId); + const pendingInput = input(threadId, 'pending'); + pendingInput.payload.attachments = [ + { id: 'pending-file', fileName: 'notes.txt', mimeType: 'text/plain', sizeBytes: 5 }, + ]; + const pending = await services.queue.enqueue(pendingInput); + expect((await services.queue.listPending(target)).items.map(({ id }) => id)).toEqual([ + pending.id, + ]); + await expect( + services.queue.removePending({ ...target, queueId: active.item.id }), + ).rejects.toThrow('already started'); + await expect( + services.queue.removePending({ ...target, userId: 'other-user', queueId: pending.id }), + ).rejects.toThrow('Session not found'); + await expect( + services.queue.listPending({ ...target, agentId: 'other-agent' }), + ).rejects.toThrow('Session not found'); + expect(services.attachmentService.deleteByIds).not.toHaveBeenCalled(); + await services.queue.removePending({ ...target, queueId: pending.id }); + expect(await services.queue.listPending(target)).toEqual({ items: [] }); + expect(services.attachmentService.deleteByIds).toHaveBeenCalledWith(['pending-file']); + expect((await repository.findOneByOrFail({ id: active.admission.executionId })).status).toBe( + 'running', + ); + await finish(services, active); + }); + + it('edits pending text without changing identity, attachments, or queue position', async () => { + const services = recordingServices(); + const threadId = uuid(); + const data = input(threadId, 'original', 'new'); + data.payload.attachments = [ + { id: 'file', fileName: 'notes.txt', mimeType: 'text/plain', sizeBytes: 5 }, + ]; + const first = await services.queue.enqueue(data); + const second = await services.queue.enqueue(input(threadId, 'second')); + const target = { + projectId, + agentId, + threadId, + userId: owner.id, + queueId: first.id, + message: 'updated', + }; + await expect( + services.queue.updatePending({ ...target, userId: 'other-user' }), + ).rejects.toThrow('Session not found'); + await expect( + services.queue.updatePending({ ...target, agentId: 'other-agent' }), + ).rejects.toThrow('Session not found'); + await expect( + services.queue.updatePending({ ...target, queueId: second.id, message: ' ' }), + ).rejects.toThrow('message or attachment'); + await services.queue.updatePending({ ...target, message: ' ' }); + expect((await services.queue.listPending(target)).items[0]).toMatchObject({ + message: '', + attachments: data.payload.attachments, + }); + await services.queue.updatePending(target); + const stored = await services.queueRepository.findOneByOrFail({ id: first.id }); + expect(stored).toMatchObject({ + id: first.id, + createdAt: first.createdAt, + source: first.source, + payload: { ...first.payload, message: 'updated' }, + }); + expect((await services.queue.listPending(target)).items.map(({ id }) => id)).toEqual([ + first.id, + second.id, + ]); + expect(services.attachmentService.deleteByIds).not.toHaveBeenCalled(); + const claimed = await claim(services, threadId); + expect(claimed.item.payload.message).toBe('updated'); + await finish(services, claimed); + await expect(services.queue.updatePending(target)).rejects.toThrow( + 'Queued message not found', + ); + }); + + it.each([ + ['claim', 'remove'], + ['remove', 'remove'], + ['claim', 'edit'], + ['edit', 'edit'], + ] as const)( + 'serializes concurrent connections when %s wins the claim-versus-%s race', + async (winner, operation) => { + const local = recordingServices(); + const remote = recordingServices(undefined, peer); + const threadId = uuid(); + const item = await local.queue.enqueue(input(threadId, 'first', 'new')); + const target = { projectId, agentId, threadId, userId: owner.id, queueId: item.id }; + const acquired = createDeferredPromise(); + const release = createDeferredPromise(); + const lock = local.threads.lockById.bind(local.threads); + vi.spyOn(local.threads, 'lockById').mockImplementationOnce(async (...args) => { + const thread = await lock(...args); + acquired.resolve(); + await release.promise; + return thread; + }); + const observe = observePeerTransaction(); + const mutate = async (services: ReturnType) => + operation === 'edit' + ? await services.queue.updatePending({ ...target, message: 'edited' }) + : await services.queue.removePending(target); + const first = + winner === 'claim' ? local.queue.claimNext(threadId, async () => true) : mutate(local); + await acquired.promise; + const second = + winner === 'claim' ? mutate(remote) : remote.queue.claimNext(threadId, async () => true); + const results = Promise.allSettled([first, second]); + try { + await observe.started; + } finally { + release.resolve(); + observe.restore(); + } + const settled = await results; + expect(settled[0].status).toBe('fulfilled'); + if (winner === 'claim') { + expect(settled[1]).toMatchObject({ + status: 'rejected', + reason: { message: 'This message has already started' }, + }); + const claimed = await first; + if (!claimed) throw new Error('Expected a claim'); + expect(await local.queueRepository.findOneByOrFail({ id: item.id })).toMatchObject({ + threadId, + executionId: claimed.admission.executionId, + }); + await finish(local, claimed); + } else if (operation === 'edit') { + const claimed = await second; + if (!claimed) throw new Error('Expected a claim'); + expect(claimed.item.payload.message).toBe('edited'); + await finish(remote, claimed); + } else { + expect(settled[1]).toEqual({ status: 'fulfilled', value: null }); + expect(await repository.countBy({ threadId })).toBe(0); + } + expect(await local.queueRepository.countBy({ threadId })).toBe(0); + }, + ); + it('serializes acceptance and claims across connections while other sessions progress', async () => { const local = recordingServices(); const remote = recordingServices(undefined, peer); diff --git a/packages/frontend/@n8n/i18n/src/locales/en.json b/packages/frontend/@n8n/i18n/src/locales/en.json index f7099414f25..3ee4335c09a 100644 --- a/packages/frontend/@n8n/i18n/src/locales/en.json +++ b/packages/frontend/@n8n/i18n/src/locales/en.json @@ -7973,6 +7973,16 @@ "agents.chat.loadHistory.error": "Could not load chat history", "agents.chat.clearHistory.error": "Could not clear chat history", "agents.chat.stop.error": "Couldn't stop the agent. Try again.", + "agents.chat.queue.title": "{count} message up next | {count} messages up next", + "agents.chat.queue.edit": "Edit queued message", + "agents.chat.queue.save": "Save changes", + "agents.chat.queue.cancelEdit": "Discard edits", + "agents.chat.queue.editUnavailable": "This message has started or was removed. Your changes weren't applied.", + "agents.chat.queue.editError": "Couldn't save your changes. Try again.", + "agents.chat.queue.remove": "Remove queued message", + "agents.chat.queue.loadError": "Couldn't load queued messages. Reopen the chat to try again.", + "agents.chat.queue.removeError": "Couldn't remove the message. Check whether it has started and try again.", + "agents.chat.queue.sendError": "Couldn't confirm the message was queued. Check the queue before sending again.", "agents.chat.backgroundTasks.resultsReceived": "Background task results received", "agents.chat.backgroundTasks.result": "{title} — {status}", "agents.chat.backgroundTasks.runningCount": "Running {count} background task | Running {count} background tasks", @@ -8022,8 +8032,6 @@ "agents.chat.approval.approved": "Approved", "agents.chat.approval.rejected": "Rejected", "agents.chat.approval.viewToolDetails": "View tool details", - "agents.chat.approval.inputPlaceholder": "Approve or reject to continue", - "agents.chat.waiting.inputPlaceholder": "Waiting on the card above to continue", "agents.chat.delegate.label": "Sub-agent · {name}", "agents.chat.delegate.labelFallback": "Sub-agent", "agents.chat.delegate.childSuspendUnsupported": "Sub-agent requested user input, which is not supported for delegated runs yet.", diff --git a/packages/frontend/editor-ui/src/features/agents/__tests__/AgentChatPanel.test.ts b/packages/frontend/editor-ui/src/features/agents/__tests__/AgentChatPanel.test.ts index 13639199d30..61f78b7b62d 100644 --- a/packages/frontend/editor-ui/src/features/agents/__tests__/AgentChatPanel.test.ts +++ b/packages/frontend/editor-ui/src/features/agents/__tests__/AgentChatPanel.test.ts @@ -4,7 +4,7 @@ import { computed, defineComponent, h, ref, nextTick } from 'vue'; import { createMemoryHistory, createRouter } from 'vue-router'; import { AGENT_SESSION_DETAIL_VIEW } from '../constants'; import { APPROVAL_TOOL_NAME, N8N_CHAT_ACTION_TOOL_NAME, WAIT_TOOL_NAME } from '@n8n/api-types'; -import type { AgentBackgroundJobDto } from '@n8n/api-types'; +import type { AgentChatQueueItem, AgentBackgroundJobDto } from '@n8n/api-types'; import type { ChatMessage } from '@/features/ai/shared/agentsChat/types'; import AgentChatPanel from '../components/AgentChatPanel.vue'; import AgentPreviewDock from '../components/AgentPreviewDock.vue'; @@ -23,6 +23,12 @@ const cancelAndSteerMock = vi.fn(); const focusInputMock = vi.fn(); const messagesMock = ref([]); const isStreamingMock = ref(false); +const isSubmittingMock = ref(false); +const isLoadingHistoryMock = ref(false); +const trackSubmittedMessageMock = vi.fn(); +const queuedMessagesMock = ref([]); +const removeQueuedMessageMock = vi.fn(); +const updateQueuedMessageMock = vi.fn(); const isCancellingMock = ref(false); const backgroundJobsMock = ref([]); vi.mock('../composables/useAgentBackgroundJobs', () => ({ @@ -45,6 +51,7 @@ vi.mock('@n8n/i18n', () => { ) => { const translations: Record = { 'agents.chat.input.placeholder.withAgent': `Message ${options?.interpolate?.agentName}…`, + 'agents.chat.queue.title': `${options?.interpolate?.count} ${options?.adjustToNumber === 1 ? 'message' : 'messages'} up next`, 'agents.chat.misconfigured.issuesPrefix': 'Check:', 'agents.chat.misconfigured.missing.tools': 'Tool configuration', 'agents.chat.misconfigured.missing.mcpServers': 'MCP server', @@ -81,7 +88,8 @@ vi.mock('@n8n/design-system', async (importOriginal) => ({ N8nLink: (await importOriginal()).N8nLink, useDropdownSearch: (await importOriginal()) .useDropdownSearch, - N8nButton: { template: '' }, + N8nInput: (await importOriginal()).N8nInput, + N8nButton: { template: '' }, N8nCallout: { template: '
' }, N8nDropdownMenu: { template: '
' }, N8nHeading: { template: '
' }, @@ -91,12 +99,6 @@ vi.mock('@n8n/design-system', async (importOriginal) => ({ template: '