refactor(core): Reference accepted messages from agent queue (#39778)

2026-10-17 05:29:44 +02:00 · 2026-09-29 08:39:25 +00:00 · 2026-09-29 08:39:25 +00:00 · 84efbac6f1
commit 84efbac6f1
parent 599df45795
26 changed files with 1407 additions and 173 deletions
--- 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
 }
--- 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
 }
--- 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
 }
--- 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
--- 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
--- 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
 }
--- 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
 }
--- 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
 }
--- a/docs/generated/sqlite-schema/agent_message_queue.md
+++ b/docs/generated/sqlite-schema/agent_message_queue.md
@ -6,7 +6,7 @@
 <summary><strong>Table Definition</strong></summary>

 ```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)
 ```

 </details>
@ -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
--- 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
--- a/packages/@n8n/db/src/migrations/postgresdb/1790605116208-AddAgentQueueMessageReferences.ts
+++ 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<typeof payloadSchema>;
+
+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<QueueRow>(
+			`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<string, string | undefined> = { 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<z.infer<typeof attachmentSchema>>) {
+		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<Array<{ id: string }>>(
+			`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');
+	}
+}
--- a/packages/@n8n/db/src/migrations/sqlite/1790605116208-AddAgentQueueMessageReferences.ts
+++ 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<Array<{ seq: string }>>(
+			'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`,
+		});
+	}
+}
--- 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<AgentMessageEntity>({ id: 'message-1' }));
 		messageRepository.findExecutionInputs.mockResolvedValue(new Map());
 		messageRepository.copyExecutionInputs.mockResolvedValue([]);

--- 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<QueuedIntegrationMessage>({
+					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<AgentMessageQueue>({
 				id: threadId,
 				threadId,
-				source: integration ? 'slack' : 'chat',
+				message: mock<AgentMessageEntity>({ origin: { source: integration ? 'slack' : 'chat' } }),
 				payload: integration
-					? mock<QueuedIntegrationMessage>({
-							kind: 'integration',
-							message: 'input',
-							credentialId: 'credential',
-							platformThreadId: 'slack:channel:thread',
-						})
-					: { kind: 'preview', message: 'input', userId: 'user', resourceId: 'draft-chat:user' },
+					? mock<AgentQueueDispatch>({ kind: 'integration', credentialId: 'credential' })
+					: { kind: 'preview' },
+			}),
+			thread: mock<AgentExecutionThread>({
+				id: threadId,
+				agentId: 'agent',
+				projectId: 'project',
+				ownerId: integration ? null : 'user',
 			}),
-			thread: mock<AgentExecutionThread>({ 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<User>({ 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<AgentChatBridge>(), 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();
--- 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<AgentExecutionReservation> {
-		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<string[]> {
 		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) {
--- 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<void> {
-		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<void> {
-		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<User> {
+		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<void> {
+		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<void> {
-		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);
 	}
 }
--- 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<QueuedPreviewMessage, 'userId'> | 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<AgentMessageQueue> {
+		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<void> {
-		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,
 		};
 	}
 }
--- 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<AgentMessageEntity>;

-	@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' })
--- 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<AgentMessageQueue> {
@ -15,22 +15,29 @@ export class AgentMessageQueueRepository extends BaseRepository<AgentMessageQueu

 	async enqueue(
 		threadId: string,
-		source: string,
-		payload: AgentQueuedMessage,
+		messageId: string,
+		payload: AgentQueueDispatch,
 		ctx: OperationContext,
 	) {
 		const repository = this.managerFor(ctx).getRepository(AgentMessageQueue);
 		return await repository.save(
-			repository.create({ threadId, source, payload, executionId: null }),
+			repository.create({ threadId, messageId, payload, executionId: null }),
 		);
 	}

 	async listPending(threadId: string) {
-		return await this.find({ where: { threadId, executionId: IsNull() }, order: { id: 'ASC' } });
+		return await this.find({
+			where: { threadId, executionId: IsNull() },
+			relations: { message: true },
+			order: { id: 'ASC' },
+		});
 	}

 	async findItem(threadId: string, id: string, ctx: OperationContext) {
-		return await this.managerFor(ctx).findOneBy(AgentMessageQueue, { threadId, id });
+		return await this.managerFor(ctx).findOne(AgentMessageQueue, {
+			where: { threadId, id },
+			relations: { message: true },
+		});
 	}

 	async removePending(threadId: string, id: string, ctx: OperationContext) {
@ -42,23 +49,10 @@ export class AgentMessageQueueRepository extends BaseRepository<AgentMessageQueu
 		return result.affected === 1;
 	}

-	async updatePendingPayload(
-		threadId: string,
-		id: string,
-		payload: AgentQueuedMessage,
-		ctx: OperationContext,
-	) {
-		const result = await this.managerFor(ctx).update(
-			AgentMessageQueue,
-			{ threadId, id, executionId: IsNull() },
-			{ payload },
-		);
-		return result.affected === 1;
-	}
-
 	async findHead(threadId: string, ctx: OperationContext) {
 		return await this.managerFor(ctx).findOne(AgentMessageQueue, {
 			where: { threadId },
+			relations: { message: true },
 			order: { id: 'ASC' },
 		});
 	}
--- a/packages/cli/src/modules/agents/repositories/agent-message.repository.ts
+++ b/packages/cli/src/modules/agents/repositories/agent-message.repository.ts
@ -25,17 +25,17 @@ export class AgentMessageRepository extends BaseRepository<AgentMessageEntity> {
 		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<string> {
+	): Promise<AgentMessageEntity> {
 		const manager = this.managerFor(ctx);
 		await manager
 			.createQueryBuilder()
@ -60,17 +60,54 @@ export class AgentMessageRepository extends BaseRepository<AgentMessageEntity> {
 			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<void> {
+		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<void> {
+		await this.managerFor(ctx).save(this.create({ id, content, modelContent: null }));
+	}
+
+	async clearPendingInput(id: string, ctx: OperationContext): Promise<void> {
+		await this.managerFor(ctx).update(AgentMessageEntity, id, {
+			content: { role: 'user', content: [] },
+			author: null,
+			modelContent: null,
+			modelContextAt: null,
+		});
 	}

 	async copyExecutionInputs(
--- 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;
--- 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<AgentMessage>([
+		{ 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');
--- 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,
+				},
+			];
+		}),
+	};
+}
--- 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<AgentExecutionUpdateBroadcaster>(),
+			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<string, string>();
 		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(
--- 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
@ -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<string, string | number | Date | null>) {
+		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<Array<Record<string, unknown>>>(
+			`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<string, unknown>();
+		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);
+		},
+	);
+});
--- 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 }),
 				);