github-mirror/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2026-10-17 05:29:44 +02:00
Code Issues Actions 14 Packages Projects Releases Wiki Activity Graph

feat(core): Add the durable agent message queue foundation (#39435)

Browse Source
This commit is contained in:
Robin Braumann 2026-09-24 12:27:30 +00:00 committed by GitHub
parent 482ffb5306
commit 6c592bdb33
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
42 changed files with 1776 additions and 385 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
docs/generated/postgres-schema/README.md generated
View File

@ -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

12
docs/generated/postgres-schema/public.agent_execution.md generated
View File

@ -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

12
docs/generated/postgres-schema/public.agent_execution_threads.md generated
View File

@ -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

104
docs/generated/postgres-schema/public.agent_message_queue.md generated Normal file
View File

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

12
docs/generated/sqlite-schema/README.md generated
View File

@ -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

12
docs/generated/sqlite-schema/agent_execution.md generated
View File

@ -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

12
docs/generated/sqlite-schema/agent_execution_threads.md generated
View File

@ -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

108
docs/generated/sqlite-schema/agent_message_queue.md generated Normal file
View File

@ -0,0 +1,108 @@
# agent_message_queue
## Description
<details>
<summary><strong>Table Definition</strong></summary>
```sql
CREATE TABLE "agent_message_queue" ("id" integer PRIMARY KEY AUTOINCREMENT NOT NULL, "threadId" varchar(128) NOT NULL, "source" varchar(32) NOT NULL, "payload" text NOT NULL, "executionId" varchar(36), "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), CONSTRAINT "FK_a74f9154a59430986112d70cb16" FOREIGN KEY ("threadId") REFERENCES "agent_execution_threads" ("id") ON DELETE CASCADE, CONSTRAINT "FK_2349d84b4f2a660fc264f38fef3" FOREIGN KEY ("executionId") REFERENCES "agent_execution" ("id") ON DELETE NO ACTION)
```
</details>
## 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)

14
packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts
View File

@ -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);
});

5
packages/@n8n/agents/src/runtime/loop/agent-runtime.ts
View File

@ -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<void> {
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;

8
packages/@n8n/agents/src/runtime/state/run-state.ts
View File

@ -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<void> {
await this.store.delete(runId);
async complete(runId: string, state?: SerializableAgentState): Promise<void> {
await this.store.delete(runId, state);
}
/** Delete a cancelled run and surface failures so its parent can remain retryable. */
async cancel(runId: string): Promise<void> {
await this.store.delete(runId);
async cancel(runId: string, state?: SerializableAgentState): Promise<void> {
await this.store.delete(runId, state);
}
}

4
packages/@n8n/agents/src/types/sdk/memory.ts
View File

@ -381,6 +381,6 @@ export interface CheckpointStore {
* false when another process has already claimed or changed the snapshot.
*/
claimForResume?(key: string, state: SerializableAgentState): Promise<boolean>;
/** Delete a snapshot by key. */
delete(key: string): Promise<void>;
/** Delete a snapshot. Hosts can use the finishing state to check ownership. */
delete(key: string, state?: SerializableAgentState): Promise<void>;
}

47
packages/@n8n/db/src/migrations/common/1790241069212-CreateAgentMessageQueue.ts Normal file
View File

@ -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');
}
}

51
packages/cli/src/modules/agents/__tests__/agent-chat-execution.service.test.ts
View File

@ -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');

14
packages/cli/src/modules/agents/__tests__/agent-checkpoint.repository.test.ts
View File

@ -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<TransactionRunner>());
});
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',

45
packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts
View File

