From 6c592bdb3306a3cfc29e25872135486c4a45940c Mon Sep 17 00:00:00 2001
From: Robin Braumann <50590409+bjorger@users.noreply.github.com>
Date: Thu, 24 Sep 2026 12:27:30 +0000
Subject: [PATCH] feat(core): Add the durable agent message queue foundation
(#39435)
---
docs/generated/postgres-schema/README.md | 12 +
.../postgres-schema/public.agent_execution.md | 12 +-
.../public.agent_execution_threads.md | 12 +-
.../public.agent_message_queue.md | 104 +++++
docs/generated/sqlite-schema/README.md | 12 +
.../sqlite-schema/agent_execution.md | 12 +-
.../sqlite-schema/agent_execution_threads.md | 12 +-
.../sqlite-schema/agent_message_queue.md | 108 +++++
.../runtime/__tests__/agent-runtime.test.ts | 14 +-
.../agents/src/runtime/loop/agent-runtime.ts | 5 +-
.../agents/src/runtime/state/run-state.ts | 8 +-
packages/@n8n/agents/src/types/sdk/memory.ts | 4 +-
.../1790241069212-CreateAgentMessageQueue.ts | 47 ++
.../agent-chat-execution.service.test.ts | 51 +-
.../agent-checkpoint.repository.test.ts | 14 +-
...ent-execution-orchestrator.service.test.ts | 45 +-
.../__tests__/agent-execution.service.test.ts | 84 +++-
...gent-interrupted-execution-sweeper.test.ts | 9 +-
...e-reconstruction-integration-tools.test.ts | 3 +
.../agent-workflow-execution.service.test.ts | 3 +
.../agents/agent-chat-execution.service.ts | 53 +--
.../agent-execution-orchestrator.service.ts | 59 ++-
.../modules/agents/agent-execution.service.ts | 212 ++++++---
.../agent-interrupted-execution-sweeper.ts | 7 +-
.../agents/agent-message-queue.service.ts | 161 +++++++
.../agent-turn-already-running.error.ts | 7 +
.../agents/agent-turn-execution.service.ts | 87 ++--
.../cli/src/modules/agents/agents.module.ts | 2 +
.../entities/agent-message-queue.entity.ts | 54 +++
.../src/modules/agents/execution-recorder.ts | 4 +-
.../__tests__/n8n-checkpoint-storage.test.ts | 42 +-
.../integrations/n8n-checkpoint-storage.ts | 119 ++++-
.../agent-checkpoint.repository.ts | 61 ++-
.../agent-execution-thread.repository.ts | 9 +
.../agent-execution.repository.ts | 36 +-
.../agent-message-queue.repository.ts | 62 +++
.../__tests__/sub-agent-runner.test.ts | 3 +
.../agents/types/agent-queued-message.ts | 47 ++
...heckpoint-pruning.task.integration.test.ts | 66 +--
.../agents/agent-conversation-state.test.ts | 14 +-
.../agent-execution.repository.test.ts | 437 ++++++++++++++++--
...1069212-create-agent-message-queue.test.ts | 48 ++
42 files changed, 1776 insertions(+), 385 deletions(-)
create mode 100644 docs/generated/postgres-schema/public.agent_message_queue.md
create mode 100644 docs/generated/sqlite-schema/agent_message_queue.md
create mode 100644 packages/@n8n/db/src/migrations/common/1790241069212-CreateAgentMessageQueue.ts
create mode 100644 packages/cli/src/modules/agents/agent-message-queue.service.ts
create mode 100644 packages/cli/src/modules/agents/agent-turn-already-running.error.ts
create mode 100644 packages/cli/src/modules/agents/entities/agent-message-queue.entity.ts
create mode 100644 packages/cli/src/modules/agents/repositories/agent-message-queue.repository.ts
create mode 100644 packages/cli/src/modules/agents/types/agent-queued-message.ts
create mode 100644 packages/cli/test/migration/1790241069212-create-agent-message-queue.test.ts
diff --git a/docs/generated/postgres-schema/README.md b/docs/generated/postgres-schema/README.md
index a588139511b..d3634934f0a 100644
--- a/docs/generated/postgres-schema/README.md
+++ b/docs/generated/postgres-schema/README.md
@@ -23,6 +23,7 @@ Auto-generated from the PostgreSQL migrations in @n8n/db. Do not edit by hand.
| [public.agent_execution_threads](public.agent_execution_threads.md) | 19 | | BASE TABLE |
| [public.agent_files](public.agent_files.md) | 10 | | BASE TABLE |
| [public.agent_history](public.agent_history.md) | 9 | | BASE TABLE |
+| [public.agent_message_queue](public.agent_message_queue.md) | 7 | | BASE TABLE |
| [public.agent_task_definition](public.agent_task_definition.md) | 8 | | BASE TABLE |
| [public.agent_task_run_lock](public.agent_task_run_lock.md) | 6 | | BASE TABLE |
| [public.agent_task_snapshot](public.agent_task_snapshot.md) | 9 | | BASE TABLE |
@@ -198,6 +199,8 @@ 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--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"
"public.agent_task_run_lock" }o--|| "public.agents" : "FOREIGN KEY (#quot;agentId#quot;) REFERENCES agents(id) ON DELETE CASCADE"
"public.agent_task_snapshot" }o--|| "public.agent_history" : "FOREIGN KEY (#quot;versionId#quot;) REFERENCES agent_history(#quot;versionId#quot;) ON DELETE CASCADE"
@@ -584,6 +587,15 @@ erDiagram
timestamp_3__with_time_zone updatedAt
varchar_36_ versionId
}
+"public.agent_message_queue" {
+ timestamp_3__with_time_zone createdAt
+ varchar_36_ executionId FK
+ bigint id
+ json payload
+ varchar_32_ source
+ varchar_128_ threadId FK
+ timestamp_3__with_time_zone updatedAt
+}
"public.agent_task_definition" {
varchar_36_ agentId FK
timestamp_3__with_time_zone createdAt
diff --git a/docs/generated/postgres-schema/public.agent_execution.md b/docs/generated/postgres-schema/public.agent_execution.md
index 849c8a86c12..8c7c8900a2b 100644
--- a/docs/generated/postgres-schema/public.agent_execution.md
+++ b/docs/generated/postgres-schema/public.agent_execution.md
@@ -13,7 +13,7 @@
| error | text | | true | | | |
| failureSummary | json | | true | | | Execution failure projection as {count, latest} for session list queries |
| hitlStatus | varchar(16) | | true | | | |
-| id | varchar(36) | | false | | | |
+| id | varchar(36) | | false | [public.agent_message_queue](public.agent_message_queue.md) | | |
| model | varchar(255) | | true | | | |
| promptTokens | integer | | true | | | |
| source | varchar(32) | | true | | | |
@@ -57,6 +57,7 @@
```mermaid
erDiagram
+"public.agent_message_queue" }o--o| "public.agent_execution" : "FOREIGN KEY (#quot;executionId#quot;) REFERENCES agent_execution(id)"
"public.agent_execution" }o--|| "public.agent_execution_threads" : "FOREIGN KEY (#quot;threadId#quot;) REFERENCES agent_execution_threads(id) ON DELETE CASCADE"
"public.agent_execution" {
@@ -83,6 +84,15 @@ erDiagram
timestamp_3__with_time_zone updatedAt
text userMessage
}
+"public.agent_message_queue" {
+ timestamp_3__with_time_zone createdAt
+ varchar_36_ executionId FK
+ bigint id
+ json payload
+ varchar_32_ source
+ varchar_128_ threadId FK
+ timestamp_3__with_time_zone updatedAt
+}
"public.agent_execution_threads" {
varchar_16_ accessScope
varchar_36_ agentId FK
diff --git a/docs/generated/postgres-schema/public.agent_execution_threads.md b/docs/generated/postgres-schema/public.agent_execution_threads.md
index 25ab87fd5a7..a00f6936dd4 100644
--- a/docs/generated/postgres-schema/public.agent_execution_threads.md
+++ b/docs/generated/postgres-schema/public.agent_execution_threads.md
@@ -9,7 +9,7 @@
| agentName | varchar(255) | | false | | | |
| createdAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | |
| emoji | varchar(8) | | true | | | |
-| id | varchar(128) | | false | [public.agent_execution](public.agent_execution.md) | | |
+| id | varchar(128) | | false | [public.agent_execution](public.agent_execution.md) [public.agent_message_queue](public.agent_message_queue.md) | | |
| ownerId | uuid | | true | | [public.user](public.user.md) | User who started this private session |
| parentAgentId | varchar(36) | | true | | | Saved agent id of the parent that delegated this subagent run. |
| parentThreadId | varchar(128) | | true | | | Parent session thread id that delegated this subagent run. |
@@ -64,6 +64,7 @@ erDiagram
"public.agent_execution_threads" }o--|| "public.agents" : "FOREIGN KEY (#quot;agentId#quot;) REFERENCES agents(id) ON DELETE CASCADE"
"public.agent_execution" }o--|| "public.agent_execution_threads" : "FOREIGN KEY (#quot;threadId#quot;) REFERENCES agent_execution_threads(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_execution_threads" }o--o| "public.user" : "FOREIGN KEY (#quot;ownerId#quot;) REFERENCES #quot;user#quot;(id) ON DELETE SET NULL"
"public.agent_execution_threads" }o--|| "public.project" : "FOREIGN KEY (#quot;projectId#quot;) REFERENCES project(id) ON DELETE CASCADE"
"public.agent_execution_threads" }o--o| "public.agent_history" : "FOREIGN KEY (#quot;taskVersionId#quot;) REFERENCES agent_history(#quot;versionId#quot;) ON DELETE SET NULL"
@@ -129,6 +130,15 @@ erDiagram
timestamp_3__with_time_zone updatedAt
text userMessage
}
+"public.agent_message_queue" {
+ timestamp_3__with_time_zone createdAt
+ varchar_36_ executionId FK
+ bigint id
+ json payload
+ varchar_32_ source
+ varchar_128_ threadId FK
+ timestamp_3__with_time_zone updatedAt
+}
"public.user" {
timestamp_3__with_time_zone createdAt
boolean disabled
diff --git a/docs/generated/postgres-schema/public.agent_message_queue.md b/docs/generated/postgres-schema/public.agent_message_queue.md
new file mode 100644
index 00000000000..7354483d848
--- /dev/null
+++ b/docs/generated/postgres-schema/public.agent_message_queue.md
@@ -0,0 +1,104 @@
+# public.agent_message_queue
+
+## Columns
+
+| Name | Type | Default | Nullable | Children | Parents | Comment |
+| ---- | ---- | ------- | -------- | -------- | ------- | ------- |
+| 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 |
+| threadId | varchar(128) | | false | | [public.agent_execution_threads](public.agent_execution_threads.md) | |
+| updatedAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | |
+
+## Constraints
+
+| Name | Type | Definition |
+| ---- | ---- | ---------- |
+| 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 |
+| 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_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" |
+
+## Indexes
+
+| Name | Definition |
+| ---- | ---------- |
+| IDX_2349d84b4f2a660fc264f38fef | CREATE INDEX "IDX_2349d84b4f2a660fc264f38fef" ON public.agent_message_queue USING btree ("executionId") |
+| 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) |
+
+## Relations
+
+```mermaid
+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.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
+ json payload
+ varchar_32_ source
+ varchar_128_ threadId FK
+ timestamp_3__with_time_zone updatedAt
+}
+"public.agent_execution" {
+ json attachments
+ json author
+ integer completionTokens
+ double_precision cost
+ timestamp_3__with_time_zone createdAt
+ integer duration
+ text error
+ json failureSummary
+ varchar_16_ hitlStatus
+ varchar_36_ id
+ varchar_255_ model
+ integer promptTokens
+ varchar_32_ source
+ timestamp_3__with_time_zone startedAt
+ varchar_16_ status
+ timestamp_3__with_time_zone stoppedAt
+ varchar_2_ storedAt
+ varchar_128_ threadId FK
+ json timeline
+ integer totalTokens
+ timestamp_3__with_time_zone updatedAt
+ text userMessage
+}
+"public.agent_execution_threads" {
+ varchar_16_ accessScope
+ varchar_36_ agentId FK
+ varchar_255_ agentName
+ timestamp_3__with_time_zone createdAt
+ varchar_8_ emoji
+ varchar_128_ id
+ uuid ownerId FK
+ varchar_36_ parentAgentId
+ varchar_128_ parentThreadId
+ varchar_255_ projectId FK
+ integer sessionNumber
+ varchar_32_ taskId
+ varchar_36_ taskVersionId FK
+ varchar_255_ title
+ integer totalCompletionTokens
+ double_precision totalCost
+ integer totalDuration
+ integer totalPromptTokens
+ timestamp_3__with_time_zone updatedAt
+}
+```
+
+---
+
+> Generated by [tbls](https://github.com/k1LoW/tbls)
diff --git a/docs/generated/sqlite-schema/README.md b/docs/generated/sqlite-schema/README.md
index 30e91b58a6d..bea77f1c993 100644
--- a/docs/generated/sqlite-schema/README.md
+++ b/docs/generated/sqlite-schema/README.md
@@ -23,6 +23,7 @@ Auto-generated from the SQLite migrations in @n8n/db. Do not edit by hand.
| [agent_execution_threads](agent_execution_threads.md) | 19 | | table |
| [agent_files](agent_files.md) | 10 | | table |
| [agent_history](agent_history.md) | 9 | | table |
+| [agent_message_queue](agent_message_queue.md) | 7 | | table |
| [agent_task_definition](agent_task_definition.md) | 8 | | table |
| [agent_task_run_lock](agent_task_run_lock.md) | 6 | | table |
| [agent_task_snapshot](agent_task_snapshot.md) | 9 | | table |
@@ -180,6 +181,8 @@ 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--|| "agent_execution_threads" : "FOREIGN KEY (threadId) REFERENCES agent_execution_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE 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"
@@ -570,6 +573,15 @@ erDiagram
datetime_3_ updatedAt
varchar_36_ versionId PK
}
+"agent_message_queue" {
+ datetime_3_ createdAt
+ varchar_36_ executionId FK
+ INTEGER id
+ TEXT payload
+ varchar_32_ source
+ varchar_128_ threadId FK
+ datetime_3_ updatedAt
+}
"agent_task_definition" {
varchar_36_ agentId FK
datetime_3_ createdAt
diff --git a/docs/generated/sqlite-schema/agent_execution.md b/docs/generated/sqlite-schema/agent_execution.md
index ac706d29144..598f0558624 100644
--- a/docs/generated/sqlite-schema/agent_execution.md
+++ b/docs/generated/sqlite-schema/agent_execution.md
@@ -24,7 +24,7 @@ CREATE TABLE "agent_execution" ("id" varchar(36) PRIMARY KEY NOT NULL, "threadId
| error | TEXT | | true | | | |
| failureSummary | TEXT | | true | | | |
| hitlStatus | varchar(16) | | true | | | |
-| id | varchar(36) | | false | | | |
+| id | varchar(36) | | false | [agent_message_queue](agent_message_queue.md) | | |
| model | varchar(255) | | true | | | |
| promptTokens | INTEGER | | true | | | |
| source | varchar(32) | | true | | | |
@@ -62,6 +62,7 @@ CREATE TABLE "agent_execution" ("id" varchar(36) PRIMARY KEY NOT NULL, "threadId
```mermaid
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_execution" }o--|| "agent_execution_threads" : "FOREIGN KEY (threadId) REFERENCES agent_execution_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"agent_execution" {
@@ -88,6 +89,15 @@ erDiagram
datetime_3_ updatedAt
TEXT userMessage
}
+"agent_message_queue" {
+ datetime_3_ createdAt
+ varchar_36_ executionId FK
+ INTEGER id
+ TEXT payload
+ varchar_32_ source
+ varchar_128_ threadId FK
+ datetime_3_ updatedAt
+}
"agent_execution_threads" {
varchar_16_ accessScope
varchar_36_ agentId FK
diff --git a/docs/generated/sqlite-schema/agent_execution_threads.md b/docs/generated/sqlite-schema/agent_execution_threads.md
index 73929ffa67e..81d951aaaff 100644
--- a/docs/generated/sqlite-schema/agent_execution_threads.md
+++ b/docs/generated/sqlite-schema/agent_execution_threads.md
@@ -20,7 +20,7 @@ CREATE TABLE "agent_execution_threads" ("id" varchar(128) PRIMARY KEY NOT NULL,
| agentName | varchar(255) | | false | | | |
| createdAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | |
| emoji | varchar(8) | | true | | | |
-| id | varchar(128) | | false | [agent_execution](agent_execution.md) | | |
+| id | varchar(128) | | false | [agent_execution](agent_execution.md) [agent_message_queue](agent_message_queue.md) | | |
| ownerId | varchar | | true | | [user](user.md) | |
| parentAgentId | varchar(36) | | true | | | |
| parentThreadId | varchar(128) | | true | | | |
@@ -64,6 +64,7 @@ erDiagram
"agent_execution_threads" }o--|| "agents" : "FOREIGN KEY (agentId) REFERENCES agents (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"agent_execution" }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--|| "agent_execution_threads" : "FOREIGN KEY (threadId) REFERENCES agent_execution_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"agent_execution_threads" }o--o| "user" : "FOREIGN KEY (ownerId) REFERENCES user (id) ON UPDATE NO ACTION ON DELETE SET NULL MATCH NONE"
"agent_execution_threads" }o--|| "project" : "FOREIGN KEY (projectId) REFERENCES project (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"agent_execution_threads" }o--o| "agent_history" : "FOREIGN KEY (taskVersionId) REFERENCES agent_history (versionId) ON UPDATE NO ACTION ON DELETE SET NULL MATCH NONE"
@@ -129,6 +130,15 @@ erDiagram
datetime_3_ updatedAt
TEXT userMessage
}
+"agent_message_queue" {
+ datetime_3_ createdAt
+ varchar_36_ executionId FK
+ INTEGER id
+ TEXT payload
+ varchar_32_ source
+ varchar_128_ threadId FK
+ datetime_3_ updatedAt
+}
"user" {
datetime_3_ createdAt
boolean disabled
diff --git a/docs/generated/sqlite-schema/agent_message_queue.md b/docs/generated/sqlite-schema/agent_message_queue.md
new file mode 100644
index 00000000000..db3fd0b8af7
--- /dev/null
+++ b/docs/generated/sqlite-schema/agent_message_queue.md
@@ -0,0 +1,108 @@
+# agent_message_queue
+
+## Description
+
+
+Table Definition
+
+```sql
+CREATE TABLE "agent_message_queue" ("id" integer PRIMARY KEY AUTOINCREMENT NOT NULL, "threadId" varchar(128) NOT NULL, "source" varchar(32) NOT NULL, "payload" text NOT NULL, "executionId" varchar(36), "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), CONSTRAINT "FK_a74f9154a59430986112d70cb16" FOREIGN KEY ("threadId") REFERENCES "agent_execution_threads" ("id") ON DELETE CASCADE, CONSTRAINT "FK_2349d84b4f2a660fc264f38fef3" FOREIGN KEY ("executionId") REFERENCES "agent_execution" ("id") ON DELETE NO ACTION)
+```
+
+
+
+## Columns
+
+| Name | Type | Default | Nullable | Children | Parents | Comment |
+| ---- | ---- | ------- | -------- | -------- | ------- | ------- |
+| createdAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | |
+| executionId | varchar(36) | | true | | [agent_execution](agent_execution.md) | |
+| id | INTEGER | | false | | | |
+| 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 | | | |
+
+## Constraints
+
+| 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: 1) | FOREIGN KEY | FOREIGN KEY (threadId) REFERENCES agent_execution_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE |
+| id | PRIMARY KEY | PRIMARY KEY (id) |
+
+## Indexes
+
+| Name | Definition |
+| ---- | ---------- |
+| IDX_2349d84b4f2a660fc264f38fef | CREATE INDEX "IDX_2349d84b4f2a660fc264f38fef" ON "agent_message_queue" ("executionId") |
+| 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") |
+
+## Relations
+
+```mermaid
+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--|| "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
+ TEXT payload
+ varchar_32_ source
+ varchar_128_ threadId FK
+ datetime_3_ updatedAt
+}
+"agent_execution" {
+ TEXT attachments
+ TEXT author
+ INTEGER completionTokens
+ REAL cost
+ datetime_3_ createdAt
+ INTEGER duration
+ TEXT error
+ TEXT failureSummary
+ varchar_16_ hitlStatus
+ varchar_36_ id PK
+ varchar_255_ model
+ INTEGER promptTokens
+ varchar_32_ source
+ datetime_3_ startedAt
+ varchar_16_ status
+ datetime_3_ stoppedAt
+ varchar_2_ storedAt
+ varchar_128_ threadId FK
+ TEXT timeline
+ INTEGER totalTokens
+ datetime_3_ updatedAt
+ TEXT userMessage
+}
+"agent_execution_threads" {
+ varchar_16_ accessScope
+ varchar_36_ agentId FK
+ varchar_255_ agentName
+ datetime_3_ createdAt
+ varchar_8_ emoji
+ varchar_128_ id PK
+ varchar ownerId FK
+ varchar_36_ parentAgentId
+ varchar_128_ parentThreadId
+ varchar_255_ projectId FK
+ INTEGER sessionNumber
+ varchar_32_ taskId
+ varchar_36_ taskVersionId FK
+ varchar_255_ title
+ INTEGER totalCompletionTokens
+ REAL totalCost
+ INTEGER totalDuration
+ INTEGER totalPromptTokens
+ datetime_3_ updatedAt
+}
+```
+
+---
+
+> Generated by [tbls](https://github.com/k1LoW/tbls)
diff --git a/packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts b/packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts
index 80d7a9dd303..83b449e3eb6 100644
--- a/packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts
+++ b/packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts
@@ -5951,7 +5951,7 @@ describe('AgentRuntime — abort during a tool batch', () => {
expect(result.pendingSuspend).toBeUndefined();
expect(runtime.getState().status).toBe('cancelled');
expect(onCancellation).toHaveBeenCalledOnce();
- expect(checkpointStore.delete).toHaveBeenCalledWith(result.runId);
+ expect(checkpointStore.delete).toHaveBeenCalledWith(result.runId, expect.any(Object));
});
it('cleans up child continuations when checkpoint persistence fails', async () => {
@@ -5980,7 +5980,7 @@ describe('AgentRuntime — abort during a tool batch', () => {
expect(result.error).toBe(persistenceError);
expect(onCancellation).toHaveBeenCalledOnce();
- expect(checkpointStore.delete).toHaveBeenCalledWith(result.runId);
+ expect(checkpointStore.delete).toHaveBeenCalledWith(result.runId, expect.any(Object));
});
});
@@ -6129,6 +6129,14 @@ describe('AgentRuntime.resume() — checkpoint lifecycle', () => {
const next = createRuntimeWithCheckpointStore([tool], checkpointStore);
await next.resume('generate', { approved: true }, { runId, toolCallId: 'tc-2' });
expect(observed[1]?.hostMetadata?.actor).toBe('tool-update');
+ expect(checkpointStore.delete).toHaveBeenCalledWith(
+ runId,
+ expect.objectContaining({
+ persistence: expect.objectContaining({
+ hostMetadata: { scope: { tenant: 'tenant-1' }, actor: 'tool-update' },
+ }),
+ }),
+ );
expect(await checkpointStore.load(runId)).toBeUndefined();
},
);
@@ -9576,7 +9584,7 @@ describe('AgentRuntime — oversized tool results', () => {
await runtime.generate('run');
- expect(checkpointStorage.delete).toHaveBeenCalledWith(runId);
+ expect(checkpointStorage.delete).toHaveBeenCalledWith(runId, expect.any(Object));
await expect(filesystem.exists(getToolResultRunDirectory(runId))).resolves.toBe(true);
});
diff --git a/packages/@n8n/agents/src/runtime/loop/agent-runtime.ts b/packages/@n8n/agents/src/runtime/loop/agent-runtime.ts
index 3c2742b1f25..aeb7bd5910d 100644
--- a/packages/@n8n/agents/src/runtime/loop/agent-runtime.ts
+++ b/packages/@n8n/agents/src/runtime/loop/agent-runtime.ts
@@ -473,6 +473,7 @@ export class AgentRuntime {
: undefined,
...mergedExecOptions,
};
+ this.updateState({ persistence: resumeOptions.persistence });
await options.onResumeClaimed?.();
abortScope = this.eventBus.createAbortScope(resumeOptions.abortSignal);
@@ -884,7 +885,7 @@ export class AgentRuntime {
abortScope.isAborted ? 'Run aborted' : 'Parent run failed before suspension',
);
try {
- await this.runState.cancel(this.runId);
+ await this.runState.cancel(this.runId, this.getState());
} catch {
// Preserve the failure that interrupted suspension finalization.
}
@@ -1233,7 +1234,7 @@ export class AgentRuntime {
/** Clean up stored state for a run when it finishes without re-suspending. */
private async cleanupRun(): Promise {
try {
- await this.runState.complete(this.runId);
+ await this.runState.complete(this.runId, this.getState());
} catch (error) {
logger.warn('Failed to clean up agent run checkpoint', { runId: this.runId, error });
return;
diff --git a/packages/@n8n/agents/src/runtime/state/run-state.ts b/packages/@n8n/agents/src/runtime/state/run-state.ts
index b1ec3c6666e..924daaa835b 100644
--- a/packages/@n8n/agents/src/runtime/state/run-state.ts
+++ b/packages/@n8n/agents/src/runtime/state/run-state.ts
@@ -109,13 +109,13 @@ export class RunStateManager {
}
/** Delete a finished run from storage. Called when a resumed run completes without re-suspending. */
- async complete(runId: string): Promise {
- await this.store.delete(runId);
+ async complete(runId: string, state?: SerializableAgentState): Promise {
+ await this.store.delete(runId, state);
}
/** Delete a cancelled run and surface failures so its parent can remain retryable. */
- async cancel(runId: string): Promise {
- await this.store.delete(runId);
+ async cancel(runId: string, state?: SerializableAgentState): Promise {
+ await this.store.delete(runId, state);
}
}
diff --git a/packages/@n8n/agents/src/types/sdk/memory.ts b/packages/@n8n/agents/src/types/sdk/memory.ts
index db6a7c661e3..5390ed26746 100644
--- a/packages/@n8n/agents/src/types/sdk/memory.ts
+++ b/packages/@n8n/agents/src/types/sdk/memory.ts
@@ -381,6 +381,6 @@ export interface CheckpointStore {
* false when another process has already claimed or changed the snapshot.
*/
claimForResume?(key: string, state: SerializableAgentState): Promise;
- /** Delete a snapshot by key. */
- delete(key: string): Promise;
+ /** Delete a snapshot. Hosts can use the finishing state to check ownership. */
+ delete(key: string, state?: SerializableAgentState): Promise;
}
diff --git a/packages/@n8n/db/src/migrations/common/1790241069212-CreateAgentMessageQueue.ts b/packages/@n8n/db/src/migrations/common/1790241069212-CreateAgentMessageQueue.ts
new file mode 100644
index 00000000000..9955453d57d
--- /dev/null
+++ b/packages/@n8n/db/src/migrations/common/1790241069212-CreateAgentMessageQueue.ts
@@ -0,0 +1,47 @@
+import type { MigrationContext, ReversibleMigration } from '../migration-types';
+
+export class CreateAgentMessageQueue1790241069212 implements ReversibleMigration {
+ async up({
+ schemaBuilder: { createTable, createIndex, column },
+ escape,
+ isSqlite,
+ }: MigrationContext) {
+ // SQLite needs AUTOINCREMENT to prevent ID reuse after queue rows are deleted.
+ const id = isSqlite
+ ? column('id').int.primary.autoGenerate
+ : column('id').bigint.primary.autoGenerate2;
+ await createTable('agent_message_queue')
+ .withColumns(
+ id.comment('Acceptance order; IDs are not reused'),
+ column('threadId').varchar(128).notNull,
+ column('source').varchar(32).notNull.comment('Preview or integration source'),
+ column('payload').json.notNull.comment(
+ 'Input, attachment references, identity, and reply context',
+ ),
+ column('executionId').varchar(36).comment('Current execution; NULL means pending'),
+ )
+ .withForeignKey('threadId', {
+ tableName: 'agent_execution_threads',
+ columnName: 'id',
+ onDelete: 'CASCADE',
+ })
+ .withForeignKey('executionId', {
+ tableName: 'agent_execution',
+ columnName: 'id',
+ onDelete: 'NO ACTION',
+ })
+ .withIndexOn(['threadId', 'id'])
+ .withIndexOn('executionId').withTimestamps;
+ await createIndex(
+ 'agent_message_queue',
+ ['threadId'],
+ true,
+ undefined,
+ `${escape.columnName('executionId')} IS NOT NULL`,
+ );
+ }
+
+ async down({ schemaBuilder: { dropTable } }: MigrationContext) {
+ await dropTable('agent_message_queue');
+ }
+}
diff --git a/packages/cli/src/modules/agents/__tests__/agent-chat-execution.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-chat-execution.service.test.ts
index 75911724ed5..bbf00b9b091 100644
--- a/packages/cli/src/modules/agents/__tests__/agent-chat-execution.service.test.ts
+++ b/packages/cli/src/modules/agents/__tests__/agent-chat-execution.service.test.ts
@@ -8,10 +8,7 @@ import { mock } from 'vitest-mock-extended';
import { NotFoundError } from '@/errors/response-errors/not-found.error';
import type { Publisher } from '@/scaling/pubsub/publisher.service';
-import {
- AgentChatExecutionService,
- AgentTurnAlreadyRunningError,
-} from '../agent-chat-execution.service';
+import { AgentChatExecutionService } from '../agent-chat-execution.service';
import type { AgentExecutionService } from '../agent-execution.service';
import type { AgentExecutionUpdateBroadcaster } from '../agent-execution-update-broadcaster';
import type { AgentExecution } from '../entities/agent-execution.entity';
@@ -88,49 +85,6 @@ function makeService() {
beforeEach(() => Container.reset());
-it('serializes thread admission without queuing and releases the lease after creation', async () => {
- const { service, repository } = makeService();
- const entered = createDeferredPromise();
- const created = createDeferredPromise();
- const admission = service.admit(context.threadId, async () => {
- entered.resolve();
- await created.promise;
- repository.existsRunningByThread.mockResolvedValue(true);
- });
- await entered.promise;
- const competing = vi.fn();
- await expect(service.admit(context.threadId, competing)).rejects.toBeInstanceOf(
- AgentTurnAlreadyRunningError,
- );
- expect(competing).not.toHaveBeenCalled();
- created.resolve();
- await admission;
- await expect(service.admit(context.threadId, competing)).rejects.toBeInstanceOf(
- AgentTurnAlreadyRunningError,
- );
- repository.existsRunningByThread.mockResolvedValue(false);
- await service.admit(context.threadId, competing);
- expect(competing).toHaveBeenCalledOnce();
-
- const automaticEntered = createDeferredPromise();
- const automaticCreated = createDeferredPromise();
- const automaticAdmission = service.admitAutomaticContinuation(
- context.threadId,
- context.agentId,
- 'run-1',
- async () => {
- automaticEntered.resolve();
- await automaticCreated.promise;
- },
- );
- await automaticEntered.promise;
- await expect(service.admit(context.threadId, competing)).rejects.toBeInstanceOf(
- AgentTurnAlreadyRunningError,
- );
- automaticCreated.resolve();
- await automaticAdmission;
-});
-
it.each([
{ userId: 'other-user' },
{ projectId: 'other-project' },
@@ -198,9 +152,6 @@ it.each([false, true])(
? expect(settlement).rejects.toBe(error)
: expect(settlement).resolves.toBeUndefined();
await deleting.promise;
- await expect(service.admit(context.threadId, vi.fn())).rejects.toBeInstanceOf(
- AgentTurnAlreadyRunningError,
- );
deleted.resolve();
await settled;
expect(checkpointStorage.delete).toHaveBeenCalledExactlyOnceWith('run-1', 'agent-1');
diff --git a/packages/cli/src/modules/agents/__tests__/agent-checkpoint.repository.test.ts b/packages/cli/src/modules/agents/__tests__/agent-checkpoint.repository.test.ts
index f9520dff7cf..aaf8c2223fa 100644
--- a/packages/cli/src/modules/agents/__tests__/agent-checkpoint.repository.test.ts
+++ b/packages/cli/src/modules/agents/__tests__/agent-checkpoint.repository.test.ts
@@ -1,4 +1,8 @@
/* eslint-disable @typescript-eslint/unbound-method -- mock-based tests intentionally reference unbound methods */
+import type { TransactionRunner } from '@n8n/db';
+import { MoreThan } from '@n8n/typeorm';
+import { mock } from 'vitest-mock-extended';
+
import { mockEntityManager } from '@test/mocking';
import { AgentCheckpoint } from '../entities/agent-checkpoint.entity';
@@ -12,7 +16,7 @@ describe('AgentCheckpointRepository', () => {
beforeEach(() => {
vi.clearAllMocks();
- repository = new AgentCheckpointRepository(mockDataSource as never);
+ repository = new AgentCheckpointRepository(mockDataSource as never, mock());
});
describe('claimForResume', () => {
@@ -25,6 +29,7 @@ describe('AgentCheckpointRepository', () => {
'agent-1',
'{"status":"suspended"}',
'{"status":"running"}',
+ new Date(0),
),
).resolves.toBe(true);
@@ -34,6 +39,7 @@ describe('AgentCheckpointRepository', () => {
agentId: 'agent-1',
expired: false,
state: '{"status":"suspended"}',
+ updatedAt: MoreThan(new Date(0)),
},
{ state: '{"status":"running"}' },
);
@@ -48,6 +54,7 @@ describe('AgentCheckpointRepository', () => {
'agent-1',
'{"status":"suspended"}',
'{"status":"running"}',
+ new Date(0),
),
).resolves.toBe(false);
});
@@ -55,13 +62,14 @@ describe('AgentCheckpointRepository', () => {
describe('cancelSuspended', () => {
it('matches checkpoints scoped to the current agent', async () => {
- vi.spyOn(repository, 'update').mockResolvedValue({ affected: 1 } as never);
+ entityManager.update.mockResolvedValue({ affected: 1 } as never);
await expect(
repository.cancelSuspended('run-1', 'agent-1', '{"status":"suspended"}'),
).resolves.toBe(true);
- expect(repository.update).toHaveBeenCalledWith(
+ expect(entityManager.update).toHaveBeenCalledWith(
+ AgentCheckpoint,
{
runId: 'run-1',
agentId: 'agent-1',
diff --git a/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts
index 00227f94c65..93752601215 100644
--- a/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts
+++ b/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts
@@ -35,6 +35,7 @@ import {
import type { AgentExecutionRepository } from '../repositories/agent-execution.repository';
import type { AgentExecutionUpdateBroadcaster } from '../agent-execution-update-broadcaster';
import type { AgentExecutionService } from '../agent-execution.service';
+import type { AgentMessageQueueService } from '../agent-message-queue.service';
import type { AgentRunTracingService } from '../agent-run-tracing.service';
import type { AgentRuntimeCacheService } from '../agent-runtime-cache.service';
import { AgentExecutionRecordingError } from '../agent-execution-recording.error';
@@ -176,6 +177,7 @@ function makeRuntime(
function makeService(sandboxEnabled = false) {
const checkpointStorage = mock();
const executionService = mock();
+ executionService.getAbortSignal.mockReturnValue(new AbortController().signal);
const executionRepository = mock();
const chatExecutionService = new AgentChatExecutionService(
Container.get(LockService),
@@ -233,7 +235,12 @@ function makeService(sandboxEnabled = false) {
mockLogger(),
checkpointStorage,
executionService,
- new AgentTurnExecutionService(mockLogger(), executionService, chatExecutionService),
+ new AgentTurnExecutionService(
+ mockLogger(),
+ executionService,
+ chatExecutionService,
+ mock(),
+ ),
telemetry,
runtimeCacheService,
integrationMessageContextService,
@@ -382,19 +389,15 @@ describe('AgentExecutionOrchestratorService', () => {
});
it('rejects a competing preview without creating an execution or claiming a resume', async () => {
- const {
- stream,
- onExecutionStarted,
- sdkStart,
- executionService,
- executionRepository,
- runtimeCacheService,
- } = makeTurn({ previewChat: true });
- executionRepository.existsRunningByThread.mockResolvedValue(true);
+ const { stream, onExecutionStarted, sdkStart, executionService, runtimeCacheService } =
+ makeTurn({ previewChat: true });
+ executionService.startExecutionRecording.mockRejectedValue(
+ new AgentTurnAlreadyRunningError(),
+ );
await expect(collect(stream)).rejects.toBeInstanceOf(AgentTurnAlreadyRunningError);
expect(onExecutionStarted).not.toHaveBeenCalled();
expect(sdkStart).not.toHaveBeenCalled();
- expect(executionService.startExecutionRecording).not.toHaveBeenCalled();
+ expect(executionService.startExecutionRecording).toHaveBeenCalledOnce();
expect(runtimeCacheService.releaseRuntimeLease).toHaveBeenCalledOnce();
});
@@ -435,6 +438,9 @@ describe('AgentExecutionOrchestratorService', () => {
);
const result = collect(stream);
const signal = await started.promise;
+ executionService.startExecutionRecording.mockRejectedValue(
+ new AgentTurnAlreadyRunningError(),
+ );
const resumeAgain = async () =>
await collect(
service.resumeForChat({
@@ -460,14 +466,14 @@ describe('AgentExecutionOrchestratorService', () => {
userId,
});
- expect(executionService.startExecutionRecording).toHaveBeenCalledOnce();
+ expect(executionService.startExecutionRecording).toHaveBeenCalledTimes(2);
expect(runtime.agent.resume).toHaveBeenCalledOnce();
expect(onExecutionStarted).not.toHaveBeenCalled();
expect(signal.aborted).toBe(true);
release.resolve();
await result;
await expect(resumeAgain()).rejects.toBeInstanceOf(AgentTurnAlreadyRunningError);
- expect(executionService.startExecutionRecording).toHaveBeenCalledOnce();
+ expect(executionService.startExecutionRecording).toHaveBeenCalledTimes(3);
expect(runtime.agent.resume).toHaveBeenCalledOnce();
expect(executionService.finalizeExecution).toHaveBeenCalledWith(
'execution-1',
@@ -630,7 +636,11 @@ describe('AgentExecutionOrchestratorService', () => {
await expect(collect(fixtures.stream)).rejects.toBe(error);
expect(fixtures.sdkStart).not.toHaveBeenCalled();
- expect(fixtures.executionService.startExecutionRecording).not.toHaveBeenCalled();
+ expect(fixtures.executionService.startExecutionRecording).toHaveBeenCalledTimes(
+ operation === 'start' ? 1 : 0,
+ );
+ if (operation === 'start')
+ expect(fixtures.executionService.finalizeExecution).toHaveBeenCalledOnce();
expect(fixtures.runtimeCacheService.releaseRuntimeLease).toHaveBeenCalledExactlyOnceWith(
fixtures.runtime.agent,
);
@@ -1137,7 +1147,7 @@ describe('AgentExecutionOrchestratorService', () => {
),
},
executionCounter: expect.any(Object),
- abortSignal: abortController.signal,
+ abortSignal: expect.any(AbortSignal),
modelStreamIdleTimeoutMs: 90_000,
modelStreamFirstOutputTimeoutMs: 180_000,
}),
@@ -1562,6 +1572,7 @@ describe('AgentExecutionOrchestratorService', () => {
hostMetadata: {
...encodeAgentSandboxHostMetadata({ projectId, principalHash: taskPrincipalHash }),
...encodeIntegrationMessageContext(selectedContext),
+ n8nExecutionId: 'execution-1',
},
},
}),
@@ -1915,7 +1926,7 @@ describe('AgentExecutionOrchestratorService', () => {
});
expect(runtime.agent.stream).toHaveBeenCalledWith(
'[]',
- expect.objectContaining({ abortSignal }),
+ expect.objectContaining({ abortSignal: expect.any(AbortSignal) }),
);
expect(executionService.startExecutionRecording).toHaveBeenCalledWith(
expect.objectContaining({ userMessage: null, sessionMode: 'existing' }),
@@ -2317,7 +2328,7 @@ describe('AgentExecutionOrchestratorService', () => {
expect.objectContaining({
runId: 'run-1',
toolCallId: 'tc-1',
- abortSignal: abortController.signal,
+ abortSignal: expect.any(AbortSignal),
}),
);
expect(externalHooks.run).not.toHaveBeenCalled();
diff --git a/packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts
index 080aeddaa77..fc6d477e98e 100644
--- a/packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts
+++ b/packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts
@@ -21,6 +21,7 @@ import type { N8NCheckpointStorage } from '../integrations/n8n-checkpoint-storag
import type { N8nMemory } from '../integrations/n8n-memory';
import type { AgentExecutionThreadRepository } from '../repositories/agent-execution-thread.repository';
import type { AgentExecutionRepository } from '../repositories/agent-execution.repository';
+import type { AgentMessageQueueRepository } from '../repositories/agent-message-queue.repository';
const previewAccess = { accessScope: 'user' as const, ownerId: 'user-1' };
@@ -78,14 +79,21 @@ describe('AgentExecutionService', () => {
const checkpointStorage = mock();
let executionUpdateBroadcaster: Mocked;
const txRunner = mock();
+ const queueRepository = mock();
beforeEach(() => {
vi.clearAllMocks();
agentExecutionRepository = mock();
+ agentExecutionRepository.findRunningByThread.mockResolvedValue([]);
+ agentExecutionRepository.touchRunning.mockResolvedValue(true);
+ queueRepository.findActive.mockResolvedValue(null);
+ queueRepository.findHead.mockResolvedValue(null);
+ checkpointStorage.findSuspendedForThread.mockResolvedValue(null);
agentExecutionRepository.updateIfRunning.mockResolvedValue(true);
agentExecutionRepository.updateTimelineIfRunning.mockResolvedValue(true);
agentExecutionThreadRepository = mock();
+ agentExecutionThreadRepository.lockById.mockResolvedValue(makeThread());
n8nMemory = mock();
memoryBackend = mock();
n8nMemory.getImplementation.mockReturnValue(memoryBackend);
@@ -110,6 +118,7 @@ describe('AgentExecutionService', () => {
executionUpdateBroadcaster,
checkpointStorage,
txRunner,
+ queueRepository,
);
});
@@ -174,7 +183,7 @@ describe('AgentExecutionService', () => {
);
});
- it('keeps a running execution alive during title lookup and until finalization', async () => {
+ it('keeps an execution alive until finalization and then synchronizes its title', async () => {
vi.useFakeTimers();
const titleLookupStarted = createDeferredPromise();
const titleLookup =
@@ -192,7 +201,7 @@ describe('AgentExecutionService', () => {
agentExecutionRepository.saveInContext.mockResolvedValue({
id: 'execution-1',
} as AgentExecution);
- agentExecutionRepository.touchRunning.mockResolvedValue();
+ agentExecutionRepository.touchRunning.mockResolvedValue(true);
agentExecutionRepository.updateIfRunning.mockResolvedValue(true);
const recording = service.startExecutionRecording(
@@ -206,11 +215,10 @@ describe('AgentExecutionService', () => {
},
new Date(),
);
- await titleLookupStarted.promise;
+ const executionId = await recording;
+ expect(memoryBackend.getThread).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(30_000);
expect(agentExecutionRepository.touchRunning).toHaveBeenCalledWith('execution-1');
- titleLookup.resolve(null);
- const executionId = await recording;
expect(executionUpdateBroadcaster.notify).toHaveBeenCalledWith({
projectId: 'project-1',
@@ -223,7 +231,7 @@ describe('AgentExecutionService', () => {
);
expect(agentExecutionRepository.touchRunning).toHaveBeenCalledWith(executionId);
- await service.finalizeExecution(executionId, {
+ const finalization = service.finalizeExecution(executionId, {
threadId: 'thread-1',
agentId: 'agent-1',
agentName: 'Agent',
@@ -231,9 +239,11 @@ describe('AgentExecutionService', () => {
userMessage: 'Run',
record: makeMessageRecord(),
});
+ await titleLookupStarted.promise;
await vi.advanceTimersByTimeAsync(30_000);
-
expect(agentExecutionRepository.touchRunning).toHaveBeenCalledOnce();
+ titleLookup.resolve(null);
+ await finalization;
} finally {
titleLookup.resolve(null);
vi.useRealTimers();
@@ -262,6 +272,44 @@ describe('AgentExecutionService', () => {
return params;
}
+ it('retries failed heartbeats and aborts only after confirmed ownership loss', async () => {
+ vi.useFakeTimers();
+ try {
+ await startSnapshotExecution();
+ const signal = service.getAbortSignal('execution-1');
+ agentExecutionRepository.touchRunning
+ .mockRejectedValueOnce(new Error('temporarily unavailable'))
+ .mockResolvedValueOnce(true)
+ .mockResolvedValue(false);
+ await vi.advanceTimersByTimeAsync(30_000);
+ expect(signal.aborted).toBe(false);
+
+ const timeline: TimelineEvent[] = [{ type: 'text', content: 'Working', timestamp: 1 }];
+ service.recordTimelineSnapshot({
+ executionId: 'execution-1',
+ projectId: 'project-1',
+ agentId: 'agent-1',
+ threadId: 'thread-1',
+ timeline,
+ });
+ await vi.advanceTimersByTimeAsync(0);
+ expect(agentExecutionRepository.updateTimelineIfRunning).toHaveBeenCalledWith(
+ 'execution-1',
+ timeline,
+ );
+
+ await vi.advanceTimersByTimeAsync(30_000);
+ expect(agentExecutionRepository.touchRunning).toHaveBeenCalledTimes(2);
+ expect(signal.aborted).toBe(false);
+ await vi.advanceTimersByTimeAsync(30_000);
+ expect(signal.aborted).toBe(true);
+ await vi.advanceTimersByTimeAsync(30_000);
+ expect(agentExecutionRepository.touchRunning).toHaveBeenCalledTimes(3);
+ } finally {
+ vi.useRealTimers();
+ }
+ });
+
it('serializes timeline snapshot updates', async () => {
const params = await startSnapshotExecution();
let releaseFirstWrite!: () => void;
@@ -468,6 +516,7 @@ describe('AgentExecutionService', () => {
executionUpdateBroadcaster,
checkpointStorage,
txRunner,
+ queueRepository,
);
const record = makeMessageRecord({
@@ -559,6 +608,7 @@ describe('AgentExecutionService', () => {
executionUpdateBroadcaster,
checkpointStorage,
txRunner,
+ queueRepository,
);
const record = makeMessageRecord({
@@ -1024,18 +1074,22 @@ describe('AgentExecutionService', () => {
executionUpdateBroadcaster,
checkpointStorage,
txRunner,
+ queueRepository,
);
const partial = [{ type: 'text', content: 'Partial', timestamp: 1, endTime: 2 }] as const;
agentExecutionRepository.updateIfRunning.mockResolvedValue(true);
agentExecutionThreadRepository.findOneBy.mockResolvedValue(makeThread());
- await service.finalizeInterruptedExecution({
- id: 'execution-1',
- threadId: 'thread-1',
- startedAt: new Date(Date.now() - 100),
- timeline: [...partial],
- thread: makeThread(),
- } as AgentExecution);
+ await service.finalizeInterruptedExecution(
+ {
+ id: 'execution-1',
+ threadId: 'thread-1',
+ startedAt: new Date(Date.now() - 100),
+ timeline: [...partial],
+ thread: makeThread(),
+ } as AgentExecution,
+ new Date(Date.now() - 120_000),
+ );
await vi.waitFor(() =>
expect(executionUpdateBroadcaster.notify).toHaveBeenCalledWith({
@@ -1062,6 +1116,8 @@ describe('AgentExecutionService', () => {
},
},
}),
+ expect.any(Date),
+ {},
);
expect(agentExecutionLogStore.write).not.toHaveBeenCalled();
});
diff --git a/packages/cli/src/modules/agents/__tests__/agent-interrupted-execution-sweeper.test.ts b/packages/cli/src/modules/agents/__tests__/agent-interrupted-execution-sweeper.test.ts
index b9036d8773b..9831e7fef28 100644
--- a/packages/cli/src/modules/agents/__tests__/agent-interrupted-execution-sweeper.test.ts
+++ b/packages/cli/src/modules/agents/__tests__/agent-interrupted-execution-sweeper.test.ts
@@ -29,7 +29,11 @@ function setup(options: { backgroundTasksEnabled?: boolean } = {}) {
}
describe('AgentInterruptedExecutionSweeper', () => {
+ afterEach(() => vi.useRealTimers());
+
it('terminalizes an abandoned running execution', async () => {
+ vi.useFakeTimers();
+ vi.setSystemTime(new Date('2026-01-01T00:02:00.000Z'));
const { sweeper, repository, executionService } = setup();
const execution = {
id: 'execution-1',
@@ -43,7 +47,10 @@ describe('AgentInterruptedExecutionSweeper', () => {
await sweeper.sweep();
- expect(executionService.finalizeInterruptedExecution).toHaveBeenCalledWith(execution);
+ expect(executionService.finalizeInterruptedExecution).toHaveBeenCalledWith(
+ execution,
+ new Date('2026-01-01T00:00:00.000Z'),
+ );
});
it('leaves a recently active execution running in another process', async () => {
diff --git a/packages/cli/src/modules/agents/__tests__/agent-runtime-reconstruction-integration-tools.test.ts b/packages/cli/src/modules/agents/__tests__/agent-runtime-reconstruction-integration-tools.test.ts
index fb4103857a9..a8b93a3189e 100644
--- a/packages/cli/src/modules/agents/__tests__/agent-runtime-reconstruction-integration-tools.test.ts
+++ b/packages/cli/src/modules/agents/__tests__/agent-runtime-reconstruction-integration-tools.test.ts
@@ -34,6 +34,7 @@ import type { NodeToolAiGatewayService } from '../json-config/node-tool-ai-gatew
import { AgentCustomToolsService } from '../agent-custom-tools.service';
import { AgentExecutionOrchestratorService } from '../agent-execution-orchestrator.service';
import type { AgentExecutionService } from '../agent-execution.service';
+import type { AgentMessageQueueService } from '../agent-message-queue.service';
import type { AgentChatExecutionService } from '../agent-chat-execution.service';
import { AgentIntegrationPersistenceService } from '../agent-integration-persistence.service';
import type { AgentKnowledgeMirrorService } from '../agent-knowledge-mirror.service';
@@ -217,6 +218,7 @@ describe('AgentRuntimeReconstructionService integration tools', () => {
n8nMemory.getImplementation.mockReturnValue(memoryBackend);
n8nCheckpointStorage = mock();
agentExecutionService = mock();
+ agentExecutionService.getAbortSignal.mockReturnValue(new AbortController().signal);
agentKnowledgeService = mock();
publisher = mock();
publisher.publishCommand.mockResolvedValue();
@@ -281,6 +283,7 @@ describe('AgentRuntimeReconstructionService integration tools', () => {
logger,
agentExecutionService,
mock(),
+ mock(),
),
telemetry,
runtimeCacheService,
diff --git a/packages/cli/src/modules/agents/__tests__/agent-workflow-execution.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-workflow-execution.service.test.ts
index dc005aa9f5b..98babc6561b 100644
--- a/packages/cli/src/modules/agents/__tests__/agent-workflow-execution.service.test.ts
+++ b/packages/cli/src/modules/agents/__tests__/agent-workflow-execution.service.test.ts
@@ -14,6 +14,7 @@ import type { ExecutionLevelTracer } from '@/modules/otel/execution-level-tracer
import type { Telemetry } from '@/telemetry';
import type { AgentExecutionService } from '../agent-execution.service';
+import type { AgentMessageQueueService } from '../agent-message-queue.service';
import type { AgentChatExecutionService } from '../agent-chat-execution.service';
import type { AgentRunTracingService } from '../agent-run-tracing.service';
import type { AgentRuntimeReconstructionService } from '../agent-runtime-reconstruction.service';
@@ -121,6 +122,7 @@ function makeRuntime(chunks: StreamChunk[] = [{ type: 'finish', finishReason: 's
function makeService() {
const agentRepository = mock();
const executionService = mock();
+ executionService.getAbortSignal.mockReturnValue(new AbortController().signal);
const telemetry = mock();
const credentialsService = mock();
const reconstructionService = mock();
@@ -146,6 +148,7 @@ function makeService() {
mockLogger(),
executionService,
mock(),
+ mock(),
),
telemetry,
credentialsService,
diff --git a/packages/cli/src/modules/agents/agent-chat-execution.service.ts b/packages/cli/src/modules/agents/agent-chat-execution.service.ts
index 4dc78c48c70..51d7de756ad 100644
--- a/packages/cli/src/modules/agents/agent-chat-execution.service.ts
+++ b/packages/cli/src/modules/agents/agent-chat-execution.service.ts
@@ -1,8 +1,7 @@
-import { LockAcquisitionTimeoutError, LockNamespace, LockService } from '@n8n/backend-common';
+import { LockNamespace, LockService } from '@n8n/backend-common';
import { OnPubSubEvent } from '@n8n/decorators';
import { Service } from '@n8n/di';
import { InstanceSettings } from 'n8n-core';
-import { UserError } from 'n8n-workflow';
import { NotFoundError } from '@/errors/response-errors/not-found.error';
import type { PubSubCommandMap } from '@/scaling/pubsub/pubsub.event-map';
@@ -27,11 +26,7 @@ export interface CancelSuspendedRunParams {
resourceId: string;
}
-export class AgentTurnAlreadyRunningError extends UserError {
- constructor() {
- super('A turn is already running in this conversation.');
- }
-}
+export { AgentTurnAlreadyRunningError } from './agent-turn-already-running.error';
@Service()
export class AgentChatExecutionService {
@@ -50,50 +45,6 @@ export class AgentChatExecutionService {
private readonly executionUpdates: AgentExecutionUpdateBroadcaster,
) {}
- async admit(threadId: string, create: () => Promise): Promise {
- return await this.withAdmissionLease(`agent-preview-turn:${threadId}`, async (signal) => {
- if (await this.executionRepository.existsRunningByThread(threadId)) {
- throw new AgentTurnAlreadyRunningError();
- }
- signal.throwIfAborted();
- return await create();
- });
- }
-
- async admitAutomaticContinuation(
- threadId: string,
- agentId: string,
- runId: string,
- createAndClaim: () => Promise,
- ): Promise {
- return await this.withAdmissionLease(`agent-preview-turn:${threadId}`, async (signal) => {
- const checkpoint = await this.checkpointStorage.getStatus(runId, agentId);
- if (
- checkpoint.status !== 'active' ||
- checkpoint.checkpoint.status !== 'suspended' ||
- checkpoint.checkpoint.persistence?.threadId !== threadId
- ) {
- throw new AgentTurnAlreadyRunningError();
- }
- signal.throwIfAborted();
- return await createAndClaim();
- });
- }
-
- private async withAdmissionLease(
- key: string,
- admit: (signal: AbortSignal) => Promise,
- ): Promise {
- try {
- return await this.lockService.withLease(LockNamespace.KNOWN_LOCKS, key, admit, {
- waitTimeoutMs: 0,
- });
- } catch (error) {
- if (error instanceof LockAcquisitionTimeoutError) throw new AgentTurnAlreadyRunningError();
- throw error;
- }
- }
-
register(context: ExecutionContext, controller: AbortController): void {
this.executions.set(context.executionId, { context, controller });
}
diff --git a/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts b/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts
index ad7d0b08663..427a808a075 100644
--- a/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts
+++ b/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts
@@ -62,6 +62,7 @@ import { modelStreamStallOptions } from './model-stream-stall-options';
import { AgentRepository } from './repositories/agent.repository';
import type { ToolRegistry } from './tool-registry';
import type { StoredAttachmentRef } from './types/agent-chat-attachment';
+import type { AgentExecutionAdmission } from './types/agent-queued-message';
import { createAgentExecutionCounter } from './utils/agent-execution-counter';
import { getPublishedAgentSnapshot } from './utils/agent-published-snapshot';
import { buildInboundUserMessage } from './utils/inbound-attachments';
@@ -82,6 +83,8 @@ interface AgentExecutionInput {
}
interface ChatExecutionInput extends AgentExecutionInput {
+ admittedExecution?: AgentExecutionAdmission;
+ abortSignal?: AbortSignal;
sessionMode?: AgentSessionMode;
/** Stored attachments for the user turn. */
attachments?: StoredAttachmentRef[];
@@ -202,6 +205,7 @@ export interface ExecuteForWakeConfig extends AgentExecutionInput {
}
export interface StreamChatResponseConfig extends ChatExecutionInput, ChatExecutionCallbacks {
+ onAdmitted?: () => Promise;
access: AgentThreadAccess;
messageContext?: IntegrationMessageContext | null;
agentInstance: RuntimeAgent;
@@ -468,6 +472,7 @@ export class AgentExecutionOrchestratorService {
source: integrationType,
access: { accessScope: 'project', ownerId: null },
sessionMode,
+ admittedExecution: config.admittedExecution,
},
),
async (runtime) => {
@@ -477,13 +482,21 @@ export class AgentExecutionOrchestratorService {
messageContext,
await this.integrationMessageContextService.getLatestForIncoming(memory.threadId),
);
- await this.integrationMessageContextService.installIncoming(
- messageContext,
- memory,
- config.contextConversation,
- );
}
+ const selectedContext = messageContext;
+ const conversation = config.contextConversation;
return this.streamChatResponse({
+ admittedExecution: config.admittedExecution,
+ abortSignal: config.abortSignal,
+ onAdmitted:
+ selectedContext && conversation
+ ? async () =>
+ await this.integrationMessageContextService.installIncoming(
+ selectedContext,
+ memory,
+ conversation,
+ )
+ : undefined,
messageContext,
access: { accessScope: 'project', ownerId: null },
agentInstance: runtime.agent,
@@ -701,6 +714,8 @@ export class AgentExecutionOrchestratorService {
*/
async *streamChatResponse(config: StreamChatResponseConfig): AsyncGenerator {
yield* this.turnExecutionService.execute({
+ admittedExecution: config.admittedExecution,
+ onAdmitted: config.onAdmitted,
agentInstance: config.agentInstance,
toolRegistry: config.toolRegistry,
mcpServerAttributions: config.mcpServerAttributions,
@@ -758,17 +773,26 @@ export class AgentExecutionOrchestratorService {
| 'taskVersionId'
| 'access'
| 'sessionMode'
+ | 'resumeRunId'
> & {
+ admittedExecution?: AgentExecutionAdmission;
onExecutionRecorded?: (executionId: string) => void;
abortSignal?: AbortSignal;
automaticContinuationRunId?: string;
},
): Promise {
- const { onExecutionRecorded, abortSignal, automaticContinuationRunId, ...recording } = session;
+ const {
+ onExecutionRecorded,
+ abortSignal,
+ automaticContinuationRunId,
+ admittedExecution,
+ ...recording
+ } = session;
abortSignal?.throwIfAborted();
try {
return await this.runtimeCacheService.getRuntime(params);
} catch (error) {
+ if (admittedExecution) throw error;
abortSignal?.throwIfAborted();
const { agentId, projectId } = params;
let agent;
@@ -830,6 +854,7 @@ export class AgentExecutionOrchestratorService {
},
{
threadId: memoryScope.threadId,
+ resumeRunId: runId,
userMessage: null,
source,
onExecutionRecorded,
@@ -1039,6 +1064,7 @@ export class AgentExecutionOrchestratorService {
{
threadId: memory.threadId,
access,
+ admittedExecution: config.admittedExecution,
userMessage: message,
attachments,
source,
@@ -1067,8 +1093,15 @@ export class AgentExecutionOrchestratorService {
previewChat,
sessionMode,
} = config;
- const messageContext = await this.installDraftMessageContext(memory, user.id);
+ const messageContext = this.createDraftMessageContext(memory, user.id);
return this.streamChatResponse({
+ admittedExecution: config.admittedExecution,
+ onAdmitted: async () =>
+ await this.integrationMessageContextService.setLatest(
+ memory.threadId,
+ memory.resourceId,
+ messageContext,
+ ),
access,
messageContext,
agentInstance: runtime.agent,
@@ -1092,23 +1125,17 @@ export class AgentExecutionOrchestratorService {
});
}
- private async installDraftMessageContext(
+ private createDraftMessageContext(
memory: AgentMemoryScope,
userId: string,
- ): Promise {
- const context: IntegrationMessageContext = {
+ ): IntegrationMessageContext {
+ return {
integrationConnectionId: N8N_CHAT_INTEGRATION_TYPE,
platform: N8N_CHAT_INTEGRATION_TYPE,
target: { type: 'dm', userId, threadId: memory.threadId },
interactingUserId: userId,
updatedAt: new Date().toISOString(),
};
- await this.integrationMessageContextService.setLatest(
- memory.threadId,
- memory.resourceId,
- context,
- );
- return context;
}
private async prepareChatTurn(config: StreamChatResponseConfig): Promise {
diff --git a/packages/cli/src/modules/agents/agent-execution.service.ts b/packages/cli/src/modules/agents/agent-execution.service.ts
index 187b4a4dfc8..4166c6e81e3 100644
--- a/packages/cli/src/modules/agents/agent-execution.service.ts
+++ b/packages/cli/src/modules/agents/agent-execution.service.ts
@@ -20,6 +20,9 @@ import { Telemetry } from '@/telemetry';
import { AgentChatAttachmentService } from './agent-chat-attachment.service';
import type { StoredAttachmentRef } from './types/agent-chat-attachment';
import { AgentExecutionUpdateBroadcaster } from './agent-execution-update-broadcaster';
+import { AgentTurnAlreadyRunningError } from './agent-turn-already-running.error';
+import { AgentMessageQueueRepository } from './repositories/agent-message-queue.repository';
+import { checkpointExecutionId } from './types/agent-queued-message';
import { buildAgentTurnMetrics } from './agent-telemetry';
import {
AgentExecutionThread,
@@ -84,6 +87,15 @@ export interface StartExecutionParams extends Omit();
+ private readonly executionControllers = new Map();
private readonly pendingTimelineSnapshots = new Map<
string,
@@ -142,49 +155,115 @@ export class AgentExecutionService {
private readonly executionUpdateBroadcaster: AgentExecutionUpdateBroadcaster,
private readonly checkpointStorage: N8NCheckpointStorage,
private readonly txRunner: TransactionRunner,
+ private readonly queueRepository: AgentMessageQueueRepository,
) {}
async startExecutionRecording(params: StartExecutionParams, startedAt: Date): Promise {
- const { inserted, created, needsTitleSync } = await this.txRunner.run({}, async (ctx) => {
- const prepared = await this.prepareThread(params, ctx);
- const execution = this.agentExecutionRepository.create({
- threadId: params.threadId,
- status: 'running',
- startedAt,
- stoppedAt: null,
- duration: 0,
- userMessage: prepared.userMessage,
- author: params.author ?? null,
- model: null,
- promptTokens: null,
- completionTokens: null,
- totalTokens: null,
- cost: null,
- // Save the background job signal before notifying clients that the execution started.
- timeline: params.initialTimeline?.length ? params.initialTimeline : null,
- storedAt: 'db',
- error: null,
- failureSummary: null,
- hitlStatus: null,
- source: params.source ?? null,
- attachments: params.attachments?.length ? params.attachments : null,
- });
- return {
- inserted: await this.agentExecutionRepository.saveInContext(execution, ctx),
- created: prepared.created,
- needsTitleSync: !prepared.created && !prepared.thread.title,
- };
+ const reservation = await this.txRunner.run(
+ {},
+ async (ctx) => await this.reserveExecution(params, startedAt, ctx),
+ );
+ this.activateExecution(reservation, params);
+ return reservation.execution.id;
+ }
+
+ async reserveExecution(
+ params: StartExecutionParams,
+ startedAt: Date,
+ ctx: OperationContext,
+ ): Promise {
+ const prepared = await this.prepareThread(params, ctx);
+ const queueItem = await this.checkAdmission(params, ctx);
+ const execution = this.agentExecutionRepository.create({
+ threadId: params.threadId,
+ status: 'running',
+ startedAt,
+ stoppedAt: null,
+ duration: 0,
+ userMessage: prepared.userMessage,
+ author: params.author ?? null,
+ model: null,
+ promptTokens: null,
+ completionTokens: null,
+ totalTokens: null,
+ cost: null,
+ // Save the background job signal before notifying clients that the execution started.
+ timeline: params.initialTimeline?.length ? params.initialTimeline : null,
+ storedAt: 'db',
+ error: null,
+ failureSummary: null,
+ hitlStatus: null,
+ source: params.source ?? null,
+ attachments: params.attachments?.length ? params.attachments : null,
});
- this.startHeartbeat(inserted.id);
- if (created) this.executionsNeedingTitleSync.add(inserted.id);
- if (needsTitleSync) await this.syncTitleFromMemory(params.threadId, params.agentId);
+ const inserted = await this.agentExecutionRepository.saveInContext(execution, ctx);
+ if (
+ queueItem &&
+ !(await this.queueRepository.linkExecution(
+ queueItem.id,
+ queueItem.executionId,
+ inserted.id,
+ ctx,
+ ))
+ ) {
+ throw new AgentTurnAlreadyRunningError();
+ }
+ return { execution: inserted, needsTitleSync: prepared.created || !prepared.thread.title };
+ }
+
+ private async checkAdmission(params: StartExecutionParams, ctx: OperationContext) {
+ const { threadId, agentId, resumeRunId, queueItemId } = params;
+ const running = await this.agentExecutionRepository.findRunningByThread(threadId, ctx);
+ const active = await this.queueRepository.findActive(threadId, ctx);
+ if (resumeRunId) {
+ const checkpoint = await this.checkpointStorage.getStatus(resumeRunId, agentId, ctx);
+ if (
+ checkpoint.status !== 'active' ||
+ checkpoint.checkpoint.status !== 'suspended' ||
+ checkpoint.checkpoint.persistence?.threadId !== threadId
+ ) {
+ throw new AgentTurnAlreadyRunningError();
+ }
+ const predecessorId = checkpointExecutionId(checkpoint.checkpoint);
+ const legacyPredecessor = !predecessorId && !active && params.allowSuspendedPredecessor;
+ if (
+ running.some(({ id }) => id !== predecessorId) &&
+ !(legacyPredecessor && running.length === 1)
+ ) {
+ throw new AgentTurnAlreadyRunningError();
+ }
+ return active;
+ }
+ if (
+ running.length ||
+ active ||
+ (await this.checkpointStorage.findSuspendedForThread(agentId, threadId, ctx))
+ ) {
+ throw new AgentTurnAlreadyRunningError();
+ }
+ const head = await this.queueRepository.findHead(threadId, ctx);
+ if (head?.id !== queueItemId && (head || queueItemId)) throw new AgentTurnAlreadyRunningError();
+ return head;
+ }
+
+ /** Activate only after the reservation transaction commits. */
+ activateExecution(
+ { execution, needsTitleSync }: AgentExecutionReservation,
+ params: StartExecutionParams,
+ ): void {
+ this.executionControllers.set(execution.id, new AbortController());
+ this.startHeartbeat(execution.id);
+ if (needsTitleSync) this.executionsNeedingTitleSync.add(execution.id);
this.executionUpdateBroadcaster.notify({
projectId: params.projectId,
agentId: params.agentId,
threadId: params.threadId,
- executionId: inserted.id,
+ executionId: execution.id,
});
- return inserted.id;
+ }
+
+ getAbortSignal(executionId: string): AbortSignal {
+ return this.executionControllers.get(executionId)?.signal ?? AbortSignal.abort();
}
recordTimelineSnapshot({ executionId, ...snapshot }: TimelineSnapshotParams): void {
@@ -246,26 +325,38 @@ export class AgentExecutionService {
}
}
- async finalizeInterruptedExecution(execution: RunningAgentExecution): Promise {
+ async finalizeInterruptedExecution(
+ execution: RunningAgentExecution,
+ staleBefore: Date,
+ ): Promise {
const timeline = execution.timeline ?? [];
const stoppedAt = new Date();
const error = 'Agent execution was interrupted by a process restart.';
const duration = execution.startedAt
? Math.max(0, stoppedAt.getTime() - execution.startedAt.getTime())
: 0;
- const finalized = await this.agentExecutionRepository.updateIfRunning(execution.id, {
- status: 'interrupted',
- stoppedAt,
- duration,
- timeline: timeline.length > 0 ? timeline : null,
- storedAt: 'db',
- error,
- failureSummary: computeExecutionFailureSummary({
- timeline,
- status: 'interrupted',
- error,
- stoppedAt: stoppedAt.getTime(),
- }),
+ const finalized = await this.txRunner.run({}, async (ctx) => {
+ if (!(await this.agentExecutionThreadRepository.lockById(execution.threadId, ctx)))
+ return false;
+ return await this.agentExecutionRepository.updateIfRunning(
+ execution.id,
+ {
+ status: 'interrupted',
+ stoppedAt,
+ duration,
+ timeline: timeline.length > 0 ? timeline : null,
+ storedAt: 'db',
+ error,
+ failureSummary: computeExecutionFailureSummary({
+ timeline,
+ status: 'interrupted',
+ error,
+ stoppedAt: stoppedAt.getTime(),
+ }),
+ },
+ staleBefore,
+ ctx,
+ );
});
if (finalized) void this.notifyInterruptedExecution(execution);
return finalized;
@@ -294,21 +385,34 @@ export class AgentExecutionService {
private startHeartbeat(executionId: string): void {
const timer = setInterval(() => {
- void this.agentExecutionRepository.touchRunning(executionId).catch((error: unknown) => {
- this.logger.warn('Failed to heartbeat a running agent execution', {
- executionId,
- error: error instanceof Error ? error.message : String(error),
+ void this.agentExecutionRepository
+ .touchRunning(executionId)
+ .then((owned) => {
+ if (!owned) this.abortLostExecution(executionId);
+ })
+ .catch((error: unknown) => {
+ this.logger.warn('Failed to heartbeat a running agent execution', {
+ executionId,
+ error: error instanceof Error ? error.message : String(error),
+ });
});
- });
}, AgentExecutionService.heartbeatIntervalMs);
timer.unref();
this.heartbeatTimers.set(executionId, timer);
}
+ private abortLostExecution(executionId: string): void {
+ this.executionControllers
+ .get(executionId)
+ ?.abort(new OperationalError('Agent execution ownership was lost'));
+ this.stopHeartbeat(executionId);
+ }
+
private stopHeartbeat(executionId: string): void {
const timer = this.heartbeatTimers.get(executionId);
if (timer) clearInterval(timer);
this.heartbeatTimers.delete(executionId);
+ this.executionControllers.delete(executionId);
}
private ensureTimelineSnapshotWrite(executionId: string): void {
@@ -334,7 +438,7 @@ export class AgentExecutionService {
}
}
- private async prepareThread(
+ async prepareThread(
params: StartExecutionParams,
ctx: OperationContext,
): Promise<{ userMessage: string | null; created: boolean; thread: AgentExecutionThread }> {
diff --git a/packages/cli/src/modules/agents/agent-interrupted-execution-sweeper.ts b/packages/cli/src/modules/agents/agent-interrupted-execution-sweeper.ts
index 6b8a8933fa3..c6161d4f9b3 100644
--- a/packages/cli/src/modules/agents/agent-interrupted-execution-sweeper.ts
+++ b/packages/cli/src/modules/agents/agent-interrupted-execution-sweeper.ts
@@ -39,7 +39,12 @@ export class AgentInterruptedExecutionSweeper {
) {
continue;
}
- if (await this.executionService.finalizeInterruptedExecution(execution)) {
+ if (
+ await this.executionService.finalizeInterruptedExecution(
+ execution,
+ new Date(Date.now() - AgentInterruptedExecutionSweeper.LIVENESS_GRACE_MS),
+ )
+ ) {
this.logger.info('Marked abandoned agent execution as interrupted', {
executionId: execution.id,
threadId: execution.threadId,
diff --git a/packages/cli/src/modules/agents/agent-message-queue.service.ts b/packages/cli/src/modules/agents/agent-message-queue.service.ts
new file mode 100644
index 00000000000..533988d8bf5
--- /dev/null
+++ b/packages/cli/src/modules/agents/agent-message-queue.service.ts
@@ -0,0 +1,161 @@
+import { TransactionRunner, type OperationContext } from '@n8n/db';
+import { Service } from '@n8n/di';
+import { OperationalError, UserError } from 'n8n-workflow';
+
+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 { 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 { AgentRepository } from './repositories/agent.repository';
+import type { AgentExecutionAdmission, AgentQueuedMessage } from './types/agent-queued-message';
+import type { AgentSessionMode } from './utils/agent-thread-access';
+
+export interface ClaimedAgentMessage {
+ item: AgentMessageQueue;
+ thread: AgentExecutionThread;
+ admission: AgentExecutionAdmission;
+ recording: StartExecutionParams;
+}
+
+@Service()
+export class AgentMessageQueueService {
+ /** The main consumer installs this callback. Webhook processes only enqueue. */
+ onAvailable?: (threadId: string) => void;
+
+ constructor(
+ private readonly txRunner: TransactionRunner,
+ private readonly repository: AgentMessageQueueRepository,
+ private readonly threadRepository: AgentExecutionThreadRepository,
+ private readonly executionRepository: AgentExecutionRepository,
+ private readonly executionService: AgentExecutionService,
+ private readonly checkpointStorage: N8NCheckpointStorage,
+ private readonly agentRepository: AgentRepository,
+ ) {}
+
+ async enqueue(
+ input: {
+ agentId: string;
+ projectId: string;
+ threadId: string;
+ sessionMode: AgentSessionMode;
+ source: string;
+ payload: AgentQueuedMessage;
+ },
+ onInserted?: (queueId: string) => void,
+ ): Promise {
+ const agent = await this.agentRepository.findByIdAndProjectId(input.agentId, input.projectId);
+ if (!agent) throw new UserError('Agent not found');
+ const { payload } = input;
+ const item = await this.txRunner.run({}, async (ctx) => {
+ await this.executionService.prepareThread(
+ {
+ ...input,
+ agentName: agent.name,
+ userMessage: payload.message,
+ access:
+ payload.kind === 'preview'
+ ? { accessScope: 'user', ownerId: payload.userId }
+ : { accessScope: 'project', ownerId: null },
+ },
+ ctx,
+ );
+ const inserted = await this.repository.enqueue(input.threadId, input.source, payload, ctx);
+ // Register delivery before another main can see the committed item.
+ onInserted?.(inserted.id);
+ return inserted;
+ });
+ this.onAvailable?.(item.threadId);
+ return item;
+ }
+
+ async claimNext(
+ threadId: string,
+ canConsume: (
+ item: AgentMessageQueue,
+ thread: AgentExecutionThread,
+ ctx: OperationContext,
+ ) => Promise,
+ ): Promise {
+ const claimed = await this.txRunner.run({}, async (ctx) => {
+ const thread = await this.threadRepository.lockById(threadId, ctx);
+ if (!thread || (await this.isBlocked(thread, ctx))) return null;
+ const active = await this.repository.findActive(threadId, ctx);
+ if (active?.executionId)
+ 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 reservation = await this.executionService.reserveExecution(recording, new Date(), ctx);
+ return { item, thread, recording, reservation };
+ });
+ if (!claimed) return null;
+ this.executionService.activateExecution(claimed.reservation, claimed.recording);
+ const { execution } = claimed.reservation;
+ if (!execution.startedAt) throw new OperationalError('Queued execution has no start time');
+ return {
+ item: claimed.item,
+ thread: claimed.thread,
+ recording: claimed.recording,
+ admission: { executionId: execution.id, startedAt: execution.startedAt },
+ };
+ }
+
+ async settle(threadId: string, executionId: string): Promise {
+ await this.txRunner.run({}, async (ctx) => {
+ const thread = await this.threadRepository.lockById(threadId, ctx);
+ if (!thread) return;
+ const active = await this.repository.findActive(threadId, ctx);
+ if (active?.executionId !== executionId || (await this.isBlocked(thread, ctx))) return;
+ await this.repository.removeActive(threadId, executionId, ctx);
+ });
+ this.onAvailable?.(threadId);
+ }
+
+ async recordFailure(
+ claim: ClaimedAgentMessage,
+ error: unknown,
+ signal?: AbortSignal,
+ ): Promise {
+ const { executionId, startedAt } = claim.admission;
+ const execution = await this.executionRepository.findExecution(executionId);
+ if (execution?.status !== 'running') return;
+ const recorder = new ExecutionRecorder(undefined, undefined, undefined, startedAt);
+ if (!signal?.aborted) recorder.record({ type: 'error', error });
+ recorder.record({ type: 'finish', finishReason: 'error' });
+ const record = recorder.getMessageRecord();
+ await this.executionService.finalizeExecution(executionId, {
+ ...claim.recording,
+ record: signal?.aborted ? { ...record, finishReason: 'cancelled', error: null } : record,
+ });
+ }
+
+ private async isBlocked(thread: AgentExecutionThread, ctx: OperationContext): Promise {
+ if (await this.executionRepository.existsRunningByThread(thread.id, ctx)) return true;
+ return (
+ (await this.checkpointStorage.findSuspendedForThread(thread.agentId, thread.id, ctx)) !== null
+ );
+ }
+
+ private recordingFor(
+ item: AgentMessageQueue,
+ thread: AgentExecutionThread,
+ ): StartExecutionParams {
+ return {
+ threadId: thread.id,
+ agentId: thread.agentId,
+ agentName: thread.agentName,
+ projectId: thread.projectId,
+ access: { accessScope: thread.accessScope, ownerId: thread.ownerId },
+ sessionMode: 'existing',
+ queueItemId: item.id,
+ userMessage: item.payload.message,
+ source: item.source,
+ attachments: item.payload.attachments,
+ author: item.payload.kind === 'integration' ? item.payload.author : undefined,
+ };
+ }
+}
diff --git a/packages/cli/src/modules/agents/agent-turn-already-running.error.ts b/packages/cli/src/modules/agents/agent-turn-already-running.error.ts
new file mode 100644
index 00000000000..565e3e35e05
--- /dev/null
+++ b/packages/cli/src/modules/agents/agent-turn-already-running.error.ts
@@ -0,0 +1,7 @@
+import { UserError } from 'n8n-workflow';
+
+export class AgentTurnAlreadyRunningError extends UserError {
+ constructor() {
+ super('A turn is already running in this conversation.');
+ }
+}
diff --git a/packages/cli/src/modules/agents/agent-turn-execution.service.ts b/packages/cli/src/modules/agents/agent-turn-execution.service.ts
index 014295f696a..55bced9c990 100644
--- a/packages/cli/src/modules/agents/agent-turn-execution.service.ts
+++ b/packages/cli/src/modules/agents/agent-turn-execution.service.ts
@@ -13,7 +13,10 @@ import { UnexpectedError } from 'n8n-workflow';
import type { AgentSessionMode } from './utils/agent-thread-access';
import { AgentExecutionRecordingError } from './agent-execution-recording.error';
+import { AgentTurnAlreadyRunningError } from './agent-turn-already-running.error';
import { AgentChatExecutionService } from './agent-chat-execution.service';
+import { AgentMessageQueueService } from './agent-message-queue.service';
+import { EXECUTION_METADATA_KEY, type AgentExecutionAdmission } from './types/agent-queued-message';
import {
AgentExecutionService,
type RecordMessageParams,
@@ -40,6 +43,8 @@ export type AgentTurnRequest = { recording: StartExecutionParams } & (
);
interface ExecuteTurnConfig {
+ admittedExecution?: AgentExecutionAdmission;
+ onAdmitted?: () => Promise;
agentInstance: RuntimeAgent;
toolRegistry: ToolRegistry;
mcpServerAttributions: Map;
@@ -103,6 +108,7 @@ export class AgentTurnExecutionService {
private readonly logger: Logger,
private readonly agentExecutionService: AgentExecutionService,
private readonly chatExecutionService: AgentChatExecutionService,
+ private readonly messageQueue: AgentMessageQueueService,
) {}
async getSessionMode(threadId: string): Promise {
@@ -112,12 +118,17 @@ export class AgentTurnExecutionService {
async *execute(config: ExecuteTurnConfig): AsyncGenerator {
let turn: AgentTurnRequest | undefined;
let previewControl: PreviewExecutionControl | undefined;
- const state: TurnExecutionState = { executionStarted: false, receivedFinish: false };
+ const state: TurnExecutionState = {
+ executionStarted: false,
+ receivedFinish: false,
+ executionId: config.admittedExecution?.executionId,
+ };
const recorder = this.createRecorder(
config.toolRegistry,
() => state.executionId,
config.context,
config.backgroundJobSignal,
+ config.admittedExecution?.startedAt,
);
try {
@@ -137,8 +148,12 @@ export class AgentTurnExecutionService {
} finally {
previewControl?.detachRequest();
if (turn && state.executionId) {
- await this.settleTurn(turn, config, recorder, state.executionId, state);
- await config.onSettled?.(recorder.suspended);
+ try {
+ await this.settleTurn(turn, config, recorder, state.executionId, state);
+ await config.onSettled?.(recorder.suspended);
+ } finally {
+ await this.messageQueue.settle(config.context.threadId, state.executionId);
+ }
}
}
}
@@ -263,6 +278,7 @@ export class AgentTurnExecutionService {
getExecutionId: () => string | undefined = () => undefined,
context?: RecordingContext,
backgroundJobSignal?: AgentBackgroundJobSignal,
+ startedAt?: Date,
): ExecutionRecorder {
return new ExecutionRecorder(
toolRegistry,
@@ -279,6 +295,7 @@ export class AgentTurnExecutionService {
}
},
backgroundJobSignal,
+ startedAt,
);
}
@@ -290,6 +307,7 @@ export class AgentTurnExecutionService {
try {
return await this.agentExecutionService.startExecutionRecording(params, startedAt);
} catch (cause) {
+ if (cause instanceof AgentTurnAlreadyRunningError) throw cause;
throw new AgentExecutionRecordingError({ phase: 'create', cause, executionError });
}
}
@@ -327,7 +345,15 @@ export class AgentTurnExecutionService {
recorder.record({ type: 'error', error: executionError });
recorder.record({ type: 'finish', finishReason: 'error' });
const recordStart = async () =>
- await this.startExecution(params, recorder.startedAt, executionError);
+ await this.startExecution(
+ {
+ ...params,
+ resumeRunId: params.resumeRunId ?? options.automaticContinuationRunId,
+ allowSuspendedPredecessor: Boolean(options.automaticContinuationRunId),
+ },
+ recorder.startedAt,
+ executionError,
+ );
const recordFailure = async (executionId: string) => {
await this.finalizeExecution({
executionId,
@@ -337,18 +363,11 @@ export class AgentTurnExecutionService {
params: { ...params, record: recorder.getMessageRecord() },
});
};
- if (options.previewChat && options.automaticContinuationRunId) {
- await this.chatExecutionService.admitAutomaticContinuation(
- params.threadId,
- params.agentId,
- options.automaticContinuationRunId,
- async () => await recordFailure(await recordStart()),
- );
- } else {
- const executionId = options.previewChat
- ? await this.chatExecutionService.admit(params.threadId, recordStart)
- : await recordStart();
+ const executionId = await recordStart();
+ try {
await recordFailure(executionId);
+ } finally {
+ await this.messageQueue.settle(params.threadId, executionId);
}
}
@@ -359,22 +378,11 @@ export class AgentTurnExecutionService {
state: TurnExecutionState,
previewControl?: PreviewExecutionControl,
): Promise> {
- const recordStart = async () => await this.recordTurnStart(turn, config, recorder, state);
- const startAccepted = async (id: string) =>
- await this.startAcceptedTurn(id, turn, config, recorder, state, previewControl);
- if (previewControl && config.automaticPreviewContinuation && turn.type === 'resume') {
- return await this.chatExecutionService.admitAutomaticContinuation(
- config.context.threadId,
- config.context.agentId,
- turn.options.runId,
- async () => await startAccepted(await recordStart()),
- );
- }
- const executionId = previewControl
- ? await this.chatExecutionService.admit(config.context.threadId, recordStart)
- : await recordStart();
+ const executionId =
+ config.admittedExecution?.executionId ??
+ (await this.recordTurnStart(turn, config, recorder, state));
state.executionId = executionId;
- return await startAccepted(executionId);
+ return await this.startAcceptedTurn(executionId, turn, config, recorder, state, previewControl);
}
private async recordTurnStart(
@@ -387,6 +395,8 @@ export class AgentTurnExecutionService {
const id = await this.startExecution(
{
...turn.recording,
+ resumeRunId: turn.type === 'resume' ? turn.options.runId : undefined,
+ allowSuspendedPredecessor: config.automaticPreviewContinuation,
...(config.backgroundJobSignal
? { initialTimeline: structuredClone(recorder.getMessageRecord().timeline) }
: {}),
@@ -406,6 +416,21 @@ export class AgentTurnExecutionService {
state: TurnExecutionState,
previewControl?: PreviewExecutionControl,
): Promise> {
+ const executionSignal = this.agentExecutionService.getAbortSignal(executionId);
+ turn.options.abortSignal = turn.options.abortSignal
+ ? AbortSignal.any([turn.options.abortSignal, executionSignal])
+ : executionSignal;
+ if (turn.type === 'start' && turn.options.persistence) {
+ turn.options.persistence.hostMetadata = {
+ ...turn.options.persistence.hostMetadata,
+ [EXECUTION_METADATA_KEY]: executionId,
+ };
+ } else if (turn.type === 'resume') {
+ turn.options.hostMetadata = {
+ ...turn.options.hostMetadata,
+ [EXECUTION_METADATA_KEY]: executionId,
+ };
+ }
if (previewControl) {
this.chatExecutionService.register(
{ ...config.context, userId: previewControl.userId, executionId },
@@ -415,6 +440,8 @@ export class AgentTurnExecutionService {
}
config.onExecutionStarted?.(executionId, config.context.threadId);
turn.options.abortSignal?.throwIfAborted();
+ await config.onAdmitted?.();
+ turn.options.abortSignal?.throwIfAborted();
return await this.startTurn(turn, config, recorder, state);
}
diff --git a/packages/cli/src/modules/agents/agents.module.ts b/packages/cli/src/modules/agents/agents.module.ts
index 7c0c413f5c7..7ceb5fc0436 100644
--- a/packages/cli/src/modules/agents/agents.module.ts
+++ b/packages/cli/src/modules/agents/agents.module.ts
@@ -231,6 +231,7 @@ export class AgentsModule implements ModuleInterface {
const { AgentMessageEntity } = await import('./entities/agent-message.entity.js');
const { AgentExecutionThread } = await import('./entities/agent-execution-thread.entity.js');
const { AgentExecution } = await import('./entities/agent-execution.entity.js');
+ const { AgentMessageQueue } = await import('./entities/agent-message-queue.entity.js');
const { AgentBackgroundJob } = await import('./entities/agent-background-job.entity.js');
const { AgentHistory } = await import('./entities/agent-history.entity.js');
const { AgentCredentialDependency } = await import(
@@ -272,6 +273,7 @@ export class AgentsModule implements ModuleInterface {
AgentMessageEntity,
AgentExecutionThread,
AgentExecution,
+ AgentMessageQueue,
AgentBackgroundJob,
AgentHistory,
AgentCredentialDependency,
diff --git a/packages/cli/src/modules/agents/entities/agent-message-queue.entity.ts b/packages/cli/src/modules/agents/entities/agent-message-queue.entity.ts
new file mode 100644
index 00000000000..cb5cc2d44a3
--- /dev/null
+++ b/packages/cli/src/modules/agents/entities/agent-message-queue.entity.ts
@@ -0,0 +1,54 @@
+import { dbType, idStringifier, JsonColumn, WithTimestamps } from '@n8n/db';
+import {
+ Column,
+ Entity,
+ Generated,
+ Index,
+ JoinColumn,
+ ManyToOne,
+ PrimaryColumn,
+ type Relation,
+} from '@n8n/typeorm';
+
+import { AgentExecutionThread } from './agent-execution-thread.entity';
+import { AgentExecution } from './agent-execution.entity';
+import type { AgentQueuedMessage } from '../types/agent-queued-message';
+
+@Entity({ name: 'agent_message_queue' })
+@Index(['threadId', 'id'])
+@Index(['threadId'], { unique: true, where: '"executionId" IS NOT NULL' })
+export class AgentMessageQueue extends WithTimestamps {
+ @Generated()
+ @PrimaryColumn({
+ type: dbType === 'sqlite' ? 'integer' : 'bigint',
+ transformer: idStringifier,
+ comment: 'Acceptance order; IDs are not reused',
+ })
+ id: string;
+
+ @ManyToOne(() => AgentExecutionThread, { onDelete: 'CASCADE' })
+ @JoinColumn({ name: 'threadId' })
+ thread: Relation;
+
+ @Column({ type: 'varchar', length: 128 })
+ threadId: string;
+
+ @Column({ type: 'varchar', length: 32, comment: 'Preview or integration source' })
+ source: string;
+
+ @JsonColumn({ comment: 'Input, attachment references, identity, and reply context' })
+ payload: AgentQueuedMessage;
+
+ @ManyToOne(() => AgentExecution, { nullable: true, onDelete: 'NO ACTION' })
+ @JoinColumn({ name: 'executionId' })
+ execution: Relation | null;
+
+ @Index()
+ @Column({
+ type: 'varchar',
+ length: 36,
+ nullable: true,
+ comment: 'Current execution; NULL means pending',
+ })
+ executionId: string | null;
+}
diff --git a/packages/cli/src/modules/agents/execution-recorder.ts b/packages/cli/src/modules/agents/execution-recorder.ts
index df96267a95e..59bd5e7b09e 100644
--- a/packages/cli/src/modules/agents/execution-recorder.ts
+++ b/packages/cli/src/modules/agents/execution-recorder.ts
@@ -329,7 +329,9 @@ export class ExecutionRecorder {
registry?: ToolRegistry,
private readonly onTimelineSnapshot?: (timeline: TimelineEvent[]) => void,
backgroundJobSignal?: AgentBackgroundJobSignal,
+ startedAt: Date = new Date(),
) {
+ this.startTime = startedAt.getTime();
this.registry = registry ?? new Map();
if (backgroundJobSignal) {
this.timeline.push({
@@ -377,7 +379,7 @@ export class ExecutionRecorder {
private error: string | null = null;
- private readonly startTime = Date.now();
+ private readonly startTime: number;
private childTraceChars = new Map();
diff --git a/packages/cli/src/modules/agents/integrations/__tests__/n8n-checkpoint-storage.test.ts b/packages/cli/src/modules/agents/integrations/__tests__/n8n-checkpoint-storage.test.ts
index 23aa71a4294..50a34cddc00 100644
--- a/packages/cli/src/modules/agents/integrations/__tests__/n8n-checkpoint-storage.test.ts
+++ b/packages/cli/src/modules/agents/integrations/__tests__/n8n-checkpoint-storage.test.ts
@@ -1,6 +1,10 @@
import type { SerializableAgentState } from '@n8n/agents';
import { mockLogger } from '@n8n/backend-test-utils';
import type { AgentsConfig } from '@n8n/config';
+import type { TransactionRunner } from '@n8n/db';
+import type { AgentExecutionRepository } from '../../repositories/agent-execution.repository';
+import type { AgentExecutionThreadRepository } from '../../repositories/agent-execution-thread.repository';
+import type { AgentMessageQueueRepository } from '../../repositories/agent-message-queue.repository';
import { mock } from 'vitest-mock-extended';
import {
@@ -31,10 +35,15 @@ const principalHash = hashAgentSandboxPrincipal({ type: 'n8n-user', userId: 'use
function makeService() {
const repository = mock();
+ repository.create.mockImplementation((data) => Object.assign(mock(), data));
const service = new N8NCheckpointStorage(
repository,
mockLogger(),
mock({ checkpointTtlSeconds: 60 }),
+ mock(),
+ mock(),
+ mock(),
+ mock(),
);
return { service, repository };
@@ -61,7 +70,7 @@ describe('N8NCheckpointStorage', () => {
expired: false,
state: JSON.stringify(suspendedState),
});
- expect(repository.save).toHaveBeenCalledWith(checkpoint);
+ expect(repository.saveCheckpoint).toHaveBeenCalledWith(checkpoint, {});
});
it.each(['thread-1', undefined])('replaces the saved thread key with %s', async (threadId) => {
@@ -81,13 +90,18 @@ describe('N8NCheckpointStorage', () => {
await service.getStorage('agent-1').save('run-1', state);
- expect(checkpoint).toMatchObject({
- agentId: 'agent-1',
- threadId: threadId ?? null,
- expired: false,
- state: JSON.stringify(state),
- });
- expect(repository.save).toHaveBeenCalledWith(checkpoint);
+ expect(repository.create).toHaveBeenCalledWith(
+ expect.objectContaining({
+ agentId: 'agent-1',
+ threadId: threadId ?? null,
+ expired: false,
+ state: JSON.stringify(state),
+ }),
+ );
+ expect(repository.saveCheckpoint).toHaveBeenCalledWith(
+ expect.objectContaining({ expired: false }),
+ {},
+ );
});
it.each(['agent-2', null])(
@@ -104,7 +118,7 @@ describe('N8NCheckpointStorage', () => {
await expect(service.getStorage('agent-1').save('run-1', suspendedState)).rejects.toThrow(
'owned by a different agent',
);
- expect(repository.save).not.toHaveBeenCalled();
+ expect(repository.saveCheckpoint).not.toHaveBeenCalled();
},
);
@@ -134,6 +148,7 @@ describe('N8NCheckpointStorage', () => {
'agent-1',
JSON.stringify(suspendedState),
JSON.stringify({ ...suspendedState, status: 'running' }),
+ expect.any(Date),
);
await expect(storage.claimForResume?.('run-1', suspendedState)).resolves.toBe(false);
});
@@ -164,7 +179,7 @@ describe('N8NCheckpointStorage', () => {
checkpoint: suspendedState,
});
- expect(repository.findByRunIdAndAgentId).toHaveBeenCalledWith('run-1', 'agent-1');
+ expect(repository.findByRunIdAndAgentId).toHaveBeenCalledWith('run-1', 'agent-1', {});
});
it('retains an expired checkpoint state while cancellation cleanup is pending', async () => {
@@ -320,7 +335,12 @@ describe('N8NCheckpointStorage', () => {
const result = await service.findSuspendedForThread('agent-1', 'thread-target');
expect(result).toEqual(parent);
- expect(repository.findActiveForThread).toHaveBeenCalledWith('agent-1', 'thread-target');
+ expect(repository.findActiveForThread).toHaveBeenCalledWith(
+ 'agent-1',
+ 'thread-target',
+ expect.any(Date),
+ {},
+ );
});
// A delegated child suspends under its parent's thread; the parent run
diff --git a/packages/cli/src/modules/agents/integrations/n8n-checkpoint-storage.ts b/packages/cli/src/modules/agents/integrations/n8n-checkpoint-storage.ts
index 681b7912025..f9d4aae784b 100644
--- a/packages/cli/src/modules/agents/integrations/n8n-checkpoint-storage.ts
+++ b/packages/cli/src/modules/agents/integrations/n8n-checkpoint-storage.ts
@@ -6,14 +6,19 @@ import {
import { Logger } from '@n8n/backend-common';
import { AgentsConfig } from '@n8n/config';
import { Time } from '@n8n/constants';
+import { TransactionRunner, type OperationContext } from '@n8n/db';
import { Service } from '@n8n/di';
-import { jsonParse, UnexpectedError, UserError } from 'n8n-workflow';
+import { jsonParse, OperationalError, UnexpectedError, UserError } from 'n8n-workflow';
import {
decodeAgentSandboxHostMetadata,
type AgentSandboxPrincipalHash,
} from '../agent-sandbox-principal';
import { AgentCheckpointRepository } from '../repositories/agent-checkpoint.repository';
+import { AgentExecutionRepository } from '../repositories/agent-execution.repository';
+import { AgentExecutionThreadRepository } from '../repositories/agent-execution-thread.repository';
+import { AgentMessageQueueRepository } from '../repositories/agent-message-queue.repository';
+import { checkpointExecutionId } from '../types/agent-queued-message';
/** File parts are checkpointed reference-only (a `Uint8Array` would not survive JSON round-tripping). */
function stripStateFileData(state: SerializableAgentState): SerializableAgentState {
@@ -47,6 +52,10 @@ export class N8NCheckpointStorage {
private readonly agentCheckpointRepository: AgentCheckpointRepository,
private readonly logger: Logger,
private readonly agentsConfig: AgentsConfig,
+ private readonly txRunner: TransactionRunner,
+ private readonly executionRepository: AgentExecutionRepository,
+ private readonly threadRepository: AgentExecutionThreadRepository,
+ private readonly queueRepository: AgentMessageQueueRepository,
) {
this.logger = this.logger.scoped('agents');
}
@@ -57,7 +66,12 @@ export class N8NCheckpointStorage {
load: async (key) => await this.load(key, agentId),
claimForResume: async (key: string, state: SerializableAgentState) =>
await this.claimForResume(key, state, agentId),
- delete: async (key) => await this.delete(key, agentId),
+ delete: async (key, state) => {
+ if (!state) return await this.delete(key, agentId);
+ await this.withExecutionOwnership(state, async (ctx) => {
+ await this.agentCheckpointRepository.expireByRunIdAndAgentId(key, agentId, ctx);
+ });
+ },
};
}
@@ -90,26 +104,51 @@ export class N8NCheckpointStorage {
async save(key: string, checkpointState: SerializableAgentState, agentId: string): Promise {
const state = stripStateFileData(checkpointState);
- const existing = await this.agentCheckpointRepository.findByRunId(key);
-
- if (existing) {
- if (existing.agentId !== agentId) {
+ await this.withExecutionOwnership(state, async (ctx) => {
+ const existing = await this.agentCheckpointRepository.findByRunId(key, ctx);
+ if (existing && existing.agentId !== agentId) {
throw new UnexpectedError('Agent checkpoint is owned by a different agent');
}
- existing.state = JSON.stringify(state);
- existing.threadId = state.persistence?.threadId ?? null;
- existing.expired = false;
- await this.agentCheckpointRepository.save(existing);
- } else {
const checkpoint = this.agentCheckpointRepository.create({
+ ...existing,
runId: key,
agentId,
threadId: state.persistence?.threadId ?? null,
state: JSON.stringify(state),
expired: false,
});
- await this.agentCheckpointRepository.save(checkpoint);
- }
+ await this.agentCheckpointRepository.saveCheckpoint(checkpoint, ctx);
+ });
+ }
+
+ private async withExecutionOwnership(
+ state: SerializableAgentState,
+ write: (ctx: OperationContext) => Promise,
+ ): Promise {
+ const executionId = checkpointExecutionId(state);
+ if (!executionId) return await write({});
+ await this.txRunner.run({}, async (ctx) => {
+ const threadId = state.persistence?.threadId;
+ if (!threadId || !(await this.threadRepository.lockById(threadId, ctx))) {
+ throw new OperationalError('Agent execution no longer owns this session');
+ }
+ const execution = await this.executionRepository.findExecution(executionId, ctx);
+ const active = await this.queueRepository.findActive(threadId, ctx);
+ if (
+ execution?.status !== 'running' ||
+ execution.threadId !== threadId ||
+ (active && active.executionId !== executionId)
+ ) {
+ throw new OperationalError('Agent execution no longer owns this session');
+ }
+ await write(ctx);
+ });
+ }
+
+ private expiryCutoff(): Date {
+ return new Date(
+ Date.now() - this.agentsConfig.checkpointTtlSeconds * Time.seconds.toMilliseconds,
+ );
}
async load(key: string, agentId: string): Promise {
@@ -117,7 +156,11 @@ export class N8NCheckpointStorage {
if (!checkpoint) return undefined;
- if (checkpoint.expired || checkpoint.state === null) {
+ if (
+ checkpoint.expired ||
+ checkpoint.state === null ||
+ checkpoint.updatedAt < this.expiryCutoff()
+ ) {
throw new UserError('This action has expired and cannot be resumed');
}
@@ -140,6 +183,7 @@ export class N8NCheckpointStorage {
agentId,
JSON.stringify(state),
JSON.stringify({ ...state, status: 'running' }),
+ this.expiryCutoff(),
);
}
@@ -149,11 +193,25 @@ export class N8NCheckpointStorage {
agentId: string,
): Promise {
if (state.status !== 'suspended') return false;
- return await this.agentCheckpointRepository.cancelSuspended(
- key,
- agentId,
- JSON.stringify(state),
- );
+ const executionId = checkpointExecutionId(state);
+ if (!executionId)
+ return await this.agentCheckpointRepository.cancelSuspended(
+ key,
+ agentId,
+ JSON.stringify(state),
+ );
+ return await this.txRunner.run({}, async (ctx) => {
+ const threadId = state.persistence?.threadId;
+ if (!threadId || !(await this.threadRepository.lockById(threadId, ctx))) return false;
+ const running = await this.executionRepository.findRunningByThread(threadId, ctx);
+ if (running.some(({ id }) => id !== executionId)) return false;
+ return await this.agentCheckpointRepository.cancelSuspended(
+ key,
+ agentId,
+ JSON.stringify(state),
+ ctx,
+ );
+ });
}
/**
@@ -165,8 +223,14 @@ export class N8NCheckpointStorage {
async findSuspendedForThread(
agentId: string,
threadId: string,
+ ctx: OperationContext = {},
): Promise {
- const rows = await this.agentCheckpointRepository.findActiveForThread(agentId, threadId);
+ const rows = await this.agentCheckpointRepository.findActiveForThread(
+ agentId,
+ threadId,
+ this.expiryCutoff(),
+ ctx,
+ );
for (const row of rows) {
const checkpoint = this.parseSuspendedState(row.state, threadId);
if (checkpoint) return checkpoint;
@@ -209,12 +273,21 @@ export class N8NCheckpointStorage {
return parsed;
}
- async getStatus(key: string, agentId: string): Promise {
- const checkpoint = await this.agentCheckpointRepository.findByRunIdAndAgentId(key, agentId);
+ async getStatus(
+ key: string,
+ agentId: string,
+ ctx: OperationContext = {},
+ ): Promise {
+ const checkpoint = await this.agentCheckpointRepository.findByRunIdAndAgentId(
+ key,
+ agentId,
+ ctx,
+ );
if (!checkpoint) return { status: 'not-found' };
if (checkpoint.state === null) return { status: 'expired' };
const state = jsonParse(checkpoint.state);
- if (checkpoint.expired) return { status: 'expired', checkpoint: state };
+ if (checkpoint.expired || checkpoint.updatedAt < this.expiryCutoff())
+ return { status: 'expired', checkpoint: state };
return { status: 'active', checkpoint: state };
}
diff --git a/packages/cli/src/modules/agents/repositories/agent-checkpoint.repository.ts b/packages/cli/src/modules/agents/repositories/agent-checkpoint.repository.ts
index 267287ff908..efa039676c7 100644
--- a/packages/cli/src/modules/agents/repositories/agent-checkpoint.repository.ts
+++ b/packages/cli/src/modules/agents/repositories/agent-checkpoint.repository.ts
@@ -1,29 +1,43 @@
+import { BaseRepository, TransactionRunner, type OperationContext } from '@n8n/db';
import { Service } from '@n8n/di';
-import { DataSource, Repository } from '@n8n/typeorm';
+import { DataSource, MoreThan } from '@n8n/typeorm';
import { AgentCheckpoint } from '../entities/agent-checkpoint.entity';
@Service()
-export class AgentCheckpointRepository extends Repository {
- constructor(dataSource: DataSource) {
- super(AgentCheckpoint, dataSource.manager);
+export class AgentCheckpointRepository extends BaseRepository {
+ constructor(dataSource: DataSource, transactionRunner: TransactionRunner) {
+ super(AgentCheckpoint, dataSource.manager, transactionRunner);
}
- async findByRunId(runId: string): Promise {
- return await this.findOneBy({ runId });
+ async findByRunId(runId: string, ctx: OperationContext = {}): Promise {
+ return await this.managerFor(ctx).findOneBy(AgentCheckpoint, { runId });
}
- async findByRunIdAndAgentId(runId: string, agentId: string): Promise {
- return await this.findOneBy({ runId, agentId });
+ async findByRunIdAndAgentId(
+ runId: string,
+ agentId: string,
+ ctx: OperationContext = {},
+ ): Promise {
+ return await this.managerFor(ctx).findOneBy(AgentCheckpoint, { runId, agentId });
}
- async findActiveForThread(agentId: string, threadId: string): Promise {
- return await this.find({
- where: { agentId, threadId, expired: false },
+ async findActiveForThread(
+ agentId: string,
+ threadId: string,
+ updatedAfter: Date,
+ ctx: OperationContext = {},
+ ): Promise {
+ return await this.managerFor(ctx).find(AgentCheckpoint, {
+ where: { agentId, threadId, expired: false, updatedAt: MoreThan(updatedAfter) },
order: { updatedAt: 'DESC' },
});
}
+ async saveCheckpoint(checkpoint: AgentCheckpoint, ctx: OperationContext): Promise {
+ await this.managerFor(ctx).save(AgentCheckpoint, checkpoint);
+ }
+
async findRetainedByThreadId(threadId: string): Promise {
return await this.find({ where: { threadId } });
}
@@ -41,17 +55,24 @@ export class AgentCheckpointRepository extends Repository {
agentId: string,
suspendedState: string,
runningState: string,
+ updatedAfter: Date,
): Promise {
const result = await this.update(
- { runId, agentId, expired: false, state: suspendedState },
+ { runId, agentId, expired: false, state: suspendedState, updatedAt: MoreThan(updatedAfter) },
{ state: runningState },
);
return (result.affected ?? 0) > 0;
}
- async cancelSuspended(runId: string, agentId: string, suspendedState: string): Promise {
- const result = await this.update(
+ async cancelSuspended(
+ runId: string,
+ agentId: string,
+ suspendedState: string,
+ ctx: OperationContext = {},
+ ): Promise {
+ const result = await this.managerFor(ctx).update(
+ AgentCheckpoint,
{ runId, agentId, expired: false, state: suspendedState },
{ expired: true },
);
@@ -59,8 +80,16 @@ export class AgentCheckpointRepository extends Repository {
return (result.affected ?? 0) > 0;
}
- async expireByRunIdAndAgentId(runId: string, agentId: string): Promise {
- await this.update({ runId, agentId }, { expired: true, state: null });
+ async expireByRunIdAndAgentId(
+ runId: string,
+ agentId: string,
+ ctx: OperationContext = {},
+ ): Promise {
+ await this.managerFor(ctx).update(
+ AgentCheckpoint,
+ { runId, agentId },
+ { expired: true, state: null },
+ );
}
async markExpired(olderThan: Date): Promise {
diff --git a/packages/cli/src/modules/agents/repositories/agent-execution-thread.repository.ts b/packages/cli/src/modules/agents/repositories/agent-execution-thread.repository.ts
index 379194e7387..95f0cf66cb3 100644
--- a/packages/cli/src/modules/agents/repositories/agent-execution-thread.repository.ts
+++ b/packages/cli/src/modules/agents/repositories/agent-execution-thread.repository.ts
@@ -42,6 +42,15 @@ export class AgentExecutionThreadRepository extends BaseRepository {
+ const manager = this.managerFor(ctx);
+ return await manager.findOne(AgentExecutionThread, {
+ where: { id: threadId },
+ lock:
+ manager.connection.options.type === 'postgres' ? { mode: 'pessimistic_write' } : undefined,
+ });
+ }
+
/**
* Find an existing thread or create a new one.
* Assign a display number on creation. Concurrent sessions can share a number.
diff --git a/packages/cli/src/modules/agents/repositories/agent-execution.repository.ts b/packages/cli/src/modules/agents/repositories/agent-execution.repository.ts
index 0cdaf40b5d8..2a410cb76b9 100644
--- a/packages/cli/src/modules/agents/repositories/agent-execution.repository.ts
+++ b/packages/cli/src/modules/agents/repositories/agent-execution.repository.ts
@@ -1,7 +1,7 @@
import type { AgentExecutionStatus } from '@n8n/api-types';
import { BaseRepository, TransactionRunner, type OperationContext } from '@n8n/db';
import { Service } from '@n8n/di';
-import { DataSource, IsNull, Not } from '@n8n/typeorm';
+import { DataSource, IsNull, LessThanOrEqual, Not } from '@n8n/typeorm';
import type { QueryDeepPartialEntity } from '@n8n/typeorm/query-builder/QueryPartialEntity';
import { AgentExecution } from '../entities/agent-execution.entity';
@@ -45,16 +45,31 @@ export class AgentExecutionRepository extends BaseRepository {
});
}
- async existsRunningByThread(threadId: string): Promise {
- return await this.existsBy({ threadId, status: 'running' });
+ async existsRunningByThread(threadId: string, ctx: OperationContext = {}): Promise {
+ return await this.managerFor(ctx).existsBy(AgentExecution, { threadId, status: 'running' });
+ }
+
+ async findRunningByThread(threadId: string, ctx: OperationContext): Promise {
+ return await this.managerFor(ctx).findBy(AgentExecution, { threadId, status: 'running' });
+ }
+
+ async findExecution(
+ executionId: string,
+ ctx: OperationContext = {},
+ ): Promise {
+ return await this.managerFor(ctx).findOneBy(AgentExecution, { id: executionId });
}
async findLatestByThreadId(threadId: string): Promise {
return await this.findOne({ where: { threadId }, order: { createdAt: 'DESC', id: 'DESC' } });
}
- async touchRunning(executionId: string): Promise {
- await this.update({ id: executionId, status: 'running' }, { updatedAt: new Date() });
+ async touchRunning(executionId: string): Promise {
+ const result = await this.update(
+ { id: executionId, status: 'running' },
+ { updatedAt: new Date() },
+ );
+ return result.affected === 1;
}
async updateTimelineIfRunning(
@@ -71,9 +86,16 @@ export class AgentExecutionRepository extends BaseRepository {
async updateIfRunning(
executionId: string,
values: AgentExecutionFinalizationValues,
+ staleBefore?: Date,
+ ctx: OperationContext = {},
): Promise {
- const result = await this.update(
- { id: executionId, status: 'running' },
+ const result = await this.managerFor(ctx).update(
+ AgentExecution,
+ {
+ id: executionId,
+ status: 'running',
+ ...(staleBefore ? { updatedAt: LessThanOrEqual(staleBefore) } : {}),
+ },
values as QueryDeepPartialEntity,
);
return result.affected === 1;
diff --git a/packages/cli/src/modules/agents/repositories/agent-message-queue.repository.ts b/packages/cli/src/modules/agents/repositories/agent-message-queue.repository.ts
new file mode 100644
index 00000000000..e3eeab07b87
--- /dev/null
+++ b/packages/cli/src/modules/agents/repositories/agent-message-queue.repository.ts
@@ -0,0 +1,62 @@
+import { BaseRepository, TransactionRunner, type OperationContext } from '@n8n/db';
+import { Service } from '@n8n/di';
+import { DataSource, IsNull, Not } from '@n8n/typeorm';
+
+import { AgentMessageQueue } from '../entities/agent-message-queue.entity';
+import type { AgentQueuedMessage } from '../types/agent-queued-message';
+
+@Service()
+export class AgentMessageQueueRepository extends BaseRepository {
+ constructor(dataSource: DataSource, transactionRunner: TransactionRunner) {
+ super(AgentMessageQueue, dataSource.manager, transactionRunner);
+ }
+
+ async enqueue(
+ threadId: string,
+ source: string,
+ payload: AgentQueuedMessage,
+ ctx: OperationContext,
+ ) {
+ const repository = this.managerFor(ctx).getRepository(AgentMessageQueue);
+ return await repository.save(
+ repository.create({ threadId, source, payload, executionId: null }),
+ );
+ }
+
+ async findHead(threadId: string, ctx: OperationContext) {
+ return await this.managerFor(ctx).findOne(AgentMessageQueue, {
+ where: { threadId },
+ order: { id: 'ASC' },
+ });
+ }
+
+ async findActive(threadId: string, ctx: OperationContext) {
+ return await this.managerFor(ctx).findOneBy(AgentMessageQueue, {
+ threadId,
+ executionId: Not(IsNull()),
+ });
+ }
+
+ async linkExecution(
+ id: string,
+ previousId: string | null,
+ executionId: string,
+ ctx: OperationContext,
+ ) {
+ const result = await this.managerFor(ctx).update(
+ AgentMessageQueue,
+ { id, executionId: previousId ?? IsNull() },
+ { executionId },
+ );
+ return result.affected === 1;
+ }
+
+ async removeActive(threadId: string, executionId: string, ctx: OperationContext) {
+ const result = await this.managerFor(ctx).delete(AgentMessageQueue, { threadId, executionId });
+ return result.affected === 1;
+ }
+
+ async findDeliveryState(id: string) {
+ return await this.findOne({ where: { id }, relations: { execution: true } });
+ }
+}
diff --git a/packages/cli/src/modules/agents/sub-agents/__tests__/sub-agent-runner.test.ts b/packages/cli/src/modules/agents/sub-agents/__tests__/sub-agent-runner.test.ts
index ead30bdec3c..20c1945ebd6 100644
--- a/packages/cli/src/modules/agents/sub-agents/__tests__/sub-agent-runner.test.ts
+++ b/packages/cli/src/modules/agents/sub-agents/__tests__/sub-agent-runner.test.ts
@@ -19,6 +19,7 @@ import type { Mocked } from 'vitest';
import { mock } from 'vitest-mock-extended';
import type { AgentExecutionService } from '../../agent-execution.service';
+import type { AgentMessageQueueService } from '../../agent-message-queue.service';
import type { AgentChatExecutionService } from '../../agent-chat-execution.service';
import { AgentTurnExecutionService } from '../../agent-turn-execution.service';
import { AgentRuntimeReconstructionService } from '../../agent-runtime-reconstruction.service';
@@ -132,6 +133,7 @@ describe('SubAgentRunner', () => {
reconstructionService = mock();
Container.set(AgentRuntimeReconstructionService, reconstructionService);
agentExecutionService = mock();
+ agentExecutionService.getAbortSignal.mockReturnValue(new AbortController().signal);
agentExecutionService.startExecutionRecording.mockResolvedValue('agent-execution-1');
agentExecutionService.finalizeExecution.mockResolvedValue('agent-execution-1');
checkpointStorage = mock();
@@ -142,6 +144,7 @@ describe('SubAgentRunner', () => {
logger,
agentExecutionService,
mock(),
+ mock(),
),
checkpointStorage,
logger,
diff --git a/packages/cli/src/modules/agents/types/agent-queued-message.ts b/packages/cli/src/modules/agents/types/agent-queued-message.ts
new file mode 100644
index 00000000000..181db02a581
--- /dev/null
+++ b/packages/cli/src/modules/agents/types/agent-queued-message.ts
@@ -0,0 +1,47 @@
+import type { SerializableAgentState } from '@n8n/agents';
+import type { AgentMessageAuthor } from '@n8n/api-types';
+import type { Author } from 'chat';
+
+import type {
+ IntegrationMessageContext,
+ SessionBinding,
+} from '../integrations/integration-tool-types';
+import type { StoredAttachmentRef } from './agent-chat-attachment';
+
+interface QueuedMessageInput {
+ message: string;
+ resourceId: string;
+ attachments?: StoredAttachmentRef[];
+}
+
+export interface QueuedPreviewMessage extends QueuedMessageInput {
+ kind: 'preview';
+ userId: string;
+}
+
+export interface QueuedIntegrationMessage extends QueuedMessageInput {
+ kind: 'integration';
+ credentialId: string;
+ platformThreadId: string;
+ modelMessage: string;
+ author: AgentMessageAuthor;
+ sender: Author;
+ messageContext: IntegrationMessageContext;
+ contextConversation: SessionBinding;
+}
+
+export type AgentQueuedMessage = QueuedPreviewMessage | QueuedIntegrationMessage;
+
+/** A committed execution reservation. Runtime preparation must reuse it. */
+export interface AgentExecutionAdmission {
+ executionId: string;
+ startedAt: Date;
+}
+
+export const EXECUTION_METADATA_KEY = 'n8nExecutionId';
+
+export function checkpointExecutionId(state: SerializableAgentState): string | undefined {
+ if (state.persistence?.delegated) return undefined;
+ const id = state.persistence?.hostMetadata?.[EXECUTION_METADATA_KEY];
+ return typeof id === 'string' ? id : undefined;
+}
diff --git a/packages/cli/test/integration/agents/agent-checkpoint-pruning.task.integration.test.ts b/packages/cli/test/integration/agents/agent-checkpoint-pruning.task.integration.test.ts
index e48b692b136..1055a20abdf 100644
--- a/packages/cli/test/integration/agents/agent-checkpoint-pruning.task.integration.test.ts
+++ b/packages/cli/test/integration/agents/agent-checkpoint-pruning.task.integration.test.ts
@@ -3,11 +3,10 @@ import type { Logger } from '@n8n/backend-common';
import { createTeamProject, testDb, testModules } from '@n8n/backend-test-utils';
import { AgentsConfig } from '@n8n/config';
import { Time } from '@n8n/constants';
-import { DbConnectionOptions } from '@n8n/db';
+import { TransactionRunner, DbConnectionOptions } from '@n8n/db';
import { Container } from '@n8n/di';
import type { QueryRunner } from '@n8n/typeorm';
import { DataSource, IsNull, Not } from '@n8n/typeorm';
-import { sleep } from '@n8n/utils/sleep';
import { v4 as uuid } from 'uuid';
import { mock } from 'vitest-mock-extended';
@@ -16,6 +15,9 @@ import type { Agent } from '@/modules/agents/entities/agent.entity';
import { N8NCheckpointStorage } from '@/modules/agents/integrations/n8n-checkpoint-storage';
import { AgentCheckpointRepository } from '@/modules/agents/repositories/agent-checkpoint.repository';
import { AgentRepository } from '@/modules/agents/repositories/agent.repository';
+import { AgentMessageQueueRepository } from '@/modules/agents/repositories/agent-message-queue.repository';
+import { AgentExecutionRepository } from '@/modules/agents/repositories/agent-execution.repository';
+import { AgentExecutionThreadRepository } from '@/modules/agents/repositories/agent-execution-thread.repository';
const suspendedState: SerializableAgentState = {
status: 'suspended',
@@ -24,9 +26,7 @@ const suspendedState: SerializableAgentState = {
pendingToolCalls: {},
};
const suspendedJson = JSON.stringify(suspendedState);
-const runningJson = JSON.stringify({ ...suspendedState, status: 'running' });
-const RESUMED = { expired: false, state: runningJson };
const PRUNED = { expired: true, state: null };
const isPostgres = process.env.DB_TYPE === 'postgresdb';
@@ -41,13 +41,25 @@ describe('AgentCheckpointPruningTask', () => {
let agentId: string;
let stale: Date;
+ function makeStorage(repository: AgentCheckpointRepository) {
+ return new N8NCheckpointStorage(
+ repository,
+ logger,
+ config,
+ Container.get(TransactionRunner),
+ Container.get(AgentExecutionRepository),
+ Container.get(AgentExecutionThreadRepository),
+ Container.get(AgentMessageQueueRepository),
+ );
+ }
+
beforeAll(async () => {
await testModules.loadModules(['agents']);
await testDb.init();
agentRepository = Container.get(AgentRepository);
checkpointRepository = Container.get(AgentCheckpointRepository);
config = Container.get(AgentsConfig);
- storage = new N8NCheckpointStorage(checkpointRepository, logger, config);
+ storage = makeStorage(checkpointRepository);
task = new AgentCheckpointPruningTask(storage);
stale = new Date(Date.now() - (config.checkpointTtlSeconds + Time.hours.toSeconds) * 1000);
});
@@ -142,10 +154,13 @@ describe('AgentCheckpointPruningTask', () => {
beforeEach(async () => {
otherRunner = otherConnection.createQueryRunner();
await otherRunner.startTransaction();
- const repository = new AgentCheckpointRepository({
- manager: otherRunner.manager,
- } as DataSource);
- otherStorage = new N8NCheckpointStorage(repository, logger, config);
+ const repository = new AgentCheckpointRepository(
+ {
+ manager: otherRunner.manager,
+ } as DataSource,
+ Container.get(TransactionRunner),
+ );
+ otherStorage = makeStorage(repository);
});
afterEach(async () => {
@@ -157,18 +172,14 @@ describe('AgentCheckpointPruningTask', () => {
await otherConnection.destroy();
});
- async function isPending(promise: Promise): Promise {
- return await Promise.race([promise.then(() => false), sleep(250).then(() => true)]);
- }
-
- it('should block a resume behind an uncommitted prune and then reject it', async () => {
+ it('rejects expired approval while pruning is uncommitted', async () => {
// Arrange: the other connection prunes the row but does not commit.
const [runId] = await insertCheckpoints('stale', 1, stale);
await new AgentCheckpointPruningTask(otherStorage).run();
- // Act: a resume on this connection waits on the row lock until the commit.
+ // Expiry rejects the claim before it needs the pruning lock.
const claim = storage.claimForResume(runId, suspendedState, agentId);
- expect(await isPending(claim)).toBe(true);
+ expect(await claim).toBe(false);
await otherRunner.commitTransaction();
// Assert: the resume sees the expired row and loses.
@@ -176,22 +187,15 @@ describe('AgentCheckpointPruningTask', () => {
expect(await checkpointRepository.findByRunId(runId)).toMatchObject(PRUNED);
});
- it('should block a prune behind an uncommitted resume and then skip the resumed row', async () => {
- // Arrange: the other connection resumes one of two stale rows but does not commit.
- const [resumed, untouched] = await insertCheckpoints('stale', 2, stale);
- expect(await otherStorage.claimForResume(resumed, suspendedState, agentId)).toBe(true);
-
- // Act: a prune on this connection waits on the row lock until the commit.
- const prune = task.run();
- expect(await isPending(prune)).toBe(true);
+ it('rejects expired approval before the pruning pass', async () => {
+ const [runId] = await insertCheckpoints('stale', 1, stale);
+ expect(await otherStorage.claimForResume(runId, suspendedState, agentId)).toBe(false);
await otherRunner.commitTransaction();
- await prune;
-
- // Assert: the prune re-reads the row after the lock, sees the fresh
- // `updatedAt`, and expires only the other row.
- expect(expiredTotal()).toBe(1);
- expect(await checkpointRepository.findByRunId(resumed)).toMatchObject(RESUMED);
- expect(await checkpointRepository.findByRunId(untouched)).toMatchObject(PRUNED);
+ expect(await checkpointRepository.findByRunId(runId)).toMatchObject({
+ expired: false,
+ state: suspendedJson,
+ });
+ await expect(storage.load(runId, agentId)).rejects.toThrow('expired');
});
});
});
diff --git a/packages/cli/test/integration/agents/agent-conversation-state.test.ts b/packages/cli/test/integration/agents/agent-conversation-state.test.ts
index e8e8f68b6a2..d8e01bd551d 100644
--- a/packages/cli/test/integration/agents/agent-conversation-state.test.ts
+++ b/packages/cli/test/integration/agents/agent-conversation-state.test.ts
@@ -3,6 +3,8 @@ import { Logger } from '@n8n/backend-common';
import { createTeamProject, testDb, testModules } from '@n8n/backend-test-utils';
import { AgentsConfig } from '@n8n/config';
import { Container } from '@n8n/di';
+import { TransactionRunner } from '@n8n/db';
+import { AgentMessageQueueRepository } from '@/modules/agents/repositories/agent-message-queue.repository';
import { DataSource } from '@n8n/typeorm';
import { randomUUID } from 'node:crypto';
@@ -48,9 +50,13 @@ describe('Agent conversation state', () => {
dropSchema: false,
}).initialize();
peerStorage = new N8NCheckpointStorage(
- new AgentCheckpointRepository(peer),
+ new AgentCheckpointRepository(peer, Container.get(TransactionRunner)),
Container.get(Logger),
Container.get(AgentsConfig),
+ Container.get(TransactionRunner),
+ Container.get(AgentExecutionRepository),
+ Container.get(AgentExecutionThreadRepository),
+ Container.get(AgentMessageQueueRepository),
);
});
@@ -104,13 +110,15 @@ describe('Agent conversation state', () => {
agentId: agent.id,
threadId: 'thread-1',
expired: false,
- updatedAt: new Date(Date.UTC(2026, 0, index + 1)),
+ updatedAt: new Date(Date.now() - (rows.length - index) * 1000),
...row,
})),
);
expect(
- (await checkpoints.findActiveForThread(agent.id, 'thread-1')).map(({ runId }) => runId),
+ (await checkpoints.findActiveForThread(agent.id, 'thread-1', new Date(0))).map(
+ ({ runId }) => runId,
+ ),
).toEqual(['cancelled', 'running', 'child', 'parent', 'older']);
expect(await storage.findSuspendedForThread(agent.id, 'thread-1')).toEqual(parent);
});
diff --git a/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts b/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts
index 803cf15ea07..802d439b32c 100644
--- a/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts
+++ b/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts
@@ -42,6 +42,13 @@ import { AgentExecutionService } from '@/modules/agents/agent-execution.service'
import type { AgentExecutionUpdateBroadcaster } from '@/modules/agents/agent-execution-update-broadcaster';
import { AgentInterruptedExecutionSweeper } from '@/modules/agents/agent-interrupted-execution-sweeper';
import { AgentTurnExecutionService } from '@/modules/agents/agent-turn-execution.service';
+import {
+ AgentMessageQueueService,
+ type ClaimedAgentMessage,
+} from '@/modules/agents/agent-message-queue.service';
+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 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';
@@ -108,6 +115,7 @@ describe('AgentExecutionRepository', () => {
});
afterEach(async () => {
+ await Container.get(AgentMessageQueueRepository).delete({});
await repository.delete({});
await threadRepo.delete({});
await agentRepo.delete({});
@@ -130,6 +138,19 @@ describe('AgentExecutionRepository', () => {
const threads = connection
? new AgentExecutionThreadRepository(connection, txRunner)
: threadRepo;
+ const queueRepository = new AgentMessageQueueRepository(
+ connection ?? repository.manager.connection,
+ txRunner,
+ );
+ const checkpointStorage = new N8NCheckpointStorage(
+ new AgentCheckpointRepository(connection ?? repository.manager.connection, txRunner),
+ mockLogger(),
+ new AgentsConfig(),
+ txRunner,
+ executions,
+ threads,
+ queueRepository,
+ );
const memory = mock();
memory.getImplementation.mockReturnValue(memoryBackend);
const attachmentService = mock();
@@ -145,12 +166,25 @@ describe('AgentExecutionRepository', () => {
mock({ modeTag: 'db' }),
mock(),
mock(),
- Container.get(N8NCheckpointStorage),
+ checkpointStorage,
txRunner,
+ queueRepository,
+ );
+ const queue = new AgentMessageQueueService(
+ txRunner,
+ queueRepository,
+ threads,
+ executions,
+ executionService,
+ checkpointStorage,
+ connection ? new AgentRepository(connection) : agentRepo,
);
return {
txRunner,
threads,
+ queue,
+ queueRepository,
+ checkpointStorage,
executionService,
attachmentService,
executionLogStore,
@@ -158,6 +192,7 @@ describe('AgentExecutionRepository', () => {
mockLogger(),
executionService,
mock(),
+ queue,
),
};
}
@@ -326,7 +361,7 @@ describe('AgentExecutionRepository', () => {
}
async function startSuspendedApprovalRun(user?: User, approvals = 1) {
- const { turns, executionService } = recordingServices();
+ const { turns, executionService, checkpointStorage: storage } = recordingServices();
const threadId = uuid();
const recording = {
access: user
@@ -339,7 +374,6 @@ describe('AgentExecutionRepository', () => {
userMessage: 'Start',
};
const checkpointRepo = Container.get(AgentCheckpointRepository);
- const storage = new N8NCheckpointStorage(checkpointRepo, mockLogger(), new AgentsConfig());
const { action, makeAgent } = createApprovalAgentFactory(threadId, user?.id, approvals);
const common = {
toolRegistry: new Map(),
@@ -400,36 +434,34 @@ describe('AgentExecutionRepository', () => {
fixture: Awaited>,
) {
const { common, makeAgent, recording, storage, suspension, turns } = fixture;
- const secondConnection = await new DataSource({
- ...repository.manager.connection.options,
- name: uuid(),
- }).initialize();
+ const { checkpointStorage: otherStorage, turns: otherTurns } = recordingServices(
+ undefined,
+ peer,
+ );
+ const admitted = createDeferredPromise();
+ const releaseAdmission = createDeferredPromise();
+ const save = repository.saveInContext.bind(repository);
+ const saveSpy = vi
+ .spyOn(repository, 'saveInContext')
+ .mockImplementationOnce(async (execution, ctx) => {
+ const saved = await save(execution, ctx);
+ admitted.resolve(ctx);
+ await releaseAdmission.promise;
+ return saved;
+ });
+ const competing = observePeerTransaction();
try {
- const otherStorage = new N8NCheckpointStorage(
- new AgentCheckpointRepository(secondConnection),
- mockLogger(),
- new AgentsConfig(),
- );
- const bothLoaded = createDeferredPromise();
- let loaded = 0;
const onResumeClaimed = vi.fn();
- const attempts = [storage, otherStorage].map(async (checkpointStorage) => {
+ const attempts = [
+ { storage, turns },
+ { storage: otherStorage, turns: otherTurns },
+ ].map(async ({ storage: checkpointStorage, turns: executingTurns }, index) => {
+ if (index === 1) await admitted.promise;
let agent: ReturnType | undefined;
try {
- const store = checkpointStorage.getStorage(agentId);
- agent = makeAgent({
- ...store,
- load: async (key) => {
- const state = await store.load(key);
- if (++loaded === 2) bothLoaded.resolve(true);
- if (!(await bothLoaded.promise)) {
- throw new Error('A resume attempt failed before both checkpoints loaded');
- }
- return state;
- },
- });
+ agent = makeAgent(checkpointStorage.getStorage(agentId));
return await collect(
- turns.execute({
+ executingTurns.execute({
...common,
agentInstance: agent,
prepare: async () => ({
@@ -444,17 +476,21 @@ describe('AgentExecutionRepository', () => {
}),
}),
);
- } catch (error) {
- bothLoaded.resolve(false);
- throw error;
} finally {
await agent?.close();
}
});
- return { outcomes: await Promise.allSettled(attempts), onResumeClaimed };
+ const outcomes = Promise.allSettled(attempts);
+ const ctx = await admitted.promise;
+ await competing.started;
+ await waitForPeerLock(ctx);
+ releaseAdmission.resolve();
+ return { outcomes: await outcomes, onResumeClaimed };
} finally {
- await secondConnection.destroy();
+ releaseAdmission.resolve();
+ saveSpy.mockRestore();
+ competing.restore();
}
}
@@ -810,7 +846,7 @@ describe('AgentExecutionRepository', () => {
}
});
- it('records one accepted resume and one failed attempt when separate connections claim the same checkpoint', async () => {
+ it('admits one resume when separate connections claim the same checkpoint', async () => {
const fixture = await startSuspendedApprovalRun();
const { outcomes, onResumeClaimed } = await resumeApprovalFromSeparateConnections(fixture);
@@ -818,16 +854,12 @@ describe('AgentExecutionRepository', () => {
expect(onResumeClaimed).toHaveBeenCalledOnce();
expect(fixture.action).toHaveBeenCalledOnce();
const executions = await repository.findByThreadIdOrdered(fixture.threadId);
- expect(executions).toHaveLength(3);
- expect(executions.map(({ status }) => status).sort()).toEqual(['error', 'success', 'success']);
+ expect(executions).toHaveLength(2);
+ expect(executions.map(({ status }) => status)).toEqual(['success', 'success']);
const resumed = executions.find(({ hitlStatus }) => hitlStatus === 'resumed');
expect(resumed).toMatchObject({ userMessage: null, status: 'success' });
expect(resumed?.timeline).toContainEqual(expect.objectContaining({ type: 'hitl-response' }));
- const rejected = executions.find(({ status }) => status === 'error');
- expect(rejected).toMatchObject({ userMessage: null, hitlStatus: null });
- expect(rejected?.timeline ?? []).not.toContainEqual(
- expect.objectContaining({ type: 'hitl-response' }),
- );
+
expect(await fixture.checkpointRepo.findByRunId(fixture.suspension.runId)).toMatchObject({
expired: true,
state: null,
@@ -981,6 +1013,329 @@ describe('AgentExecutionRepository', () => {
expect(await repository.findOneByOrFail({ id: executionId })).toEqual(recovered);
});
+ describe('durable message queue', () => {
+ let owner: User;
+ beforeEach(async () => {
+ owner = await createMember();
+ });
+
+ function input(
+ threadId: string,
+ message: string,
+ sessionMode: 'new' | 'existing' = 'existing',
+ ): Parameters[0] {
+ return {
+ agentId,
+ projectId,
+ threadId,
+ sessionMode,
+ source: 'chat',
+ payload: {
+ kind: 'preview',
+ message,
+ userId: owner.id,
+ resourceId: `draft-chat:${owner.id}`,
+ },
+ };
+ }
+
+ async function claim(services: ReturnType, threadId: string) {
+ const item = await services.queue.claimNext(threadId, async () => true);
+ if (!item) throw new Error('Expected a queue claim');
+ return item;
+ }
+
+ async function finish(
+ services: ReturnType,
+ item: ClaimedAgentMessage,
+ ) {
+ const recorder = new ExecutionRecorder();
+ recorder.record({ type: 'finish', finishReason: 'stop' });
+ await services.executionService.finalizeExecution(item.admission.executionId, {
+ ...item.recording,
+ record: recorder.getMessageRecord(),
+ });
+ await services.queue.settle(item.thread.id, item.admission.executionId);
+ }
+
+ it('serializes acceptance and claims across connections while other sessions progress', async () => {
+ const local = recordingServices();
+ const remote = recordingServices(undefined, peer);
+ const threadId = uuid();
+ const first = await local.queue.enqueue(input(threadId, 'first', 'new'));
+ const inserted = createDeferredPromise();
+ const release = createDeferredPromise();
+ const insert = local.queueRepository.enqueue.bind(local.queueRepository);
+ const spy = vi
+ .spyOn(local.queueRepository, 'enqueue')
+ .mockImplementationOnce(async (...args) => {
+ const item = await insert(...args);
+ inserted.resolve(args[3]);
+ await release.promise;
+ return item;
+ });
+ const competing = observePeerTransaction();
+ try {
+ const second = local.queue.enqueue(input(threadId, 'second'));
+ const ctx = await inserted.promise;
+ const third = remote.queue.enqueue(input(threadId, 'third'));
+ await competing.started;
+ await waitForPeerLock(ctx);
+ release.resolve();
+ const accepted = [first, ...(await Promise.all([second, third]))];
+ expect(accepted.map(({ id }) => BigInt(id))).toEqual(
+ accepted.map(({ id }) => BigInt(id)).sort((a, b) => (a < b ? -1 : 1)),
+ );
+ } finally {
+ release.resolve();
+ spy.mockRestore();
+ competing.restore();
+ }
+ const claims = await Promise.all(
+ [local, remote].map(async (services) => ({
+ services,
+ item: await services.queue.claimNext(threadId, async () => true),
+ })),
+ );
+ const winner = claims.find(({ item }) => item !== null);
+ if (!winner?.item) throw new Error('Expected one consumer');
+ 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);
+ await expect(
+ local.executionService.startExecutionRecording(winner.item.recording, new Date()),
+ ).rejects.toBeInstanceOf(AgentTurnAlreadyRunningError);
+ const independentId = uuid();
+ await remote.queue.enqueue(input(independentId, 'independent', 'new'));
+ const independent = await claim(remote, independentId);
+ await finish(remote, independent);
+ 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);
+ await finish(local, next);
+ }
+ expect(await local.queueRepository.count()).toBe(0);
+ const later = await local.queue.enqueue(input(threadId, 'later'));
+ expect(BigInt(later.id)).toBeGreaterThan(BigInt(independent.item.id));
+ });
+
+ it('rolls back acceptance and claim without partial session or execution records', async () => {
+ const services = recordingServices();
+ const threadId = uuid();
+ const insert = services.queueRepository.enqueue.bind(services.queueRepository);
+ const failedInsert = vi
+ .spyOn(services.queueRepository, 'enqueue')
+ .mockImplementationOnce(async (...args) => {
+ await insert(...args);
+ throw new Error('acceptance failed');
+ });
+ await expect(services.queue.enqueue(input(threadId, 'first', 'new'))).rejects.toThrow(
+ 'acceptance failed',
+ );
+ failedInsert.mockRestore();
+ expect(await threadRepo.findOneBy({ id: threadId })).toBeNull();
+ expect(await services.queueRepository.count()).toBe(0);
+ const accepted = await services.queue.enqueue(input(threadId, 'first', 'new'));
+ const failedLink = vi
+ .spyOn(services.queueRepository, 'linkExecution')
+ .mockResolvedValueOnce(false);
+ await expect(claim(services, threadId)).rejects.toBeInstanceOf(AgentTurnAlreadyRunningError);
+ failedLink.mockRestore();
+ expect(await repository.countBy({ threadId })).toBe(0);
+ expect(await services.queueRepository.findDeliveryState(accepted.id)).toMatchObject({
+ executionId: null,
+ });
+ await finish(services, await claim(services, threadId));
+ });
+
+ it('enforces active-slot uniqueness and prevents execution deletion from requeuing a message', async () => {
+ const services = recordingServices();
+ const threadId = uuid();
+ await services.queue.enqueue(input(threadId, 'first', 'new'));
+ const pending = await services.queue.enqueue(input(threadId, 'second'));
+ const active = await claim(services, threadId);
+ await expect(
+ services.queueRepository.update(pending.id, { executionId: active.admission.executionId }),
+ ).rejects.toThrow();
+ await expect(repository.delete(active.admission.executionId)).rejects.toThrow();
+ expect(await services.queueRepository.findDeliveryState(active.item.id)).toMatchObject({
+ executionId: active.admission.executionId,
+ });
+ await finish(services, active);
+ });
+
+ it('recovers pending and abandoned work without replaying the interrupted input', async () => {
+ const local = recordingServices();
+ const remote = recordingServices(undefined, peer);
+ const threadId = uuid();
+ await local.queue.enqueue(input(threadId, 'interrupted', 'new'));
+ await local.queue.enqueue(input(threadId, 'next'));
+ const active = await claim(remote, threadId);
+ const cutoff = new Date(Date.now() - 120_000);
+ await repository.update(active.admission.executionId, {
+ updatedAt: new Date(cutoff.getTime() - 1),
+ });
+ const stale = await repository.findOneByOrFail({ id: active.admission.executionId });
+ await repository.touchRunning(stale.id);
+ expect(await local.executionService.finalizeInterruptedExecution(stale, cutoff)).toBe(false);
+ 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');
+ await expect(finish(remote, active)).rejects.toThrow('no longer running');
+ await finish(local, next);
+ expect(
+ (await repository.findByThreadIdOrdered(threadId)).map(({ userMessage, status }) => ({
+ userMessage,
+ status,
+ })),
+ ).toEqual([
+ { userMessage: 'interrupted', status: 'interrupted' },
+ { userMessage: 'next', status: 'success' },
+ ]);
+ });
+
+ it.each(['settled', 'running', 'interrupted'] as const)(
+ 'keeps approval ownership when its predecessor is %s and ignores late callbacks',
+ async (predecessor) => {
+ const local = recordingServices();
+ const remote = recordingServices(undefined, peer);
+ const threadId = uuid();
+ await local.queue.enqueue(input(threadId, 'approval', 'new'));
+ const pending = await local.queue.enqueue(input(threadId, 'later'));
+ const active = await claim(local, threadId);
+ const runId = uuid();
+ const state: SerializableAgentState = {
+ status: 'suspended',
+ messageList: { messages: [], historyIds: [], inputIds: [], responseIds: [] },
+ pendingToolCalls: {},
+ persistence: {
+ threadId,
+ resourceId: `draft-chat:${owner.id}`,
+ hostMetadata: { [EXECUTION_METADATA_KEY]: active.admission.executionId },
+ },
+ };
+ await local.checkpointStorage.save(runId, state, agentId);
+ if (predecessor === 'settled') await finish(local, active);
+ if (predecessor === 'interrupted') {
+ const cutoff = new Date(Date.now() - 120_000);
+ await repository.update(active.admission.executionId, {
+ updatedAt: new Date(cutoff.getTime() - 1),
+ });
+ const stale = await repository.findOneByOrFail({ id: active.admission.executionId });
+ expect(await remote.executionService.finalizeInterruptedExecution(stale, cutoff)).toBe(
+ true,
+ );
+ }
+ expect(await remote.queue.claimNext(threadId, async () => true)).toBeNull();
+ expect(await remote.queueRepository.findDeliveryState(active.item.id)).not.toBeNull();
+ const resumedId = await remote.executionService.startExecutionRecording(
+ { ...active.recording, userMessage: null, resumeRunId: runId },
+ new Date(),
+ );
+ await local.queue.settle(threadId, active.admission.executionId);
+ expect(await local.queueRepository.findDeliveryState(active.item.id)).toMatchObject({
+ executionId: resumedId,
+ });
+ await expect(local.checkpointStorage.save(runId, state, agentId)).rejects.toThrow(
+ 'no longer owns',
+ );
+ await expect(
+ local.checkpointStorage.getStorage(agentId).delete(runId, state),
+ ).rejects.toThrow('no longer owns');
+ expect(await local.checkpointStorage.cancelSuspended(runId, state, agentId)).toBe(false);
+ if (predecessor === 'running') await finish(local, active);
+ const resumedState = {
+ ...state,
+ persistence: {
+ ...state.persistence!,
+ hostMetadata: { [EXECUTION_METADATA_KEY]: resumedId },
+ },
+ };
+ await remote.checkpointStorage.getStorage(agentId).delete(runId, resumedState);
+ await finish(remote, {
+ ...active,
+ admission: { executionId: resumedId, startedAt: new Date() },
+ });
+ const next = await claim(local, threadId);
+ expect(next.item.id).toBe(pending.id);
+ await local.queue.settle(threadId, active.admission.executionId);
+ expect(await local.queueRepository.findDeliveryState(next.item.id)).toMatchObject({
+ executionId: next.admission.executionId,
+ });
+ await finish(local, next);
+ },
+ );
+
+ it('advances after approval expiry without waiting for pruning', async () => {
+ const services = recordingServices();
+ const threadId = uuid();
+ await services.queue.enqueue(input(threadId, 'approval', 'new'));
+ const pending = await services.queue.enqueue(input(threadId, 'later'));
+ const active = await claim(services, threadId);
+ const runId = uuid();
+ await services.checkpointStorage.save(
+ runId,
+ {
+ status: 'suspended',
+ persistence: { threadId, resourceId: owner.id },
+ messageList: { messages: [], historyIds: [], inputIds: [], responseIds: [] },
+ pendingToolCalls: {},
+ },
+ agentId,
+ );
+ await finish(services, active);
+ expect(await services.queue.claimNext(threadId, async () => true)).toBeNull();
+ await Container.get(AgentCheckpointRepository).update(runId, {
+ updatedAt: new Date(Date.now() - (new AgentsConfig().checkpointTtlSeconds + 1) * 1000),
+ });
+ const next = await claim(services, threadId);
+ expect(next.item.id).toBe(pending.id);
+ await expect(services.checkpointStorage.load(runId, agentId)).rejects.toThrow('expired');
+ expect(await Container.get(AgentCheckpointRepository).findByRunId(runId)).toMatchObject({
+ expired: false,
+ });
+ await finish(services, next);
+ });
+
+ it('removes pending input and attachments on session deletion and never recreates it', async () => {
+ const services = recordingServices();
+ const threadId = uuid();
+ const attachment = await attachmentRepo.save(buildAttachment(threadId));
+ const accepted = input(threadId, 'pending', 'new');
+ accepted.payload.attachments = [
+ {
+ id: attachment.id,
+ fileName: attachment.fileName,
+ mimeType: attachment.mimeType,
+ sizeBytes: attachment.fileSizeBytes,
+ },
+ ];
+ await services.queue.enqueue(accepted);
+ expect(
+ await services.executionService.deleteThread(projectId, agentId, threadId, owner.id),
+ ).toBe(true);
+ expect(await services.queueRepository.countBy({ threadId })).toBe(0);
+ expect(await attachmentRepo.findOneBy({ id: attachment.id })).toBeNull();
+ expect(services.attachmentService.deleteStoredData).toHaveBeenCalledWith(
+ [attachment.binaryDataId],
+ { threadId },
+ );
+ expect(await services.queue.claimNext(threadId, async () => true)).toBeNull();
+ await expect(services.queue.enqueue(input(threadId, 'late'))).rejects.toThrow();
+ expect(await threadRepo.findOneBy({ id: threadId })).toBeNull();
+ });
+
+ it('cascades agent deletion through pending queue rows', async () => {
+ const services = recordingServices();
+ const threadId = uuid();
+ await services.queue.enqueue(input(threadId, 'pending', 'new'));
+ await agentRepo.delete(agentId);
+ expect(await services.queueRepository.countBy({ threadId })).toBe(0);
+ });
+ });
+
const createThread = async (overrides: Partial = {}) => {
const thread = threadRepo.create({
accessScope: 'project',
diff --git a/packages/cli/test/migration/1790241069212-create-agent-message-queue.test.ts b/packages/cli/test/migration/1790241069212-create-agent-message-queue.test.ts
new file mode 100644
index 00000000000..72878d13719
--- /dev/null
+++ b/packages/cli/test/migration/1790241069212-create-agent-message-queue.test.ts
@@ -0,0 +1,48 @@
+import {
+ createTestMigrationContext,
+ initDbUpToMigration,
+ runSingleMigration,
+} from '@n8n/backend-test-utils';
+import { DbConnection } from '@n8n/db';
+import { Container } from '@n8n/di';
+import { DataSource } from '@n8n/typeorm';
+
+const MIGRATION_NAME = 'CreateAgentMessageQueue1790241069212';
+
+describe('CreateAgentMessageQueue migration', () => {
+ let dataSource: DataSource;
+
+ beforeAll(async () => {
+ await Container.get(DbConnection).init();
+ dataSource = Container.get(DataSource);
+ const context = createTestMigrationContext(dataSource);
+ await context.queryRunner.clearDatabase();
+ await context.queryRunner.release();
+ await initDbUpToMigration(MIGRATION_NAME);
+ });
+
+ afterAll(async () => {
+ await Container.get(DbConnection).close();
+ });
+
+ async function hasTable(name: string): Promise {
+ const context = createTestMigrationContext(dataSource);
+ try {
+ return await context.queryRunner.hasTable(`${context.tablePrefix}${name}`);
+ } finally {
+ await context.queryRunner.release();
+ }
+ }
+
+ it('adds, removes, and restores the queue without removing execution history tables', async () => {
+ await runSingleMigration(MIGRATION_NAME);
+ expect(await hasTable('agent_message_queue')).toBe(true);
+ await dataSource.undoLastMigration({ transaction: 'each' });
+ expect(await hasTable('agent_message_queue')).toBe(false);
+ for (const table of ['agent_execution_threads', 'agent_execution', 'agent_checkpoints']) {
+ expect(await hasTable(table)).toBe(true);
+ }
+ await runSingleMigration(MIGRATION_NAME);
+ expect(await hasTable('agent_message_queue')).toBe(true);
+ });
+});