From 3fbb249e4f7dae97b5675a92a0960d7cb0a88cfd Mon Sep 17 00:00:00 2001 From: Robin Braumann Date: Mon, 14 Sep 2026 11:58:36 +0200 Subject: [PATCH] feat(editor): Show and edit queued messages in the agent preview chat A message sent while the thread runs a turn becomes an `agent_execution` row with status `queued`. The row is visible in every tab, survives a reload, and runs after the current turn ends. - Migration adds the `queued` status and an `agent_execution.resourceId` column that records the sender of a queued turn. - `AgentChatQueueService` stores, edits, removes, and drains queued rows. Draining runs oldest-first through `executeForQueued`, before any background wake; the sweeper drains as a fallback. - New routes: `POST /chat/queue`, `PATCH /chat/queue/:executionId`, `DELETE /chat/queue/:executionId`. - The transcript carries `executionStatus: 'queued'`; the sessions list and the LangSmith export ignore queued rows. - The composer stays live while a turn runs: it enqueues instead of waiting. Queued bubbles show inline edit and remove actions. Co-authored-by: Cursor --- docs/generated/postgres-schema/README.md | 3 +- .../postgres-schema/public.agent_execution.md | 4 +- .../public.agent_execution_threads.md | 1 + docs/generated/sqlite-schema/README.md | 3 +- .../sqlite-schema/agent_execution.md | 10 +- .../sqlite-schema/agent_execution_threads.md | 1 + packages/@n8n/api-types/src/agents/dto.ts | 38 ++- packages/@n8n/api-types/src/agents/types.ts | 4 +- ...5477575-AddQueuedStatusToAgentExecution.ts | 67 +++++ .../agent-chat-queue.service.test.ts | 244 ++++++++++++++++++ .../__tests__/agent-chat.controller.test.ts | 76 ++++++ ...ent-execution-orchestrator.service.test.ts | 112 ++++++++ ...gent-interrupted-execution-sweeper.test.ts | 2 + .../execution-to-message-mapper.test.ts | 16 ++ .../agents/agent-chat-queue.service.ts | 217 ++++++++++++++++ .../modules/agents/agent-chat.controller.ts | 114 +++++++- .../agent-execution-orchestrator.service.ts | 233 +++++++++++++---- .../modules/agents/agent-execution.service.ts | 110 +++++++- .../agent-interrupted-execution-sweeper.ts | 9 + .../agent-session-langsmith-export.service.ts | 4 +- .../agents/entities/agent-execution.entity.ts | 15 +- .../agent-execution-thread.repository.ts | 22 +- .../agent-execution.repository.ts | 92 ++++++- .../utils/execution-to-message-mapper.ts | 2 + .../agent-execution.repository.test.ts | 65 ++++- ...d-queued-status-to-agent-execution.test.ts | 124 +++++++++ .../frontend/@n8n/i18n/src/locales/en.json | 9 + .../__tests__/AgentChatMessageList.test.ts | 36 +++ .../agents/__tests__/AgentChatPanel.test.ts | 82 ++++-- .../__tests__/useAgentChatStream.test.ts | 130 +++++++++- .../components/AgentChatMessageList.vue | 151 ++++++++++- .../agents/components/AgentChatPanel.vue | 63 +++-- .../agents/composables/useAgentApi.ts | 44 ++++ .../agents/composables/useAgentChatStream.ts | 188 +++++++++++--- .../ai/shared/agentsChat/constants.ts | 3 + .../ai/shared/agentsChat/messageMappers.ts | 14 +- 36 files changed, 2139 insertions(+), 169 deletions(-) create mode 100644 packages/@n8n/db/src/migrations/common/1789375477575-AddQueuedStatusToAgentExecution.ts create mode 100644 packages/cli/src/modules/agents/__tests__/agent-chat-queue.service.test.ts create mode 100644 packages/cli/src/modules/agents/agent-chat-queue.service.ts create mode 100644 packages/cli/test/migration/1789375477575-add-queued-status-to-agent-execution.test.ts diff --git a/docs/generated/postgres-schema/README.md b/docs/generated/postgres-schema/README.md index 79849c3000e..b169d9b0c0e 100644 --- a/docs/generated/postgres-schema/README.md +++ b/docs/generated/postgres-schema/README.md @@ -19,7 +19,7 @@ Auto-generated from the PostgreSQL migrations in @n8n/db. Do not edit by hand. | [public.agent_eval_rating](public.agent_eval_rating.md) | 8 | | BASE TABLE | | [public.agent_eval_result](public.agent_eval_result.md) | 15 | | BASE TABLE | | [public.agent_eval_run](public.agent_eval_run.md) | 14 | | BASE TABLE | -| [public.agent_execution](public.agent_execution.md) | 23 | | BASE TABLE | +| [public.agent_execution](public.agent_execution.md) | 24 | | BASE TABLE | | [public.agent_execution_threads](public.agent_execution_threads.md) | 17 | | BASE TABLE | | [public.agent_files](public.agent_files.md) | 10 | | BASE TABLE | | [public.agent_history](public.agent_history.md) | 9 | | BASE TABLE | @@ -525,6 +525,7 @@ erDiagram varchar_36_ id varchar_255_ model integer promptTokens + varchar_255_ resourceId varchar_32_ source timestamp_3__with_time_zone startedAt varchar_16_ status diff --git a/docs/generated/postgres-schema/public.agent_execution.md b/docs/generated/postgres-schema/public.agent_execution.md index 29aaef9a68f..34f57cf2d90 100644 --- a/docs/generated/postgres-schema/public.agent_execution.md +++ b/docs/generated/postgres-schema/public.agent_execution.md @@ -17,6 +17,7 @@ | id | varchar(36) | | false | | | | | model | varchar(255) | | true | | | | | promptTokens | integer | | true | | | | +| resourceId | varchar(255) | | true | | | Memory resource id of the sender; set on queued preview turns so the drain resolves the user | | source | varchar(32) | | true | | | | | startedAt | timestamp(3) with time zone | | true | | | | | status | varchar(16) | | false | | | | @@ -33,7 +34,7 @@ | Name | Type | Definition | | ---- | ---- | ---------- | | CHK_agent_execution_hitlStatus | CHECK | CHECK ((("hitlStatus")::text = ANY ((ARRAY['suspended'::character varying, 'resumed'::character varying])::text[]))) | -| CHK_agent_execution_status | CHECK | CHECK (((status)::text = ANY ((ARRAY['running'::character varying, 'success'::character varying, 'error'::character varying, 'cancelled'::character varying, 'interrupted'::character varying])::text[]))) | +| CHK_agent_execution_status | CHECK | CHECK (((status)::text = ANY ((ARRAY['queued'::character varying, 'running'::character varying, 'success'::character varying, 'error'::character varying, 'cancelled'::character varying, 'interrupted'::character varying])::text[]))) | | CHK_agent_execution_storedAt | CHECK | CHECK ((("storedAt")::text = ANY ((ARRAY['db'::character varying, 'fs'::character varying, 's3'::character varying, 'az'::character varying])::text[]))) | | FK_add2432fb6034cc18b6af299dce | FOREIGN KEY | FOREIGN KEY ("threadId") REFERENCES agent_execution_threads(id) ON DELETE CASCADE | | PK_ba438acc8532addc12d1ef17049 | PRIMARY KEY | PRIMARY KEY (id) | @@ -75,6 +76,7 @@ erDiagram varchar_36_ id varchar_255_ model integer promptTokens + varchar_255_ resourceId varchar_32_ source timestamp_3__with_time_zone startedAt varchar_16_ status diff --git a/docs/generated/postgres-schema/public.agent_execution_threads.md b/docs/generated/postgres-schema/public.agent_execution_threads.md index 3d87b411dca..4ac15da81c0 100644 --- a/docs/generated/postgres-schema/public.agent_execution_threads.md +++ b/docs/generated/postgres-schema/public.agent_execution_threads.md @@ -110,6 +110,7 @@ erDiagram varchar_36_ id varchar_255_ model integer promptTokens + varchar_255_ resourceId varchar_32_ source timestamp_3__with_time_zone startedAt varchar_16_ status diff --git a/docs/generated/sqlite-schema/README.md b/docs/generated/sqlite-schema/README.md index e0e48af44f8..09cb13ff632 100644 --- a/docs/generated/sqlite-schema/README.md +++ b/docs/generated/sqlite-schema/README.md @@ -19,7 +19,7 @@ Auto-generated from the SQLite migrations in @n8n/db. Do not edit by hand. | [agent_eval_rating](agent_eval_rating.md) | 8 | | table | | [agent_eval_result](agent_eval_result.md) | 15 | | table | | [agent_eval_run](agent_eval_run.md) | 14 | | table | -| [agent_execution](agent_execution.md) | 23 | | table | +| [agent_execution](agent_execution.md) | 24 | | table | | [agent_execution_threads](agent_execution_threads.md) | 17 | | table | | [agent_files](agent_files.md) | 10 | | table | | [agent_history](agent_history.md) | 9 | | table | @@ -512,6 +512,7 @@ erDiagram varchar_36_ id PK varchar_255_ model INTEGER promptTokens + varchar_255_ resourceId varchar_32_ source datetime_3_ startedAt varchar_16_ status diff --git a/docs/generated/sqlite-schema/agent_execution.md b/docs/generated/sqlite-schema/agent_execution.md index 1f576881123..9cfb401ab31 100644 --- a/docs/generated/sqlite-schema/agent_execution.md +++ b/docs/generated/sqlite-schema/agent_execution.md @@ -6,7 +6,7 @@ Table Definition ```sql -CREATE TABLE "agent_execution" ("id" varchar(36) PRIMARY KEY NOT NULL, "threadId" varchar(128) NOT NULL, "status" varchar(16) NOT NULL, "startedAt" datetime(3), "stoppedAt" datetime(3), "duration" integer NOT NULL DEFAULT (0), "userMessage" text, "model" varchar(255), "promptTokens" integer, "completionTokens" integer, "totalTokens" integer, "cost" real, "timeline" text, "error" text, "hitlStatus" varchar(16), "source" varchar(32), "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "storedAt" varchar(2) NOT NULL DEFAULT ('db'), "attachments" text, "failureSummary" text, "author" text, "activeThreadId" varchar(128), CONSTRAINT "CHK_agent_execution_storedAt" CHECK ((("storedAt" IN ('db', 'fs', 's3', 'az')))), CONSTRAINT "CHK_agent_execution_hitlStatus" CHECK (((((("hitlStatus" IN ('suspended', 'resumed'))))))), CONSTRAINT "CHK_agent_execution_status" CHECK (("status" IN ('running', 'success', 'error', 'cancelled', 'interrupted'))), CONSTRAINT "FK_add2432fb6034cc18b6af299dce" FOREIGN KEY ("threadId") REFERENCES "agent_execution_threads" ("id") ON DELETE CASCADE ON UPDATE NO ACTION) +CREATE TABLE "agent_execution" ("id" varchar(36) PRIMARY KEY NOT NULL, "threadId" varchar(128) NOT NULL, "status" varchar(16) NOT NULL, "startedAt" datetime(3), "stoppedAt" datetime(3), "duration" integer NOT NULL DEFAULT (0), "userMessage" text, "model" varchar(255), "promptTokens" integer, "completionTokens" integer, "totalTokens" integer, "cost" real, "timeline" text, "error" text, "hitlStatus" varchar(16), "source" varchar(32), "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "storedAt" varchar(2) NOT NULL DEFAULT ('db'), "attachments" text, "failureSummary" text, "author" text, "activeThreadId" varchar(128), "resourceId" varchar(255), CONSTRAINT "CHK_agent_execution_storedAt" CHECK (((("storedAt" IN ('db', 'fs', 's3', 'az'))))), CONSTRAINT "CHK_agent_execution_hitlStatus" CHECK ((((((("hitlStatus" IN ('suspended', 'resumed')))))))), CONSTRAINT "CHK_agent_execution_status" CHECK ("status" IN ('queued', 'running', 'success', 'error', 'cancelled', 'interrupted')), CONSTRAINT "FK_add2432fb6034cc18b6af299dce" FOREIGN KEY ("threadId") REFERENCES "agent_execution_threads" ("id") ON DELETE CASCADE ON UPDATE NO ACTION) ``` @@ -28,6 +28,7 @@ CREATE TABLE "agent_execution" ("id" varchar(36) PRIMARY KEY NOT NULL, "threadId | id | varchar(36) | | false | | | | | model | varchar(255) | | true | | | | | promptTokens | INTEGER | | true | | | | +| resourceId | varchar(255) | | true | | | | | source | varchar(32) | | true | | | | | startedAt | datetime(3) | | true | | | | | status | varchar(16) | | false | | | | @@ -43,9 +44,9 @@ CREATE TABLE "agent_execution" ("id" varchar(36) PRIMARY KEY NOT NULL, "threadId | Name | Type | Definition | | ---- | ---- | ---------- | -| - | CHECK | CHECK ((("storedAt" IN ('db', 'fs', 's3', 'az')))) | -| - | CHECK | CHECK (((((("hitlStatus" IN ('suspended', 'resumed'))))))) | -| - | CHECK | CHECK (("status" IN ('running', 'success', 'error', 'cancelled', 'interrupted'))) | +| - | CHECK | CHECK (((("storedAt" IN ('db', 'fs', 's3', 'az'))))) | +| - | CHECK | CHECK ((((((("hitlStatus" IN ('suspended', 'resumed')))))))) | +| - | CHECK | CHECK ("status" IN ('queued', 'running', 'success', 'error', 'cancelled', 'interrupted')) | | - (Foreign key ID: 0) | FOREIGN KEY | FOREIGN KEY (threadId) REFERENCES agent_execution_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE | | id | PRIMARY KEY | PRIMARY KEY (id) | | sqlite_autoindex_agent_execution_1 | PRIMARY KEY | PRIMARY KEY (id) | @@ -80,6 +81,7 @@ erDiagram varchar_36_ id PK varchar_255_ model INTEGER promptTokens + varchar_255_ resourceId varchar_32_ source datetime_3_ startedAt varchar_16_ status diff --git a/docs/generated/sqlite-schema/agent_execution_threads.md b/docs/generated/sqlite-schema/agent_execution_threads.md index 552723322f6..8c5b8e8d6b0 100644 --- a/docs/generated/sqlite-schema/agent_execution_threads.md +++ b/docs/generated/sqlite-schema/agent_execution_threads.md @@ -111,6 +111,7 @@ erDiagram varchar_36_ id PK varchar_255_ model INTEGER promptTokens + varchar_255_ resourceId varchar_32_ source datetime_3_ startedAt varchar_16_ status diff --git a/packages/@n8n/api-types/src/agents/dto.ts b/packages/@n8n/api-types/src/agents/dto.ts index ee8cc3bb17a..e73797b9584 100644 --- a/packages/@n8n/api-types/src/agents/dto.ts +++ b/packages/@n8n/api-types/src/agents/dto.ts @@ -228,12 +228,16 @@ const agentChatMessageShape = { .optional(), }; +const hasTextOrAttachment = (value: { message: string; attachments?: unknown[] }) => + value.message.trim().length > 0 || (value.attachments?.length ?? 0) > 0; +const textOrAttachmentIssue = { + message: 'Message text or at least one attachment is required', + path: ['message'], +}; + const agentChatMessageSchema = z .object(agentChatMessageShape) - .refine((value) => value.message.trim().length > 0 || (value.attachments?.length ?? 0) > 0, { - message: 'Message text or at least one attachment is required', - path: ['message'], - }); + .refine(hasTextOrAttachment, textOrAttachmentIssue); /** * Validate via `parse`/`safeParse` (what the controller registry's `@Body` @@ -255,6 +259,32 @@ export class AgentChatMessageDto extends Z.class(agentChatMessageShape) { } } +// A message queues behind the running turn of an existing thread only. +const agentChatQueueMessageShape = { ...agentChatMessageShape, sessionId: z.string().min(1) }; + +const agentChatQueueMessageSchema = z + .object(agentChatQueueMessageShape) + .refine(hasTextOrAttachment, textOrAttachmentIssue); + +/** See {@link AgentChatMessageDto} for why `parse`/`safeParse` are overridden. */ +export class AgentChatQueueMessageDto extends Z.class(agentChatQueueMessageShape) { + constructor(data: z.infer) { + super(agentChatQueueMessageSchema.parse(data)); + } + + static override safeParse(data: unknown) { + return agentChatQueueMessageSchema.safeParse(data); + } + + static override parse(data: unknown) { + return agentChatQueueMessageSchema.parse(data); + } +} + +export class AgentChatEditQueuedMessageDto extends Z.class({ + message: z.string().trim().min(1), +}) {} + 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 4599311a4bb..583e86f2239 100644 --- a/packages/@n8n/api-types/src/agents/types.ts +++ b/packages/@n8n/api-types/src/agents/types.ts @@ -341,8 +341,8 @@ export interface AgentPersistedMessageDto { author?: AgentMessageAuthor; /** Agent-execution turn id when this message was produced from an execution transcript. */ executionId?: string; - /** Outcome of the execution that produced this message. */ - executionStatus?: 'running' | 'success' | 'error' | 'cancelled' | 'interrupted'; + /** Status of the execution that produced this message; `queued` marks a user turn that waits for its run. */ + executionStatus?: 'queued' | 'running' | 'success' | 'error' | 'cancelled' | 'interrupted'; /** * The recorded run error for a turn that ended in `error` or `interrupted`, * so history renders the same error bubble the live stream showed. diff --git a/packages/@n8n/db/src/migrations/common/1789375477575-AddQueuedStatusToAgentExecution.ts b/packages/@n8n/db/src/migrations/common/1789375477575-AddQueuedStatusToAgentExecution.ts new file mode 100644 index 00000000000..d26da1b3dc8 --- /dev/null +++ b/packages/@n8n/db/src/migrations/common/1789375477575-AddQueuedStatusToAgentExecution.ts @@ -0,0 +1,67 @@ +import type { MigrationContext, ReversibleMigration } from '../migration-types'; + +const executionTable = 'agent_execution'; +const resourceIdColumn = 'resourceId'; +const statusesBefore = ['running', 'success', 'error', 'cancelled', 'interrupted']; +const statusesAfter = ['queued', ...statusesBefore]; + +/** + * A preview-chat message sent while a turn runs is stored as a `queued` + * execution row and runs once the thread is idle. `resourceId` names the + * sender's memory resource, so the drain can run the turn as that user. + */ +export class AddQueuedStatusToAgentExecution1789375477575 implements ReversibleMigration { + async up({ + isSqlite, + escape, + runQuery, + schemaBuilder: { addColumns, column, addEnumCheck, dropEnumCheck }, + }: MigrationContext) { + await dropEnumCheck(executionTable, 'status', { recreatesOnSqlite: true }); + await addEnumCheck(executionTable, 'status', statusesAfter, { recreatesOnSqlite: true }); + + // The schema builder rebuilds the table on SQLite. + // Add this nullable column in place. + if (isSqlite) { + await runQuery( + `ALTER TABLE ${escape.tableName(executionTable)} ADD COLUMN ${escape.columnName(resourceIdColumn)} varchar(255)`, + ); + } else { + await addColumns( + executionTable, + [ + column(resourceIdColumn) + .varchar(255) + .comment( + 'Memory resource id of the sender; set on queued preview turns so the drain resolves the user', + ), + ], + { recreatesOnSqlite: true }, + ); + } + } + + async down({ + isSqlite, + escape, + runQuery, + schemaBuilder: { dropColumns, addEnumCheck, dropEnumCheck }, + }: MigrationContext) { + if (isSqlite) { + await runQuery( + `ALTER TABLE ${escape.tableName(executionTable)} DROP COLUMN ${escape.columnName(resourceIdColumn)}`, + ); + } else { + await dropColumns(executionTable, [resourceIdColumn], { recreatesOnSqlite: true }); + } + + // A queued row never ran, so it ends as cancelled once the status is gone. + await runQuery( + `UPDATE ${escape.tableName(executionTable)} ` + + `SET ${escape.columnName('status')} = 'cancelled' ` + + `WHERE ${escape.columnName('status')} = 'queued'`, + ); + await dropEnumCheck(executionTable, 'status', { recreatesOnSqlite: true }); + await addEnumCheck(executionTable, 'status', statusesBefore, { recreatesOnSqlite: true }); + } +} diff --git a/packages/cli/src/modules/agents/__tests__/agent-chat-queue.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-chat-queue.service.test.ts new file mode 100644 index 00000000000..4d27d1925e8 --- /dev/null +++ b/packages/cli/src/modules/agents/__tests__/agent-chat-queue.service.test.ts @@ -0,0 +1,244 @@ +import type { Logger } from '@n8n/backend-common'; +import type { UserRepository } from '@n8n/db'; +import { mock } from 'vitest-mock-extended'; + +import { ConflictError } from '@/errors/response-errors/conflict.error'; +import { NotFoundError } from '@/errors/response-errors/not-found.error'; +import { userHasScopes } from '@/permissions.ee/check-access'; + +import type { AgentChatAttachmentService } from '../agent-chat-attachment.service'; +import { AgentChatQueueService, type QueuedMessageRef } from '../agent-chat-queue.service'; +import type { AgentExecutionOrchestratorService } from '../agent-execution-orchestrator.service'; +import type { AgentExecutionUpdateBroadcaster } from '../agent-execution-update-broadcaster'; +import type { AgentExecutionService } from '../agent-execution.service'; +import { MAX_AGENT_THREAD_WAITERS } from '../agent-thread-turn-coordinator'; +import type { AgentExecutionThread } from '../entities/agent-execution-thread.entity'; +import type { AgentExecutionThreadRepository } from '../repositories/agent-execution-thread.repository'; +import type { + AgentExecutionRepository, + AgentExecutionUserTurn, +} from '../repositories/agent-execution.repository'; + +vi.mock('@/permissions.ee/check-access', () => ({ + userHasScopes: vi.fn().mockResolvedValue(true), +})); + +const user = { id: 'user-1', disabled: false }; +const thread = { + id: 'thread-1', + agentId: 'agent-1', + projectId: 'project-1', +} as AgentExecutionThread; +const ownRef: QueuedMessageRef = { + executionId: 'exec-1', + agentId: thread.agentId, + projectId: thread.projectId, + resourceId: `draft-chat:${user.id}`, +}; + +function queuedRow(overrides: Partial = {}): AgentExecutionUserTurn { + return { + id: 'exec-1', + threadId: thread.id, + status: 'queued', + resourceId: `draft-chat:${user.id}`, + attachments: null, + ...overrides, + }; +} + +function setup() { + const executionRepository = mock(); + const threadRepository = mock(); + const executionService = mock(); + const attachmentService = mock(); + const broadcaster = mock(); + const userRepository = mock(); + const orchestrator = mock(); + const logger = mock(); + logger.scoped.mockReturnValue(logger); + + threadRepository.findOneBy.mockResolvedValue(thread); + userRepository.findByIdWithRole.mockResolvedValue(user as never); + executionRepository.findOldestQueuedByThread.mockResolvedValue(null); + + const service = new AgentChatQueueService( + logger, + executionRepository, + threadRepository, + executionService, + attachmentService, + broadcaster, + userRepository, + orchestrator, + ); + return { + service, + executionRepository, + executionService, + attachmentService, + broadcaster, + userRepository, + orchestrator, + }; +} + +describe('AgentChatQueueService', () => { + beforeEach(() => { + vi.mocked(userHasScopes).mockResolvedValue(true); + }); + + describe('enqueue', () => { + it('rejects the message with 409 and stores nothing once the thread has its maximum of waiting messages', async () => { + const { service, executionRepository, executionService } = setup(); + executionRepository.countQueuedByThread.mockResolvedValue(MAX_AGENT_THREAD_WAITERS); + + await expect( + service.enqueue({ + agentId: thread.agentId, + agentName: 'Agent', + projectId: thread.projectId, + threadId: thread.id, + message: 'one too many', + resourceId: ownRef.resourceId, + }), + ).rejects.toBeInstanceOf(ConflictError); + + expect(executionService.recordQueuedExecution).not.toHaveBeenCalled(); + }); + }); + + describe('editQueued and removeQueued', () => { + it.each([ + ['a row of another sender', {}, { resourceId: 'draft-chat:user-2' }, NotFoundError], + ['a row through another agent', {}, { agentId: 'agent-2' }, NotFoundError], + ['a row that already started', { status: 'running' as const }, {}, ConflictError], + ])('reject %s and change nothing', async (_, rowOverrides, refOverrides, errorClass) => { + const { service, executionRepository, attachmentService, broadcaster } = setup(); + executionRepository.findUserTurnById.mockResolvedValue( + queuedRow({ attachments: [{ id: 'att-1' }] as never, ...rowOverrides }), + ); + const ref = { ...ownRef, ...refOverrides }; + + await expect(service.editQueued(ref, 'changed')).rejects.toBeInstanceOf(errorClass); + await expect(service.removeQueued(ref)).rejects.toBeInstanceOf(errorClass); + + expect(executionRepository.updateQueuedMessage).not.toHaveBeenCalled(); + expect(executionRepository.deleteQueued).not.toHaveBeenCalled(); + expect(attachmentService.deleteByIds).not.toHaveBeenCalled(); + expect(broadcaster.notify).not.toHaveBeenCalled(); + }); + + it('answers 409 when the row starts between the ownership check and the write', async () => { + const { service, executionRepository, broadcaster } = setup(); + executionRepository.findUserTurnById.mockResolvedValue(queuedRow()); + executionRepository.updateQueuedMessage.mockResolvedValue(false); + executionRepository.deleteQueued.mockResolvedValue(false); + + await expect(service.editQueued(ownRef, 'changed')).rejects.toBeInstanceOf(ConflictError); + await expect(service.removeQueued(ownRef)).rejects.toBeInstanceOf(ConflictError); + expect(broadcaster.notify).not.toHaveBeenCalled(); + }); + + it('removes the row with its attachments and tells every tab', async () => { + const { service, executionRepository, attachmentService, broadcaster } = setup(); + executionRepository.findUserTurnById.mockResolvedValue( + queuedRow({ attachments: [{ id: 'att-1' }, { id: 'att-2' }] as never }), + ); + executionRepository.deleteQueued.mockResolvedValue(true); + + await service.removeQueued(ownRef); + + expect(attachmentService.deleteByIds).toHaveBeenCalledWith(['att-1', 'att-2']); + expect(broadcaster.notify).toHaveBeenCalledWith({ + projectId: thread.projectId, + agentId: thread.agentId, + threadId: thread.id, + executionId: 'exec-1', + }); + }); + }); + + describe('requestDrain', () => { + it('runs the waiting rows oldest first, skips the ones that left the queue and stops at a deferred one', async () => { + const { service, executionRepository, orchestrator } = setup(); + const rows = [ + queuedRow({ id: 'exec-1' }), + queuedRow({ id: 'exec-2' }), + queuedRow({ id: 'exec-3' }), + queuedRow({ id: 'exec-4' }), + ]; + const outcomes: Record = { + 'exec-1': 'ran', + 'exec-2': 'skipped', + 'exec-3': 'deferred', + }; + executionRepository.findOldestQueuedByThread.mockImplementation(async () => rows[0] ?? null); + orchestrator.executeForQueued.mockImplementation(async ({ executionId }) => { + const outcome = outcomes[executionId]; + // A deferred row stays queued; a drain that keeps retrying it must fail loudly, not spin. + if (outcome !== 'deferred') rows.shift(); + if (orchestrator.executeForQueued.mock.calls.length > 3) { + throw new Error('the drain did not stop at the deferred row'); + } + return outcome; + }); + + service.requestDrain(thread.id); + + await vi.waitFor(() => expect(orchestrator.executeForQueued).toHaveBeenCalledTimes(3)); + await new Promise((resolve) => setImmediate(resolve)); + expect( + orchestrator.executeForQueued.mock.calls.map(([config]) => config.executionId), + ).toEqual(['exec-1', 'exec-2', 'exec-3']); + expect(orchestrator.executeForQueued).toHaveBeenCalledWith( + expect.objectContaining({ + agentId: thread.agentId, + projectId: thread.projectId, + threadId: thread.id, + user, + resourceId: `draft-chat:${user.id}`, + }), + ); + expect(rows.map((row) => row.id)).toEqual(['exec-3', 'exec-4']); + }); + + it('ends a row whose sender can no longer run the agent instead of retrying it', async () => { + const { service, executionRepository, executionService, orchestrator, userRepository } = + setup(); + const rows = [ + queuedRow({ id: 'exec-gone', resourceId: 'draft-chat:user-gone' }), + queuedRow({ id: 'exec-2' }), + ]; + executionRepository.findOldestQueuedByThread.mockImplementation(async () => rows[0] ?? null); + userRepository.findByIdWithRole.mockImplementation(async (id) => + id === user.id ? (user as never) : null, + ); + executionService.failQueuedExecution.mockImplementation(async () => { + rows.shift(); + }); + orchestrator.executeForQueued.mockImplementation(async () => { + rows.shift(); + return 'ran'; + }); + + service.requestDrain(thread.id); + + await vi.waitFor(() => expect(orchestrator.executeForQueued).toHaveBeenCalledTimes(1)); + expect(executionService.failQueuedExecution).toHaveBeenCalledWith( + { + id: 'exec-gone', + threadId: thread.id, + agentId: thread.agentId, + projectId: thread.projectId, + }, + expect.objectContaining({ + message: 'The user who sent this message is no longer active', + }), + ); + expect(orchestrator.executeForQueued).toHaveBeenCalledWith( + expect.objectContaining({ executionId: 'exec-2' }), + ); + }); + }); +}); 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 6bf04e8b655..4aac85be405 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 @@ -5,9 +5,11 @@ import { mock } from 'vitest-mock-extended'; import { FileNotFoundError } from 'n8n-core'; import type { CredentialsService } from '@/credentials/credentials.service'; +import { ConflictError } from '@/errors/response-errors/conflict.error'; import { NotFoundError } from '@/errors/response-errors/not-found.error'; import type { AgentChatAttachmentService } from '../agent-chat-attachment.service'; +import type { AgentChatQueueService } from '../agent-chat-queue.service'; import { AgentChatController } from '../agent-chat.controller'; import type { AgentExecutionOrchestratorService } from '../agent-execution-orchestrator.service'; import type { FlushableResponse } from '../agent-sse-stream'; @@ -31,6 +33,7 @@ function makeController() { const agentsBuilderService = mock(); const agentTestRunService = mock(); const agentChatAttachmentService = mock(); + const agentChatQueueService = mock(); agentTestRunService.prepareDraftRun.mockResolvedValue({ status: 'ready', sessionId: 'thread-1', @@ -53,6 +56,7 @@ function makeController() { mock(), agentsService as unknown as AgentsService, agentChatAttachmentService, + agentChatQueueService, ); return { @@ -60,6 +64,7 @@ function makeController() { agentExecutionOrchestratorService, agentTestRunService, agentChatAttachmentService, + agentChatQueueService, agentsService: { findById: agentsService.findById, getConversationHistory: agentExecutionOrchestratorService.getConversationHistory, @@ -104,6 +109,9 @@ describe('AgentChatController route access scopes', () => { ['chat', 'agent:execute'], ['chatResume', 'agent:execute'], ['cancelChatRun', 'agent:execute'], + ['queueChatMessage', 'agent:execute'], + ['editQueuedChatMessage', 'agent:execute'], + ['removeQueuedChatMessage', 'agent:execute'], ['getChatMessages', 'agent:read'], ['getTestChatMessages', 'agent:read'], ['clearTestChatMessages', 'agent:update'], @@ -359,6 +367,74 @@ describe('AgentChatController full thread queue', () => { }); }); +describe('AgentChatController queued messages', () => { + const req = { params: { projectId: 'project-1' }, user: { id: 'user-1' } } as never; + + it('stores the message for the session as the sender and returns the queued row id', async () => { + const { controller, agentsService, agentChatQueueService } = makeController(); + agentsService.findById.mockResolvedValue({ id: 'agent-1', name: 'Agent' } as never); + agentChatQueueService.enqueue.mockResolvedValue('queued-1'); + + await expect( + controller.queueChatMessage(req, {} as never, 'agent-1', { + message: 'later', + sessionId: 'thread-1', + } as never), + ).resolves.toEqual({ executionId: 'queued-1' }); + + expect(agentChatQueueService.enqueue).toHaveBeenCalledWith({ + agentId: 'agent-1', + agentName: 'Agent', + projectId: 'project-1', + threadId: 'thread-1', + message: 'later', + attachments: undefined, + resourceId: 'draft-chat:user-1', + }); + }); + + it('answers 404 for a session that is not the sender’s and stores nothing', async () => { + const { controller, agentsService, agentTestRunService, agentChatQueueService } = + makeController(); + agentsService.findById.mockResolvedValue({ id: 'agent-1', name: 'Agent' } as never); + agentTestRunService.prepareDraftRun.mockResolvedValue({ status: 'session_not_found' }); + + await expect( + controller.queueChatMessage(req, {} as never, 'agent-1', { + message: 'later', + sessionId: 'someone-elses-thread', + } as never), + ).rejects.toBeInstanceOf(NotFoundError); + expect(agentChatQueueService.enqueue).not.toHaveBeenCalled(); + }); + + it('drops the stored attachments when the queue rejects the message', async () => { + const { controller, agentsService, agentChatQueueService, agentChatAttachmentService } = + makeController(); + agentsService.findById.mockResolvedValue({ id: 'agent-1', name: 'Agent' } as never); + agentChatAttachmentService.storeInbound.mockResolvedValue({ id: 'att-1' } as never); + agentChatAttachmentService.deleteByIds.mockResolvedValue(undefined); + const full = new ConflictError(new AgentThreadQueueFullError().message); + agentChatQueueService.enqueue.mockRejectedValue(full); + + await expect( + controller.queueChatMessage(req, {} as never, 'agent-1', { + message: 'one too many', + sessionId: 'thread-1', + attachments: [ + { + fileName: 'notes.txt', + mimeType: 'text/plain', + data: Buffer.from('hi').toString('base64'), + }, + ], + } as never), + ).rejects.toBe(full); + + expect(agentChatAttachmentService.deleteByIds).toHaveBeenCalledWith(['att-1']); + }); +}); + describe('AgentChatController attachment cleanup on failed turns', () => { const textAttachment = (fileName: string) => ({ fileName, diff --git a/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts index e3c929cd851..7279eecd52e 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts @@ -16,6 +16,7 @@ import { mock } from 'vitest-mock-extended'; import type { ExternalHooks } from '@/external-hooks'; import type { Telemetry } from '@/telemetry'; +import { AgentChatQueueService } from '../agent-chat-queue.service'; import { AgentExecutionOrchestratorService } from '../agent-execution-orchestrator.service'; import type { AgentExecutionService } from '../agent-execution.service'; import { AgentThreadClaimConflictError } from '../repositories/agent-execution.repository'; @@ -153,6 +154,8 @@ function makeService(sandboxEnabled = false) { }); const wakeService = mock(); Container.set(AgentWakeService, wakeService); + const chatQueueService = mock(); + Container.set(AgentChatQueueService, chatQueueService); const turnCoordinator = createTestTurnCoordinator(); executionService.startExecutionRecording.mockResolvedValue('execution-1'); @@ -190,6 +193,7 @@ function makeService(sandboxEnabled = false) { agentSandboxRuntimeService, agentRepository, wakeService, + chatQueueService, chatIntegrationService, bridge, turnCoordinator, @@ -868,6 +872,11 @@ describe('AgentExecutionOrchestratorService', () => { ); expect(first.wakeService.onParentTurnFinished).toHaveBeenCalledWith('thread-1'); + // Messages queued during the turn run before the job results. + expect(first.chatQueueService.requestDrain).toHaveBeenCalledWith('thread-1'); + expect(first.chatQueueService.requestDrain.mock.invocationCallOrder[0]).toBeLessThan( + first.wakeService.onParentTurnFinished.mock.invocationCallOrder[0], + ); const failing = makeService(); failing.runtimeCacheService.getRuntime.mockResolvedValue(makeRuntime()); @@ -964,6 +973,109 @@ describe('AgentExecutionOrchestratorService', () => { expect(runtime.agent.stream).not.toHaveBeenCalled(); }); + it('runs a queued message as its own turn on the promoted row with the text read at pickup', async () => { + const { service, runtimeCacheService, executionService, turnCoordinator } = makeService(); + const runtime = makeRuntime([ + { type: 'text-start', id: 'text-1' }, + { type: 'text-delta', id: 'text-1', delta: 'Answer to the queued question.' }, + { type: 'finish', finishReason: 'stop' }, + ]); + runtimeCacheService.getRuntime.mockResolvedValue(runtime); + executionService.isQueued.mockResolvedValue(true); + executionService.claimQueuedExecution.mockResolvedValue({ + executionId: 'queued-1', + claimLost: new AbortController().signal, + userMessage: 'edited before pickup', + attachments: null, + }); + + // The running turn holds the thread; the queued message waits for it. + const runningTurn = await turnCoordinator.coordinator.acquire('thread-1'); + const queued = service.executeForQueued({ + agentId, + projectId, + executionId: 'queued-1', + threadId: 'thread-1', + user, + resourceId: 'draft-chat:user-1', + }); + await new Promise((resolve) => setImmediate(resolve)); + expect(runtime.agent.stream).not.toHaveBeenCalled(); + runningTurn.release(); + + await expect(queued).resolves.toBe('ran'); + expect(executionService.claimQueuedExecution).toHaveBeenCalledWith( + { id: 'queued-1', threadId: 'thread-1', agentId, projectId }, + expect.any(Date), + ); + expect(executionService.startClaimedExecutionRecording).not.toHaveBeenCalled(); + expect(runtimeCacheService.getRuntime).toHaveBeenCalledWith( + expect.objectContaining({ user, previewChat: true }), + ); + expect(runtime.agent.stream).toHaveBeenCalledWith( + 'edited before pickup', + expect.objectContaining({ + persistence: expect.objectContaining({ resourceId: 'draft-chat:user-1' }), + }), + ); + expect(executionService.finalizeExecution).toHaveBeenCalledWith( + 'queued-1', + expect.objectContaining({ + userMessage: 'edited before pickup', + record: expect.objectContaining({ assistantResponse: 'Answer to the queued question.' }), + }), + ); + expect(runtimeCacheService.releaseRuntimeLease).toHaveBeenCalledWith(runtime.agent); + }); + + it('leaves a queued message alone while the thread awaits a human response and skips one that left the queue', async () => { + const { service, runtimeCacheService, executionService, checkpointStorage } = makeService(); + const config = { + agentId, + projectId, + executionId: 'queued-1', + threadId: 'thread-1', + user, + resourceId: 'draft-chat:user-1', + }; + + executionService.hasSuspendedRun.mockResolvedValue(true); + checkpointStorage.findSuspendedForThread.mockResolvedValue(makeCheckpoint()); + await expect(service.executeForQueued(config)).resolves.toBe('deferred'); + + executionService.hasSuspendedRun.mockResolvedValue(false); + executionService.isQueued.mockResolvedValue(false); + await expect(service.executeForQueued(config)).resolves.toBe('skipped'); + + expect(runtimeCacheService.getRuntime).not.toHaveBeenCalled(); + expect(executionService.claimQueuedExecution).not.toHaveBeenCalled(); + expect(executionService.failQueuedExecution).not.toHaveBeenCalled(); + }); + + it('ends a queued message whose runtime cannot be built and moves on', async () => { + const { service, runtimeCacheService, executionService } = makeService(); + const cause = new Error('credential gone'); + executionService.isQueued.mockResolvedValue(true); + runtimeCacheService.getRuntime.mockRejectedValue(cause); + + await expect( + service.executeForQueued({ + agentId, + projectId, + executionId: 'queued-1', + threadId: 'thread-1', + user, + resourceId: 'draft-chat:user-1', + }), + ).resolves.toBe('skipped'); + + expect(executionService.failQueuedExecution).toHaveBeenCalledWith( + { id: 'queued-1', threadId: 'thread-1', agentId, projectId }, + cause, + ); + expect(executionService.claimQueuedExecution).not.toHaveBeenCalled(); + }); + it.each(['draft', 'published'] as const)( 'records a failed %s wake without delivering its output', async (type) => { diff --git a/packages/cli/src/modules/agents/__tests__/agent-interrupted-execution-sweeper.test.ts b/packages/cli/src/modules/agents/__tests__/agent-interrupted-execution-sweeper.test.ts index b9036d8773b..f9f7a7a780e 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-interrupted-execution-sweeper.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-interrupted-execution-sweeper.test.ts @@ -2,6 +2,7 @@ import { mockLogger } from '@n8n/backend-test-utils'; import type { AgentsConfig } from '@n8n/config'; import { mock } from 'vitest-mock-extended'; +import type { AgentChatQueueService } from '../agent-chat-queue.service'; import type { AgentExecutionService } from '../agent-execution.service'; import { AgentInterruptedExecutionSweeper } from '../agent-interrupted-execution-sweeper'; import type { AgentBackgroundJobService } from '../background/agent-background-job.service'; @@ -24,6 +25,7 @@ function setup(options: { backgroundTasksEnabled?: boolean } = {}) { backgroundJobService, agentWakeService, agentsConfig, + mock(), ); return { sweeper, repository, executionService, backgroundJobService, agentWakeService }; } diff --git a/packages/cli/src/modules/agents/__tests__/execution-to-message-mapper.test.ts b/packages/cli/src/modules/agents/__tests__/execution-to-message-mapper.test.ts index 1006a925ce8..5c37de241ae 100644 --- a/packages/cli/src/modules/agents/__tests__/execution-to-message-mapper.test.ts +++ b/packages/cli/src/modules/agents/__tests__/execution-to-message-mapper.test.ts @@ -30,6 +30,22 @@ describe('execution-to-message-mapper', () => { }); }); + it('marks a queued turn on its user message and gives it no assistant message', () => { + const result = executionsToMessagesDto([ + execution({ status: 'queued', userMessage: 'Later', timeline: null, error: null }), + ]); + + expect(result).toEqual([ + { + id: 'execution-1:user', + role: 'user', + content: [{ type: 'text', text: 'Later' }], + executionId: 'execution-1', + executionStatus: 'queued', + }, + ]); + }); + it('carries the integration author on the user message and omits it when absent', () => { const author = { id: 'U1', name: 'alice' }; diff --git a/packages/cli/src/modules/agents/agent-chat-queue.service.ts b/packages/cli/src/modules/agents/agent-chat-queue.service.ts new file mode 100644 index 00000000000..e379f49b63c --- /dev/null +++ b/packages/cli/src/modules/agents/agent-chat-queue.service.ts @@ -0,0 +1,217 @@ +import { Logger } from '@n8n/backend-common'; +import { type User, UserRepository } from '@n8n/db'; +import { Service } from '@n8n/di'; +import { OperationalError } from 'n8n-workflow'; + +import { ConflictError } from '@/errors/response-errors/conflict.error'; +import { NotFoundError } from '@/errors/response-errors/not-found.error'; +import { userHasScopes } from '@/permissions.ee/check-access'; + +import { + AgentChatAttachmentService, + type StoredAttachmentRef, +} from './agent-chat-attachment.service'; +import { AgentExecutionOrchestratorService } from './agent-execution-orchestrator.service'; +import { AgentExecutionUpdateBroadcaster } from './agent-execution-update-broadcaster'; +import { AgentExecutionService, threadBelongsTo } from './agent-execution.service'; +import { + AgentThreadQueueFullError, + MAX_AGENT_THREAD_WAITERS, +} from './agent-thread-turn-coordinator'; +import type { AgentExecutionThread } from './entities/agent-execution-thread.entity'; +import { AgentExecutionThreadRepository } from './repositories/agent-execution-thread.repository'; +import { + AgentExecutionRepository, + type AgentExecutionUserTurn, +} from './repositories/agent-execution.repository'; +import { + draftChatMemoryResourceId, + userIdFromDraftChatMemoryResourceId, +} from './utils/agent-memory-scope'; + +export interface EnqueueChatMessageParams { + agentId: string; + agentName: string; + projectId: string; + threadId: string; + message: string; + attachments?: StoredAttachmentRef[]; + /** Memory resource id of the sender (`draft-chat:`). */ + resourceId: string; +} + +/** Scopes a queued row to the sender and to the agent named in the request. */ +export interface QueuedMessageRef { + executionId: string; + agentId: string; + projectId: string; + resourceId: string; +} + +/** + * Preview-chat messages sent while the thread's turn runs. Each waits as a + * `queued` execution row and runs as its own turn, oldest first, once the + * thread is idle. Every tab sees the rows through the usual execution update + * push, and the sender can edit or remove them until the drain picks them up. + */ +@Service() +export class AgentChatQueueService { + /** Threads with a drain in this process; `again` reruns it once the current pass ends. */ + private readonly drains = new Map(); + + constructor( + private readonly logger: Logger, + private readonly executionRepository: AgentExecutionRepository, + private readonly threadRepository: AgentExecutionThreadRepository, + private readonly executionService: AgentExecutionService, + private readonly attachmentService: AgentChatAttachmentService, + private readonly broadcaster: AgentExecutionUpdateBroadcaster, + private readonly userRepository: UserRepository, + private readonly orchestrator: AgentExecutionOrchestratorService, + ) { + this.logger = this.logger.scoped('agents'); + } + + /** Store a message for the thread. Answers 409 when the thread has its maximum of waiting messages. */ + async enqueue(params: EnqueueChatMessageParams): Promise { + const { message, ...scope } = params; + const waiting = await this.executionRepository.countQueuedByThread(scope.threadId); + if (waiting >= MAX_AGENT_THREAD_WAITERS) { + throw new ConflictError(new AgentThreadQueueFullError().message); + } + const executionId = await this.executionService.recordQueuedExecution({ + ...scope, + userMessage: message, + }); + this.requestDrain(scope.threadId); + return executionId; + } + + /** Replace the text of a waiting message. 404 for an unknown or foreign row, 409 once it started. */ + async editQueued(ref: QueuedMessageRef, message: string): Promise { + const { row, thread } = await this.findOwnQueued(ref); + if (!(await this.executionRepository.updateQueuedMessage(row.id, message))) { + throw new ConflictError('This message has already started'); + } + this.notify(thread, row.id); + } + + /** Remove a waiting message and its attachments. 404 for an unknown or foreign row, 409 once it started. */ + async removeQueued(ref: QueuedMessageRef): Promise { + const { row, thread } = await this.findOwnQueued(ref); + if (!(await this.executionRepository.deleteQueued(row.id))) { + throw new ConflictError('This message has already started'); + } + if (row.attachments?.length) { + await this.attachmentService.deleteByIds(row.attachments.map(({ id }) => id)); + } + this.notify(thread, row.id); + } + + /** Run the thread's waiting messages in the background, oldest first. Cheap to call often. */ + requestDrain(threadId: string): void { + const active = this.drains.get(threadId); + if (active) { + active.again = true; + return; + } + const state = { again: true }; + this.drains.set(threadId, state); + void (async () => { + while (state.again) { + state.again = false; + await this.drain(threadId); + } + })() + .catch((error: unknown) => { + this.logger.warn('Failed to run queued agent chat messages', { threadId, error }); + }) + .finally(() => this.drains.delete(threadId)); + } + + /** Restart the drains of every thread with waiting messages, e.g. after a main restarted. */ + async drainAll(): Promise { + for (const threadId of await this.executionRepository.findThreadIdsWithQueued()) { + this.requestDrain(threadId); + } + } + + private async drain(threadId: string): Promise { + const thread = await this.threadRepository.findOneBy({ id: threadId }); + if (!thread) return; + for (;;) { + const row = await this.executionRepository.findOldestQueuedByThread(threadId); + if (!row) return; + if ((await this.runQueued(row, thread)) === 'deferred') return; + } + } + + private async runQueued( + row: AgentExecutionUserTurn, + thread: AgentExecutionThread, + ): Promise<'ran' | 'skipped' | 'deferred'> { + let user: User; + try { + user = await this.resolveSender(row.resourceId, thread.projectId); + } catch (error) { + // Without a runnable sender the row can never start; end it so the drain moves on. + await this.executionService.failQueuedExecution( + { + id: row.id, + threadId: row.threadId, + agentId: thread.agentId, + projectId: thread.projectId, + }, + error, + ); + return 'skipped'; + } + return await this.orchestrator.executeForQueued({ + agentId: thread.agentId, + projectId: thread.projectId, + executionId: row.id, + threadId: row.threadId, + user, + resourceId: draftChatMemoryResourceId(user.id), + }); + } + + /** The sender must still be able to run the agent; the drain has no request that checked that. */ + private async resolveSender(resourceId: string | null, projectId: string): Promise { + const userId = resourceId ? userIdFromDraftChatMemoryResourceId(resourceId) : undefined; + const user = userId ? await this.userRepository.findByIdWithRole(userId) : null; + if (!user || user.disabled) { + throw new OperationalError('The user who sent this message is no longer active'); + } + if (!(await userHasScopes(user, ['agent:execute'], false, { projectId }))) { + throw new OperationalError('The user who sent this message can no longer run this agent'); + } + return user; + } + + private async findOwnQueued( + ref: QueuedMessageRef, + ): Promise<{ row: AgentExecutionUserTurn; thread: AgentExecutionThread }> { + const row = await this.executionRepository.findUserTurnById(ref.executionId); + const thread = row ? await this.threadRepository.findOneBy({ id: row.threadId }) : null; + if ( + !row || + !thread || + row.resourceId !== ref.resourceId || + !threadBelongsTo(thread, ref.projectId, ref.agentId) + ) { + throw new NotFoundError(`Queued message "${ref.executionId}" not found`); + } + if (row.status !== 'queued') throw new ConflictError('This message has already started'); + return { row, thread }; + } + + private notify(thread: AgentExecutionThread, executionId: string): void { + this.broadcaster.notify({ + projectId: thread.projectId, + agentId: thread.agentId, + threadId: thread.id, + executionId, + }); + } +} diff --git a/packages/cli/src/modules/agents/agent-chat.controller.ts b/packages/cli/src/modules/agents/agent-chat.controller.ts index 41aadec8b49..ccbbaae5f7e 100644 --- a/packages/cli/src/modules/agents/agent-chat.controller.ts +++ b/packages/cli/src/modules/agents/agent-chat.controller.ts @@ -1,7 +1,9 @@ import { type AgentChatAttachmentPayload, + AgentChatEditQueuedMessageDto, AgentChatMessageDto, type AgentChatMessagesResponse, + AgentChatQueueMessageDto, AgentChatResumeDto, type AgentSseEvent, MAX_AGENT_CHAT_ATTACHMENT_SIZE_BYTES, @@ -10,7 +12,16 @@ import { 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 { sanitizeFilename } from '@n8n/utils/files/sanitize-filename'; import type { Response } from 'express'; import { FileNotFoundError, getHtmlSandboxCSP } from 'n8n-core'; @@ -25,6 +36,7 @@ import { AgentChatAttachmentService, type StoredAttachmentRef, } from './agent-chat-attachment.service'; +import { AgentChatQueueService } from './agent-chat-queue.service'; import { AgentExecutionOrchestratorService } from './agent-execution-orchestrator.service'; import { messagesToDto } from './agent-message-mapper'; import { AgentThreadQueueFullError } from './agent-thread-turn-coordinator'; @@ -55,6 +67,7 @@ export class AgentChatController { private readonly credentialsService: CredentialsService, private readonly agentsService: AgentsService, private readonly agentChatAttachmentService: AgentChatAttachmentService, + private readonly agentChatQueueService: AgentChatQueueService, ) {} /** Decode, sniff, and persist inbound chat attachments; returns refs for the user turn. */ @@ -247,6 +260,105 @@ export class AgentChatController { } } + /** Store a message that runs after the thread's running turn ends. */ + @Post('/:agentId/chat/queue') + @ProjectScope('agent:execute') + async queueChatMessage( + req: AuthenticatedRequest<{ projectId: string }>, + _res: Response, + @Param('agentId') agentId: string, + @Body payload: AgentChatQueueMessageDto, + ) { + const { projectId } = req.params; + const { message, sessionId, attachments } = payload; + const agent = await this.agentsService.findById(agentId, projectId); + if (!agent) throw new NotFoundError(`Agent "${agentId}" not found`); + + const prepared = await this.agentTestRunService.prepareDraftRun({ + agentId, + projectId, + sessionId, + credentialProvider: new AgentsCredentialProvider( + this.credentialsService, + projectId, + req.user, + ), + }); + if (prepared.status === 'session_not_found') { + throw new NotFoundError(`Session "${sessionId}" not found`); + } + if (prepared.status === 'agent_misconfigured') { + throw new BadRequestError('This agent is not ready to run yet.'); + } + + const resourceId = draftChatMemoryResourceId(req.user.id); + const storedAttachments = await this.storeChatAttachments({ + attachments, + agentId, + projectId, + threadId: sessionId, + resourceId, + }); + try { + const executionId = await this.agentChatQueueService.enqueue({ + agentId, + agentName: agent.name, + projectId, + threadId: sessionId, + message, + attachments: storedAttachments, + resourceId, + }); + return { executionId }; + } catch (error) { + // A rejected message owns nothing; drop the attachments it brought. + if (storedAttachments?.length) { + await this.agentChatAttachmentService + .deleteByIds(storedAttachments.map((ref) => ref.id)) + .catch(() => {}); + } + throw error; + } + } + + @Patch('/:agentId/chat/queue/:executionId') + @ProjectScope('agent:execute') + async editQueuedChatMessage( + req: AuthenticatedRequest<{ projectId: string }>, + _res: Response, + @Param('agentId') agentId: string, + @Param('executionId') executionId: string, + @Body payload: AgentChatEditQueuedMessageDto, + ) { + await this.agentChatQueueService.editQueued( + { + executionId, + agentId, + projectId: req.params.projectId, + resourceId: draftChatMemoryResourceId(req.user.id), + }, + payload.message, + ); + return { executionId }; + } + + @Delete('/:agentId/chat/queue/:executionId') + @ProjectScope('agent:execute') + async removeQueuedChatMessage( + req: AuthenticatedRequest<{ projectId: string }>, + _res: Response, + @Param('agentId') agentId: string, + @Param('executionId') executionId: string, + ) { + await this.agentChatQueueService.removeQueued({ + executionId, + agentId, + projectId: req.params.projectId, + resourceId: draftChatMemoryResourceId(req.user.id), + }); + return { removed: true }; + } + @Delete('/:agentId/chat/runs/:runId') @ProjectScope('agent:execute') async cancelChatRun( diff --git a/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts b/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts index 5d035482011..0955a7df6e8 100644 --- a/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts +++ b/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts @@ -205,6 +205,17 @@ export interface ExecuteForWakeConfig { }; } +export interface ExecuteForQueuedConfig { + agentId: string; + projectId: string; + /** The queued execution row to run. */ + executionId: string; + threadId: string; + /** The sender, resolved from the row's `resourceId`. */ + user: User; + resourceId: string; +} + export interface StreamChatResponseConfig { agentInstance: RuntimeAgent; toolRegistry: ToolRegistry; @@ -238,6 +249,8 @@ export interface StreamChatResponseConfig { isWakeRun?: boolean; /** The admitted turn for `memory.threadId`; the run claims the thread under it. */ permit: AgentThreadTurnPermit; + /** A queued row the caller already promoted to the claimed running row; skips the insert. */ + claimedExecution?: { executionId: string; abortSignal: AbortSignal }; } function withApprovalToolDetails(chunk: StreamChunk, toolRegistry: ToolRegistry): StreamChunk { @@ -610,8 +623,12 @@ export class AgentExecutionOrchestratorService { }, }, }); - // After the resumed turn, request any job results that arrived during the approval wait. - if (!recorder.suspended) await this.requestPendingBackgroundWake(threadId); + // After the resumed turn, run the messages queued during the approval wait, + // then request any job results that arrived meanwhile. + if (!recorder.suspended) { + await this.requestQueueDrain(threadId); + await this.requestPendingBackgroundWake(threadId); + } } finally { this.runtimeCacheService.releaseRuntimeLease(agentInstance); } @@ -645,35 +662,15 @@ export class AgentExecutionOrchestratorService { yield* this.streamAdmittedTurn( memory.threadId, abortSignal, - async () => { - const runtime = await this.runtimeCacheService.getRuntime({ + async () => + await this.getDraftChatRuntime({ agentId, projectId, - integrationType: N8N_CHAT_INTEGRATION_TYPE, user, sandboxPrincipalHash, + memory, previewChat, - }); - try { - // Message context is written by the admitted turn only, so a - // waiting message never redirects the running turn's replies. - await this.integrationMessageContextService.setLatest( - memory.threadId, - memory.resourceId, - { - integrationConnectionId: N8N_CHAT_INTEGRATION_TYPE, - platform: N8N_CHAT_INTEGRATION_TYPE, - target: { type: 'dm', userId: user.id, threadId: memory.threadId }, - interactingUserId: user.id, - updatedAt: new Date().toISOString(), - }, - ); - return runtime; - } catch (error) { - this.runtimeCacheService.releaseRuntimeLease(runtime.agent); - throw error; - } - }, + }), (runtime, permit) => ({ agentInstance: runtime.agent, toolRegistry: runtime.toolRegistry, @@ -697,6 +694,129 @@ export class AgentExecutionOrchestratorService { ); } + /** Build the draft runtime for an n8n user and point the in-app chat's message context at them. */ + private async getDraftChatRuntime(params: { + agentId: string; + projectId: string; + user: User; + sandboxPrincipalHash: AgentSandboxPrincipalHash; + memory: AgentMemoryScope; + previewChat?: boolean; + }): Promise { + const { agentId, projectId, user, sandboxPrincipalHash, memory, previewChat } = params; + const runtime = await this.runtimeCacheService.getRuntime({ + agentId, + projectId, + integrationType: N8N_CHAT_INTEGRATION_TYPE, + user, + sandboxPrincipalHash, + previewChat, + }); + try { + // Message context is written by the admitted turn only, so a + // waiting message never redirects the running turn's replies. + await this.integrationMessageContextService.setLatest(memory.threadId, memory.resourceId, { + integrationConnectionId: N8N_CHAT_INTEGRATION_TYPE, + platform: N8N_CHAT_INTEGRATION_TYPE, + target: { type: 'dm', userId: user.id, threadId: memory.threadId }, + interactingUserId: user.id, + updatedAt: new Date().toISOString(), + }); + return runtime; + } catch (error) { + this.runtimeCacheService.releaseRuntimeLease(runtime.agent); + throw error; + } + } + + /** + * Run one queued preview-chat message as the thread's next turn, after any + * running turn. `skipped`: the row left `queued` before pickup (removed, or + * taken by another main) or cannot start, so the drain moves on. `deferred`: + * the thread waits for a human response; the row stays queued and the drain + * stops. Progress reaches every tab through the timeline snapshot pushes. + */ + async executeForQueued(config: ExecuteForQueuedConfig): Promise<'ran' | 'skipped' | 'deferred'> { + const { agentId, projectId, executionId, threadId, user, resourceId } = config; + const memory = { threadId, resourceId }; + const ref = { id: executionId, threadId, agentId, projectId }; + const sandboxPrincipalHash = hashAgentSandboxPrincipal({ type: 'n8n-user', userId: user.id }); + + return await this.turnCoordinator.run(threadId, undefined, async (permit) => { + if (await this.awaitsHumanResponse(agentId, threadId)) return 'deferred'; + if (!(await this.agentExecutionService.isQueued(executionId))) return 'skipped'; + + let runtime: AgentRuntime; + try { + runtime = await this.getDraftChatRuntime({ + agentId, + projectId, + user, + sandboxPrincipalHash, + memory, + previewChat: true, + }); + } catch (error) { + await this.agentExecutionService.failQueuedExecution(ref, error); + return 'skipped'; + } + try { + // The claim makes the row final: later edits and removals answer 409, + // so the text read with it is what the model receives. + const startedAt = new Date(); + const claimed = await this.retryWhileThreadClaimed( + threadId, + agentId, + undefined, + async () => await this.agentExecutionService.claimQueuedExecution(ref, startedAt), + ); + if (!claimed) return 'skipped'; + const stream = this.streamChatResponse({ + agentInstance: runtime.agent, + toolRegistry: runtime.toolRegistry, + agentId, + userId: user.id, + message: claimed.userMessage ?? '', + attachments: claimed.attachments ?? undefined, + memory, + projectId: runtime.projectId, + telemetry: { runType: 'test', configuration: runtime.telemetryConfiguration }, + includeHitlToolDetails: true, + sandboxPrincipalHash, + claimedExecution: { executionId, abortSignal: claimed.claimLost }, + permit, + }); + try { + for await (const _chunk of stream) { + // No client listens; the recorded run is the result. + } + } catch (error) { + // The row already carries the failure; the next message still runs. + this.logger.warn('Queued agent chat message failed', { + agentId, + threadId, + executionId, + error: error instanceof Error ? error.message : String(error), + }); + } + return 'ran'; + } finally { + this.runtimeCacheService.releaseRuntimeLease(runtime.agent); + } + }); + } + + /** + * A thread with an open checkpoint belongs to its pending resume. Execution + * rows keep `suspended` after a resume, so the checkpoint store decides. + */ + private async awaitsHumanResponse(agentId: string, threadId: string): Promise { + return ( + (await this.agentExecutionService.hasSuspendedRun(threadId)) && + (await this.n8nCheckpointStorage.findSuspendedForThread(agentId, threadId)) !== null + ); + } + /** Admit one turn on `threadId`, then build its runtime and stream the turn under the permit. */ private async *streamAdmittedTurn( threadId: string, @@ -902,15 +1022,8 @@ export class AgentExecutionOrchestratorService { const integrationType = isDraft ? N8N_CHAT_INTEGRATION_TYPE : identity.integrationType; return await this.turnCoordinator.run(memory.threadId, abortSignal, async (permit) => { - // A suspended thread belongs to its pending resume, which may have - // started while this wake waited. Execution rows keep `suspended` - // after a resume, so the checkpoint store decides. - if ( - (await this.agentExecutionService.hasSuspendedRun(memory.threadId)) && - (await this.n8nCheckpointStorage.findSuspendedForThread(agentId, memory.threadId)) !== null - ) { - return 'skipped'; - } + // The pending resume may have started while this wake waited. + if (await this.awaitsHumanResponse(agentId, memory.threadId)) return 'skipped'; const delivery = isDraft ? undefined @@ -1020,13 +1133,14 @@ export class AgentExecutionOrchestratorService { hideUserMessageFromTranscript, isWakeRun, permit, + claimedExecution, } = config; const { threadId, resourceId } = memory; if (permit.threadId !== threadId) { throw new UnexpectedError('Agent turn permit does not belong to this thread'); } - let executionId: string | undefined; + let executionId = claimedExecution?.executionId; const recorder = this.createRecorder(toolRegistry, () => executionId, { projectId, agentId, @@ -1065,7 +1179,8 @@ export class AgentExecutionOrchestratorService { telemetry: { ...telemetry, userId }, }; // The claimed running row is the fail-closed fence: no row, no turn. - const claimed = await this.claimTurn(startParams, startedAt, abortSignal); + const claimed = + claimedExecution ?? (await this.claimTurn(startParams, startedAt, abortSignal)); executionId = claimed.executionId; const resultStream = await agentInstance.stream(input, { persistence: { threadId, resourceId, hostMetadata }, @@ -1124,10 +1239,21 @@ export class AgentExecutionOrchestratorService { telemetry: { ...telemetry, userId }, }, }); + // Queued user messages run before background job results. + await this.requestQueueDrain(threadId); if (!isWakeRun) await this.requestPendingBackgroundWake(threadId); } } + private async requestQueueDrain(threadId: string): Promise { + try { + const { AgentChatQueueService } = await import('./agent-chat-queue.service.js'); + Container.get(AgentChatQueueService).requestDrain(threadId); + } catch (error) { + this.logger.warn('Failed to request the run of queued chat messages', { threadId, error }); + } + } + private async requestPendingBackgroundWake(threadId: string): Promise { try { const { AgentWakeService } = await import('./background/agent-wake.service.js'); @@ -1238,22 +1364,37 @@ export class AgentExecutionOrchestratorService { startedAt: Date, abortSignal?: AbortSignal, ): Promise<{ executionId: string; abortSignal: AbortSignal }> { + const { executionId, claimLost } = await this.retryWhileThreadClaimed( + params.threadId, + params.agentId, + abortSignal, + async () => + await this.agentExecutionService.startClaimedExecutionRecording(params, startedAt), + ); + return { + executionId, + abortSignal: AbortSignal.any([abortSignal, claimLost].filter((s) => s !== undefined)), + }; + } + + /** Retry `claim` each time another run's claimed row on the thread ends. */ + private async retryWhileThreadClaimed( + threadId: string, + agentId: string, + abortSignal: AbortSignal | undefined, + claim: () => Promise, + ): Promise { for (;;) { abortSignal?.throwIfAborted(); try { - const { executionId, claimLost } = - await this.agentExecutionService.startClaimedExecutionRecording(params, startedAt); - return { - executionId, - abortSignal: AbortSignal.any([abortSignal, claimLost].filter((s) => s !== undefined)), - }; + return await claim(); } catch (error) { if (!(error instanceof AgentThreadClaimConflictError)) throw error; this.logger.info('Agent thread is claimed by another run, waiting for it to end', { - agentId: params.agentId, - threadId: params.threadId, + agentId, + threadId, }); - await this.turnCoordinator.waitUntilThreadIdle(params.threadId, abortSignal); + await this.turnCoordinator.waitUntilThreadIdle(threadId, abortSignal); } } } diff --git a/packages/cli/src/modules/agents/agent-execution.service.ts b/packages/cli/src/modules/agents/agent-execution.service.ts index fa68580d41c..5ef96639ffa 100644 --- a/packages/cli/src/modules/agents/agent-execution.service.ts +++ b/packages/cli/src/modules/agents/agent-execution.service.ts @@ -71,6 +71,18 @@ export interface ClaimedExecutionRecording { claimLost: AbortSignal; } +/** A queued row plus the scope its update events carry. */ +export type QueuedExecutionRef = Pick & { + agentId: string; + projectId: string; +}; + +/** The claim on a promoted queued row, with the user turn as it was at pickup. */ +export interface ClaimedQueuedExecution extends ClaimedExecutionRecording { + userMessage: string | null; + attachments: AgentExecution['attachments']; +} + interface TimelineSnapshotParams { executionId: string; projectId: string; @@ -130,7 +142,12 @@ export class AgentExecutionService { /** Record a running run that does not take part in the per-thread turn queue. */ async startExecutionRecording(params: StartExecutionParams, startedAt: Date): Promise { - const executionId = await this.insertRunningExecution(params, startedAt, null); + const executionId = await this.insertExecution(params, { + status: 'running', + activeThreadId: null, + startedAt, + resourceId: null, + }); this.startHeartbeat(executionId); return executionId; } @@ -145,26 +162,97 @@ export class AgentExecutionService { params: StartExecutionParams, startedAt: Date, ): Promise { - const executionId = await this.insertRunningExecution(params, startedAt, params.threadId); + const executionId = await this.insertExecution(params, { + status: 'running', + activeThreadId: params.threadId, + startedAt, + resourceId: null, + }); + return this.holdClaim(executionId, params.threadId); + } + + /** + * Store a preview-chat message that waits for the thread's running turn. + * `resourceId` names the sender, so the drain can run the turn as that user. + */ + async recordQueuedExecution( + params: StartExecutionParams & { resourceId: string }, + ): Promise { + return await this.insertExecution(params, { + status: 'queued', + activeThreadId: null, + startedAt: null, + resourceId: params.resourceId, + }); + } + + async isQueued(executionId: string): Promise { + return await this.agentExecutionRepository.existsBy({ id: executionId, status: 'queued' }); + } + + /** + * Promote a queued row to the thread's claimed running row, like + * {@link startClaimedExecutionRecording} without the insert. `null` when the + * row is no longer queued. The user turn is read after the promote, so no + * later edit can change what the model receives. + */ + async claimQueuedExecution( + execution: QueuedExecutionRef, + startedAt: Date, + ): Promise { + const promoted = await this.agentExecutionRepository.promoteQueuedToRunning( + execution.id, + execution.threadId, + startedAt, + ); + if (!promoted) return null; + this.executionUpdateBroadcaster.notify({ + projectId: execution.projectId, + agentId: execution.agentId, + threadId: execution.threadId, + executionId: execution.id, + }); + const row = await this.agentExecutionRepository.findOneBy({ id: execution.id }); + return { + ...this.holdClaim(execution.id, execution.threadId), + userMessage: row?.userMessage ?? null, + attachments: row?.attachments ?? null, + }; + } + + /** End a queued row that cannot run, e.g. because its sender is gone. */ + async failQueuedExecution(execution: QueuedExecutionRef, error: unknown): Promise { + const failed = await this.agentExecutionRepository.failQueued( + execution.id, + error instanceof Error ? error.message : String(error), + new Date(), + ); + if (!failed) return; + this.executionUpdateBroadcaster.notify({ + projectId: execution.projectId, + agentId: execution.agentId, + threadId: execution.threadId, + executionId: execution.id, + }); + } + + private holdClaim(executionId: string, threadId: string): ClaimedExecutionRecording { const claim = new AbortController(); this.startHeartbeat(executionId, { - threadId: params.threadId, + threadId, onLost: (reason) => claim.abort(new OperationalError(reason)), }); return { executionId, claimLost: claim.signal }; } - private async insertRunningExecution( + private async insertExecution( params: StartExecutionParams, - startedAt: Date, - activeThreadId: string | null, + row: Pick, ): Promise { const { userMessage, created } = await this.prepareThread(params); const inserted = await this.agentExecutionRepository.insertRunning({ threadId: params.threadId, - activeThreadId, - status: 'running', - startedAt, + ...row, stoppedAt: null, duration: 0, userMessage, @@ -696,7 +784,7 @@ export class AgentExecutionService { } function toSessionStatus( - latestStatus: AgentExecutionStatus | undefined, + latestStatus: Exclude | undefined, hasFailureSummary: boolean, ): AgentSessionStatus | null { if (!latestStatus) return null; @@ -713,7 +801,7 @@ function cleanUserMessage(message: string | null, agentName: string): string | n return cleaned.length > 0 ? cleaned : null; } -function executionStatus(record: MessageRecord): AgentExecution['status'] { +function executionStatus(record: MessageRecord): Exclude { if (record.error !== null || record.finishReason === 'error') return 'error'; if (record.finishReason === 'cancelled') return 'cancelled'; return 'success'; diff --git a/packages/cli/src/modules/agents/agent-interrupted-execution-sweeper.ts b/packages/cli/src/modules/agents/agent-interrupted-execution-sweeper.ts index 37b5e135945..f3cdf4a001d 100644 --- a/packages/cli/src/modules/agents/agent-interrupted-execution-sweeper.ts +++ b/packages/cli/src/modules/agents/agent-interrupted-execution-sweeper.ts @@ -2,6 +2,7 @@ import { Logger } from '@n8n/backend-common'; import { AgentsConfig } from '@n8n/config'; import { Service } from '@n8n/di'; +import { AgentChatQueueService } from './agent-chat-queue.service'; import { AgentExecutionService } from './agent-execution.service'; import { AgentBackgroundJobService } from './background/agent-background-job.service'; import { AgentWakeService } from './background/agent-wake.service'; @@ -18,6 +19,7 @@ export class AgentInterruptedExecutionSweeper { private readonly backgroundJobService: AgentBackgroundJobService, private readonly agentWakeService: AgentWakeService, private readonly agentsConfig: AgentsConfig, + private readonly chatQueueService: AgentChatQueueService, ) { this.logger = this.logger.scoped('agents'); } @@ -74,5 +76,12 @@ export class AgentInterruptedExecutionSweeper { } catch (error) { this.logger.error('Failed to schedule delivery of pending background job results', { error }); } + + // Queued chat messages whose drain died with a main restart resume here. + try { + await this.chatQueueService.drainAll(); + } catch (error) { + this.logger.error('Failed to schedule queued agent chat messages', { error }); + } } } diff --git a/packages/cli/src/modules/agents/agent-session-langsmith-export.service.ts b/packages/cli/src/modules/agents/agent-session-langsmith-export.service.ts index 66d9596b410..d6495db2923 100644 --- a/packages/cli/src/modules/agents/agent-session-langsmith-export.service.ts +++ b/packages/cli/src/modules/agents/agent-session-langsmith-export.service.ts @@ -157,7 +157,9 @@ export class AgentSessionLangSmithExportService { children.push(await this.loadSessionTree(child.id, projectId, child.agentId, visited)); } - return { ...detail, children }; + // Queued messages have not run, so they have no trace. + const executions = detail.executions.filter((execution) => execution.status !== 'queued'); + return { ...detail, executions, children }; } private ensureSettled(detail: ThreadDetail) { diff --git a/packages/cli/src/modules/agents/entities/agent-execution.entity.ts b/packages/cli/src/modules/agents/entities/agent-execution.entity.ts index 61cf0bb05ea..60c6948fc87 100644 --- a/packages/cli/src/modules/agents/entities/agent-execution.entity.ts +++ b/packages/cli/src/modules/agents/entities/agent-execution.entity.ts @@ -11,7 +11,13 @@ import { AgentExecutionThread } from './agent-execution-thread.entity'; import type { TimelineEvent } from '../execution-recorder'; import type { AgentExecutionFailureSummary } from '../utils/execution-failure-summary'; -export type AgentExecutionStatus = 'running' | 'success' | 'error' | 'cancelled' | 'interrupted'; +export type AgentExecutionStatus = + | 'queued' + | 'running' + | 'success' + | 'error' + | 'cancelled' + | 'interrupted'; export type AgentExecutionHitlStatus = 'suspended' | 'resumed'; /** @@ -54,6 +60,13 @@ export class AgentExecution extends WithTimestampsAndStringId { @Column({ type: 'varchar', length: 128, nullable: true }) activeThreadId: string | null; + /** + * Memory resource id of the sender (`draft-chat:`). Set on queued + * preview turns so the drain can run the turn as that user; null otherwise. + */ + @Column({ type: 'varchar', length: 255, nullable: true }) + resourceId: string | null; + @DateTimeColumn({ precision: 3, nullable: true }) startedAt: Date | null; diff --git a/packages/cli/src/modules/agents/repositories/agent-execution-thread.repository.ts b/packages/cli/src/modules/agents/repositories/agent-execution-thread.repository.ts index 0825c1660f3..c8e2b91d61e 100644 --- a/packages/cli/src/modules/agents/repositories/agent-execution-thread.repository.ts +++ b/packages/cli/src/modules/agents/repositories/agent-execution-thread.repository.ts @@ -158,15 +158,19 @@ export class AgentExecutionThreadRepository extends Repository): string { - return query - .subQuery() - .select('latestExecution.status') - .from(AgentExecution, 'latestExecution') - .where('latestExecution.threadId = thread.id') - .orderBy('latestExecution.createdAt', 'DESC') - .addOrderBy('latestExecution.id', 'DESC') - .limit(1) - .getQuery(); + return ( + query + .subQuery() + .select('latestExecution.status') + .from(AgentExecution, 'latestExecution') + .where('latestExecution.threadId = thread.id') + // A queued row has not run yet; the filter matches the last actual run. + .andWhere("latestExecution.status != 'queued'") + .orderBy('latestExecution.createdAt', 'DESC') + .addOrderBy('latestExecution.id', 'DESC') + .limit(1) + .getQuery() + ); } private failureExistsSubquery(query: SelectQueryBuilder): string { 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 368b0750ad3..c042f0e003b 100644 --- a/packages/cli/src/modules/agents/repositories/agent-execution.repository.ts +++ b/packages/cli/src/modules/agents/repositories/agent-execution.repository.ts @@ -19,11 +19,19 @@ export class AgentThreadClaimConflictError extends OperationalError { } } -type NewRunningAgentExecution = Omit< +type NewAgentExecution = Omit< AgentExecution, 'id' | 'createdAt' | 'updatedAt' | 'thread' | 'generateId' | 'setUpdateDate' >; +/** A user turn as the chat queue sees it: who sent it, where it waits, and what it carries. */ +export type AgentExecutionUserTurn = Pick< + AgentExecution, + 'id' | 'threadId' | 'status' | 'resourceId' | 'attachments' +>; + +const userTurnColumns = ['id', 'threadId', 'status', 'resourceId', 'attachments'] as const; + type AgentExecutionFinalizationValues = Pick< AgentExecution, 'status' | 'stoppedAt' | 'duration' | 'timeline' | 'storedAt' | 'error' | 'failureSummary' @@ -58,10 +66,10 @@ export class AgentExecutionRepository extends Repository { } /** - * Insert a running row. With `activeThreadId` set, the partial unique index + * Insert a running or queued row. With `activeThreadId` set, the partial unique index * turns a second claim on the same thread into {@link AgentThreadClaimConflictError}. */ - async insertRunning(values: NewRunningAgentExecution): Promise { + async insertRunning(values: NewAgentExecution): Promise { try { return await this.save(this.create(values)); } catch (error) { @@ -72,6 +80,74 @@ export class AgentExecutionRepository extends Repository { } } + async countQueuedByThread(threadId: string): Promise { + return await this.countBy({ threadId, status: 'queued' }); + } + + /** The next queued message of a thread, in arrival order. */ + async findOldestQueuedByThread(threadId: string): Promise { + return await this.findOne({ + select: [...userTurnColumns], + where: { threadId, status: 'queued' }, + order: { createdAt: 'ASC', id: 'ASC' }, + }); + } + + async findUserTurnById(executionId: string): Promise { + return await this.findOne({ select: [...userTurnColumns], where: { id: executionId } }); + } + + async findThreadIdsWithQueued(): Promise { + const rows = await this.find({ select: ['threadId'], where: { status: 'queued' } }); + return [...new Set(rows.map((row) => row.threadId))]; + } + + /** + * Turn a queued row into the thread's claimed running row. `false` when the + * row is no longer queued. A unique-index conflict means another claimed + * running row holds the thread: {@link AgentThreadClaimConflictError}. + */ + async promoteQueuedToRunning( + executionId: string, + threadId: string, + startedAt: Date, + ): Promise { + try { + const result = await this.update( + { id: executionId, threadId, status: 'queued' }, + { status: 'running', activeThreadId: threadId, startedAt, updatedAt: startedAt }, + ); + return result.affected === 1; + } catch (error) { + if (isUniqueConstraintError(error)) throw new AgentThreadClaimConflictError(); + throw error; + } + } + + /** Replace the text of a queued row. `false` once the row left `queued`. */ + async updateQueuedMessage(executionId: string, userMessage: string): Promise { + const result = await this.update( + { id: executionId, status: 'queued' }, + { userMessage, updatedAt: new Date() }, + ); + return result.affected === 1; + } + + /** Remove a queued row. `false` once the row left `queued`. */ + async deleteQueued(executionId: string): Promise { + const result = await this.delete({ id: executionId, status: 'queued' }); + return result.affected === 1; + } + + /** End a queued row that cannot run. `false` once the row left `queued`. */ + async failQueued(executionId: string, error: string, stoppedAt: Date): Promise { + const result = await this.update( + { id: executionId, status: 'queued' }, + { status: 'error', error, stoppedAt, updatedAt: stoppedAt }, + ); + return result.affected === 1; + } + /** * Refresh the liveness timestamp. With `activeThreadId`, the update matches * only while the row still holds that claim, so `false` means the claim @@ -179,9 +255,13 @@ export class AgentExecutionRepository extends Repository { return new Map(rows.map((r) => [r.threadId, r.source])); } + /** + * Status of the newest run in each thread. Queued rows have not run yet, so + * they do not count: the thread keeps the status of its last actual run. + */ async findLatestStatusesByThreadIds( threadIds: string[], - ): Promise> { + ): Promise>> { if (threadIds.length === 0) return new Map(); const tableName = this.metadata.tablePath; @@ -190,10 +270,10 @@ export class AgentExecutionRepository extends Repository { .where('e."threadId" IN (:...threadIds)', { threadIds }) .andWhere( `e.id = (SELECT e2.id FROM ${tableName} e2 ` + - 'WHERE e2."threadId" = e."threadId" ' + + 'WHERE e2."threadId" = e."threadId" AND e2."status" != \'queued\' ' + 'ORDER BY e2."createdAt" DESC, e2.id DESC LIMIT 1)', ) - .getRawMany<{ threadId: string; status: AgentExecutionStatus }>(); + .getRawMany<{ threadId: string; status: Exclude }>(); return new Map(rows.map((row) => [row.threadId, row.status])); } diff --git a/packages/cli/src/modules/agents/utils/execution-to-message-mapper.ts b/packages/cli/src/modules/agents/utils/execution-to-message-mapper.ts index fad4526ebaa..27cb74c1a6f 100644 --- a/packages/cli/src/modules/agents/utils/execution-to-message-mapper.ts +++ b/packages/cli/src/modules/agents/utils/execution-to-message-mapper.ts @@ -164,6 +164,8 @@ export function executionToMessagesDto(execution: ExecutionTranscript): AgentPer content: userContent, ...(execution.author ? { author: execution.author } : {}), executionId: execution.id, + // The status tells the client which user turns still wait or run. + ...(execution.status ? { executionStatus: execution.status } : {}), }); } 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 4266afb414c..43a51824be3 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 @@ -241,6 +241,51 @@ describe('AgentExecutionRepository', () => { hitlStatus: null, source: null, attachments: null, + resourceId: null, + }); + + it('promotes a queued row into the thread claim only while the row is queued and the thread is free', async () => { + const thread = await createThread(); + const queued = await repository.insertRunning({ + ...runningValues(thread.id, null), + status: 'queued', + startedAt: null, + resourceId: 'draft-chat:user-1', + }); + const running = await repository.insertRunning(runningValues(thread.id, thread.id)); + const startedAt = new Date('2026-09-14T10:00:00.000Z'); + + await expect( + repository.promoteQueuedToRunning(queued.id, thread.id, startedAt), + ).rejects.toBeInstanceOf(AgentThreadClaimConflictError); + expect(await repository.findOneByOrFail({ id: queued.id })).toMatchObject({ + status: 'queued', + activeThreadId: null, + }); + + await repository.updateIfRunning(running.id, { + status: 'success', + stoppedAt: new Date(), + duration: 1, + timeline: null, + storedAt: 'db', + error: null, + failureSummary: null, + }); + expect(await repository.promoteQueuedToRunning(queued.id, thread.id, startedAt)).toBe(true); + expect(await repository.findOneByOrFail({ id: queued.id })).toMatchObject({ + status: 'running', + activeThreadId: thread.id, + startedAt, + }); + + // A promoted row is no longer queued: a second pickup, an edit and a removal all miss it. + expect(await repository.promoteQueuedToRunning(queued.id, thread.id, startedAt)).toBe(false); + expect(await repository.updateQueuedMessage(queued.id, 'changed')).toBe(false); + expect(await repository.deleteQueued(queued.id)).toBe(false); + expect(await repository.findOneByOrFail({ id: queued.id })).toMatchObject({ + userMessage: 'run', + }); }); it('allows one claimed running row per thread and releases the claim when the row ends', async () => { @@ -375,7 +420,19 @@ describe('AgentExecutionRepository', () => { failureSummary: null, createdAt: new Date('2026-01-02T00:00:00Z'), }); - await createExecution({ threadId: succeeded.id, status: 'success', failureSummary: null }); + await createExecution({ + threadId: succeeded.id, + status: 'success', + failureSummary: null, + createdAt: new Date('2026-01-01T00:00:00Z'), + }); + // A message waiting behind the last run does not change the thread's status. + await createExecution({ + threadId: succeeded.id, + status: 'queued', + failureSummary: null, + createdAt: new Date('2026-01-03T00:00:00Z'), + }); await createExecution({ threadId: recovered.id, status: 'success', @@ -394,8 +451,12 @@ describe('AgentExecutionRepository', () => { expect(await idsFor('succeeded')).toEqual([succeeded.id]); expect(new Set(await idsFor('error'))).toEqual(new Set([recovered.id, errored.id])); - const latestStatuses = await repository.findLatestStatusesByThreadIds([running.id]); + const latestStatuses = await repository.findLatestStatusesByThreadIds([ + running.id, + succeeded.id, + ]); expect(latestStatuses.get(running.id)).toBe('running'); + expect(latestStatuses.get(succeeded.id)).toBe('success'); }); it('mirrors the displayed origin precedence', async () => { diff --git a/packages/cli/test/migration/1789375477575-add-queued-status-to-agent-execution.test.ts b/packages/cli/test/migration/1789375477575-add-queued-status-to-agent-execution.test.ts new file mode 100644 index 00000000000..f432e33310b --- /dev/null +++ b/packages/cli/test/migration/1789375477575-add-queued-status-to-agent-execution.test.ts @@ -0,0 +1,124 @@ +import { + createTestMigrationContext, + initDbUpToMigration, + runSingleMigration, + undoLastSingleMigration, + type TestMigrationContext, +} from '@n8n/backend-test-utils'; +import { DbConnection } from '@n8n/db'; +import { Container } from '@n8n/di'; +import { DataSource } from '@n8n/typeorm'; +import { randomUUID } from 'node:crypto'; + +const MIGRATION_NAME = 'AddQueuedStatusToAgentExecution1789375477575'; + +describe('AddQueuedStatusToAgentExecution migration', () => { + let dataSource: DataSource; + + async function withContext(fn: (context: TestMigrationContext) => Promise): Promise { + const context = createTestMigrationContext(dataSource); + try { + return await fn(context); + } finally { + await context.queryRunner.release(); + } + } + + beforeAll(async () => { + const dbConnection = Container.get(DbConnection); + await dbConnection.init(); + dataSource = Container.get(DataSource); + await withContext(async (context) => { + await context.queryRunner.clearDatabase(); + }); + await initDbUpToMigration(MIGRATION_NAME); + }); + + afterAll(async () => { + await Container.get(DbConnection).close(); + }); + + it('accepts queued rows with a sender after up and cancels them on down', async () => { + const ids = { project: randomUUID(), agent: randomUUID(), thread: randomUUID() }; + const now = new Date('2026-09-14T10:00:00.000Z'); + + await withContext(async (context) => { + await context.runQuery( + `INSERT INTO ${context.escape.tableName('project')} ("id", "name", "type", "createdAt", "updatedAt") + VALUES (:id, :name, :type, :createdAt, :updatedAt)`, + { id: ids.project, name: 'Test project', type: 'team', createdAt: now, updatedAt: now }, + ); + await context.runQuery( + `INSERT INTO ${context.escape.tableName('agents')} ("id", "name", "projectId", "integrations", "tools", "skills", "createdAt", "updatedAt") + VALUES (:id, :name, :projectId, :integrations, :tools, :skills, :createdAt, :updatedAt)`, + { + id: ids.agent, + name: 'Test agent', + projectId: ids.project, + integrations: '[]', + tools: '{}', + skills: '{}', + createdAt: now, + updatedAt: now, + }, + ); + await context.runQuery( + `INSERT INTO ${context.escape.tableName('agent_execution_threads')} ("id", "agentId", "agentName", "projectId", "createdAt", "updatedAt") + VALUES (:id, :agentId, :agentName, :projectId, :createdAt, :updatedAt)`, + { + id: ids.thread, + agentId: ids.agent, + agentName: 'Test agent', + projectId: ids.project, + createdAt: now, + updatedAt: now, + }, + ); + }); + + await runSingleMigration(MIGRATION_NAME); + dataSource = Container.get(DataSource); + + const queuedId = randomUUID(); + await withContext(async (context) => { + const table = context.escape.tableName('agent_execution'); + await context.runQuery( + `INSERT INTO ${table} ("id", "threadId", "status", "resourceId", "createdAt", "updatedAt") + VALUES (:id, :threadId, 'queued', :resourceId, :now, :now)`, + { id: queuedId, threadId: ids.thread, resourceId: 'draft-chat:user-1', now }, + ); + const rows = await context.runQuery>( + `SELECT "status", "resourceId" FROM ${table} WHERE "id" = :id`, + { id: queuedId }, + ); + expect(rows).toEqual([{ status: 'queued', resourceId: 'draft-chat:user-1' }]); + }); + + await undoLastSingleMigration(); + dataSource = Container.get(DataSource); + await withContext(async (context) => { + const table = context.escape.tableName('agent_execution'); + const rows = await context.runQuery>( + `SELECT "status" FROM ${table} WHERE "id" = :id`, + { id: queuedId }, + ); + expect(rows).toEqual([{ status: 'cancelled' }]); + const columns = await context.queryRunner.getTable(`${context.tablePrefix}agent_execution`); + expect(columns?.findColumnByName('resourceId')).toBeUndefined(); + await expect( + context.runQuery( + `INSERT INTO ${table} ("id", "threadId", "status", "createdAt", "updatedAt") + VALUES (:id, :threadId, 'queued', :now, :now)`, + { id: randomUUID(), threadId: ids.thread, now }, + ), + ).rejects.toThrow(); + }); + + await runSingleMigration(MIGRATION_NAME); + dataSource = Container.get(DataSource); + await withContext(async (context) => { + const table = await context.queryRunner.getTable(`${context.tablePrefix}agent_execution`); + expect(table?.findColumnByName('resourceId')?.isNullable).toBe(true); + }); + }); +}); diff --git a/packages/frontend/@n8n/i18n/src/locales/en.json b/packages/frontend/@n8n/i18n/src/locales/en.json index 41c7fa7268c..8c28710aa7a 100644 --- a/packages/frontend/@n8n/i18n/src/locales/en.json +++ b/packages/frontend/@n8n/i18n/src/locales/en.json @@ -7568,6 +7568,15 @@ "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.queued": "Queued", + "agents.chat.queue.edit": "Edit queued message", + "agents.chat.queue.remove": "Remove queued message", + "agents.chat.queue.save": "Save", + "agents.chat.queue.cancel": "Cancel", + "agents.chat.queue.error": "Couldn't queue the message", + "agents.chat.queue.editError": "Couldn't edit the queued message", + "agents.chat.queue.removeError": "Couldn't remove the queued message", + "agents.chat.queue.alreadyStarted": "The agent has already started on this message", "agents.chat.streamInterrupted": "The connection was interrupted. Refresh the page to see the latest response.", "agents.chat.toolError.generic": "Something went wrong while running this tool", "agents.chat.clearHistory": "Clear chat history", diff --git a/packages/frontend/editor-ui/src/features/agents/__tests__/AgentChatMessageList.test.ts b/packages/frontend/editor-ui/src/features/agents/__tests__/AgentChatMessageList.test.ts index 7ba8357aaa4..6c42e4eb33d 100644 --- a/packages/frontend/editor-ui/src/features/agents/__tests__/AgentChatMessageList.test.ts +++ b/packages/frontend/editor-ui/src/features/agents/__tests__/AgentChatMessageList.test.ts @@ -48,6 +48,12 @@ vi.mock('@n8n/design-system', () => ({ emits: ['click'], props: ['variant', 'size'], }, + N8nInput: { + template: + '