@ -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<N8NCheckpointStorage>();
const executionService = mock<AgentExecutionService>();
executionService.getAbortSignal.mockReturnValue(new AbortController().signal);
const executionRepository = mock<AgentExecutionRepository>();
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<AgentMessageQueueService>(),
),
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(
'<background-jobs-settled>[]</background-jobs-settled>',
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();

84
packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts
View File

@ -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<N8NCheckpointStorage>();
let executionUpdateBroadcaster: Mocked<AgentExecutionUpdateBroadcaster>;
const txRunner = mock<TransactionRunner>();
const queueRepository = mock<AgentMessageQueueRepository>();
beforeEach(() => {
vi.clearAllMocks();
agentExecutionRepository = mock<AgentExecutionRepository>();
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>();
agentExecutionThreadRepository.lockById.mockResolvedValue(makeThread());
n8nMemory = mock<N8nMemory>();
memoryBackend = mock<N8nMemoryImplementation>();
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();
});

9
packages/cli/src/modules/agents/__tests__/agent-interrupted-execution-sweeper.test.ts
View File

@ -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 () => {

3
packages/cli/src/modules/agents/__tests__/agent-runtime-reconstruction-integration-tools.test.ts
View File

@ -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<N8NCheckpointStorage>();
agentExecutionService = mock<AgentExecutionService>();
agentExecutionService.getAbortSignal.mockReturnValue(new AbortController().signal);
agentKnowledgeService = mock<AgentKnowledgeService>();
publisher = mock<Publisher>();
publisher.publishCommand.mockResolvedValue();
@ -281,6 +283,7 @@ describe('AgentRuntimeReconstructionService integration tools', () => {
logger,
agentExecutionService,
mock<AgentChatExecutionService>(),
mock<AgentMessageQueueService>(),
),
telemetry,
runtimeCacheService,

3
packages/cli/src/modules/agents/__tests__/agent-workflow-execution.service.test.ts
View File

@ -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<AgentRepository>();
const executionService = mock<AgentExecutionService>();
executionService.getAbortSignal.mockReturnValue(new AbortController().signal);
const telemetry = mock<Telemetry>();
const credentialsService = mock<CredentialsService>();
const reconstructionService = mock<AgentRuntimeReconstructionService>();
@ -146,6 +148,7 @@ function makeService() {
mockLogger(),
executionService,
mock<AgentChatExecutionService>(),
mock<AgentMessageQueueService>(),
),
telemetry,
credentialsService,

53
packages/cli/src/modules/agents/agent-chat-execution.service.ts
View File

@ -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<T>(threadId: string, create: () => Promise<T>): Promise<T> {
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<T>(
threadId: string,
agentId: string,
runId: string,
createAndClaim: () => Promise<T>,
): Promise<T> {
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<T>(
key: string,
admit: (signal: AbortSignal) => Promise<T>,
): Promise<T> {
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 });
}

59
packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts
View File

@ -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<void>;
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<StreamChunk> {
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<AgentRuntime> {
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<IntegrationMessageContext> {
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<AgentTurnRequest> {

212
packages/cli/src/modules/agents/agent-execution.service.ts
View File

@ -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<RecordMessageParams, 'record'
access: AgentThreadAccess;
sessionMode?: AgentSessionMode;
initialTimeline?: TimelineEvent[];
/** Internal admission data. These fields are not stored on the execution. */
queueItemId?: string;
resumeRunId?: string;
allowSuspendedPredecessor?: boolean;
}
export interface AgentExecutionReservation {
execution: AgentExecution;
needsTitleSync: boolean;
}
interface TimelineSnapshotParams extends AgentExecutionLogRef {
@ -119,6 +131,7 @@ export class AgentExecutionService {
private static readonly heartbeatIntervalMs = 30_000;
private readonly heartbeatTimers = new Map<string, NodeJS.Timeout>();
private readonly executionControllers = new Map<string, AbortController>();
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<string> {
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<AgentExecutionReservation> {
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<boolean> {
async finalizeInterruptedExecution(
execution: RunningAgentExecution,
staleBefore: Date,
): Promise<boolean> {
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 }> {

7
packages/cli/src/modules/agents/agent-interrupted-execution-sweeper.ts
View File

@ -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,

161
packages/cli/src/modules/agents/agent-message-queue.service.ts Normal file
View File

@ -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<AgentMessageQueue> {
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<boolean>,
): Promise<ClaimedAgentMessage | null> {
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<void> {
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<void> {
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<boolean> {
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,
};
}
}

7
packages/cli/src/modules/agents/agent-turn-already-running.error.ts Normal file
View File

@ -0,0 +1,7 @@
import { UserError } from 'n8n-workflow';
export class AgentTurnAlreadyRunningError extends UserError {
constructor() {
super('A turn is already running in this conversation.');
}
}

87
packages/cli/src/modules/agents/agent-turn-execution.service.ts
View File

@ -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<void>;
agentInstance: RuntimeAgent;
toolRegistry: ToolRegistry;
mcpServerAttributions: Map<string, string>;
@ -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<AgentSessionMode> {
@ -112,12 +118,17 @@ export class AgentTurnExecutionService {
async *execute(config: ExecuteTurnConfig): AsyncGenerator<StreamChunk> {
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<ReadableStream<StreamChunk>> {
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<ReadableStream<StreamChunk>> {
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);
}

2
packages/cli/src/modules/agents/agents.module.ts
View File

@ -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,

54
packages/cli/src/modules/agents/entities/agent-message-queue.entity.ts Normal file
View File

@ -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<AgentExecutionThread>;
@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<AgentExecution> | null;
@Index()
@Column({
type: 'varchar',
length: 36,
nullable: true,
comment: 'Current execution; NULL means pending',
})
executionId: string | null;
}

4
packages/cli/src/modules/agents/execution-recorder.ts
View File

@ -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<string, number>();

42
packages/cli/src/modules/agents/integrations/__tests__/n8n-checkpoint-storage.test.ts
View File

@ -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<AgentCheckpointRepository>();
repository.create.mockImplementation((data) => Object.assign(mock<AgentCheckpoint>(), data));
const service = new N8NCheckpointStorage(
repository,
mockLogger(),
mock<AgentsConfig>({ checkpointTtlSeconds: 60 }),
mock<TransactionRunner>(),
mock<AgentExecutionRepository>(),
mock<AgentExecutionThreadRepository>(),
mock<AgentMessageQueueRepository>(),
);
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

119
packages/cli/src/modules/agents/integrations/n8n-checkpoint-storage.ts
View File

@ -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<void> {
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<void>,
): Promise<void> {
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<SerializableAgentState | undefined> {
@ -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<boolean> {
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<SerializableAgentState | null> {
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<CheckpointStatus> {
const checkpoint = await this.agentCheckpointRepository.findByRunIdAndAgentId(key, agentId);
async getStatus(
key: string,
agentId: string,
ctx: OperationContext = {},
): Promise<CheckpointStatus> {
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<SerializableAgentState>(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 };
}

61
packages/cli/src/modules/agents/repositories/agent-checkpoint.repository.ts
View File

@ -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<AgentCheckpoint> {
constructor(dataSource: DataSource) {
super(AgentCheckpoint, dataSource.manager);
export class AgentCheckpointRepository extends BaseRepository<AgentCheckpoint> {
constructor(dataSource: DataSource, transactionRunner: TransactionRunner) {
super(AgentCheckpoint, dataSource.manager, transactionRunner);
}
async findByRunId(runId: string): Promise<AgentCheckpoint | null> {
return await this.findOneBy({ runId });
async findByRunId(runId: string, ctx: OperationContext = {}): Promise<AgentCheckpoint | null> {
return await this.managerFor(ctx).findOneBy(AgentCheckpoint, { runId });
}
async findByRunIdAndAgentId(runId: string, agentId: string): Promise<AgentCheckpoint | null> {
return await this.findOneBy({ runId, agentId });
async findByRunIdAndAgentId(
runId: string,
agentId: string,
ctx: OperationContext = {},
): Promise<AgentCheckpoint | null> {
return await this.managerFor(ctx).findOneBy(AgentCheckpoint, { runId, agentId });
}
async findActiveForThread(agentId: string, threadId: string): Promise<AgentCheckpoint[]> {
return await this.find({
where: { agentId, threadId, expired: false },
async findActiveForThread(
agentId: string,
threadId: string,
updatedAfter: Date,
ctx: OperationContext = {},
): Promise<AgentCheckpoint[]> {
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<void> {
await this.managerFor(ctx).save(AgentCheckpoint, checkpoint);
}
async findRetainedByThreadId(threadId: string): Promise<AgentCheckpoint[]> {
return await this.find({ where: { threadId } });
}
@ -41,17 +55,24 @@ export class AgentCheckpointRepository extends Repository<AgentCheckpoint> {
agentId: string,
suspendedState: string,
runningState: string,
updatedAfter: Date,
): Promise<boolean> {
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<boolean> {
const result = await this.update(
async cancelSuspended(
runId: string,
agentId: string,
suspendedState: string,
ctx: OperationContext = {},
): Promise<boolean> {
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<AgentCheckpoint> {
return (result.affected ?? 0) > 0;
}
async expireByRunIdAndAgentId(runId: string, agentId: string): Promise<void> {
await this.update({ runId, agentId }, { expired: true, state: null });
async expireByRunIdAndAgentId(
runId: string,
agentId: string,
ctx: OperationContext = {},
): Promise<void> {
await this.managerFor(ctx).update(
AgentCheckpoint,
{ runId, agentId },
{ expired: true, state: null },
);
}
async markExpired(olderThan: Date): Promise<number> {

9
packages/cli/src/modules/agents/repositories/agent-execution-thread.repository.ts
View File

@ -42,6 +42,15 @@ export class AgentExecutionThreadRepository extends BaseRepository<AgentExecutio
super(AgentExecutionThread, dataSource.manager, transactionRunner);
}
async lockById(threadId: string, ctx: OperationContext): Promise<AgentExecutionThread | null> {
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.

36
packages/cli/src/modules/agents/repositories/agent-execution.repository.ts
View File

@ -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<AgentExecution> {
});
}
async existsRunningByThread(threadId: string): Promise<boolean> {
return await this.existsBy({ threadId, status: 'running' });
async existsRunningByThread(threadId: string, ctx: OperationContext = {}): Promise<boolean> {
return await this.managerFor(ctx).existsBy(AgentExecution, { threadId, status: 'running' });
}
async findRunningByThread(threadId: string, ctx: OperationContext): Promise<AgentExecution[]> {
return await this.managerFor(ctx).findBy(AgentExecution, { threadId, status: 'running' });
}
async findExecution(
executionId: string,
ctx: OperationContext = {},
): Promise<AgentExecution | null> {
return await this.managerFor(ctx).findOneBy(AgentExecution, { id: executionId });
}
async findLatestByThreadId(threadId: string): Promise<AgentExecution | null> {
return await this.findOne({ where: { threadId }, order: { createdAt: 'DESC', id: 'DESC' } });
}
async touchRunning(executionId: string): Promise<void> {
await this.update({ id: executionId, status: 'running' }, { updatedAt: new Date() });
async touchRunning(executionId: string): Promise<boolean> {
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<AgentExecution> {
async updateIfRunning(
executionId: string,
values: AgentExecutionFinalizationValues,
staleBefore?: Date,
ctx: OperationContext = {},
): Promise<boolean> {
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<AgentExecution>,
);
return result.affected === 1;

62
packages/cli/src/modules/agents/repositories/agent-message-queue.repository.ts Normal file
View File

@ -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<AgentMessageQueue> {
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 } });
}
}

3
packages/cli/src/modules/agents/sub-agents/__tests__/sub-agent-runner.test.ts
View File

@ -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<AgentRuntimeReconstructionService>();
Container.set(AgentRuntimeReconstructionService, reconstructionService);
agentExecutionService = mock<AgentExecutionService>();
agentExecutionService.getAbortSignal.mockReturnValue(new AbortController().signal);
agentExecutionService.startExecutionRecording.mockResolvedValue('agent-execution-1');
agentExecutionService.finalizeExecution.mockResolvedValue('agent-execution-1');
checkpointStorage = mock<N8NCheckpointStorage>();
@ -142,6 +144,7 @@ describe('SubAgentRunner', () => {
logger,
agentExecutionService,
mock<AgentChatExecutionService>(),
mock<AgentMessageQueueService>(),
),
checkpointStorage,
logger,

47
packages/cli/src/modules/agents/types/agent-queued-message.ts Normal file
View File

@ -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;
}

66
packages/cli/test/integration/agents/agent-checkpoint-pruning.task.integration.test.ts
View File

@ -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<unknown>): Promise<boolean> {
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');
});
});
});

14
packages/cli/test/integration/agents/agent-conversation-state.test.ts
View File

@ -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);
});

437
packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts
View File

@ -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<N8nMemory>();
memory.getImplementation.mockReturnValue(memoryBackend);
const attachmentService = mock<AgentChatAttachmentService>();
@ -145,12 +166,25 @@ describe('AgentExecutionRepository', () => {
mock<StorageConfig>({ modeTag: 'db' }),
mock<ErrorReporter>(),
mock<AgentExecutionUpdateBroadcaster>(),
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<AgentChatExecutionService>(),
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<ReturnType<typeof startSuspendedApprovalRun>>,
) {
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<OperationContext>();
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<boolean>();
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<typeof makeAgent> | 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<AgentMessageQueueService['enqueue']>[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<typeof recordingServices>, 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<typeof recordingServices>,
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<OperationContext>();
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<AgentExecutionThread> = {}) => {
const thread = threadRepo.create({
accessScope: 'project',

48
packages/cli/test/migration/1790241069212-create-agent-message-queue.test.ts Normal file
View File

@ -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<boolean> {
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);
});
});