From 84efbac6f12958a8f134d82eead91e7cfad20d02 Mon Sep 17 00:00:00 2001
From: Robin Braumann <50590409+bjorger@users.noreply.github.com>
Date: Tue, 29 Sep 2026 08:39:25 +0000
Subject: [PATCH] refactor(core): Reference accepted messages from agent queue
(#39778)
---
docs/generated/postgres-schema/README.md | 3 +-
.../postgres-schema/public.agent_execution.md | 2 +-
.../public.agent_execution_threads.md | 2 +-
.../public.agent_message_queue.md | 25 +-
.../postgres-schema/public.agents_messages.md | 12 +-
docs/generated/sqlite-schema/README.md | 5 +-
.../sqlite-schema/agent_execution.md | 2 +-
.../sqlite-schema/agent_execution_threads.md | 2 +-
.../sqlite-schema/agent_message_queue.md | 25 +-
.../sqlite-schema/agents_messages.md | 12 +-
...05116208-AddAgentQueueMessageReferences.ts | 239 ++++++++++
...05116208-AddAgentQueueMessageReferences.ts | 65 +++
.../__tests__/agent-execution.service.test.ts | 2 +-
...ent-message-queue-consumer.service.test.ts | 131 ++++--
.../modules/agents/agent-execution.service.ts | 25 +-
.../agent-message-queue-consumer.service.ts | 62 +--
.../agents/agent-message-queue.service.ts | 148 +++++--
.../entities/agent-message-queue.entity.ts | 22 +-
.../agent-message-queue.repository.ts | 34 +-
.../repositories/agent-message.repository.ts | 53 ++-
.../agents/types/agent-queued-message.ts | 13 +
.../__tests__/inbound-attachments.test.ts | 28 +-
.../agents/utils/inbound-attachments.ts | 30 +-
.../agent-execution.repository.test.ts | 194 +++++++-
...add-agent-queue-message-references.test.ts | 413 ++++++++++++++++++
.../agents/message-queue.spec.ts | 31 +-
26 files changed, 1407 insertions(+), 173 deletions(-)
create mode 100644 packages/@n8n/db/src/migrations/postgresdb/1790605116208-AddAgentQueueMessageReferences.ts
create mode 100644 packages/@n8n/db/src/migrations/sqlite/1790605116208-AddAgentQueueMessageReferences.ts
create mode 100644 packages/cli/test/migration/1790605116208-add-agent-queue-message-references.test.ts
diff --git a/docs/generated/postgres-schema/README.md b/docs/generated/postgres-schema/README.md
index 58d6ecf65f1..f21ce36148b 100644
--- a/docs/generated/postgres-schema/README.md
+++ b/docs/generated/postgres-schema/README.md
@@ -205,6 +205,7 @@ erDiagram
"public.agent_files" }o--|| "public.agents" : "FOREIGN KEY (#quot;agentId#quot;) REFERENCES agents(id) ON DELETE CASCADE"
"public.agent_history" }o--o| "public.user" : "FOREIGN KEY (#quot;publishedById#quot;) REFERENCES #quot;user#quot;(id) ON DELETE SET NULL"
"public.agent_history" }o--|| "public.agents" : "FOREIGN KEY (#quot;agentId#quot;) REFERENCES agents(id) ON DELETE CASCADE"
+"public.agent_message_queue" }o--|| "public.agents_messages" : "FOREIGN KEY (#quot;messageId#quot;) REFERENCES agents_messages(id) ON DELETE CASCADE"
"public.agent_message_queue" }o--o| "public.agent_execution" : "FOREIGN KEY (#quot;executionId#quot;) REFERENCES agent_execution(id)"
"public.agent_message_queue" }o--|| "public.agent_execution_threads" : "FOREIGN KEY (#quot;threadId#quot;) REFERENCES agent_execution_threads(id) ON DELETE CASCADE"
"public.agent_task_definition" }o--|| "public.agents" : "FOREIGN KEY (#quot;agentId#quot;) REFERENCES agents(id) ON DELETE CASCADE"
@@ -607,8 +608,8 @@ erDiagram
timestamp_3__with_time_zone createdAt
varchar_36_ executionId FK
bigint id
+ varchar_36_ messageId FK
json payload
- varchar_32_ source
varchar_128_ threadId FK
timestamp_3__with_time_zone updatedAt
}
diff --git a/docs/generated/postgres-schema/public.agent_execution.md b/docs/generated/postgres-schema/public.agent_execution.md
index c48414745ae..0b5a5bd3bc3 100644
--- a/docs/generated/postgres-schema/public.agent_execution.md
+++ b/docs/generated/postgres-schema/public.agent_execution.md
@@ -96,8 +96,8 @@ erDiagram
timestamp_3__with_time_zone createdAt
varchar_36_ executionId FK
bigint id
+ varchar_36_ messageId FK
json payload
- varchar_32_ source
varchar_128_ threadId FK
timestamp_3__with_time_zone updatedAt
}
diff --git a/docs/generated/postgres-schema/public.agent_execution_threads.md b/docs/generated/postgres-schema/public.agent_execution_threads.md
index a00f6936dd4..08046a75c2c 100644
--- a/docs/generated/postgres-schema/public.agent_execution_threads.md
+++ b/docs/generated/postgres-schema/public.agent_execution_threads.md
@@ -134,8 +134,8 @@ erDiagram
timestamp_3__with_time_zone createdAt
varchar_36_ executionId FK
bigint id
+ varchar_36_ messageId FK
json payload
- varchar_32_ source
varchar_128_ threadId FK
timestamp_3__with_time_zone updatedAt
}
diff --git a/docs/generated/postgres-schema/public.agent_message_queue.md b/docs/generated/postgres-schema/public.agent_message_queue.md
index 7354483d848..9f6064e82c1 100644
--- a/docs/generated/postgres-schema/public.agent_message_queue.md
+++ b/docs/generated/postgres-schema/public.agent_message_queue.md
@@ -7,8 +7,8 @@
| createdAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | |
| executionId | varchar(36) | | true | | [public.agent_execution](public.agent_execution.md) | Current execution; NULL means pending |
| id | bigint | | false | | | Acceptance order; IDs are not reused |
-| payload | json | | false | | | Input, attachment references, identity, and reply context |
-| source | varchar(32) | | false | | | Preview or integration source |
+| messageId | varchar(36) | | false | | [public.agents_messages](public.agents_messages.md) | Canonical input created when the queue accepts it |
+| payload | json | | false | | | Dispatch, authorization, and reply context. Input is stored on the message |
| threadId | varchar(128) | | false | | [public.agent_execution_threads](public.agent_execution_threads.md) | |
| updatedAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | |
@@ -18,11 +18,12 @@
| ---- | ---- | ---------- |
| FK_2349d84b4f2a660fc264f38fef3 | FOREIGN KEY | FOREIGN KEY ("executionId") REFERENCES agent_execution(id) |
| FK_a74f9154a59430986112d70cb16 | FOREIGN KEY | FOREIGN KEY ("threadId") REFERENCES agent_execution_threads(id) ON DELETE CASCADE |
+| FK_agent_message_queue_messageId | FOREIGN KEY | FOREIGN KEY ("messageId") REFERENCES agents_messages(id) ON DELETE CASCADE |
| PK_733d8c959a6057f04f4da5ab721 | PRIMARY KEY | PRIMARY KEY (id) |
| agent_message_queue_createdAt_not_null | n | NOT NULL "createdAt" |
| agent_message_queue_id_not_null | n | NOT NULL id |
+| agent_message_queue_messageId_not_null | n | NOT NULL "messageId" |
| agent_message_queue_payload_not_null | n | NOT NULL payload |
-| agent_message_queue_source_not_null | n | NOT NULL source |
| agent_message_queue_threadId_not_null | n | NOT NULL "threadId" |
| agent_message_queue_updatedAt_not_null | n | NOT NULL "updatedAt" |
@@ -31,6 +32,7 @@
| Name | Definition |
| ---- | ---------- |
| IDX_2349d84b4f2a660fc264f38fef | CREATE INDEX "IDX_2349d84b4f2a660fc264f38fef" ON public.agent_message_queue USING btree ("executionId") |
+| IDX_agent_message_queue_messageId | CREATE UNIQUE INDEX "IDX_agent_message_queue_messageId" ON public.agent_message_queue USING btree ("messageId") |
| IDX_agent_message_queue_threadId | CREATE UNIQUE INDEX "IDX_agent_message_queue_threadId" ON public.agent_message_queue USING btree ("threadId") WHERE ("executionId" IS NOT NULL) |
| IDX_c1db6ea2d031cc49100535e6c6 | CREATE INDEX "IDX_c1db6ea2d031cc49100535e6c6" ON public.agent_message_queue USING btree ("threadId", id) |
| PK_733d8c959a6057f04f4da5ab721 | CREATE UNIQUE INDEX "PK_733d8c959a6057f04f4da5ab721" ON public.agent_message_queue USING btree (id) |
@@ -41,14 +43,15 @@
erDiagram
"public.agent_message_queue" }o--o| "public.agent_execution" : "FOREIGN KEY (#quot;executionId#quot;) REFERENCES agent_execution(id)"
+"public.agent_message_queue" }o--|| "public.agents_messages" : "FOREIGN KEY (#quot;messageId#quot;) REFERENCES agents_messages(id) ON DELETE CASCADE"
"public.agent_message_queue" }o--|| "public.agent_execution_threads" : "FOREIGN KEY (#quot;threadId#quot;) REFERENCES agent_execution_threads(id) ON DELETE CASCADE"
"public.agent_message_queue" {
timestamp_3__with_time_zone createdAt
varchar_36_ executionId FK
bigint id
+ varchar_36_ messageId FK
json payload
- varchar_32_ source
varchar_128_ threadId FK
timestamp_3__with_time_zone updatedAt
}
@@ -76,6 +79,20 @@ erDiagram
timestamp_3__with_time_zone updatedAt
text userMessage
}
+"public.agents_messages" {
+ json author
+ json content
+ timestamp_3__with_time_zone createdAt
+ varchar_36_ id
+ json modelContent
+ timestamp_3__with_time_zone modelContextAt
+ json origin
+ varchar_255_ resourceId
+ varchar_36_ role
+ varchar_255_ threadId FK
+ varchar_36_ type
+ timestamp_3__with_time_zone updatedAt
+}
"public.agent_execution_threads" {
varchar_16_ accessScope
varchar_36_ agentId FK
diff --git a/docs/generated/postgres-schema/public.agents_messages.md b/docs/generated/postgres-schema/public.agents_messages.md
index c60d77fa57e..7741ec2b276 100644
--- a/docs/generated/postgres-schema/public.agents_messages.md
+++ b/docs/generated/postgres-schema/public.agents_messages.md
@@ -7,7 +7,7 @@
| author | json | | true | | | Original message author supplied by the chat platform |
| content | json | | false | | | |
| createdAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | |
-| id | varchar(36) | | false | [public.agent_execution_message_links](public.agent_execution_message_links.md) [public.agents_memory_entry_candidates](public.agents_memory_entry_candidates.md) | | |
+| id | varchar(36) | | false | [public.agent_execution_message_links](public.agent_execution_message_links.md) [public.agent_message_queue](public.agent_message_queue.md) [public.agents_memory_entry_candidates](public.agents_memory_entry_candidates.md) | | |
| modelContent | json | | true | | | Enriched model input when it differs from the original content |
| modelContextAt | timestamp(3) with time zone | | true | | | Runtime ordering timestamp. NULL inputs are excluded from model context |
| origin | json | | true | | | Source identifiers and transcript visibility. NULL for legacy or SDK-only messages |
@@ -47,6 +47,7 @@
erDiagram
"public.agent_execution_message_links" }o--|| "public.agents_messages" : "FOREIGN KEY (#quot;messageId#quot;) REFERENCES agents_messages(id) ON DELETE CASCADE"
+"public.agent_message_queue" }o--|| "public.agents_messages" : "FOREIGN KEY (#quot;messageId#quot;) REFERENCES agents_messages(id) ON DELETE CASCADE"
"public.agents_memory_entry_candidates" }o--o| "public.agents_messages" : "FOREIGN KEY (#quot;sourceMessageId#quot;) REFERENCES agents_messages(id) ON DELETE SET NULL"
"public.agents_messages" }o--|| "public.agents_threads" : "FOREIGN KEY (#quot;threadId#quot;) REFERENCES agents_threads(id) ON DELETE CASCADE"
@@ -71,6 +72,15 @@ erDiagram
varchar_36_ messageId FK
integer position
}
+"public.agent_message_queue" {
+ timestamp_3__with_time_zone createdAt
+ varchar_36_ executionId FK
+ bigint id
+ varchar_36_ messageId FK
+ json payload
+ varchar_128_ threadId FK
+ timestamp_3__with_time_zone updatedAt
+}
"public.agents_memory_entry_candidates" {
varchar_36_ agentId FK
smallint attemptCount
diff --git a/docs/generated/sqlite-schema/README.md b/docs/generated/sqlite-schema/README.md
index 34c3eb17f70..3ba496efa16 100644
--- a/docs/generated/sqlite-schema/README.md
+++ b/docs/generated/sqlite-schema/README.md
@@ -187,8 +187,9 @@ erDiagram
"agent_files" }o--|| "agents" : "FOREIGN KEY (agentId) REFERENCES agents (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"agent_history" }o--o| "user" : "FOREIGN KEY (publishedById) REFERENCES user (id) ON UPDATE NO ACTION ON DELETE SET NULL MATCH NONE"
"agent_history" }o--|| "agents" : "FOREIGN KEY (agentId) REFERENCES agents (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
-"agent_message_queue" }o--o| "agent_execution" : "FOREIGN KEY (executionId) REFERENCES agent_execution (id) ON UPDATE NO ACTION ON DELETE NO ACTION MATCH NONE"
+"agent_message_queue" }o--|| "agents_messages" : "FOREIGN KEY (messageId) REFERENCES agents_messages (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"agent_message_queue" }o--|| "agent_execution_threads" : "FOREIGN KEY (threadId) REFERENCES agent_execution_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
+"agent_message_queue" }o--o| "agent_execution" : "FOREIGN KEY (executionId) REFERENCES agent_execution (id) ON UPDATE NO ACTION ON DELETE NO ACTION MATCH NONE"
"agent_task_definition" }o--|| "agents" : "FOREIGN KEY (agentId) REFERENCES agents (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"agent_task_run_lock" |o--|| "agents" : "FOREIGN KEY (agentId) REFERENCES agents (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"agent_task_snapshot" |o--|| "agent_history" : "FOREIGN KEY (versionId) REFERENCES agent_history (versionId) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
@@ -593,8 +594,8 @@ erDiagram
datetime_3_ createdAt
varchar_36_ executionId FK
INTEGER id
+ varchar_36_ messageId FK
TEXT payload
- varchar_32_ source
varchar_128_ threadId FK
datetime_3_ updatedAt
}
diff --git a/docs/generated/sqlite-schema/agent_execution.md b/docs/generated/sqlite-schema/agent_execution.md
index ec7393ac62f..ee08257d4d9 100644
--- a/docs/generated/sqlite-schema/agent_execution.md
+++ b/docs/generated/sqlite-schema/agent_execution.md
@@ -101,8 +101,8 @@ erDiagram
datetime_3_ createdAt
varchar_36_ executionId FK
INTEGER id
+ varchar_36_ messageId FK
TEXT payload
- varchar_32_ source
varchar_128_ threadId FK
datetime_3_ updatedAt
}
diff --git a/docs/generated/sqlite-schema/agent_execution_threads.md b/docs/generated/sqlite-schema/agent_execution_threads.md
index 81d951aaaff..c31f00dfcc2 100644
--- a/docs/generated/sqlite-schema/agent_execution_threads.md
+++ b/docs/generated/sqlite-schema/agent_execution_threads.md
@@ -134,8 +134,8 @@ erDiagram
datetime_3_ createdAt
varchar_36_ executionId FK
INTEGER id
+ varchar_36_ messageId FK
TEXT payload
- varchar_32_ source
varchar_128_ threadId FK
datetime_3_ updatedAt
}
diff --git a/docs/generated/sqlite-schema/agent_message_queue.md b/docs/generated/sqlite-schema/agent_message_queue.md
index db3fd0b8af7..e283d081cdc 100644
--- a/docs/generated/sqlite-schema/agent_message_queue.md
+++ b/docs/generated/sqlite-schema/agent_message_queue.md
@@ -6,7 +6,7 @@
Table Definition
```sql
-CREATE TABLE "agent_message_queue" ("id" integer PRIMARY KEY AUTOINCREMENT NOT NULL, "threadId" varchar(128) NOT NULL, "source" varchar(32) NOT NULL, "payload" text NOT NULL, "executionId" varchar(36), "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')), CONSTRAINT "FK_a74f9154a59430986112d70cb16" FOREIGN KEY ("threadId") REFERENCES "agent_execution_threads" ("id") ON DELETE CASCADE, CONSTRAINT "FK_2349d84b4f2a660fc264f38fef3" FOREIGN KEY ("executionId") REFERENCES "agent_execution" ("id") ON DELETE NO ACTION)
+CREATE TABLE "agent_message_queue" ("id" integer PRIMARY KEY AUTOINCREMENT NOT NULL, "threadId" varchar(128) NOT NULL, "payload" text NOT NULL, "executionId" varchar(36), "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')), "messageId" varchar(36) NOT NULL, CONSTRAINT "FK_2349d84b4f2a660fc264f38fef3" FOREIGN KEY ("executionId") REFERENCES "agent_execution" ("id") ON DELETE NO ACTION ON UPDATE NO ACTION, CONSTRAINT "FK_a74f9154a59430986112d70cb16" FOREIGN KEY ("threadId") REFERENCES "agent_execution_threads" ("id") ON DELETE CASCADE ON UPDATE NO ACTION, CONSTRAINT "FK_agent_message_queue_messageId" FOREIGN KEY ("messageId") REFERENCES "agents_messages" ("id") ON DELETE CASCADE)
```
@@ -18,8 +18,8 @@ CREATE TABLE "agent_message_queue" ("id" integer PRIMARY KEY AUTOINCREMENT NOT N
| createdAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | |
| executionId | varchar(36) | | true | | [agent_execution](agent_execution.md) | |
| id | INTEGER | | false | | | |
+| messageId | varchar(36) | | false | | [agents_messages](agents_messages.md) | |
| payload | TEXT | | false | | | |
-| source | varchar(32) | | false | | | |
| threadId | varchar(128) | | false | | [agent_execution_threads](agent_execution_threads.md) | |
| updatedAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | |
@@ -27,8 +27,9 @@ CREATE TABLE "agent_message_queue" ("id" integer PRIMARY KEY AUTOINCREMENT NOT N
| Name | Type | Definition |
| ---- | ---- | ---------- |
-| - (Foreign key ID: 0) | FOREIGN KEY | FOREIGN KEY (executionId) REFERENCES agent_execution (id) ON UPDATE NO ACTION ON DELETE NO ACTION MATCH NONE |
+| - (Foreign key ID: 0) | FOREIGN KEY | FOREIGN KEY (messageId) REFERENCES agents_messages (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE |
| - (Foreign key ID: 1) | FOREIGN KEY | FOREIGN KEY (threadId) REFERENCES agent_execution_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE |
+| - (Foreign key ID: 2) | FOREIGN KEY | FOREIGN KEY (executionId) REFERENCES agent_execution (id) ON UPDATE NO ACTION ON DELETE NO ACTION MATCH NONE |
| id | PRIMARY KEY | PRIMARY KEY (id) |
## Indexes
@@ -36,6 +37,7 @@ CREATE TABLE "agent_message_queue" ("id" integer PRIMARY KEY AUTOINCREMENT NOT N
| Name | Definition |
| ---- | ---------- |
| IDX_2349d84b4f2a660fc264f38fef | CREATE INDEX "IDX_2349d84b4f2a660fc264f38fef" ON "agent_message_queue" ("executionId") |
+| IDX_agent_message_queue_messageId | CREATE UNIQUE INDEX "IDX_agent_message_queue_messageId" ON "agent_message_queue" ("messageId") |
| IDX_agent_message_queue_threadId | CREATE UNIQUE INDEX "IDX_agent_message_queue_threadId" ON "agent_message_queue" ("threadId") WHERE "executionId" IS NOT NULL |
| IDX_c1db6ea2d031cc49100535e6c6 | CREATE INDEX "IDX_c1db6ea2d031cc49100535e6c6" ON "agent_message_queue" ("threadId", "id") |
@@ -45,14 +47,15 @@ CREATE TABLE "agent_message_queue" ("id" integer PRIMARY KEY AUTOINCREMENT NOT N
erDiagram
"agent_message_queue" }o--o| "agent_execution" : "FOREIGN KEY (executionId) REFERENCES agent_execution (id) ON UPDATE NO ACTION ON DELETE NO ACTION MATCH NONE"
+"agent_message_queue" }o--|| "agents_messages" : "FOREIGN KEY (messageId) REFERENCES agents_messages (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"agent_message_queue" }o--|| "agent_execution_threads" : "FOREIGN KEY (threadId) REFERENCES agent_execution_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"agent_message_queue" {
datetime_3_ createdAt
varchar_36_ executionId FK
INTEGER id
+ varchar_36_ messageId FK
TEXT payload
- varchar_32_ source
varchar_128_ threadId FK
datetime_3_ updatedAt
}
@@ -80,6 +83,20 @@ erDiagram
datetime_3_ updatedAt
TEXT userMessage
}
+"agents_messages" {
+ TEXT author
+ TEXT content
+ datetime_3_ createdAt
+ varchar_36_ id PK
+ TEXT modelContent
+ DATETIME modelContextAt
+ TEXT origin
+ varchar_255_ resourceId
+ varchar_36_ role
+ varchar_255_ threadId FK
+ varchar_36_ type
+ datetime_3_ updatedAt
+}
"agent_execution_threads" {
varchar_16_ accessScope
varchar_36_ agentId FK
diff --git a/docs/generated/sqlite-schema/agents_messages.md b/docs/generated/sqlite-schema/agents_messages.md
index f3cedce58bd..b55b3676787 100644
--- a/docs/generated/sqlite-schema/agents_messages.md
+++ b/docs/generated/sqlite-schema/agents_messages.md
@@ -18,7 +18,7 @@ CREATE TABLE "agents_messages" ("id" varchar(36) PRIMARY KEY NOT NULL, "threadId
| author | TEXT | | true | | | |
| content | TEXT | | false | | | |
| createdAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | |
-| id | varchar(36) | | false | [agent_execution_message_links](agent_execution_message_links.md) [agents_memory_entry_candidates](agents_memory_entry_candidates.md) | | |
+| id | varchar(36) | | false | [agent_execution_message_links](agent_execution_message_links.md) [agent_message_queue](agent_message_queue.md) [agents_memory_entry_candidates](agents_memory_entry_candidates.md) | | |
| modelContent | TEXT | | true | | | |
| modelContextAt | DATETIME | | true | | | |
| origin | TEXT | | true | | | |
@@ -52,6 +52,7 @@ CREATE TABLE "agents_messages" ("id" varchar(36) PRIMARY KEY NOT NULL, "threadId
erDiagram
"agent_execution_message_links" |o--|| "agents_messages" : "FOREIGN KEY (messageId) REFERENCES agents_messages (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
+"agent_message_queue" }o--|| "agents_messages" : "FOREIGN KEY (messageId) REFERENCES agents_messages (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"agents_memory_entry_candidates" }o--o| "agents_messages" : "FOREIGN KEY (sourceMessageId) REFERENCES agents_messages (id) ON UPDATE NO ACTION ON DELETE SET NULL MATCH NONE"
"agents_messages" }o--|| "agents_threads" : "FOREIGN KEY (threadId) REFERENCES agents_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
@@ -76,6 +77,15 @@ erDiagram
varchar_36_ messageId PK
INTEGER position
}
+"agent_message_queue" {
+ datetime_3_ createdAt
+ varchar_36_ executionId FK
+ INTEGER id
+ varchar_36_ messageId FK
+ TEXT payload
+ varchar_128_ threadId FK
+ datetime_3_ updatedAt
+}
"agents_memory_entry_candidates" {
varchar_36_ agentId FK
smallint attemptCount
diff --git a/packages/@n8n/db/src/migrations/postgresdb/1790605116208-AddAgentQueueMessageReferences.ts b/packages/@n8n/db/src/migrations/postgresdb/1790605116208-AddAgentQueueMessageReferences.ts
new file mode 100644
index 00000000000..2858d0b43ce
--- /dev/null
+++ b/packages/@n8n/db/src/migrations/postgresdb/1790605116208-AddAgentQueueMessageReferences.ts
@@ -0,0 +1,239 @@
+import { UserError } from 'n8n-workflow';
+import { randomUUID } from 'node:crypto';
+import { z } from 'zod';
+
+import type { MigrationContext, ReversibleMigration } from '../migration-types';
+
+const attachmentSchema = z.object({
+ id: z.string().min(1),
+ fileName: z.string(),
+ mimeType: z.string(),
+ sizeBytes: z.number(),
+});
+const inputSchema = z.object({
+ message: z.string(),
+ resourceId: z.string().min(1),
+ attachments: z.array(attachmentSchema).optional(),
+});
+const payloadSchema = z.discriminatedUnion('kind', [
+ inputSchema.extend({ kind: z.literal('preview') }).passthrough(),
+ inputSchema
+ .extend({
+ kind: z.literal('integration'),
+ modelMessage: z.string(),
+ author: z.record(z.unknown()),
+ platformThreadId: z.string().min(1),
+ messageContext: z
+ .object({
+ platform: z.string(),
+ integrationConnectionId: z.string().min(1),
+ messageId: z.string().optional(),
+ })
+ .passthrough(),
+ })
+ .passthrough(),
+]);
+type QueuePayload = z.infer;
+
+interface QueueRow {
+ id: string;
+ threadId: string;
+ source: string;
+ payload: unknown;
+ executionId: string | null;
+ createdAt: Date | string;
+ updatedAt: Date | string;
+ inputMessageId: string | null;
+ inputResourceId: string | null;
+}
+
+export class AddAgentQueueMessageReferences1790605116208 implements ReversibleMigration {
+ async up(ctx: MigrationContext) {
+ const {
+ schemaBuilder: { addColumns, column, addNotNull, dropColumns, addForeignKey, createIndex },
+ escape,
+ runQuery,
+ } = ctx;
+ await addColumns(
+ 'agent_message_queue',
+ [
+ column('messageId')
+ .varchar(36)
+ .comment('Canonical input created when the queue accepts it'),
+ ],
+ { recreatesOnSqlite: true },
+ );
+ await this.backfillQueue(ctx);
+ await dropColumns('agent_message_queue', ['source'], { recreatesOnSqlite: true });
+ await addNotNull('agent_message_queue', 'messageId', { recreatesOnSqlite: true });
+ await addForeignKey(
+ 'agent_message_queue',
+ 'messageId',
+ ['agents_messages', 'id'],
+ 'FK_agent_message_queue_messageId',
+ 'CASCADE',
+ );
+ await runQuery(`COMMENT ON COLUMN ${escape.tableName('agent_message_queue')}.${escape.columnName('payload')}
+ IS 'Dispatch, authorization, and reply context. Input is stored on the message'`);
+ await createIndex('agent_message_queue', ['messageId'], true);
+ }
+
+ protected async backfillQueue(ctx: MigrationContext) {
+ const { escape, runInBatches } = ctx;
+ await runInBatches(
+ `SELECT CAST(queue.${escape.columnName('id')} AS TEXT) AS ${escape.columnName('id')},
+ queue.${escape.columnName('threadId')}, queue.${escape.columnName('source')},
+ queue.${escape.columnName('payload')}, queue.${escape.columnName('executionId')},
+ queue.${escape.columnName('createdAt')}, queue.${escape.columnName('updatedAt')},
+ message.${escape.columnName('id')} AS ${escape.columnName('inputMessageId')},
+ message.${escape.columnName('resourceId')} AS ${escape.columnName('inputResourceId')}
+ FROM ${escape.tableName('agent_message_queue')} queue
+ LEFT JOIN ${escape.tableName('agent_execution_message_links')} link
+ ON link.${escape.columnName('executionId')} = queue.${escape.columnName('executionId')}
+ AND link.${escape.columnName('direction')} = 'input' AND link.${escape.columnName('position')} = 0
+ LEFT JOIN ${escape.tableName('agents_messages')} message
+ ON message.${escape.columnName('id')} = link.${escape.columnName('messageId')}
+ AND message.${escape.columnName('threadId')} = queue.${escape.columnName('threadId')}
+ ORDER BY queue.${escape.columnName('id')}`,
+ async (rows) => {
+ for (const row of rows) await this.backfillItem(ctx, row);
+ },
+ );
+ }
+
+ private async backfillItem(ctx: MigrationContext, row: QueueRow) {
+ const { escape, runQuery } = ctx;
+ let payload: QueuePayload;
+ try {
+ payload = payloadSchema.parse(ctx.parseJson(row.payload));
+ } catch {
+ // An incomplete conversion must roll back instead of losing an accepted message.
+ throw new UserError(`Cannot migrate queued agent message ${row.id}: invalid payload`);
+ }
+ let messageId = row.inputMessageId;
+ if (row.executionId !== null) {
+ if (!messageId || row.inputResourceId !== payload.resourceId) {
+ throw new UserError(
+ `Cannot migrate queued agent message ${row.id}: input link is missing or invalid`,
+ );
+ }
+ } else {
+ messageId = await this.createInput(ctx, row, payload);
+ }
+ await runQuery(
+ `UPDATE ${escape.tableName('agent_message_queue')}
+ SET ${escape.columnName('messageId')} = :messageId, ${escape.columnName('payload')} = :payload
+ WHERE ${escape.columnName('id')} = :id`,
+ { id: row.id, messageId, payload: JSON.stringify(this.dispatchPayload(payload)) },
+ );
+ }
+
+ private async createInput(ctx: MigrationContext, row: QueueRow, payload: QueuePayload) {
+ const { escape, runQuery } = ctx;
+ const id = randomUUID();
+ const parameters = {
+ id,
+ threadId: row.threadId,
+ resourceId: payload.resourceId,
+ createdAt: row.createdAt,
+ updatedAt: row.updatedAt,
+ };
+ await runQuery(
+ `INSERT INTO ${escape.tableName('agents_resources')}
+ (${['id', 'createdAt', 'updatedAt'].map(escape.columnName).join(', ')})
+ VALUES (:resourceId, :createdAt, :updatedAt) ON CONFLICT (${escape.columnName('id')}) DO NOTHING`,
+ parameters,
+ );
+ await runQuery(
+ `INSERT INTO ${escape.tableName('agents_threads')}
+ (${['id', 'resourceId', 'createdAt', 'updatedAt'].map(escape.columnName).join(', ')})
+ VALUES (:threadId, :resourceId, :createdAt, :updatedAt) ON CONFLICT (${escape.columnName('id')}) DO NOTHING`,
+ parameters,
+ );
+ const origin: Record = { source: row.source };
+ if (payload.kind === 'integration') {
+ origin.integrationConnectionId = payload.messageContext.integrationConnectionId;
+ origin.platformMessageId = payload.messageContext.messageId;
+ origin.platformThreadId = payload.platformThreadId;
+ }
+ const modelContent =
+ payload.kind === 'integration' && payload.modelMessage !== payload.message
+ ? JSON.stringify(this.messageContent(payload.modelMessage, payload.attachments ?? []))
+ : null;
+ await runQuery(
+ `INSERT INTO ${escape.tableName('agents_messages')}
+ (${['id', 'threadId', 'resourceId', 'role', 'type', 'content', 'author', 'origin', 'modelContent', 'modelContextAt', 'createdAt', 'updatedAt'].map(escape.columnName).join(', ')})
+ VALUES (:id, :threadId, :resourceId, 'user', NULL, :content, :author, :origin, :modelContent, NULL, :createdAt, :updatedAt)`,
+ {
+ ...parameters,
+ content: JSON.stringify(this.messageContent(payload.message, payload.attachments ?? [])),
+ author: payload.kind === 'integration' ? JSON.stringify(payload.author) : null,
+ origin: JSON.stringify(origin),
+ modelContent,
+ },
+ );
+ return id;
+ }
+
+ private messageContent(text: string, attachments: Array>) {
+ const files = attachments.map(({ mimeType, ...fileRef }) => ({
+ type: 'file',
+ mediaType: mimeType,
+ fileRef,
+ }));
+ return { role: 'user', content: [...(text ? [{ type: 'text', text }] : []), ...files] };
+ }
+
+ private dispatchPayload(payload: QueuePayload) {
+ if (payload.kind === 'preview') return { kind: 'preview' };
+ const {
+ message: _message,
+ resourceId: _resourceId,
+ attachments: _attachments,
+ modelMessage: _modelMessage,
+ author: _author,
+ platformThreadId: _platformThreadId,
+ messageContext: fullContext,
+ ...dispatch
+ } = payload;
+ const {
+ platform: _platform,
+ integrationConnectionId: _connectionId,
+ messageId: _messageId,
+ ...messageContext
+ } = fullContext;
+ return { ...dispatch, messageContext };
+ }
+
+ async down(ctx: MigrationContext) {
+ const {
+ schemaBuilder: { dropIndex, dropForeignKey, dropColumns, addColumns, column },
+ escape,
+ runQuery,
+ } = ctx;
+ await this.requireEmptyQueue(ctx);
+ await dropIndex('agent_message_queue', ['messageId']);
+ await dropForeignKey(
+ 'agent_message_queue',
+ 'messageId',
+ ['agents_messages', 'id'],
+ 'FK_agent_message_queue_messageId',
+ );
+ await dropColumns('agent_message_queue', ['messageId'], { recreatesOnSqlite: true });
+ await addColumns(
+ 'agent_message_queue',
+ [column('source').varchar(32).notNull.comment('Preview or integration source')],
+ { recreatesOnSqlite: true },
+ );
+ await runQuery(`COMMENT ON COLUMN ${escape.tableName('agent_message_queue')}.${escape.columnName('payload')}
+ IS 'Input, attachment references, identity, and reply context'`);
+ }
+
+ protected async requireEmptyQueue({ runQuery, escape }: MigrationContext) {
+ const rows = await runQuery>(
+ `SELECT ${escape.columnName('id')} FROM ${escape.tableName('agent_message_queue')} LIMIT 1`,
+ );
+ if (rows.length)
+ throw new UserError('The agent message queue must be empty to change its message references');
+ }
+}
diff --git a/packages/@n8n/db/src/migrations/sqlite/1790605116208-AddAgentQueueMessageReferences.ts b/packages/@n8n/db/src/migrations/sqlite/1790605116208-AddAgentQueueMessageReferences.ts
new file mode 100644
index 00000000000..eacda21f7a9
--- /dev/null
+++ b/packages/@n8n/db/src/migrations/sqlite/1790605116208-AddAgentQueueMessageReferences.ts
@@ -0,0 +1,65 @@
+import type { MigrationContext } from '../migration-types';
+import { AddAgentQueueMessageReferences1790605116208 as BaseMigration } from '../postgresdb/1790605116208-AddAgentQueueMessageReferences';
+
+export class AddAgentQueueMessageReferences1790605116208 extends BaseMigration {
+ async up(ctx: MigrationContext) {
+ const {
+ schemaBuilder: { addNotNull, addForeignKey, createIndex },
+ escape,
+ runQuery,
+ } = ctx;
+ // SQLite rebuilds the table for NOT NULL and otherwise forgets deleted queue IDs.
+ const sequence = await this.readQueueSequence(ctx);
+ await runQuery(`ALTER TABLE ${escape.tableName('agent_message_queue')}
+ ADD COLUMN ${escape.columnName('messageId')} varchar(36)`);
+ await this.backfillQueue(ctx);
+ await runQuery(
+ `ALTER TABLE ${escape.tableName('agent_message_queue')} DROP COLUMN ${escape.columnName('source')}`,
+ );
+ await addNotNull('agent_message_queue', 'messageId', { recreatesOnSqlite: true });
+ await addForeignKey(
+ 'agent_message_queue',
+ 'messageId',
+ ['agents_messages', 'id'],
+ 'FK_agent_message_queue_messageId',
+ 'CASCADE',
+ );
+ await createIndex('agent_message_queue', ['messageId'], true);
+ await this.restoreQueueSequence(ctx, sequence);
+ }
+
+ async down(ctx: MigrationContext) {
+ const {
+ schemaBuilder: { dropIndex, dropColumns },
+ escape,
+ runQuery,
+ } = ctx;
+ await this.requireEmptyQueue(ctx);
+ const sequence = await this.readQueueSequence(ctx);
+ await dropIndex('agent_message_queue', ['messageId']);
+ await dropColumns('agent_message_queue', ['messageId'], { recreatesOnSqlite: true });
+ await runQuery(
+ `ALTER TABLE ${escape.tableName('agent_message_queue')} ADD COLUMN ${escape.columnName('source')} varchar(32) NOT NULL`,
+ );
+ await this.restoreQueueSequence(ctx, sequence);
+ }
+
+ private async readQueueSequence({ runQuery, tablePrefix }: MigrationContext) {
+ const [sequence] = await runQuery>(
+ 'SELECT CAST(seq AS TEXT) AS seq FROM sqlite_sequence WHERE name = :tableName',
+ { tableName: `${tablePrefix}agent_message_queue` },
+ );
+ return sequence?.seq;
+ }
+
+ private async restoreQueueSequence(
+ { runQuery, tablePrefix }: MigrationContext,
+ sequence: string | undefined,
+ ) {
+ if (sequence === undefined) return;
+ await runQuery('UPDATE sqlite_sequence SET seq = :seq WHERE name = :tableName', {
+ seq: sequence,
+ tableName: `${tablePrefix}agent_message_queue`,
+ });
+ }
+}
diff --git a/packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts
index 9709457fc26..3f789cef813 100644
--- a/packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts
+++ b/packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts
@@ -86,7 +86,7 @@ describe('AgentExecutionService', () => {
beforeEach(() => {
vi.clearAllMocks();
- messageRepository.createExecutionInput.mockResolvedValue('message-1');
+ messageRepository.createInput.mockResolvedValue(mock({ id: 'message-1' }));
messageRepository.findExecutionInputs.mockResolvedValue(new Map());
messageRepository.copyExecutionInputs.mockResolvedValue([]);
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 b9d1289f5cb..dd5302382b7 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
@@ -15,10 +15,11 @@ import type { AgentQueuedPreviewStreamService } from '../agent-queued-preview-st
import type { AgentTestRunService } from '../agent-test-run.service';
import type { AgentExecutionThread } from '../entities/agent-execution-thread.entity';
import type { AgentMessageQueue } from '../entities/agent-message-queue.entity';
+import type { AgentMessageEntity } from '../entities/agent-message.entity';
import type { AgentChatBridge } from '../integrations/agent-chat-bridge';
import type { ChatIntegrationService } from '../integrations/chat-integration.service';
import type { AgentMessageQueueRepository } from '../repositories/agent-message-queue.repository';
-import type { QueuedIntegrationMessage } from '../types/agent-queued-message';
+import type { AgentQueueDispatch, QueuedIntegrationMessage } from '../types/agent-queued-message';
vi.mock('@/permissions.ee/check-access', () => ({ userHasScopes: vi.fn() }));
@@ -35,21 +36,36 @@ describe('AgentMessageQueueConsumer', () => {
let consumer: AgentMessageQueueConsumer;
function claim(threadId: string, integration = false): ClaimedAgentMessage {
+ const payload: ClaimedAgentMessage['payload'] = integration
+ ? mock({
+ kind: 'integration',
+ message: 'input',
+ credentialId: 'credential',
+ platformThreadId: 'slack:channel:thread',
+ messageContext: {
+ platform: 'slack',
+ integrationConnectionId: 'slack:credential',
+ target: { type: 'thread', threadId: 'slack:channel:thread' },
+ updatedAt: '2026-06-01T12:00:00.000Z',
+ },
+ })
+ : { kind: 'preview', message: 'input', resourceId: 'draft-chat:user' };
return {
+ payload,
item: mock({
id: threadId,
threadId,
- source: integration ? 'slack' : 'chat',
+ message: mock({ origin: { source: integration ? 'slack' : 'chat' } }),
payload: integration
- ? mock({
- kind: 'integration',
- message: 'input',
- credentialId: 'credential',
- platformThreadId: 'slack:channel:thread',
- })
- : { kind: 'preview', message: 'input', userId: 'user', resourceId: 'draft-chat:user' },
+ ? mock({ kind: 'integration', credentialId: 'credential' })
+ : { kind: 'preview' },
+ }),
+ thread: mock({
+ id: threadId,
+ agentId: 'agent',
+ projectId: 'project',
+ ownerId: integration ? null : 'user',
}),
- thread: mock({ id: threadId, agentId: 'agent', projectId: 'project' }),
admission: {
executionId: `execution-${threadId}`,
startedAt: new Date(),
@@ -62,6 +78,7 @@ describe('AgentMessageQueueConsumer', () => {
agentName: 'Agent',
projectId: 'project',
userMessage: 'input',
+ source: integration ? 'slack' : 'chat',
access: { accessScope: 'user', ownerId: 'user' },
},
};
@@ -103,8 +120,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;
+ first.payload.message = 'edited first message';
+ first.recording.userMessage = first.payload.message;
const second = claim('second');
const waiting = createDeferredPromise();
repository.findThreadIds.mockResolvedValue(['first', 'second']);
@@ -126,6 +143,11 @@ describe('AgentMessageQueueConsumer', () => {
expect(queue.settle).toHaveBeenCalledWith('second', second.admission.executionId),
);
expect(queue.settle).not.toHaveBeenCalledWith('first', first.admission.executionId);
+ expect(users.findByIdWithRole).toHaveBeenCalledWith(first.thread.ownerId);
+ expect(chatExecutions.register).toHaveBeenCalledWith(
+ expect.objectContaining({ userId: first.thread.ownerId, threadId: 'first' }),
+ expect.any(AbortController),
+ );
expect(sender.send).toHaveBeenCalledWith({
type: 'execution-started',
inputMessageIds: ['message-1'],
@@ -143,6 +165,7 @@ describe('AgentMessageQueueConsumer', () => {
expect.objectContaining({
sessionId: 'first',
sessionMode: 'existing',
+ source: 'chat',
admittedExecution: first.admission,
}),
);
@@ -154,31 +177,58 @@ describe('AgentMessageQueueConsumer', () => {
);
});
- it.each(['revoked access', 'invalid draft'] as const)(
- 'records %s on the claimed execution and releases the item',
- async (failure) => {
- const item = claim('session');
- if (failure === 'revoked access') vi.mocked(userHasScopes).mockResolvedValue(false);
- else
+ it.each([
+ 'missing owner',
+ 'deleted owner',
+ 'disabled owner',
+ 'revoked access',
+ 'invalid draft',
+ ] as const)('records %s on the claimed execution and releases the item', async (failure) => {
+ const item = claim('session');
+ switch (failure) {
+ case 'missing owner':
+ item.thread.ownerId = null;
+ break;
+ case 'deleted owner':
+ users.findByIdWithRole.mockResolvedValue(null);
+ break;
+ case 'disabled owner':
+ users.findByIdWithRole.mockResolvedValue(mock({ id: 'user', disabled: true }));
+ break;
+ case 'revoked access':
+ vi.mocked(userHasScopes).mockResolvedValue(false);
+ break;
+ case 'invalid draft':
testRuns.prepareDraftRun.mockResolvedValue({
status: 'agent_misconfigured',
missing: ['model'],
});
- repository.findThreadIds.mockResolvedValue(['session']);
- queue.claimNext.mockResolvedValueOnce(item).mockResolvedValue(null);
- consumer.start();
- await vi.waitFor(() =>
- expect(queue.settle).toHaveBeenCalledWith('session', item.admission.executionId),
- );
- expect(queue.recordFailure).toHaveBeenCalledWith(
- item,
- expect.any(Error),
- expect.any(AbortSignal),
- );
- expect(testRuns.executePreparedDraftRun).not.toHaveBeenCalled();
- expect(sender.send).toHaveBeenCalledWith(expect.objectContaining({ type: 'error' }));
- },
- );
+ break;
+ }
+ repository.findThreadIds.mockResolvedValue(['session']);
+ queue.claimNext.mockResolvedValueOnce(item).mockResolvedValue(null);
+ consumer.start();
+ await vi.waitFor(() =>
+ expect(queue.settle).toHaveBeenCalledWith('session', item.admission.executionId),
+ );
+ expect(queue.recordFailure).toHaveBeenCalledWith(
+ item,
+ expect.objectContaining({
+ message:
+ failure === 'invalid draft'
+ ? 'This agent is not ready to run yet.'
+ : 'You can no longer execute this agent',
+ }),
+ expect.any(AbortSignal),
+ );
+ expect(testRuns.executePreparedDraftRun).not.toHaveBeenCalled();
+ if (failure !== 'invalid draft') {
+ expect(chatExecutions.register).not.toHaveBeenCalled();
+ expect(testRuns.prepareDraftRun).not.toHaveBeenCalled();
+ }
+ if (failure === 'missing owner') expect(users.findByIdWithRole).not.toHaveBeenCalled();
+ expect(sender.send).toHaveBeenCalledWith(expect.objectContaining({ type: 'error' }));
+ });
it('leaves integration work pending until its bridge is available and stops admission during shutdown', async () => {
const item = claim('session', true);
@@ -187,9 +237,16 @@ describe('AgentMessageQueueConsumer', () => {
const release = vi.fn();
repository.findThreadIds.mockResolvedValue(['session']);
queue.claimNext.mockImplementationOnce(async (_threadId, canConsume) => {
+ item.item.message.origin = null;
+ await expect(canConsume(item.item, item.thread, {})).rejects.toThrow(
+ 'Queued integration input has no source',
+ );
+ expect(repository.findPublishedConnection).not.toHaveBeenCalled();
+ item.item.message.origin = { source: 'slack' };
expect(await canConsume(item.item, item.thread, {})).toBe(false);
integrations.acquireQueueBridge.mockReturnValue({ bridge: mock(), release });
expect(await canConsume(item.item, item.thread, {})).toBe(true);
+ expect(integrations.acquireQueueBridge).toHaveBeenCalledWith('agent', 'slack', 'credential');
checked.resolve();
await resume.promise;
expect(await canConsume(item.item, item.thread, {})).toBe(false);
@@ -234,7 +291,7 @@ describe('AgentMessageQueueConsumer', () => {
if (outcome === 'accepted') {
await vi.waitFor(() =>
expect(bridge.consumeQueuedMessage).toHaveBeenCalledWith(
- item.item.payload,
+ item.payload,
item.thread.id,
item.admission,
expect.any(AbortSignal),
@@ -242,6 +299,12 @@ describe('AgentMessageQueueConsumer', () => {
),
);
expect(release).not.toHaveBeenCalled();
+ expect(repository.findPublishedConnection).toHaveBeenCalledWith(
+ 'agent',
+ 'project',
+ 'slack',
+ 'credential',
+ );
}
} finally {
running.resolve();
diff --git a/packages/cli/src/modules/agents/agent-execution.service.ts b/packages/cli/src/modules/agents/agent-execution.service.ts
index defdff62679..38bd12a9054 100644
--- a/packages/cli/src/modules/agents/agent-execution.service.ts
+++ b/packages/cli/src/modules/agents/agent-execution.service.ts
@@ -192,8 +192,12 @@ export class AgentExecutionService {
params: StartExecutionParams,
startedAt: Date,
ctx: OperationContext,
+ lockedQueueThread?: AgentExecutionThread,
): Promise {
- const prepared = await this.prepareThread(params, ctx);
+ // The queue already locked the session. Its consumer validates the owner before execution.
+ const prepared = lockedQueueThread
+ ? { thread: lockedQueueThread, created: false }
+ : await this.prepareThread(params, ctx);
const { queueItem, predecessorId } = await this.checkAdmission(params, ctx);
const execution = this.agentExecutionRepository.create({
threadId: params.threadId,
@@ -218,7 +222,13 @@ export class AgentExecutionService {
attachments: null,
});
const inserted = await this.agentExecutionRepository.saveInContext(execution, ctx);
- const inputMessageIds = await this.reserveInput(params, inserted.id, predecessorId, ctx);
+ const inputMessageIds = await this.reserveInput(
+ params,
+ inserted.id,
+ predecessorId,
+ queueItem?.messageId,
+ ctx,
+ );
if (
queueItem &&
!(await this.queueRepository.linkExecution(
@@ -241,6 +251,7 @@ export class AgentExecutionService {
params: StartExecutionParams,
executionId: string,
predecessorId: string | undefined,
+ queuedMessageId: string | undefined,
ctx: OperationContext,
): Promise {
if (predecessorId) {
@@ -252,10 +263,13 @@ export class AgentExecutionService {
);
}
if (params.resumeRunId || params.userMessage === null) return [];
+ if (queuedMessageId) {
+ await this.messageRepository.linkExecutionInput(executionId, queuedMessageId, params, ctx);
+ return [queuedMessageId];
+ }
const [content] = buildInboundUserMessage(params.userMessage, params.attachments ?? []);
- const id = await this.messageRepository.createExecutionInput(
+ const message = await this.messageRepository.createInput(
{
- executionId,
threadId: params.threadId,
resourceId: params.resourceId,
content,
@@ -268,7 +282,8 @@ export class AgentExecutionService {
},
ctx,
);
- return [id];
+ await this.messageRepository.linkExecutionInput(executionId, message.id, params, ctx);
+ return [message.id];
}
private async checkAdmission(params: StartExecutionParams, ctx: OperationContext) {
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 d090d515597..ccba5b2c232 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
@@ -1,9 +1,9 @@
import type { AgentSseEvent } from '@n8n/api-types';
import { Logger } from '@n8n/backend-common';
-import { UserRepository } from '@n8n/db';
+import { UserRepository, type User } from '@n8n/db';
import { OnShutdown } from '@n8n/decorators';
import { Service } from '@n8n/di';
-import { OperationalError, UserError } from 'n8n-workflow';
+import { OperationalError, UnexpectedError, UserError } from 'n8n-workflow';
import { CredentialsService } from '@/credentials/credentials.service';
import { userHasScopes } from '@/permissions.ee/check-access';
@@ -15,6 +15,7 @@ import { AgentMessageQueueService, type ClaimedAgentMessage } from './agent-mess
import { AgentQueuedPreviewStreamService } from './agent-queued-preview-stream.service';
import { emitChunkEvents } from './agent-sse-stream';
import { AgentTestRunService } from './agent-test-run.service';
+import type { AgentExecutionThread } from './entities/agent-execution-thread.entity';
import type { AgentChatBridge } from './integrations/agent-chat-bridge';
import { ChatIntegrationService } from './integrations/chat-integration.service';
import { AgentMessageQueueRepository } from './repositories/agent-message-queue.repository';
@@ -95,10 +96,12 @@ export class AgentMessageQueueConsumer {
const claim = await this.queue.claimNext(threadId, async (item, thread, ctx) => {
if (this.stopped) return false;
if (item.payload.kind === 'preview') return true;
+ const source = item.message.origin?.source;
+ if (!source) throw new UnexpectedError('Queued integration input has no source');
const connection = await this.repository.findPublishedConnection(
thread.agentId,
thread.projectId,
- item.source,
+ source,
item.payload.credentialId,
ctx,
);
@@ -107,7 +110,7 @@ export class AgentMessageQueueConsumer {
if (!connection) return true;
lease = this.integrations.acquireQueueBridge(
thread.agentId,
- item.source,
+ source,
item.payload.credentialId,
);
return lease !== undefined;
@@ -122,22 +125,23 @@ export class AgentMessageQueueConsumer {
/** Execute a claimed message, deliver its output, and settle its queue item. */
private async consume(claim: ClaimedAgentMessage, bridge?: AgentChatBridge): Promise {
- const { item, thread, admission } = claim;
+ const { item, thread, admission, payload } = claim;
const controller = new AbortController();
const signal = AbortSignal.any([
controller.signal,
this.executionService.getAbortSignal(admission.executionId),
]);
const sender =
- item.payload.kind === 'preview' ? this.previewStreams.createSender(item.id) : undefined;
+ payload.kind === 'preview' ? this.previewStreams.createSender(item.id) : undefined;
try {
- if (item.payload.kind === 'preview' && sender) {
+ if (payload.kind === 'preview' && sender) {
+ const user = await this.getPreviewOwner(thread);
this.chatExecutionService.register(
{
projectId: thread.projectId,
agentId: thread.agentId,
threadId: thread.id,
- userId: item.payload.userId,
+ userId: user.id,
executionId: admission.executionId,
},
controller,
@@ -147,11 +151,11 @@ export class AgentMessageQueueConsumer {
executionId: admission.executionId,
sessionId: thread.id,
inputMessageIds: admission.inputMessageIds,
- message: item.payload.message,
+ message: payload.message,
});
await this.chatExecutionService.settle(
admission.executionId,
- async () => await this.consumePreview(claim, signal, sender.send),
+ async () => await this.consumePreview(claim, user, signal, sender.send),
);
} else {
await this.consumeIntegration(claim, signal, bridge);
@@ -173,14 +177,9 @@ export class AgentMessageQueueConsumer {
}
}
- private async consumePreview(
- claim: ClaimedAgentMessage,
- signal: AbortSignal,
- send: (event: AgentSseEvent) => void,
- ): Promise {
- const { item, thread, admission } = claim;
- if (item.payload.kind !== 'preview') return;
- const user = await this.userRepository.findByIdWithRole(item.payload.userId);
+ private async getPreviewOwner(thread: AgentExecutionThread): Promise {
+ if (!thread.ownerId) throw new UserError('You can no longer execute this agent');
+ const user = await this.userRepository.findByIdWithRole(thread.ownerId);
if (
!user ||
user.disabled ||
@@ -188,6 +187,17 @@ export class AgentMessageQueueConsumer {
) {
throw new UserError('You can no longer execute this agent');
}
+ return user;
+ }
+
+ private async consumePreview(
+ claim: ClaimedAgentMessage,
+ user: User,
+ signal: AbortSignal,
+ send: (event: AgentSseEvent) => void,
+ ): Promise {
+ const { thread, admission, payload } = claim;
+ if (payload.kind !== 'preview') return;
signal.throwIfAborted();
const prepared = await this.testRunService.prepareDraftRun({
agentId: thread.agentId,
@@ -222,12 +232,12 @@ export class AgentMessageQueueConsumer {
agentId: thread.agentId,
projectId: thread.projectId,
user,
- message: item.payload.message,
- attachments: item.payload.attachments,
+ message: payload.message,
+ attachments: payload.attachments,
sessionId: thread.id,
sessionMode: 'existing',
previewChat: true,
- source: item.source,
+ source: claim.recording.source,
admittedExecution: admission,
abortSignal: signal,
errorMode: 'forward',
@@ -242,16 +252,16 @@ export class AgentMessageQueueConsumer {
signal: AbortSignal,
bridge?: AgentChatBridge,
): Promise {
- const { item, thread } = claim;
- if (item.payload.kind !== 'integration') return;
+ const { thread, payload } = claim;
+ if (payload.kind !== 'integration') return;
const connection = await this.repository.findPublishedConnection(
thread.agentId,
thread.projectId,
- item.source,
- item.payload.credentialId,
+ payload.messageContext.platform,
+ payload.credentialId,
);
if (!connection) throw new UserError('The message integration is no longer configured');
if (!bridge) throw new OperationalError('The message integration is unavailable');
- await bridge.consumeQueuedMessage(item.payload, thread.id, claim.admission, signal, connection);
+ await bridge.consumeQueuedMessage(payload, thread.id, claim.admission, signal, connection);
}
}
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 e9182d63c93..7db1e9a0c19 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,8 @@
-import type { AgentChatQueueResponse } from '@n8n/api-types';
+import type { Message } from '@n8n/agents';
+import type { AgentChatQueueResponse, AgentMessageAuthor } from '@n8n/api-types';
import { TransactionRunner, type OperationContext } from '@n8n/db';
import { Service } from '@n8n/di';
-import { OperationalError, UserError } from 'n8n-workflow';
+import { OperationalError, UnexpectedError, UserError } from 'n8n-workflow';
import { ConflictError, BadRequestError, NotFoundError } from '@n8n/errors';
@@ -10,17 +11,27 @@ import { AgentExecutionUpdateBroadcaster } from './agent-execution-update-broadc
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';
+import type { AgentMessageOrigin } from './entities/agent-message.entity';
import { ExecutionRecorder } from './execution-recorder';
import { N8NCheckpointStorage } from './integrations/n8n-checkpoint-storage';
import { AgentExecutionRepository } from './repositories/agent-execution.repository';
import { AgentExecutionThreadRepository } from './repositories/agent-execution-thread.repository';
import { AgentMessageQueueRepository } from './repositories/agent-message-queue.repository';
+import { AgentMessageRepository } from './repositories/agent-message.repository';
import { AgentRepository } from './repositories/agent.repository';
-import type { AgentExecutionAdmission, AgentQueuedMessage } from './types/agent-queued-message';
+import type {
+ AgentExecutionAdmission,
+ AgentQueuedMessage,
+ AgentQueueDispatch,
+ QueuedIntegrationMessage,
+ QueuedPreviewMessage,
+} from './types/agent-queued-message';
import { canContinueThreadInPreview, type AgentSessionMode } from './utils/agent-thread-access';
+import { buildInboundUserMessage, readInboundUserMessage } from './utils/inbound-attachments';
export interface ClaimedAgentMessage {
item: AgentMessageQueue;
+ payload: Omit | QueuedIntegrationMessage;
thread: AgentExecutionThread;
/** Admission identifies the committed execution that the turn pipeline must reuse. */
admission: AgentExecutionAdmission;
@@ -42,6 +53,7 @@ export class AgentMessageQueueService {
private readonly agentRepository: AgentRepository,
private readonly attachments: AgentChatAttachmentService,
private readonly updates: AgentExecutionUpdateBroadcaster,
+ private readonly messages: AgentMessageRepository,
) {}
/** Save a pending message. It is durably accepted when the transaction commits. */
@@ -73,7 +85,7 @@ export class AgentMessageQueueService {
},
ctx,
);
- const inserted = await this.repository.enqueue(input.threadId, input.source, payload, ctx);
+ const inserted = await this.enqueueMessage(input.threadId, input.source, payload, ctx);
// Register delivery before another main can see the committed item.
onInserted?.(inserted.id);
return inserted;
@@ -83,6 +95,49 @@ export class AgentMessageQueueService {
return item;
}
+ private async enqueueMessage(
+ threadId: string,
+ source: string,
+ payload: AgentQueuedMessage,
+ ctx: OperationContext,
+ ): Promise {
+ const { message, resourceId, attachments = [], ...dispatch } = payload;
+ const [content] = buildInboundUserMessage(message, attachments);
+ let queueDispatch: AgentQueueDispatch = { kind: 'preview' };
+ let modelContent: Message | undefined;
+ let author: AgentMessageAuthor | undefined;
+ let origin: AgentMessageOrigin = { source };
+ if (dispatch.kind === 'integration') {
+ const {
+ modelMessage,
+ author: inputAuthor,
+ platformThreadId,
+ messageContext: fullContext,
+ ...integrationDispatch
+ } = dispatch;
+ const {
+ platform: _platform,
+ integrationConnectionId,
+ messageId,
+ ...messageContext
+ } = fullContext;
+ queueDispatch = { ...integrationDispatch, messageContext };
+ [modelContent] = buildInboundUserMessage(modelMessage, attachments);
+ author = inputAuthor;
+ origin = {
+ source,
+ integrationConnectionId,
+ platformMessageId: messageId,
+ platformThreadId,
+ };
+ }
+ const input = await this.messages.createInput(
+ { threadId, resourceId, content, modelContent, author, origin },
+ ctx,
+ );
+ return await this.repository.enqueue(threadId, input.id, queueDispatch, ctx);
+ }
+
async listPending(input: {
projectId: string;
agentId: string;
@@ -99,8 +154,7 @@ export class AgentMessageQueueService {
.filter((item) => item.payload.kind === 'preview')
.map((item) => ({
id: item.id,
- message: item.payload.message,
- attachments: item.payload.attachments,
+ ...readInboundUserMessage(item.message.content),
createdAt: item.createdAt.toISOString(),
})),
};
@@ -113,7 +167,7 @@ export class AgentMessageQueueService {
userId: string;
queueId: string;
}): Promise {
- const item = await this.txRunner.run({}, async (ctx) => {
+ const removed = 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);
@@ -126,10 +180,14 @@ export class AgentMessageQueueService {
) {
throw new ConflictError('This message has already started');
}
- return item;
+ const attachmentIds = readInboundUserMessage(item.message.content).attachments.map(
+ ({ id }) => id,
+ );
+ await this.messages.clearPendingInput(item.messageId, ctx);
+ return { threadId: item.threadId, attachmentIds };
});
- this.updates.notifyQueueUpdated(item.threadId);
- await this.attachments.deleteByIds(item.payload.attachments?.map(({ id }) => id) ?? []);
+ this.updates.notifyQueueUpdated(removed.threadId);
+ await this.attachments.deleteByIds(removed.attachmentIds);
}
async updatePending(input: {
@@ -149,15 +207,11 @@ export class AgentMessageQueueService {
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)
+ const { attachments } = readInboundUserMessage(item.message.content);
+ if (!message && !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');
+ const [content] = buildInboundUserMessage(message, attachments);
+ await this.messages.updatePendingInput(item.messageId, content, ctx);
});
this.updates.notifyQueueUpdated(input.threadId);
}
@@ -197,11 +251,18 @@ export class AgentMessageQueueService {
await this.repository.removeActive(threadId, active.executionId, ctx);
const item = await this.repository.findHead(threadId, ctx);
if (!item || !(await canConsume(item, thread, ctx))) return null;
- const recording = this.recordingFor(item, thread);
+ const payload = this.restoreInput(item);
+ const recording = this.recordingFor(item, thread, payload);
+ await this.threadRepository.bumpUpdatedAt(threadId, ctx);
// Reservation creates the running execution and links it to this item in one transaction.
// Runtime work starts only after the transaction commits.
- const reservation = await this.executionService.reserveExecution(recording, new Date(), ctx);
- return { item, thread, recording, reservation };
+ const reservation = await this.executionService.reserveExecution(
+ recording,
+ new Date(),
+ ctx,
+ thread,
+ );
+ return { item, thread, payload, recording, reservation };
});
if (!claimed) return null;
this.executionService.activateExecution(claimed.reservation, claimed.recording);
@@ -209,6 +270,7 @@ export class AgentMessageQueueService {
if (!execution.startedAt) throw new OperationalError('Queued execution has no start time');
return {
item: claimed.item,
+ payload: claimed.payload,
thread: claimed.thread,
recording: claimed.recording,
admission: {
@@ -262,6 +324,7 @@ export class AgentMessageQueueService {
private recordingFor(
item: AgentMessageQueue,
thread: AgentExecutionThread,
+ payload: ClaimedAgentMessage['payload'],
): StartExecutionParams {
return {
threadId: thread.id,
@@ -271,18 +334,37 @@ export class AgentMessageQueueService {
access: { accessScope: thread.accessScope, ownerId: thread.ownerId },
sessionMode: 'existing',
queueItemId: item.id,
- userMessage: item.payload.message,
- resourceId: item.payload.resourceId,
- ...(item.payload.kind === 'integration' && {
- messageOrigin: {
- integrationConnectionId: item.payload.messageContext.integrationConnectionId,
- platformMessageId: item.payload.messageContext.messageId,
- platformThreadId: item.payload.platformThreadId,
- },
- }),
- source: item.source,
- attachments: item.payload.attachments,
- author: item.payload.kind === 'integration' ? item.payload.author : undefined,
+ userMessage: payload.message,
+ resourceId: payload.resourceId,
+ source: item.message.origin?.source ?? undefined,
+ attachments: payload.attachments,
+ author: payload.kind === 'integration' ? payload.author : undefined,
+ };
+ }
+
+ private restoreInput(item: AgentMessageQueue): ClaimedAgentMessage['payload'] {
+ const input = {
+ ...readInboundUserMessage(item.message.content),
+ resourceId: item.message.resourceId,
+ };
+ if (item.payload.kind === 'preview') return { ...item.payload, ...input };
+ if (!item.message.author) throw new UnexpectedError('Queued integration input has no author');
+ const { origin } = item.message;
+ if (!origin?.source || !origin.integrationConnectionId || !origin.platformThreadId)
+ throw new UnexpectedError('Queued integration input has incomplete origin');
+ return {
+ ...item.payload,
+ ...input,
+ platformThreadId: origin.platformThreadId,
+ messageContext: {
+ ...item.payload.messageContext,
+ platform: origin.source,
+ integrationConnectionId: origin.integrationConnectionId,
+ ...(origin.platformMessageId !== undefined ? { messageId: origin.platformMessageId } : {}),
+ },
+ author: item.message.author,
+ modelMessage: readInboundUserMessage(item.message.modelContent ?? item.message.content)
+ .message,
};
}
}
diff --git a/packages/cli/src/modules/agents/entities/agent-message-queue.entity.ts b/packages/cli/src/modules/agents/entities/agent-message-queue.entity.ts
index cb5cc2d44a3..4444fe11b96 100644
--- a/packages/cli/src/modules/agents/entities/agent-message-queue.entity.ts
+++ b/packages/cli/src/modules/agents/entities/agent-message-queue.entity.ts
@@ -12,7 +12,8 @@ import {
import { AgentExecutionThread } from './agent-execution-thread.entity';
import { AgentExecution } from './agent-execution.entity';
-import type { AgentQueuedMessage } from '../types/agent-queued-message';
+import { AgentMessageEntity } from './agent-message.entity';
+import type { AgentQueueDispatch } from '../types/agent-queued-message';
@Entity({ name: 'agent_message_queue' })
@Index(['threadId', 'id'])
@@ -33,11 +34,22 @@ export class AgentMessageQueue extends WithTimestamps {
@Column({ type: 'varchar', length: 128 })
threadId: string;
- @Column({ type: 'varchar', length: 32, comment: 'Preview or integration source' })
- source: string;
+ @ManyToOne(() => AgentMessageEntity, { nullable: false, onDelete: 'CASCADE' })
+ @JoinColumn({ name: 'messageId', foreignKeyConstraintName: 'FK_agent_message_queue_messageId' })
+ message: Relation;
- @JsonColumn({ comment: 'Input, attachment references, identity, and reply context' })
- payload: AgentQueuedMessage;
+ @Index({ unique: true })
+ @Column({
+ type: 'varchar',
+ length: 36,
+ comment: 'Canonical input created when the queue accepts it',
+ })
+ messageId: string;
+
+ @JsonColumn({
+ comment: 'Dispatch, authorization, and reply context. Input is stored on the message',
+ })
+ payload: AgentQueueDispatch;
@ManyToOne(() => AgentExecution, { nullable: true, onDelete: 'NO ACTION' })
@JoinColumn({ name: 'executionId' })
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 e2e7e22f83a..e18ab059c8f 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
@@ -5,7 +5,7 @@ import { isDraftIntegration } from '@n8n/api-types';
import { AgentMessageQueue } from '../entities/agent-message-queue.entity';
import { Agent } from '../entities/agent.entity';
-import type { AgentQueuedMessage } from '../types/agent-queued-message';
+import type { AgentQueueDispatch } from '../types/agent-queued-message';
@Service()
export class AgentMessageQueueRepository extends BaseRepository {
@@ -15,22 +15,29 @@ export class AgentMessageQueueRepository extends BaseRepository {
super(AgentMessageEntity, dataSource.manager, transactionRunner);
}
- async createExecutionInput(
+ async createInput(
params: {
- executionId: string;
threadId: string;
resourceId: string;
content: AgentMessage;
+ modelContent?: AgentMessage;
author?: AgentMessageAuthor;
origin: AgentMessageOrigin;
},
ctx: OperationContext,
- ): Promise {
+ ): Promise {
const manager = this.managerFor(ctx);
await manager
.createQueryBuilder()
@@ -60,17 +60,54 @@ export class AgentMessageRepository extends BaseRepository {
content: params.content,
author: params.author ?? null,
origin: params.origin,
- modelContent: null,
+ modelContent:
+ params.modelContent && !isDeepStrictEqual(params.content, params.modelContent)
+ ? params.modelContent
+ : null,
modelContextAt: null,
});
- await manager.save(message);
+ return await manager.save(message);
+ }
+
+ async linkExecutionInput(
+ executionId: string,
+ messageId: string,
+ scope: { threadId: string; resourceId: string },
+ ctx: OperationContext,
+ ): Promise {
+ const manager = this.managerFor(ctx);
+ if (
+ !(await manager.existsBy(AgentMessageEntity, {
+ id: messageId,
+ threadId: scope.threadId,
+ resourceId: scope.resourceId,
+ }))
+ ) {
+ throw new UnexpectedError('The input message does not belong to this session and resource');
+ }
await manager.insert(AgentExecutionMessageLink, {
- executionId: params.executionId,
- messageId: message.id,
+ executionId,
+ messageId,
direction: 'input',
position: 0,
});
- return message.id;
+ }
+
+ async updatePendingInput(
+ id: string,
+ content: AgentMessage,
+ ctx: OperationContext,
+ ): Promise {
+ await this.managerFor(ctx).save(this.create({ id, content, modelContent: null }));
+ }
+
+ async clearPendingInput(id: string, ctx: OperationContext): Promise {
+ await this.managerFor(ctx).update(AgentMessageEntity, id, {
+ content: { role: 'user', content: [] },
+ author: null,
+ modelContent: null,
+ modelContextAt: null,
+ });
}
async copyExecutionInputs(
diff --git a/packages/cli/src/modules/agents/types/agent-queued-message.ts b/packages/cli/src/modules/agents/types/agent-queued-message.ts
index 45612d52919..759572ec4e4 100644
--- a/packages/cli/src/modules/agents/types/agent-queued-message.ts
+++ b/packages/cli/src/modules/agents/types/agent-queued-message.ts
@@ -35,6 +35,19 @@ export interface QueuedIntegrationMessage extends QueuedMessageInput {
export type AgentQueuedMessage = QueuedPreviewMessage | QueuedIntegrationMessage;
+/** Queue storage keeps dispatch data. Conversation input belongs to the referenced message. */
+export type AgentQueueDispatch =
+ | { kind: 'preview' }
+ | (Omit<
+ QueuedIntegrationMessage,
+ keyof QueuedMessageInput | 'modelMessage' | 'author' | 'platformThreadId' | 'messageContext'
+ > & {
+ messageContext: Omit<
+ IntegrationMessageContext,
+ 'platform' | 'integrationConnectionId' | 'messageId'
+ >;
+ });
+
/** A committed execution reservation. Runtime preparation must reuse it. */
export interface AgentExecutionAdmission {
executionId: string;
diff --git a/packages/cli/src/modules/agents/utils/__tests__/inbound-attachments.test.ts b/packages/cli/src/modules/agents/utils/__tests__/inbound-attachments.test.ts
index 6cf881efd88..a08c0863391 100644
--- a/packages/cli/src/modules/agents/utils/__tests__/inbound-attachments.test.ts
+++ b/packages/cli/src/modules/agents/utils/__tests__/inbound-attachments.test.ts
@@ -1,7 +1,12 @@
-import type { Message } from '@n8n/agents';
+import type { AgentMessage, Message } from '@n8n/agents';
import { MAX_AGENT_CHAT_ATTACHMENT_MIMETYPE_LENGTH } from '@n8n/api-types';
+import { UnexpectedError } from 'n8n-workflow';
-import { buildInboundUserMessage, resolveInboundMimeType } from '../inbound-attachments';
+import {
+ buildInboundUserMessage,
+ readInboundUserMessage,
+ resolveInboundMimeType,
+} from '../inbound-attachments';
// Minimal real PNG header (magic bytes + IHDR chunk) so file-type can sniff it.
const PNG_BYTES = Buffer.from([
@@ -33,6 +38,13 @@ describe('buildInboundUserMessage', () => {
fileRef: { id: 'att-2', fileName: 'voice.ogg', sizeBytes: 100 },
},
]);
+ expect(readInboundUserMessage(message)).toEqual({
+ message: 'look at these',
+ attachments: [
+ { id: 'att-1', fileName: 'photo.png', mimeType: 'image/png', sizeBytes: 33 },
+ { id: 'att-2', fileName: 'voice.ogg', mimeType: 'audio/ogg', sizeBytes: 100 },
+ ],
+ });
});
it('omits the text part for attachment-only sends', () => {
@@ -45,6 +57,18 @@ describe('buildInboundUserMessage', () => {
});
});
+describe('readInboundUserMessage', () => {
+ it.each([
+ { role: 'assistant', content: [] },
+ { type: 'custom', data: { dummy: 'message' } },
+ ])('rejects input without a user role: %j', (input) => {
+ const readInput = () => readInboundUserMessage(input);
+
+ expect(readInput).toThrow(UnexpectedError);
+ expect(readInput).toThrow('Queued input must be a user message');
+ });
+});
+
describe('resolveInboundMimeType', () => {
it('confirms a declared image type against magic bytes', async () => {
expect(await resolveInboundMimeType('image/png', PNG_BYTES)).toBe('image/png');
diff --git a/packages/cli/src/modules/agents/utils/inbound-attachments.ts b/packages/cli/src/modules/agents/utils/inbound-attachments.ts
index af4dba43a14..6f1b2cba783 100644
--- a/packages/cli/src/modules/agents/utils/inbound-attachments.ts
+++ b/packages/cli/src/modules/agents/utils/inbound-attachments.ts
@@ -1,5 +1,6 @@
-import type { Message } from '@n8n/agents';
+import type { AgentMessage, Message } from '@n8n/agents';
import { MAX_AGENT_CHAT_ATTACHMENT_MIMETYPE_LENGTH } from '@n8n/api-types';
+import { UnexpectedError } from 'n8n-workflow';
import type { StoredAttachmentRef } from '../types/agent-chat-attachment';
@@ -69,3 +70,30 @@ export function buildInboundUserMessage(
}
return [{ role: 'user', content }];
}
+
+/** Read the original input without display normalization or file hydration. */
+export function readInboundUserMessage(input: AgentMessage): {
+ message: string;
+ attachments: StoredAttachmentRef[];
+} {
+ if (!('role' in input) || input.role !== 'user') {
+ throw new UnexpectedError('Queued input must be a user message');
+ }
+ return {
+ message: input.content
+ .filter((part) => part.type === 'text')
+ .map((part) => part.text)
+ .join('\n'),
+ attachments: input.content.flatMap((part) => {
+ if (part.type !== 'file' || !part.fileRef) return [];
+ return [
+ {
+ id: part.fileRef.id,
+ fileName: part.fileRef.fileName ?? '',
+ mimeType: part.mediaType ?? 'application/octet-stream',
+ sizeBytes: part.fileRef.sizeBytes ?? 0,
+ },
+ ];
+ }),
+ };
+}
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 7bd2a5c12d3..0870c8e0e51 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
@@ -11,7 +11,7 @@ import {
} from '@n8n/agents';
import { createTeamProject, mockLogger, testDb, testModules } from '@n8n/backend-test-utils';
import { AgentsConfig, AiConfig } from '@n8n/config';
-import type { OperationContext, User } from '@n8n/db';
+import { UserRepository, type OperationContext, type User } from '@n8n/db';
import { Container } from '@n8n/di';
import { DataSource, EntityManager } from '@n8n/typeorm';
import { generateNanoId } from '@n8n/utils/generate-nano-id';
@@ -54,7 +54,11 @@ import { AgentThreadRepository } from '@/modules/agents/repositories/agent-threa
import { AgentExecutionMessageLink } from '@/modules/agents/entities/agent-execution-message-link.entity';
import { AgentMessageQueueRepository } from '@/modules/agents/repositories/agent-message-queue.repository';
import { AgentTurnAlreadyRunningError } from '@/modules/agents/agent-turn-already-running.error';
-import { EXECUTION_METADATA_KEY } from '@/modules/agents/types/agent-queued-message';
+import {
+ EXECUTION_METADATA_KEY,
+ type AgentExecutionAdmission,
+ type QueuedIntegrationMessage,
+} from '@/modules/agents/types/agent-queued-message';
import type { AgentBackgroundJobService } from '@/modules/agents/background/agent-background-job.service';
import type { AgentWakeService } from '@/modules/agents/background/agent-wake.service';
import { ExecutionRecorder, type TimelineEvent } from '@/modules/agents/execution-recorder';
@@ -193,6 +197,7 @@ describe('AgentExecutionRepository', () => {
connection ? new AgentRepository(connection) : agentRepo,
attachmentService,
mock(),
+ messageRepository,
);
return {
txRunner,
@@ -1152,7 +1157,7 @@ describe('AgentExecutionRepository', () => {
}
it('lists only pending Preview input and removes its attachments without affecting the active run', async () => {
- const services = recordingServices();
+ const services = recordingServices(Container.get(N8nMemory).getImplementation(agentId));
const threadId = uuid();
const target = { projectId, agentId, threadId, userId: owner.id };
await services.queue.enqueue(input(threadId, 'active', 'new'));
@@ -1162,6 +1167,7 @@ describe('AgentExecutionRepository', () => {
{ id: 'pending-file', fileName: 'notes.txt', mimeType: 'text/plain', sizeBytes: 5 },
];
const pending = await services.queue.enqueue(pendingInput);
+ const original = await services.messageRepository.findOneByOrFail({ id: pending.messageId });
expect((await services.queue.listPending(target)).items.map(({ id }) => id)).toEqual([
pending.id,
]);
@@ -1178,10 +1184,26 @@ describe('AgentExecutionRepository', () => {
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 services.messageRepository.findOneByOrFail({ id: pending.messageId }),
+ ).toMatchObject({
+ id: original.id,
+ threadId,
+ resourceId: original.resourceId,
+ createdAt: original.createdAt,
+ origin: original.origin,
+ content: { role: 'user', content: [] },
+ author: null,
+ modelContent: null,
+ modelContextAt: null,
+ });
+ expect(await services.messageRepository.findRuntimeMessages({ threadId })).toEqual([]);
expect((await repository.findOneByOrFail({ id: active.admission.executionId })).status).toBe(
'running',
);
await finish(services, active);
+ await services.executionService.deleteThread(projectId, agentId, threadId, owner.id);
+ expect(await services.messageRepository.countBy({ threadId })).toBe(0);
});
it('edits pending text without changing identity, attachments, or queue position', async () => {
@@ -1192,6 +1214,7 @@ describe('AgentExecutionRepository', () => {
{ id: 'file', fileName: 'notes.txt', mimeType: 'text/plain', sizeBytes: 5 },
];
const first = await services.queue.enqueue(data);
+ const original = await services.messageRepository.findOneByOrFail({ id: first.messageId });
const second = await services.queue.enqueue(input(threadId, 'second'));
const target = {
projectId,
@@ -1220,16 +1243,31 @@ describe('AgentExecutionRepository', () => {
expect(stored).toMatchObject({
id: first.id,
createdAt: first.createdAt,
- source: first.source,
- payload: { ...first.payload, message: 'updated' },
+ payload: { kind: 'preview' },
+ messageId: first.messageId,
});
+ expect(stored.payload).toEqual({ kind: 'preview' });
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');
+ expect(claimed.payload).toEqual({
+ kind: 'preview',
+ message: 'updated',
+ resourceId: data.payload.resourceId,
+ attachments: data.payload.attachments,
+ });
+ expect(claimed.thread.ownerId).toBe(owner.id);
+ expect(claimed.admission.inputMessageIds).toEqual([first.messageId]);
+ expect(
+ await services.messageRepository.findOneByOrFail({ id: first.messageId }),
+ ).toMatchObject({
+ createdAt: original.createdAt,
+ content: buildInboundUserMessage('updated', data.payload.attachments)[0],
+ modelContextAt: null,
+ });
await finish(services, claimed);
await expect(services.queue.updatePending(target)).rejects.toThrow(
'Queued message not found',
@@ -1292,7 +1330,7 @@ describe('AgentExecutionRepository', () => {
} else if (operation === 'edit') {
const claimed = await second;
if (!claimed) throw new Error('Expected a claim');
- expect(claimed.item.payload.message).toBe('edited');
+ expect(claimed.payload.message).toBe('edited');
await finish(remote, claimed);
} else {
expect(settled[1]).toEqual({ status: 'fulfilled', value: null });
@@ -1346,7 +1384,7 @@ describe('AgentExecutionRepository', () => {
expect(claims.filter(({ item }) => item !== null)).toHaveLength(1);
expect(winner.item.item.id).toBe(first.id);
expect(await repository.countBy({ threadId, status: 'running' })).toBe(1);
- expect(await local.messageRepository.countBy({ threadId })).toBe(1);
+ expect(await local.messageRepository.countBy({ threadId })).toBe(3);
await expect(
local.executionService.startExecutionRecording(winner.item.recording, new Date()),
).rejects.toBeInstanceOf(AgentTurnAlreadyRunningError);
@@ -1357,7 +1395,7 @@ describe('AgentExecutionRepository', () => {
await finish(winner.services, winner.item);
for (const message of ['second', 'third']) {
const next = await claim(local, threadId);
- expect(next.item.payload.message).toBe(message);
+ expect(next.payload.message).toBe(message);
await finish(local, next);
}
expect(await local.queueRepository.count()).toBe(0);
@@ -1380,6 +1418,8 @@ describe('AgentExecutionRepository', () => {
);
failedInsert.mockRestore();
expect(await threadRepo.findOneBy({ id: threadId })).toBeNull();
+ expect(await services.messageRepository.countBy({ threadId })).toBe(0);
+ expect(await Container.get(AgentThreadRepository).findOneBy({ id: threadId })).toBeNull();
expect(await services.queueRepository.count()).toBe(0);
const accepted = await services.queue.enqueue(input(threadId, 'first', 'new'));
const failedLink = vi
@@ -1387,16 +1427,41 @@ describe('AgentExecutionRepository', () => {
.mockResolvedValueOnce(false);
await expect(claim(services, threadId)).rejects.toBeInstanceOf(AgentTurnAlreadyRunningError);
failedLink.mockRestore();
- expect(await services.messageRepository.countBy({ threadId })).toBe(0);
- expect(await Container.get(AgentThreadRepository).findOneBy({ id: threadId })).toBeNull();
+ expect(await services.messageRepository.countBy({ threadId })).toBe(1);
+ expect(await Container.get(AgentThreadRepository).findOneBy({ id: threadId })).not.toBeNull();
expect(await repository.manager.count(AgentExecutionMessageLink)).toBe(0);
expect(await repository.countBy({ threadId })).toBe(0);
expect(await services.queueRepository.findDeliveryState(accepted.id)).toMatchObject({
executionId: null,
+ messageId: accepted.messageId,
});
await finish(services, await claim(services, threadId));
});
+ it('settles accepted Preview input after its session owner is deleted', async () => {
+ const services = recordingServices();
+ const threadId = uuid();
+ const queued = await services.queue.enqueue(input(threadId, 'pending', 'new'));
+ await Container.get(UserRepository).delete(owner.id);
+
+ const active = await claim(services, threadId);
+ expect(active.thread.ownerId).toBeNull();
+ await services.queue.recordFailure(active, new Error('Owner deleted'));
+ await services.queue.settle(threadId, active.admission.executionId);
+
+ expect(await services.queueRepository.countBy({ threadId })).toBe(0);
+ expect(await repository.findOneByOrFail({ id: active.admission.executionId })).toMatchObject({
+ status: 'error',
+ error: 'Owner deleted',
+ });
+ expect(
+ await services.messageRepository.findOneByOrFail({ id: queued.messageId }),
+ ).toMatchObject({
+ content: buildInboundUserMessage('pending', [])[0],
+ modelContextAt: null,
+ });
+ });
+
it('enforces active-slot uniqueness and prevents execution deletion from requeuing a message', async () => {
const services = recordingServices();
const threadId = uuid();
@@ -1430,7 +1495,7 @@ describe('AgentExecutionRepository', () => {
await repository.update(stale.id, { updatedAt: new Date(cutoff.getTime() - 1) });
expect(await local.executionService.finalizeInterruptedExecution(stale, cutoff)).toBe(true);
const next = await claim(local, threadId);
- expect(next.item.payload.message).toBe('next');
+ expect(next.payload.message).toBe('next');
await expect(finish(remote, active)).rejects.toThrow('no longer running');
await finish(local, next);
expect(
@@ -1551,7 +1616,7 @@ describe('AgentExecutionRepository', () => {
});
it('removes pending input and attachments on session deletion and never recreates it', async () => {
- const services = recordingServices();
+ const services = recordingServices(Container.get(N8nMemory).getImplementation(agentId));
const threadId = uuid();
const attachment = await attachmentRepo.save(buildAttachment(threadId));
const accepted = input(threadId, 'pending', 'new');
@@ -1569,6 +1634,7 @@ describe('AgentExecutionRepository', () => {
).toBe(true);
expect(await services.queueRepository.countBy({ threadId })).toBe(0);
expect(await attachmentRepo.findOneBy({ id: attachment.id })).toBeNull();
+ expect(await services.messageRepository.countBy({ threadId })).toBe(0);
expect(services.attachmentService.deleteStoredData).toHaveBeenCalledWith(
[attachment.binaryDataId],
{ threadId },
@@ -1594,9 +1660,8 @@ describe('AgentExecutionRepository', () => {
const expected = new Map();
for (let position = 0; position < 2; position++) {
const execution = await createExecution({ threadId: thread.id });
- const messageId = await messageRepository.createExecutionInput(
+ const message = await messageRepository.createInput(
{
- executionId: execution.id,
threadId: thread.id,
resourceId: 'user-1',
content: { role: 'user', content: [{ type: 'text', text: 'Input' }] },
@@ -1605,7 +1670,13 @@ describe('AgentExecutionRepository', () => {
{},
);
executionIds.push(execution.id);
- expected.set(execution.id, messageId);
+ await messageRepository.linkExecutionInput(
+ execution.id,
+ message.id,
+ { threadId: thread.id, resourceId: 'user-1' },
+ {},
+ );
+ expected.set(execution.id, message.id);
}
executionIds.splice(1, 0, ...Array.from({ length: 33_000 }, () => uuid()));
@@ -1617,7 +1688,11 @@ describe('AgentExecutionRepository', () => {
}
});
- it('keeps canonical input through enrichment, runtime persistence, mixed history, and session deletion', async () => {
+ it.each([
+ ['direct', 'message-42'],
+ ['queued', 'message-42'],
+ ['queued', undefined],
+ ] as const)('preserves %s canonical input (%s)', async (path, platformMessageId) => {
const thread = await createThread();
const threadId = thread.id;
const resourceId = 'user-1';
@@ -1664,17 +1739,96 @@ describe('AgentExecutionRepository', () => {
source: 'slack',
messageOrigin: {
integrationConnectionId: 'slack:credential',
- platformMessageId: 'message-42',
+ ...(platformMessageId !== undefined ? { platformMessageId } : {}),
},
};
- const admission = await services.executionService.startExecutionRecording(params, new Date());
+ let admission: AgentExecutionAdmission;
+ if (path === 'direct') {
+ admission = await services.executionService.startExecutionRecording(params, new Date());
+ } else {
+ const payload: QueuedIntegrationMessage = {
+ kind: 'integration',
+ message: params.userMessage,
+ resourceId,
+ attachments,
+ modelMessage: '[Ada]: Original input',
+ author: params.author,
+ credentialId: 'credential',
+ platformThreadId: 'slack:channel:thread',
+ sender: {
+ userId: 'author-1',
+ userName: 'ada',
+ fullName: 'Ada',
+ isBot: false,
+ isMe: false,
+ },
+ messageContext: {
+ integrationConnectionId: 'slack:credential',
+ ...(platformMessageId !== undefined ? { messageId: platformMessageId } : {}),
+ platform: 'slack',
+ target: { type: 'thread', threadId: 'slack:channel:thread' },
+ replyTarget: { type: 'thread', threadId: 'slack:channel:reply' },
+ replyMessageId: 'reply-41',
+ updatedAt: '2026-06-01T12:00:00.000Z',
+ },
+ contextConversation: { threadId: 'integration-context', resourceId: 'integration-user' },
+ forceBuffered: true,
+ };
+ const queued = await services.queue.enqueue({
+ agentId,
+ projectId,
+ threadId,
+ sessionMode: 'existing',
+ source: 'slack',
+ payload,
+ });
+ for (const field of [
+ 'message',
+ 'resourceId',
+ 'attachments',
+ 'author',
+ 'modelMessage',
+ 'platformThreadId',
+ 'messageContext.platform',
+ 'messageContext.integrationConnectionId',
+ 'messageContext.messageId',
+ ]) {
+ expect(queued.payload).not.toHaveProperty(field);
+ }
+ expect((await memory.getMessages(threadId)).map(({ id }) => id)).not.toContain(
+ queued.messageId,
+ );
+ const origin = {
+ source: 'slack',
+ ...params.messageOrigin,
+ platformThreadId: payload.platformThreadId,
+ };
+ for (const invalidOrigin of [
+ null,
+ { ...origin, source: null },
+ { ...origin, integrationConnectionId: undefined },
+ { ...origin, platformThreadId: undefined },
+ ]) {
+ await services.messageRepository.update(queued.messageId, { origin: invalidOrigin });
+ await expect(services.queue.claimNext(threadId, async () => true)).rejects.toThrow(
+ 'Queued integration input has incomplete origin',
+ );
+ }
+ await services.messageRepository.update(queued.messageId, { origin });
+ const claimed = await services.queue.claimNext(threadId, async () => true);
+ expect(claimed?.payload).toStrictEqual(payload);
+ expect(claimed?.recording.source).toBe('slack');
+ admission = claimed!.admission;
+ expect(admission.inputMessageIds).toEqual([queued.messageId]);
+ }
const [id] = admission.inputMessageIds;
const canonical = await services.messageRepository.findOneByOrFail({ id });
expect(canonical).toMatchObject({
content: buildInboundUserMessage(params.userMessage, attachments)[0],
author: params.author,
modelContextAt: null,
- modelContent: null,
+ modelContent:
+ path === 'queued' ? buildInboundUserMessage('[Ada]: Original input', attachments)[0] : null,
origin: { source: 'slack', ...params.messageOrigin },
});
expect(
diff --git a/packages/cli/test/migration/1790605116208-add-agent-queue-message-references.test.ts b/packages/cli/test/migration/1790605116208-add-agent-queue-message-references.test.ts
new file mode 100644
index 00000000000..862c2018692
--- /dev/null
+++ b/packages/cli/test/migration/1790605116208-add-agent-queue-message-references.test.ts
@@ -0,0 +1,413 @@
+import {
+ createTestMigrationContext,
+ initDbUpToMigration,
+ runSingleMigration,
+ type TestMigrationContext,
+} from '@n8n/backend-test-utils';
+import { DbConnection } from '@n8n/db';
+import { Container } from '@n8n/di';
+import { DataSource } from '@n8n/typeorm';
+import { jsonParse, UserError } from 'n8n-workflow';
+import { randomUUID } from 'node:crypto';
+
+const MIGRATION_NAME = 'AddAgentQueueMessageReferences1790605116208';
+const createdAt = new Date('2026-05-12T10:00:00.123Z');
+const updatedAt = new Date('2026-05-12T10:01:00.456Z');
+const timestamps = { createdAt, updatedAt };
+const previewPayload = {
+ kind: 'preview',
+ message: 'Pending input',
+ resourceId: 'preview-resource',
+ userId: 'owner',
+};
+
+function readJson(value: unknown): unknown {
+ return typeof value === 'string' ? jsonParse(value) : value;
+}
+
+describe('AddAgentQueueMessageReferences migration', () => {
+ let dataSource: DataSource;
+ let context: TestMigrationContext;
+ let projectId: string;
+ let agentId: string;
+
+ beforeAll(async () => {
+ await Container.get(DbConnection).init();
+ dataSource = Container.get(DataSource);
+ });
+
+ beforeEach(async () => {
+ const setup = createTestMigrationContext(dataSource);
+ await setup.queryRunner.clearDatabase();
+ await setup.queryRunner.release();
+ await initDbUpToMigration(MIGRATION_NAME);
+ context = createTestMigrationContext(dataSource);
+ projectId = randomUUID();
+ agentId = randomUUID();
+ await insert('project', { id: projectId, name: 'Test project', type: 'team', ...timestamps });
+ await insert('agents', {
+ id: agentId,
+ name: 'Test agent',
+ projectId,
+ integrations: '[]',
+ tools: '{}',
+ skills: '{}',
+ ...timestamps,
+ });
+ });
+
+ afterEach(async () => {
+ await context?.queryRunner.release();
+ });
+
+ afterAll(async () => {
+ await Container.get(DbConnection).close();
+ });
+
+ async function insert(table: string, values: Record) {
+ const keys = Object.keys(values);
+ await context.runQuery(
+ `INSERT INTO ${context.escape.tableName(table)} (${keys.map(context.escape.columnName).join(', ')}) VALUES (${keys.map((key) => ':' + key).join(', ')})`,
+ values,
+ );
+ }
+
+ async function rows(table: string) {
+ return await context.runQuery>>(
+ `SELECT * FROM ${context.escape.tableName(table)} ORDER BY ${context.escape.columnName('id')}`,
+ );
+ }
+
+ async function createThread() {
+ const threadId = randomUUID();
+ await insert('agent_execution_threads', {
+ id: threadId,
+ agentId,
+ agentName: 'Test agent',
+ projectId,
+ accessScope: 'project',
+ ...timestamps,
+ });
+ return threadId;
+ }
+
+ async function migrate() {
+ await context.queryRunner.release();
+ try {
+ await runSingleMigration(MIGRATION_NAME);
+ } finally {
+ context = createTestMigrationContext(dataSource);
+ }
+ }
+
+ it('backfills queued input, preserves started input, and keeps constraints through rollback', async () => {
+ const previewThread = await createThread();
+ const integrationThread = await createThread();
+ const activeThread = await createThread();
+ const executionId = randomUUID();
+ const messageId = randomUUID();
+ const attachment = {
+ id: 'attachment',
+ fileName: 'image.png',
+ mimeType: 'image/png',
+ sizeBytes: 3,
+ };
+ const filePart = {
+ type: 'file',
+ mediaType: 'image/png',
+ fileRef: { id: 'attachment', fileName: 'image.png', sizeBytes: 3 },
+ };
+ const author = { userId: 'sender', displayName: 'Sender' };
+ const messageContext = {
+ platform: 'slack',
+ integrationConnectionId: 'connection',
+ messageId: 'platform-message',
+ target: { type: 'thread', threadId: 'platform-thread' },
+ replyMessageId: 'reply-message',
+ replyExpectation: 'optional',
+ updatedAt: createdAt.toISOString(),
+ };
+ const integrationPayload = {
+ kind: 'integration',
+ message: 'Original input',
+ resourceId: 'integration-resource',
+ modelMessage: 'Enriched input',
+ attachments: [attachment],
+ author,
+ credentialId: 'credential',
+ platformThreadId: 'platform-thread',
+ sender: { userId: 'sender', isMe: false },
+ messageContext,
+ contextConversation: { threadId: integrationThread },
+ forceBuffered: false,
+ slackThreadContext: { channelId: 'channel', threadTs: '123.456' },
+ };
+ await insert('agent_message_queue', {
+ threadId: previewThread,
+ source: 'chat',
+ payload: JSON.stringify({ ...previewPayload, message: '', attachments: [attachment] }),
+ ...timestamps,
+ });
+ await insert('agent_message_queue', {
+ threadId: integrationThread,
+ source: 'slack',
+ payload: JSON.stringify(integrationPayload),
+ ...timestamps,
+ });
+ await insert('agents_resources', { id: 'active-resource', ...timestamps });
+ await insert('agents_threads', {
+ id: activeThread,
+ resourceId: 'active-resource',
+ ...timestamps,
+ });
+ await insert('agents_messages', {
+ id: messageId,
+ threadId: activeThread,
+ resourceId: 'active-resource',
+ role: 'user',
+ content: JSON.stringify({ role: 'user', content: [{ type: 'text', text: 'Started input' }] }),
+ modelContent: JSON.stringify({
+ role: 'user',
+ content: [{ type: 'text', text: 'Runtime enrichment' }],
+ }),
+ modelContextAt: updatedAt,
+ origin: '{"source":"chat"}',
+ ...timestamps,
+ });
+ await insert('agent_execution', {
+ id: executionId,
+ threadId: activeThread,
+ status: 'success',
+ hitlStatus: 'suspended',
+ ...timestamps,
+ });
+ await insert('agent_execution_message_links', {
+ executionId,
+ messageId,
+ direction: 'input',
+ position: 0,
+ createdAt,
+ });
+ await insert('agent_checkpoints', {
+ runId: 'suspended-run',
+ agentId,
+ threadId: activeThread,
+ state: JSON.stringify({
+ status: 'suspended',
+ persistence: { threadId: activeThread, hostMetadata: { n8nExecutionId: executionId } },
+ messages: [{ id: messageId }],
+ }),
+ ...timestamps,
+ });
+ await insert('agent_message_queue', {
+ threadId: activeThread,
+ executionId,
+ source: 'chat',
+ payload: JSON.stringify({
+ ...previewPayload,
+ resourceId: 'active-resource',
+ message: 'Started input',
+ }),
+ ...timestamps,
+ });
+ // Cross a batch boundary while earlier queue rows are updated in place.
+ for (let index = 0; index < 101; index++) {
+ await insert('agent_message_queue', {
+ threadId: previewThread,
+ source: 'chat',
+ payload: JSON.stringify({ ...previewPayload, message: `Pending ${index}` }),
+ ...timestamps,
+ });
+ }
+ await insert('agent_message_queue', {
+ threadId: previewThread,
+ source: 'chat',
+ payload: JSON.stringify(previewPayload),
+ ...timestamps,
+ });
+ const originalQueue = await rows('agent_message_queue');
+ const deletedId = originalQueue.pop()!.id;
+ await context.runQuery(
+ `DELETE FROM ${context.escape.tableName('agent_message_queue')} WHERE ${context.escape.columnName('id')} = :id`,
+ { id: deletedId },
+ );
+ const [originalMessage] = await rows('agents_messages');
+ const snapshots = new Map();
+ for (const table of [
+ 'agent_execution_threads',
+ 'agent_execution',
+ 'agent_checkpoints',
+ 'agent_execution_message_links',
+ ]) {
+ snapshots.set(
+ table,
+ await context.runQuery(`SELECT * FROM ${context.escape.tableName(table)}`),
+ );
+ }
+
+ await migrate();
+
+ const queue = await rows('agent_message_queue');
+ const messages = await rows('agents_messages');
+ expect(messages).toHaveLength(originalQueue.length);
+ expect(queue).toHaveLength(originalQueue.length);
+ for (const [index, item] of queue.entries()) {
+ const { source: _source, payload: _payload, ...original } = originalQueue[index];
+ expect(item).toMatchObject(original);
+ expect(item).not.toHaveProperty('source');
+ expect(item.messageId).toEqual(expect.any(String));
+ }
+ const preview = messages.find(({ id }) => id === queue[0].messageId)!;
+ expect(preview).toMatchObject({
+ threadId: previewThread,
+ resourceId: 'preview-resource',
+ role: 'user',
+ type: null,
+ author: null,
+ modelContent: null,
+ modelContextAt: null,
+ createdAt: queue[0].createdAt,
+ updatedAt: queue[0].updatedAt,
+ });
+ expect(readJson(preview.content)).toEqual({ role: 'user', content: [filePart] });
+ expect(readJson(preview.origin)).toEqual({ source: 'chat' });
+ expect(readJson(queue[0].payload)).toEqual({ kind: 'preview' });
+ const integration = messages.find(({ id }) => id === queue[1].messageId)!;
+ expect(integration).toMatchObject({
+ threadId: integrationThread,
+ resourceId: 'integration-resource',
+ modelContextAt: null,
+ });
+ expect(readJson(integration.content)).toEqual({
+ role: 'user',
+ content: [{ type: 'text', text: 'Original input' }, filePart],
+ });
+ expect(readJson(integration.modelContent)).toEqual({
+ role: 'user',
+ content: [{ type: 'text', text: 'Enriched input' }, filePart],
+ });
+ expect(readJson(integration.author)).toEqual(author);
+ expect(readJson(integration.origin)).toEqual({
+ source: 'slack',
+ integrationConnectionId: 'connection',
+ platformMessageId: 'platform-message',
+ platformThreadId: 'platform-thread',
+ });
+ expect(readJson(queue[1].payload)).toEqual({
+ kind: 'integration',
+ credentialId: 'credential',
+ sender: integrationPayload.sender,
+ contextConversation: integrationPayload.contextConversation,
+ forceBuffered: false,
+ slackThreadContext: integrationPayload.slackThreadContext,
+ messageContext: {
+ target: messageContext.target,
+ replyMessageId: 'reply-message',
+ replyExpectation: 'optional',
+ updatedAt: messageContext.updatedAt,
+ },
+ });
+ expect(queue[2].messageId).toBe(messageId);
+ expect(messages.find(({ id }) => id === messageId)).toEqual(originalMessage);
+ const lastMessage = messages.find(({ id }) => id === queue.at(-1)!.messageId)!;
+ expect(readJson(lastMessage.content)).toEqual({
+ role: 'user',
+ content: [{ type: 'text', text: 'Pending 100' }],
+ });
+ expect(await rows('agents_resources')).toHaveLength(3);
+ expect(await rows('agents_threads')).toHaveLength(3);
+ for (const [table, snapshot] of snapshots) {
+ expect(await context.runQuery(`SELECT * FROM ${context.escape.tableName(table)}`)).toEqual(
+ snapshot,
+ );
+ }
+
+ const queueInput = {
+ threadId: previewThread,
+ messageId: preview.id as string,
+ payload: '{"kind":"preview"}',
+ };
+ await expect(
+ insert('agent_message_queue', { ...queueInput, messageId: null }),
+ ).rejects.toThrow();
+ await expect(
+ insert('agent_message_queue', { ...queueInput, messageId: randomUUID() }),
+ ).rejects.toThrow();
+ await expect(insert('agent_message_queue', queueInput)).rejects.toThrow();
+ await context.queryRunner.release();
+ await expect(dataSource.undoLastMigration({ transaction: 'each' })).rejects.toThrow(UserError);
+ context = createTestMigrationContext(dataSource);
+ expect(await rows('agent_message_queue')).toEqual(queue);
+ await context.runQuery(
+ `DELETE FROM ${context.escape.tableName('agents_messages')} WHERE ${context.escape.columnName('id')} = :id`,
+ { id: preview.id },
+ );
+ expect(await rows('agent_message_queue')).toHaveLength(queue.length - 1);
+ await context.runQuery(`DELETE FROM ${context.escape.tableName('agent_message_queue')}`);
+ await insert('agent_message_queue', { ...queueInput, messageId: lastMessage.id as string });
+ const [newItem] = await rows('agent_message_queue');
+ expect(BigInt(String(newItem.id))).toBeGreaterThan(BigInt(String(deletedId)));
+ await context.runQuery(`DELETE FROM ${context.escape.tableName('agent_message_queue')}`);
+ const retainedMessages = await rows('agents_messages');
+ await context.queryRunner.release();
+ await dataSource.undoLastMigration({ transaction: 'each' });
+ context = createTestMigrationContext(dataSource);
+ expect(await rows('agents_messages')).toEqual(retainedMessages);
+ const reverted = await context.queryRunner.getTable(
+ `${context.tablePrefix}agent_message_queue`,
+ );
+ expect(reverted?.findColumnByName('source')).toMatchObject({ length: '32', isNullable: false });
+ await insert('agent_message_queue', {
+ threadId: previewThread,
+ source: 'chat',
+ payload: JSON.stringify(previewPayload),
+ });
+ const [revertedItem] = await rows('agent_message_queue');
+ expect(BigInt(String(revertedItem.id))).toBeGreaterThan(BigInt(String(newItem.id)));
+ await context.runQuery(`DELETE FROM ${context.escape.tableName('agent_message_queue')}`);
+ await migrate();
+ expect(await rows('agents_messages')).toEqual(retainedMessages);
+ await insert('agent_message_queue', { ...queueInput, messageId: lastMessage.id as string });
+ const [reappliedItem] = await rows('agent_message_queue');
+ expect(BigInt(String(reappliedItem.id))).toBeGreaterThan(BigInt(String(revertedItem.id)));
+ });
+
+ it.each(['invalid payload', 'missing input link'])(
+ 'preserves queue data when an item has %s',
+ async (problem) => {
+ const threadId = await createThread();
+ const executionId = randomUUID();
+ await insert('agent_execution', {
+ id: executionId,
+ threadId,
+ status: 'running',
+ ...timestamps,
+ });
+ await insert('agent_message_queue', {
+ threadId,
+ source: 'chat',
+ payload: JSON.stringify(previewPayload),
+ ...timestamps,
+ });
+ await insert('agent_message_queue', {
+ threadId,
+ source: 'chat',
+ executionId: problem === 'missing input link' ? executionId : null,
+ payload: problem === 'invalid payload' ? 'null' : JSON.stringify(previewPayload),
+ ...timestamps,
+ });
+ const original = await rows('agent_message_queue');
+ await expect(migrate()).rejects.toThrow(UserError);
+ expect(await rows('agent_message_queue')).toEqual(original);
+ expect(await rows('agents_messages')).toEqual([]);
+ expect(await rows('agents_threads')).toEqual([]);
+ expect(await rows('agents_resources')).toEqual([]);
+ expect(
+ await context.queryRunner.hasColumn(
+ `${context.tablePrefix}agent_message_queue`,
+ 'messageId',
+ ),
+ ).toBe(false);
+ },
+ );
+});
diff --git a/packages/testing/playwright/tests/infrastructure/agents/message-queue.spec.ts b/packages/testing/playwright/tests/infrastructure/agents/message-queue.spec.ts
index 3616ac94b04..06b92997a8d 100644
--- a/packages/testing/playwright/tests/infrastructure/agents/message-queue.spec.ts
+++ b/packages/testing/playwright/tests/infrastructure/agents/message-queue.spec.ts
@@ -160,6 +160,12 @@ test.describe(
`SELECT count(*) FROM agent_message_queue WHERE "threadId" = '${threadId}' AND "executionId" IS NULL`,
),
);
+ const queuedMessageId = async (queueId: string) =>
+ (
+ await n8nContainer.services.postgres.exec(
+ `SELECT "messageId" FROM agent_message_queue WHERE id = '${queueId}'`,
+ )
+ ).trim();
try {
const threadId = randomUUID();
await signalMain(n8nContainer, 0, 'SIGSTOP');
@@ -177,6 +183,7 @@ test.describe(
.poll(() => removable.events.find((event) => event.type === 'message-queued'))
.toBeTruthy();
const restored = await consumer.agents.queuedMessages(project.id, agent.id, threadId);
+ const remoteInputId = await queuedMessageId(restored.items[0].id);
expect(restored.items.map(({ message }) => message)).toEqual([
'fifo-remote',
'third',
@@ -227,8 +234,17 @@ test.describe(
.toEqual(['cancelled', 'success', 'success']);
expect(await second.done).toBeUndefined();
expect(second.events).toContainEqual(
- expect.objectContaining({ type: 'execution-started', sessionId: threadId }),
+ expect.objectContaining({
+ type: 'execution-started',
+ sessionId: threadId,
+ inputMessageIds: [remoteInputId],
+ }),
);
+ expect(
+ (await ingress.agents.history(project.id, agent.id, threadId)).messages.filter(
+ ({ id }) => id === remoteInputId,
+ ),
+ ).toHaveLength(1);
expect(
second.events
.filter((event) => event.type === 'text-delta')
@@ -251,6 +267,8 @@ test.describe(
await signalMain(n8nContainer, 0, 'SIGCONT');
const afterCrash = await open(0, crashThread, 'survives restart');
await expect.poll(async () => await queued(crashThread)).toBe(1);
+ const pending = await ingress.agents.queuedMessages(project.id, agent.id, crashThread);
+ const recoveredInputId = await queuedMessageId(pending.items[0].id);
const [main] = n8nContainer.findContainers('-n8n-main-2$');
await main.restart({ timeout: 0 });
await expect
@@ -263,6 +281,17 @@ test.describe(
)
.toEqual(['interrupted', 'success']);
expect(await afterCrash.done).toBeUndefined();
+ expect(afterCrash.events).toContainEqual(
+ expect.objectContaining({
+ type: 'execution-started',
+ inputMessageIds: [recoveredInputId],
+ }),
+ );
+ expect(
+ (await ingress.agents.history(project.id, agent.id, crashThread)).messages.filter(
+ ({ id }) => id === recoveredInputId,
+ ),
+ ).toHaveLength(1);
expect(afterCrash.events).toContainEqual(
expect.objectContaining({ type: 'done', sessionId: crashThread }),
);