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 agent input boundaries and steering storage (no-changelog) (#39529)

Browse Source
This commit is contained in:
Robin Braumann 2026-09-29 12:41:34 +00:00 committed by GitHub
parent 7b12ec165d
commit b09fd0dc69
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
43 changed files with 1220 additions and 160 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -19,12 +19,12 @@ Auto-generated from the PostgreSQL migrations in @n8n/db. Do not edit by hand.
| [public.agent_eval_rating](public.agent_eval_rating.md) | 8 | | BASE TABLE |
| [public.agent_eval_result](public.agent_eval_result.md) | 15 | | BASE TABLE |
| [public.agent_eval_run](public.agent_eval_run.md) | 14 | | BASE TABLE |
| [public.agent_execution](public.agent_execution.md) | 22 | | BASE TABLE |
| [public.agent_execution](public.agent_execution.md) | 23 | | BASE TABLE |
| [public.agent_execution_message_links](public.agent_execution_message_links.md) | 5 | | BASE TABLE |
| [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_message_queue](public.agent_message_queue.md) | 9 | | 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 |
@ -209,6 +209,7 @@ erDiagram
"public.agent_history" }o--|| "public.agents" : "FOREIGN KEY (#quot;agentId#quot;) REFERENCES agents(id) ON DELETE CASCADE"
"public.agent_message_queue" }o--|| "public.agents_messages" : "FOREIGN KEY (#quot;messageId#quot;) REFERENCES agents_messages(id) ON DELETE CASCADE"
"public.agent_message_queue" }o--o| "public.agent_execution" : "FOREIGN KEY (#quot;executionId#quot;) REFERENCES agent_execution(id)"
"public.agent_message_queue" }o--o| "public.agent_execution" : "FOREIGN KEY (#quot;steeringExecutionId#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"
@ -534,6 +535,7 @@ erDiagram
timestamp_3__with_time_zone updatedAt
}
"public.agent_execution" {
boolean acceptsSteering
json attachments
json author
integer completionTokens
@ -614,6 +616,8 @@ erDiagram
bigint id
varchar_36_ messageId FK
json payload
varchar_36_ steeringExecutionId FK
integer steeringOrder
varchar_128_ threadId FK
timestamp_3__with_time_zone updatedAt
}

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

@ -4,6 +4,7 @@
| Name | Type | Default | Nullable | Children | Parents | Comment |
| ---- | ---- | ------- | -------- | -------- | ------- | ------- |
| acceptsSteering | boolean | false | false | | | Accept input until the runtime closes admission |
| attachments | json | | true | | | Metadata of files attached to the user turn ({id, fileName, mimeType, sizeBytes}[]); bytes live in BinaryDataService |
| author | json | | true | | | Chat platform user who wrote the turn as {id, name}; null for runs outside chat integrations |
| completionTokens | integer | | true | | | |
@ -36,6 +37,7 @@
| CHK_agent_execution_storedAt | CHECK | CHECK ((("storedAt")::text = ANY ((ARRAY['db'::character varying, 'fs'::character varying, 's3'::character varying, 'az'::character varying])::text[]))) |
| FK_add2432fb6034cc18b6af299dce | FOREIGN KEY | FOREIGN KEY ("threadId") REFERENCES agent_execution_threads(id) ON DELETE CASCADE |
| PK_ba438acc8532addc12d1ef17049 | PRIMARY KEY | PRIMARY KEY (id) |
| agent_execution_acceptsSteering_not_null | n | NOT NULL "acceptsSteering" |
| agent_execution_createdAt_not_null | n | NOT NULL "createdAt" |
| agent_execution_duration_not_null | n | NOT NULL duration |
| agent_execution_id_not_null | n | NOT NULL id |
@ -59,9 +61,11 @@ erDiagram
"public.agent_execution_message_links" }o--|| "public.agent_execution" : "FOREIGN KEY (#quot;executionId#quot;) REFERENCES agent_execution(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--o| "public.agent_execution" : "FOREIGN KEY (#quot;steeringExecutionId#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" {
boolean acceptsSteering
json attachments
json author
integer completionTokens
@ -98,6 +102,8 @@ erDiagram
bigint id
varchar_36_ messageId FK
json payload
varchar_36_ steeringExecutionId FK
integer steeringOrder
varchar_128_ threadId FK
timestamp_3__with_time_zone updatedAt
}

1
docs/generated/postgres-schema/public.agent_execution_message_links.md generated
View File

@ -48,6 +48,7 @@ erDiagram
integer position
}
"public.agent_execution" {
boolean acceptsSteering
json attachments
json author
integer completionTokens

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

@ -107,6 +107,7 @@ erDiagram
varchar_36_ versionId
}
"public.agent_execution" {
boolean acceptsSteering
json attachments
json author
integer completionTokens
@ -136,6 +137,8 @@ erDiagram
bigint id
varchar_36_ messageId FK
json payload
varchar_36_ steeringExecutionId FK
integer steeringOrder
varchar_128_ threadId FK
timestamp_3__with_time_zone updatedAt
}

8
docs/generated/postgres-schema/public.agent_message_queue.md generated
View File

@ -9,6 +9,8 @@
| id | bigint | | false | | | Acceptance order; IDs are not reused |
| messageId | varchar(36) | | false | | [public.agents_messages](public.agents_messages.md) | Canonical input created when the queue accepts it |
| payload | json | | false | | | Dispatch, authorization, and reply context. Input is stored on the message |
| steeringExecutionId | varchar(36) | | true | | [public.agent_execution](public.agent_execution.md) | Execution reserved to consume this input |
| steeringOrder | integer | | true | | | Acceptance order among outstanding steers |
| threadId | varchar(128) | | false | | [public.agent_execution_threads](public.agent_execution_threads.md) | |
| updatedAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | |
@ -19,6 +21,7 @@
| FK_2349d84b4f2a660fc264f38fef3 | FOREIGN KEY | FOREIGN KEY ("executionId") REFERENCES agent_execution(id) |
| FK_a74f9154a59430986112d70cb16 | FOREIGN KEY | FOREIGN KEY ("threadId") REFERENCES agent_execution_threads(id) ON DELETE CASCADE |
| FK_agent_message_queue_messageId | FOREIGN KEY | FOREIGN KEY ("messageId") REFERENCES agents_messages(id) ON DELETE CASCADE |
| FK_eff54787927c2968dc24495c911 | FOREIGN KEY | FOREIGN KEY ("steeringExecutionId") REFERENCES agent_execution(id) |
| 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 |
@ -33,6 +36,7 @@
| ---- | ---------- |
| IDX_2349d84b4f2a660fc264f38fef | CREATE INDEX "IDX_2349d84b4f2a660fc264f38fef" ON public.agent_message_queue USING btree ("executionId") |
| IDX_agent_message_queue_messageId | CREATE UNIQUE INDEX "IDX_agent_message_queue_messageId" ON public.agent_message_queue USING btree ("messageId") |
| IDX_agent_message_queue_steeringExecutionId_steeringOrder | CREATE UNIQUE INDEX "IDX_agent_message_queue_steeringExecutionId_steeringOrder" ON public.agent_message_queue USING btree ("steeringExecutionId", "steeringOrder") WHERE ("steeringExecutionId" IS NOT NULL) |
| 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) |
@ -44,6 +48,7 @@ erDiagram
"public.agent_message_queue" }o--o| "public.agent_execution" : "FOREIGN KEY (#quot;executionId#quot;) REFERENCES agent_execution(id)"
"public.agent_message_queue" }o--|| "public.agents_messages" : "FOREIGN KEY (#quot;messageId#quot;) REFERENCES agents_messages(id) ON DELETE CASCADE"
"public.agent_message_queue" }o--o| "public.agent_execution" : "FOREIGN KEY (#quot;steeringExecutionId#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" {
@ -52,10 +57,13 @@ erDiagram
bigint id
varchar_36_ messageId FK
json payload
varchar_36_ steeringExecutionId FK
integer steeringOrder
varchar_128_ threadId FK
timestamp_3__with_time_zone updatedAt
}
"public.agent_execution" {
boolean acceptsSteering
json attachments
json author
integer completionTokens

2
docs/generated/postgres-schema/public.agents_messages.md generated
View File

@ -78,6 +78,8 @@ erDiagram
bigint id
varchar_36_ messageId FK
json payload
varchar_36_ steeringExecutionId FK
integer steeringOrder
varchar_128_ threadId FK
timestamp_3__with_time_zone updatedAt
}

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

@ -19,12 +19,12 @@ Auto-generated from the SQLite migrations in @n8n/db. Do not edit by hand.
| [agent_eval_rating](agent_eval_rating.md) | 8 | | table |
| [agent_eval_result](agent_eval_result.md) | 15 | | table |
| [agent_eval_run](agent_eval_run.md) | 14 | | table |
| [agent_execution](agent_execution.md) | 22 | | table |
| [agent_execution](agent_execution.md) | 23 | | table |
| [agent_execution_message_links](agent_execution_message_links.md) | 5 | | table |
| [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_message_queue](agent_message_queue.md) | 9 | | 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 |
@ -189,6 +189,7 @@ 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 (steeringExecutionId) REFERENCES agent_execution (id) ON UPDATE NO ACTION ON DELETE NO ACTION MATCH NONE"
"agent_message_queue" }o--|| "agents_messages" : "FOREIGN KEY (messageId) REFERENCES agents_messages (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"agent_message_queue" }o--|| "agent_execution_threads" : "FOREIGN KEY (threadId) REFERENCES agent_execution_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"agent_message_queue" }o--o| "agent_execution" : "FOREIGN KEY (executionId) REFERENCES agent_execution (id) ON UPDATE NO ACTION ON DELETE NO ACTION MATCH NONE"
@ -520,6 +521,7 @@ erDiagram
datetime_3_ updatedAt
}
"agent_execution" {
BOOLEAN acceptsSteering
TEXT attachments
TEXT author
INTEGER completionTokens
@ -600,6 +602,8 @@ erDiagram
INTEGER id
varchar_36_ messageId FK
TEXT payload
varchar_36_ steeringExecutionId FK
INTEGER steeringOrder
varchar_128_ threadId FK
datetime_3_ updatedAt
}

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

@ -6,7 +6,7 @@
<summary><strong>Table Definition</strong></summary>
```sql
CREATE TABLE "agent_execution" ("id" varchar(36) PRIMARY KEY NOT NULL, "threadId" varchar(128) NOT NULL, "status" varchar(16) NOT NULL, "startedAt" datetime(3), "stoppedAt" datetime(3), "duration" integer NOT NULL DEFAULT (0), "userMessage" text, "model" varchar(255), "promptTokens" integer, "completionTokens" integer, "totalTokens" integer, "cost" real, "timeline" text, "error" text, "hitlStatus" varchar(16), "source" varchar(32), "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "storedAt" varchar(2) NOT NULL DEFAULT ('db'), "attachments" text, "failureSummary" text, "author" text, CONSTRAINT "CHK_agent_execution_storedAt" CHECK ((("storedAt" IN ('db', 'fs', 's3', 'az')))), CONSTRAINT "CHK_agent_execution_hitlStatus" CHECK (((((("hitlStatus" IN ('suspended', 'resumed'))))))), CONSTRAINT "CHK_agent_execution_status" CHECK (("status" IN ('running', 'success', 'error', 'cancelled', 'interrupted'))), CONSTRAINT "FK_add2432fb6034cc18b6af299dce" FOREIGN KEY ("threadId") REFERENCES "agent_execution_threads" ("id") ON DELETE CASCADE ON UPDATE NO ACTION)
CREATE TABLE "agent_execution" ("id" varchar(36) PRIMARY KEY NOT NULL, "threadId" varchar(128) NOT NULL, "status" varchar(16) NOT NULL, "startedAt" datetime(3), "stoppedAt" datetime(3), "duration" integer NOT NULL DEFAULT (0), "userMessage" text, "model" varchar(255), "promptTokens" integer, "completionTokens" integer, "totalTokens" integer, "cost" real, "timeline" text, "error" text, "hitlStatus" varchar(16), "source" varchar(32), "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "storedAt" varchar(2) NOT NULL DEFAULT ('db'), "attachments" text, "failureSummary" text, "author" text, "acceptsSteering" BOOLEAN NOT NULL DEFAULT FALSE, CONSTRAINT "CHK_agent_execution_storedAt" CHECK ((("storedAt" IN ('db', 'fs', 's3', 'az')))), CONSTRAINT "CHK_agent_execution_hitlStatus" CHECK (((((("hitlStatus" IN ('suspended', 'resumed'))))))), CONSTRAINT "CHK_agent_execution_status" CHECK (("status" IN ('running', 'success', 'error', 'cancelled', 'interrupted'))), CONSTRAINT "FK_add2432fb6034cc18b6af299dce" FOREIGN KEY ("threadId") REFERENCES "agent_execution_threads" ("id") ON DELETE CASCADE ON UPDATE NO ACTION)
```
</details>
@ -15,6 +15,7 @@ CREATE TABLE "agent_execution" ("id" varchar(36) PRIMARY KEY NOT NULL, "threadId
| Name | Type | Default | Nullable | Children | Parents | Comment |
| ---- | ---- | ------- | -------- | -------- | ------- | ------- |
| acceptsSteering | BOOLEAN | FALSE | false | | | |
| attachments | TEXT | | true | | | |
| author | TEXT | | true | | | |
| completionTokens | INTEGER | | true | | | |
@ -63,10 +64,12 @@ CREATE TABLE "agent_execution" ("id" varchar(36) PRIMARY KEY NOT NULL, "threadId
erDiagram
"agent_execution_message_links" |o--|| "agent_execution" : "FOREIGN KEY (executionId) REFERENCES agent_execution (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"agent_message_queue" }o--o| "agent_execution" : "FOREIGN KEY (steeringExecutionId) REFERENCES agent_execution (id) ON UPDATE NO ACTION ON DELETE NO ACTION 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_execution" }o--|| "agent_execution_threads" : "FOREIGN KEY (threadId) REFERENCES agent_execution_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"agent_execution" {
BOOLEAN acceptsSteering
TEXT attachments
TEXT author
INTEGER completionTokens
@ -103,6 +106,8 @@ erDiagram
INTEGER id
varchar_36_ messageId FK
TEXT payload
varchar_36_ steeringExecutionId FK
INTEGER steeringOrder
varchar_128_ threadId FK
datetime_3_ updatedAt
}

1
docs/generated/sqlite-schema/agent_execution_message_links.md generated
View File

@ -56,6 +56,7 @@ erDiagram
INTEGER position
}
"agent_execution" {
BOOLEAN acceptsSteering
TEXT attachments
TEXT author
INTEGER completionTokens

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

@ -107,6 +107,7 @@ erDiagram
varchar_36_ versionId
}
"agent_execution" {
BOOLEAN acceptsSteering
TEXT attachments
TEXT author
INTEGER completionTokens
@ -136,6 +137,8 @@ erDiagram
INTEGER id
varchar_36_ messageId FK
TEXT payload
varchar_36_ steeringExecutionId FK
INTEGER steeringOrder
varchar_128_ threadId FK
datetime_3_ updatedAt
}

16
docs/generated/sqlite-schema/agent_message_queue.md generated
View File

@ -6,7 +6,7 @@
<summary><strong>Table Definition</strong></summary>
```sql
CREATE TABLE "agent_message_queue" ("id" integer PRIMARY KEY AUTOINCREMENT NOT NULL, "threadId" varchar(128) NOT NULL, "payload" text NOT NULL, "executionId" varchar(36), "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "messageId" varchar(36) NOT NULL, CONSTRAINT "FK_2349d84b4f2a660fc264f38fef3" FOREIGN KEY ("executionId") REFERENCES "agent_execution" ("id") ON DELETE NO ACTION ON UPDATE NO ACTION, CONSTRAINT "FK_a74f9154a59430986112d70cb16" FOREIGN KEY ("threadId") REFERENCES "agent_execution_threads" ("id") ON DELETE CASCADE ON UPDATE NO ACTION, CONSTRAINT "FK_agent_message_queue_messageId" FOREIGN KEY ("messageId") REFERENCES "agents_messages" ("id") ON DELETE CASCADE)
CREATE TABLE "agent_message_queue" ("id" integer PRIMARY KEY AUTOINCREMENT NOT NULL, "threadId" varchar(128) NOT NULL, "payload" text NOT NULL, "executionId" varchar(36), "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "messageId" varchar(36) NOT NULL, "steeringExecutionId" varchar(36), "steeringOrder" integer, CONSTRAINT "FK_2349d84b4f2a660fc264f38fef3" FOREIGN KEY ("executionId") REFERENCES "agent_execution" ("id") ON DELETE NO ACTION ON UPDATE NO ACTION, CONSTRAINT "FK_a74f9154a59430986112d70cb16" FOREIGN KEY ("threadId") REFERENCES "agent_execution_threads" ("id") ON DELETE CASCADE ON UPDATE NO ACTION, CONSTRAINT "FK_agent_message_queue_messageId" FOREIGN KEY ("messageId") REFERENCES "agents_messages" ("id") ON DELETE CASCADE, CONSTRAINT "FK_eff54787927c2968dc24495c911" FOREIGN KEY ("steeringExecutionId") REFERENCES "agent_execution" ("id") ON DELETE NO ACTION)
```
</details>
@ -20,6 +20,8 @@ CREATE TABLE "agent_message_queue" ("id" integer PRIMARY KEY AUTOINCREMENT NOT N
| id | INTEGER | | false | | | |
| messageId | varchar(36) | | false | | [agents_messages](agents_messages.md) | |
| payload | TEXT | | false | | | |
| steeringExecutionId | varchar(36) | | true | | [agent_execution](agent_execution.md) | |
| steeringOrder | INTEGER | | true | | | |
| threadId | varchar(128) | | false | | [agent_execution_threads](agent_execution_threads.md) | |
| updatedAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | |
@ -27,9 +29,10 @@ CREATE TABLE "agent_message_queue" ("id" integer PRIMARY KEY AUTOINCREMENT NOT N
| Name | Type | Definition |
| ---- | ---- | ---------- |
| - (Foreign key ID: 0) | FOREIGN KEY | FOREIGN KEY (messageId) REFERENCES agents_messages (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE |
| - (Foreign key ID: 1) | FOREIGN KEY | FOREIGN KEY (threadId) REFERENCES agent_execution_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE |
| - (Foreign key ID: 2) | FOREIGN KEY | FOREIGN KEY (executionId) REFERENCES agent_execution (id) ON UPDATE NO ACTION ON DELETE NO ACTION MATCH NONE |
| - (Foreign key ID: 0) | FOREIGN KEY | FOREIGN KEY (steeringExecutionId) REFERENCES agent_execution (id) ON UPDATE NO ACTION ON DELETE NO ACTION MATCH NONE |
| - (Foreign key ID: 1) | FOREIGN KEY | FOREIGN KEY (messageId) REFERENCES agents_messages (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE |
| - (Foreign key ID: 2) | FOREIGN KEY | FOREIGN KEY (threadId) REFERENCES agent_execution_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE |
| - (Foreign key ID: 3) | FOREIGN KEY | FOREIGN KEY (executionId) REFERENCES agent_execution (id) ON UPDATE NO ACTION ON DELETE NO ACTION MATCH NONE |
| id | PRIMARY KEY | PRIMARY KEY (id) |
## Indexes
@ -38,6 +41,7 @@ CREATE TABLE "agent_message_queue" ("id" integer PRIMARY KEY AUTOINCREMENT NOT N
| ---- | ---------- |
| IDX_2349d84b4f2a660fc264f38fef | CREATE INDEX "IDX_2349d84b4f2a660fc264f38fef" ON "agent_message_queue" ("executionId") |
| IDX_agent_message_queue_messageId | CREATE UNIQUE INDEX "IDX_agent_message_queue_messageId" ON "agent_message_queue" ("messageId") |
| IDX_agent_message_queue_steeringExecutionId_steeringOrder | CREATE UNIQUE INDEX "IDX_agent_message_queue_steeringExecutionId_steeringOrder" ON "agent_message_queue" ("steeringExecutionId", "steeringOrder") WHERE "steeringExecutionId" IS NOT NULL |
| 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") |
@ -48,6 +52,7 @@ erDiagram
"agent_message_queue" }o--o| "agent_execution" : "FOREIGN KEY (executionId) REFERENCES agent_execution (id) ON UPDATE NO ACTION ON DELETE NO ACTION MATCH NONE"
"agent_message_queue" }o--|| "agents_messages" : "FOREIGN KEY (messageId) REFERENCES agents_messages (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
"agent_message_queue" }o--o| "agent_execution" : "FOREIGN KEY (steeringExecutionId) 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" {
@ -56,10 +61,13 @@ erDiagram
INTEGER id
varchar_36_ messageId FK
TEXT payload
varchar_36_ steeringExecutionId FK
INTEGER steeringOrder
varchar_128_ threadId FK
datetime_3_ updatedAt
}
"agent_execution" {
BOOLEAN acceptsSteering
TEXT attachments
TEXT author
INTEGER completionTokens

2
docs/generated/sqlite-schema/agents_messages.md generated
View File

@ -83,6 +83,8 @@ erDiagram
INTEGER id
varchar_36_ messageId FK
TEXT payload
varchar_36_ steeringExecutionId FK
INTEGER steeringOrder
varchar_128_ threadId FK
datetime_3_ updatedAt
}

1
packages/@n8n/agents/src/index.ts
View File

@ -76,6 +76,7 @@ export type {
McpVerifyResult,
ModelConfig,
ExecutionOptions,
AgentInputBoundary,
SmoothStreamOptions,
TokenUsage,
AgentExecutionCounter,

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

@ -16,7 +16,7 @@ import type { RuntimeSkillSource } from '../../skills/types';
import type { CheckpointStore, ModelConfig, SerializableAgentState } from '../../types';
import { AgentEvent } from '../../types/runtime/event';
import type { AgentEventData } from '../../types/runtime/event';
import type { StreamChunk } from '../../types/sdk/agent';
import type { ExecutionOptions, StreamChunk } from '../../types/sdk/agent';
import type { AgentDbMessage, ContentToolCall, Message } from '../../types/sdk/message';
import type { BuiltTool, InterruptibleToolContext, ToolContext } from '../../types/sdk/tool';
import type { BuiltTelemetry } from '../../types/telemetry';
@ -173,7 +173,9 @@ async function collectChunks(stream: ReadableStream<unknown>): Promise<StreamChu
while (true) {
const { done, value } = await reader.read();
if (done) break;
chunks.push(value as StreamChunk);
const chunk = value as StreamChunk;
chunks.push(chunk);
if (chunk.type === 'input-boundary') chunk.acknowledge();
}
return chunks;
}
@ -409,13 +411,13 @@ describe('AgentRuntime — rejected attachments', () => {
runtime: AgentRuntime,
mode: 'generate' | 'stream',
messages: Message[] | string = [structuredClone(input)],
abortSignal?: AbortSignal,
options: Pick<ExecutionOptions, 'abortSignal' | 'onInputBoundary'> = {},
) {
if (mode === 'generate') {
const result = await runtime.generate(messages, { persistence, abortSignal });
const result = await runtime.generate(messages, { persistence, ...options });
return result.error;
}
const result = await runtime.stream(messages, { persistence, abortSignal });
const result = await runtime.stream(messages, { persistence, ...options });
const chunks = await collectChunks(result.stream);
return chunks.find((chunk) => chunk.type === 'error');
}
@ -608,7 +610,9 @@ describe('AgentRuntime — rejected attachments', () => {
await deleteMessages(ids);
});
rejectModel();
await run(buildRuntime(memory), mode, [structuredClone(input)], controller.signal);
await run(buildRuntime(memory), mode, [structuredClone(input)], {
abortSignal: controller.signal,
});
expect(await memory.getMessages(persistence.threadId)).toEqual([]);
},
);
@ -637,37 +641,79 @@ describe('AgentRuntime — rejected attachments', () => {
},
);
it.each(['generate', 'stream'] as const)('%s preserves input after a tool call', async (mode) => {
const memory = new InMemoryMemory();
const remove = vi.spyOn(memory, 'deleteMessages');
const handler = vi.fn().mockResolvedValue('Done');
const tool: BuiltTool = {
name: 'search',
description: 'Search',
inputSchema: z.object({}),
handler,
};
const response = makeGenerateWithToolCalls([
{ toolCallId: 'search-1', toolName: 'search', args: {} },
]);
generateText.mockResolvedValueOnce(response);
streamText.mockReturnValueOnce({
...makeStreamSuccess(),
stream: makeChunkStream([{ type: 'tool-call', ...response.toolCalls[0] }]),
finishReason: Promise.resolve('tool-calls'),
response: Promise.resolve(response.response),
toolCalls: Promise.resolve(response.toolCalls),
});
rejectModel();
await run(buildRuntime(memory, [tool]), mode);
expect(handler).toHaveBeenCalledOnce();
expect(remove).not.toHaveBeenCalled();
expect(await memory.getMessages(persistence.threadId)).toEqual(
expect.arrayContaining([expect.objectContaining({ id: input.id })]),
);
});
it.each([
{ mode: 'generate', steer: false },
{ mode: 'generate', steer: true },
{ mode: 'stream', steer: false },
{ mode: 'stream', steer: true },
] as const)(
'$mode preserves accepted input after a tool call (steer: $steer)',
async ({ mode, steer }) => {
const memory = new InMemoryMemory();
const remove = vi.spyOn(memory, 'deleteMessages');
const steered: AgentDbMessage = {
...structuredClone(input),
id: 'steered-input',
createdAt: new Date(Date.now() + 100),
};
steered.content[0] = { type: 'text', text: 'Steered image' };
const handler = vi.fn().mockResolvedValue('Done');
const tool: BuiltTool = {
name: 'search',
description: 'Search',
inputSchema: z.object({}),
handler,
};
const response = makeGenerateWithToolCalls([
{ toolCallId: 'search-1', toolName: 'search', args: {} },
]);
generateText.mockResolvedValueOnce(response);
streamText.mockReturnValueOnce({
...makeStreamSuccess(),
stream: makeChunkStream([{ type: 'tool-call', ...response.toolCalls[0] }]),
finishReason: Promise.resolve('tool-calls'),
response: Promise.resolve(response.response),
toolCalls: Promise.resolve(response.toolCalls),
});
rejectModel();
const onInputBoundary = vi.fn(async () => {
if (!steer || handler.mock.calls.length === 0) return [];
await memory.saveMessages({ ...persistence, messages: [steered] });
return [steered];
});
const runtime = buildRuntime(memory, [tool]);
expect(await run(runtime, mode, [structuredClone(input)], { onInputBoundary })).toBeDefined();
expect(handler).toHaveBeenCalledOnce();
if (steer) {
expect(remove).toHaveBeenCalledExactlyOnceWith([steered.id]);
expect(runtime.getState().messageList.inputIds).toEqual([input.id]);
} else expect(remove).not.toHaveBeenCalled();
expect(await memory.getMessages(persistence.threadId)).toEqual(
expect.arrayContaining([expect.objectContaining({ id: input.id })]),
);
expect(await memory.getMessages(persistence.threadId)).not.toEqual(
expect.arrayContaining([expect.objectContaining({ id: steered.id })]),
);
generateText.mockResolvedValueOnce(makeGenerateSuccess());
streamText.mockReturnValueOnce(makeStreamSuccess());
expect(await run(buildRuntime(memory), mode, 'Continue')).toBeUndefined();
const model = mode === 'generate' ? generateText : streamText;
expect(model.mock.calls[2][0].messages).not.toEqual(
expect.arrayContaining([
expect.objectContaining({
content: expect.arrayContaining([{ type: 'text', text: 'Steered image' }]),
}),
]),
);
},
);
it.each(['generate', 'stream'] as const)('%s leaves resumed input unchanged', async (mode) => {
it.each([
{ mode: 'generate', steer: false },
{ mode: 'generate', steer: true },
{ mode: 'stream', steer: false },
{ mode: 'stream', steer: true },
] as const)('$mode preserves resumed input (steer: $steer)', async ({ mode, steer }) => {
const memory = new InMemoryMemory();
const remove = vi.spyOn(memory, 'deleteMessages');
const runtime = buildRuntime(memory, [makeInterruptibleTool()]);
@ -678,20 +724,265 @@ describe('AgentRuntime — rejected attachments', () => {
);
const first = await runtime.generate([structuredClone(input)], { persistence });
const { runId, toolCallId } = first.pendingSuspend![0];
const steered: AgentDbMessage = {
...structuredClone(input),
id: 'steered-input',
createdAt: new Date(Date.now() + 100),
};
const onInputBoundary = async () => {
if (!steer) return [];
await memory.saveMessages({ ...persistence, messages: [steered] });
return [steered];
};
rejectModel();
if (mode === 'generate') {
await runtime.resume('generate', { approved: true }, { runId, toolCallId });
await runtime.resume('generate', { approved: true }, { runId, toolCallId, onInputBoundary });
} else {
const resumed = await runtime.resume('stream', { approved: true }, { runId, toolCallId });
const resumed = await runtime.resume(
'stream',
{ approved: true },
{ runId, toolCallId, onInputBoundary },
);
await collectChunks(resumed.stream);
}
expect(remove).not.toHaveBeenCalled();
if (steer) expect(remove).toHaveBeenCalledExactlyOnceWith([steered.id]);
else expect(remove).not.toHaveBeenCalled();
expect(runtime.getState().messageList.inputIds).toEqual([input.id]);
expect(await memory.getMessages(persistence.threadId)).toEqual(
expect.arrayContaining([expect.objectContaining({ id: input.id })]),
);
expect(await memory.getMessages(persistence.threadId)).not.toEqual(
expect.arrayContaining([expect.objectContaining({ id: steered.id })]),
);
});
});
describe('AgentRuntime — additional input', () => {
beforeEach(() => {
generateText.mockReset();
streamText.mockReset();
});
function input(text: string): AgentDbMessage {
return {
id: crypto.randomUUID(),
role: 'user',
content: [{ type: 'text', text }],
createdAt: new Date(Date.now() + 100),
};
}
it('waits for the complete tool batch before applying input in order', async () => {
const started = createDeferredPromise();
const release = createDeferredPromise();
const handler = vi.fn(async () => {
started.resolve();
await release.promise;
return 'tool result';
});
const tool: BuiltTool = {
name: 'search',
description: 'Search',
inputSchema: z.object({}),
handler,
};
const response = makeGenerateWithToolCall('search-1', 'search', {});
streamText
.mockReturnValueOnce({
...makeStreamSuccess(),
stream: makeChunkStream([{ type: 'tool-call', ...response.toolCalls[0] }]),
finishReason: Promise.resolve('tool-calls'),
response: Promise.resolve(response.response),
toolCalls: Promise.resolve(response.toolCalls),
})
.mockReturnValue(makeStreamSuccess('after'));
const runtime = new AgentRuntime({
name: 'test',
model: 'openai/gpt-4o-mini',
instructions: 'Test',
tools: [tool],
eventBus: new AgentEventBus(),
});
const pending = [input('C'), input('D')];
let available = false;
const onInputBoundary = vi.fn(async () => {
if (!available) return [];
available = false;
return pending;
});
const counter = makeExecutionCounter();
const result = await runtime.stream('A', { onInputBoundary, executionCounter: counter });
const chunksPromise = collectChunks(result.stream);
await started.promise;
available = true;
expect(onInputBoundary).toHaveBeenCalledTimes(1);
expect(streamText).toHaveBeenCalledTimes(1);
release.resolve();
const chunks = await chunksPromise;
const messages = streamText.mock.calls[1][0].messages as Array<{
role: string;
content: unknown;
}>;
expect(messages.filter(({ role }) => role === 'user').map(({ content }) => content)).toEqual([
[{ type: 'text', text: 'A' }],
[{ type: 'text', text: 'C' }],
[{ type: 'text', text: 'D' }],
]);
expect(messages.findIndex(({ role }) => role === 'tool')).toBeLessThan(
messages.findIndex(({ content }) => JSON.stringify(content).includes('"C"')),
);
expect(chunks.filter((chunk) => chunk.type === 'input').map(({ message }) => message)).toEqual(
pending,
);
expect(counter.incrementMessageCount).toHaveBeenCalledTimes(1);
expect(counter.incrementToolCallCount).toHaveBeenCalledTimes(1);
});
it.each([
{ maxIterations: 1, finishReason: 'stop' },
{ maxIterations: 2, finishReason: 'stop' },
{ maxIterations: 1, finishReason: 'error' },
{ maxIterations: 2, finishReason: 'error' },
])(
'handles $finishReason completion with a $maxIterations-call limit',
async ({ maxIterations, finishReason }) => {
streamText.mockReturnValue(makeStreamSuccess('response'));
const error = new Error('Model stream failed');
if (finishReason === 'error') {
streamText.mockReturnValueOnce({
...makeStreamSuccess('partial response'),
stream: makeChunkStream([
{ type: 'text-delta', id: 'text-1', text: 'partial response' },
{ type: 'error', error },
]),
finishReason: Promise.resolve('error'),
});
}
const pending = input('C');
let consumed = false;
const onInputBoundary = vi.fn(
async ({ completing, canContinue }: { completing: boolean; canContinue: boolean }) => {
if (!completing || !canContinue || consumed) return [];
consumed = true;
return [pending];
},
);
const { runtime } = createRuntime();
const chunks = await collectChunks(
(await runtime.stream('A', { maxIterations, onInputBoundary })).stream,
);
expect(streamText).toHaveBeenCalledTimes(finishReason === 'error' ? 1 : maxIterations);
expect(consumed).toBe(finishReason === 'stop' && maxIterations === 2);
expect(chunks.filter((chunk) => chunk.type === 'input')).toHaveLength(consumed ? 1 : 0);
if (finishReason === 'error') {
expect(onInputBoundary).toHaveBeenCalledTimes(1);
expect(chunks).toContainEqual({ type: 'error', error });
} else {
expect(onInputBoundary).toHaveBeenLastCalledWith(
expect.objectContaining({ completing: true, canContinue: false }),
);
}
expect(chunks.at(-1)).toMatchObject({ type: 'finish', finishReason });
},
);
it('applies the attachment budget to earlier input and steered input together', async () => {
const bytes = new Uint8Array(40 * 1024 * 1024);
const runtime = new AgentRuntime({
name: 'attachment-test',
model: 'openai/gpt-4o-mini',
instructions: 'Test',
fileStore: { load: vi.fn().mockResolvedValue(bytes) },
});
const earlier = input('A');
const steered = input('C');
for (const message of [earlier, steered]) {
if ('role' in message) {
message.content.push({
type: 'file',
mediaType: 'image/png',
fileRef: { id: message.id, sizeBytes: bytes.byteLength },
});
}
}
generateText.mockResolvedValue(makeGenerateSuccess());
const onInputBoundary = vi
.fn()
.mockResolvedValueOnce([])
.mockResolvedValueOnce([steered])
.mockResolvedValue([]);
const result = await runtime.generate([earlier], { maxIterations: 2, onInputBoundary });
expect(result.error).toBeUndefined();
expect(generateText).toHaveBeenCalledTimes(2);
const messages: aiModule.ModelMessage[] = generateText.mock.calls[1][0].messages ?? [];
const files = messages.flatMap((message) => {
if (!Array.isArray(message.content)) return [];
return message.content.filter((part) => part.type === 'file');
});
expect(files).toHaveLength(1);
expect(files[0].data).toBe(bytes);
});
it('discards earlier structured output when steered input reaches the iteration limit', async () => {
const tool: BuiltTool = {
name: 'search',
description: 'Search',
inputSchema: z.object({}),
handler: async () => 'tool result',
};
const response = makeGenerateWithToolCall('search-1', 'search', {});
streamText
.mockReturnValueOnce(makeStreamSuccessWithOutput({ answer: 'A', score: 1 }))
.mockReturnValueOnce({
...makeStreamSuccess(),
stream: makeChunkStream([{ type: 'tool-call', ...response.toolCalls[0] }]),
finishReason: Promise.resolve('tool-calls'),
response: Promise.resolve(response.response),
toolCalls: Promise.resolve(response.toolCalls),
});
const onInputBoundary = vi
.fn()
.mockResolvedValueOnce([])
.mockResolvedValueOnce([input('C')])
.mockResolvedValue([]);
const { runtime } = createStructuredRuntime({ tools: [tool] });
const chunks = await collectChunks(
(await runtime.stream('A', { maxIterations: 2, onInputBoundary })).stream,
);
expect(streamText).toHaveBeenCalledTimes(2);
expect(chunks.at(-1)).toMatchObject({
type: 'finish',
finishReason: 'max-iterations',
});
expect(chunks.at(-1)).not.toHaveProperty('structuredOutput');
});
it.each(['abort', 'disconnect'] as const)(
'releases an unacknowledged boundary on %s',
async (action) => {
streamText.mockReturnValue(makeStreamSuccess());
const onInputBoundary = vi.fn().mockResolvedValue([]);
const controller = new AbortController();
const { runtime } = createRuntime();
const result = await runtime.stream('A', { abortSignal: controller.signal, onInputBoundary });
const reader = result.stream.getReader();
expect((await reader.read()).value).toMatchObject({ type: 'input-boundary' });
expect(onInputBoundary).not.toHaveBeenCalled();
if (action === 'disconnect') await reader.cancel();
else {
controller.abort();
while (!(await reader.read()).done) {
/* Drain terminal chunks. */
}
}
expect(streamText).not.toHaveBeenCalled();
},
);
});
describe('AgentRuntime — execution counters', () => {
beforeEach(() => {
generateText.mockReset();

21
packages/@n8n/agents/src/runtime/__tests__/file-parts.test.ts
View File

@ -96,20 +96,37 @@ describe('hydrateFileParts', () => {
expect(store.load).toHaveBeenCalledWith(block.fileRef, { threadId: 'thread-1' });
});
it('hydrates only the newest parts up to the count cap', async () => {
it('keeps only the newest parts within the count cap across hydration calls', async () => {
const blocks = Array.from({ length: MAX_HYDRATED_FILE_PARTS + 2 }, (_, index) =>
refFileBlock({ fileRef: { id: `att-${index}`, sizeBytes: 1 } }),
);
const store: BuiltFileStore = { load: vi.fn().mockResolvedValue(bytes) };
await hydrateFileParts([userMessage(blocks.slice(0, MAX_HYDRATED_FILE_PARTS))], store);
await hydrateFileParts([userMessage(blocks)], store);
expect(store.load).toHaveBeenCalledTimes(MAX_HYDRATED_FILE_PARTS);
expect(store.load).toHaveBeenCalledTimes(MAX_HYDRATED_FILE_PARTS + 2);
expect(blocks[0].data).toBeUndefined();
expect(blocks[1].data).toBeUndefined();
expect(blocks.at(-1)!.data).toBe(bytes);
});
it.each([bytes, bytes.buffer, 'AQID'])(
'charges already-loaded data against the byte budget (%s)',
async (data) => {
const old = refFileBlock({ data });
const newest = refFileBlock({ fileRef: { id: 'att-new' } });
const large = new Uint8Array(MAX_HYDRATED_FILE_BYTES - 2);
const store: BuiltFileStore = { load: vi.fn().mockResolvedValue(large) };
await hydrateFileParts([userMessage([old, newest])], store);
expect(old.data).toBeUndefined();
expect(old.fileRef?.id).toBe('att-1');
expect(newest.data).toBe(large);
},
);
it('spends the byte budget newest-first and skips parts that would exceed it', async () => {
const oldSmall = refFileBlock({ fileRef: { id: 'att-old', sizeBytes: 3 } });
const hugeDeclared = refFileBlock({

24
packages/@n8n/agents/src/runtime/__tests__/memory-orchestrator-persist-input.test.ts
View File

@ -46,7 +46,7 @@ function textsOf(messages: AgentDbMessage[]): string[] {
}
describe('MemoryOrchestrator.discardRejectedInput', () => {
it('removes input from storage, deltas, and serialized state without changing history', async () => {
it('removes selected input from storage and state without changing history or accepted input', async () => {
const store = new InMemoryMemory();
await store.saveThread({ id: THREAD_ID, resourceId: RESOURCE_ID });
const history: AgentDbMessage = {
@ -57,25 +57,29 @@ describe('MemoryOrchestrator.discardRejectedInput', () => {
await store.saveMessages({ threadId: THREAD_ID, resourceId: RESOURCE_ID, messages: [history] });
const list = new AgentMessageList();
list.addHistory([history]);
list.addInput([userMsg('Rejected message'), userMsg('Additional input')]);
list.addInput([
{ ...userMsg('Accepted message'), id: 'accepted', createdAt: new Date() },
{ ...userMsg('Rejected message'), id: 'rejected', createdAt: new Date() },
]);
const accepted = list.inputDelta()[0];
const orchestrator = buildOrchestrator(store);
await orchestrator.persistInputMessages(list, PERSIST);
await orchestrator.discardRejectedInput(list, PERSIST);
await orchestrator.discardRejectedInput(list, PERSIST);
await orchestrator.discardRejectedInput(list, ['rejected'], PERSIST);
await orchestrator.discardRejectedInput(list, ['rejected'], PERSIST);
await orchestrator.persistInputMessages(list, PERSIST);
await orchestrator.persistTurnDelta(list, PERSIST);
await orchestrator.saveToMemory(list, PERSIST);
expect(list.inputDelta()).toEqual([]);
expect(list.turnDelta()).toEqual([]);
expect(list.inputDelta()).toEqual([accepted]);
expect(list.turnDelta()).toEqual([accepted]);
expect(list.serialize()).toMatchObject({
messages: [history],
inputIds: [],
messages: [history, accepted],
inputIds: ['accepted'],
historyIds: ['history'],
});
expect(AgentMessageList.deserialize(list.serialize()).turnDelta()).toEqual([]);
expect(await store.getMessages(THREAD_ID)).toEqual([history]);
expect(AgentMessageList.deserialize(list.serialize()).turnDelta()).toEqual([accepted]);
expect(await store.getMessages(THREAD_ID)).toEqual([history, accepted]);
});
});

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

@ -53,6 +53,7 @@ import { getModelIdString } from '../../utils/model';
import { removeToolResultRun } from '../../workspace';
import { createFilteredLogger } from '../logger';
import { MemoryOrchestrator } from '../memory/memory-orchestrator';
import { sanitizeOffloadedToolResultsForMemory } from '../memory/tool-result-memory';
import { generateThreadTitle } from '../memory/title-generation';
import { AgentMessageList, type SerializedMessageList } from '../model/message-list';
import { createModelTokenCounter } from '../model/model-token-counter';
@ -107,7 +108,7 @@ interface PreparedLoopContext extends LoopContext {
staticContext: StaticLoopContext;
instructionProviderOptions: ProviderOptions | undefined;
runTelemetry: BuiltTelemetry | undefined;
canDiscardRejectedInput: boolean;
acceptedInputIds: Set<string>;
}
interface LoopState {
@ -722,11 +723,17 @@ export class AgentRuntime {
}
for (; state.iterationCount < state.maxIterations; state.iterationCount++) {
const hasInput = await this.consumeAdditionalInput(prepared, sink, state);
if (state.reachedStopCondition && !hasInput) break;
state.reachedStopCondition = false;
const settlement = await this.runLoopIteration(prepared, sink, state);
if (settlement.suspended) return settlement.result;
if (state.reachedStopCondition) break;
if (state.reachedStopCondition && !prepared.options?.onInputBoundary) break;
}
if (state.iterationCount >= state.maxIterations) {
await this.consumeAdditionalInput(prepared, sink, state);
}
if (!state.reachedStopCondition && state.iterationCount >= state.maxIterations) {
state.lastFinishReason = 'max-iterations';
}
@ -739,6 +746,35 @@ export class AgentRuntime {
});
}
private async consumeAdditionalInput<T>(
ctx: PreparedLoopContext,
sink: RunOutputSink<T>,
state: LoopState,
): Promise<boolean> {
const receive = ctx.options?.onInputBoundary;
if (!receive || state.lastFinishReason === 'error') return false;
this.assertNotAborted(ctx.abortScope);
await sink.inputBoundary?.(ctx.abortScope.signal);
this.assertNotAborted(ctx.abortScope);
const messages = await receive({
messages: sanitizeOffloadedToolResultsForMemory(ctx.list.turnDelta()),
lastCreatedAt: ctx.list.messages().at(-1)?.createdAt.getTime() ?? 0,
completing: state.reachedStopCondition,
canContinue: state.iterationCount < state.maxIterations,
});
// Only committed input enters the live list. Its IDs and timestamps stay stable.
ctx.list.addInput(messages);
for (const message of messages) await sink.emitInput?.(message);
this.assertNotAborted(ctx.abortScope);
if (messages.length > 0) {
state.structuredOutput = undefined;
await hydrateFileParts(ctx.list.messages(), this.config.fileStore, {
threadId: ctx.options?.persistence?.threadId,
});
}
return messages.length > 0;
}
private async prepareLoop(ctx: LoopContext): Promise<{
prepared: PreparedLoopContext;
state: LoopState;
@ -769,28 +805,32 @@ export class AgentRuntime {
runTelemetry,
staticContext,
instructionProviderOptions,
canDiscardRejectedInput: this.canDiscardRejectedInput(ctx),
acceptedInputIds: new Set(
ctx.isFreshRun ? [] : list.inputDelta().map((message) => message.id),
),
},
state,
};
}
private canDiscardRejectedInput(ctx: LoopContext): boolean {
const inputMessages = new Set(ctx.list.inputDelta());
const inputIds = new Set([...inputMessages].map((message) => message.id));
return (
ctx.isFreshRun === true &&
[...inputMessages].some(
(message) =>
'role' in message &&
message.role === 'user' &&
Array.isArray(message.content) &&
message.content.some((part) => part.type === 'file' && part.data !== undefined),
) &&
!ctx.list
.messages()
.some((message) => !inputMessages.has(message) && inputIds.has(message.id))
private getRejectableInputIds(ctx: PreparedLoopContext): string[] {
const inputMessages = new Set(
ctx.list.inputDelta().filter((message) => !ctx.acceptedInputIds.has(message.id)),
);
const inputIds = new Set([...inputMessages].map((message) => message.id));
const hasFiles = [...inputMessages].some(
(message) =>
'role' in message &&
message.role === 'user' &&
Array.isArray(message.content) &&
message.content.some((part) => part.type === 'file' && part.data !== undefined),
);
if (!hasFiles) return [];
if (
ctx.list.messages().some((message) => !inputMessages.has(message) && inputIds.has(message.id))
)
return [];
return [...inputIds];
}
private buildToolBatchContext(
@ -846,9 +886,10 @@ export class AgentRuntime {
): Promise<ToolBatchSettlement<T>> {
this.assertNotAborted(ctx.abortScope);
this.eventBus.emit({ type: AgentEvent.TurnStart });
const { toolMap, modelCallContext } = await this.prepareModelCall(ctx, state.iterationCount);
const { toolMap, modelCallContext } = await this.prepareModelCall(ctx);
const turn = await this.callModelWithRetries(ctx, sink, state, modelCallContext);
this.assertNotAborted(ctx.abortScope);
for (const message of ctx.list.inputDelta()) ctx.acceptedInputIds.add(message.id);
state.lastFinishReason = turn.finishReason;
if (!isReasoningOnlyStop(turn)) ctx.list.addResponse(turn.newMessages);
@ -880,7 +921,7 @@ export class AgentRuntime {
return settlement;
}
private async prepareModelCall(ctx: PreparedLoopContext, iterationCount: number) {
private async prepareModelCall(ctx: PreparedLoopContext) {
const { list, options, abortScope, staticContext } = ctx;
for (const toolName of this.activeSkills?.toolDependencies() ?? []) {
this.deferredToolManager?.load(toolName);
@ -911,19 +952,19 @@ export class AgentRuntime {
outputSpec: staticContext.outputSpec,
maxOutputTokens: staticContext.maxOutputTokens,
aiSdkOptions: this.buildAiSdkOptions(tools.toolMap, options),
onInputRejected: this.createInputRejectionHandler(ctx, iterationCount),
onInputRejected: this.createInputRejectionHandler(ctx),
};
return { toolMap: tools.toolMap, modelCallContext };
}
private createInputRejectionHandler(
ctx: PreparedLoopContext,
iterationCount: number,
): ModelCallContext['onInputRejected'] {
if (!ctx.canDiscardRejectedInput || iterationCount !== 0) return undefined;
const messageIds = this.getRejectableInputIds(ctx);
if (messageIds.length === 0) return undefined;
return async () => {
if (ctx.abortScope.isAborted) return;
await this.memory.discardRejectedInput(ctx.list, ctx.options);
await this.memory.discardRejectedInput(ctx.list, messageIds, ctx.options);
this.updateState({ messageList: ctx.list.serialize() });
};
}

20
packages/@n8n/agents/src/runtime/loop/hydrate-file-parts.ts
View File

@ -3,7 +3,7 @@ import type { AgentDbMessage, ContentFile } from '../../types/sdk/message';
/** Only the newest file parts are hydrated; older ones stay reference-only. */
export const MAX_HYDRATED_FILE_PARTS = 20;
/** Cumulative cap on hydrated bytes per turn. */
/** Cumulative cap on hydrated bytes in the live message list. */
export const MAX_HYDRATED_FILE_BYTES = 50 * 1024 * 1024;
/**
@ -32,25 +32,31 @@ export async function hydrateFileParts(
for (const message of messages) {
if (!('content' in message) || !Array.isArray(message.content)) continue;
for (const block of message.content) {
if (block.type !== 'file' || !block.fileRef || block.data !== undefined) continue;
if (block.type !== 'file' || !block.fileRef) continue;
if (fileStore.isMediaTypeSupported && !fileStore.isMediaTypeSupported(block.mediaType)) {
delete block.data;
continue;
}
blocks.push(block);
}
}
for (const block of blocks.slice(0, -MAX_HYDRATED_FILE_PARTS)) delete block.data;
let budget = MAX_HYDRATED_FILE_BYTES;
const newestFirst = blocks.slice(-MAX_HYDRATED_FILE_PARTS).reverse();
for (const block of newestFirst) {
if (budget <= 0) break;
const existingData = block.data;
delete block.data;
if (budget <= 0) continue;
const declaredSize = block.fileRef!.sizeBytes;
if (declaredSize !== undefined && declaredSize > budget) continue;
if (existingData === undefined && declaredSize !== undefined && declaredSize > budget) continue;
try {
const data = await fileStore.load(block.fileRef!, scope);
if (!data || data.byteLength > budget) continue;
const data = existingData ?? (await fileStore.load(block.fileRef!, scope));
if (!data) continue;
const size = typeof data === 'string' ? Buffer.byteLength(data, 'base64') : data.byteLength;
if (size > budget) continue;
block.data = data;
budget -= data.byteLength;
budget -= size;
} catch {
// Leave the block reference-only; the model sees its text metadata.
}

23
packages/@n8n/agents/src/runtime/loop/stream-sink.ts
View File

@ -1,4 +1,6 @@
import type { StreamTextTransform, TextStreamPart, ToolSet } from 'ai';
import { createDeferredPromise } from '@n8n/utils/promise/deferred-promise';
import { raceWithAbort } from '../../sdk/abort';
import { finalizeRun } from './run-output-sink';
import type {
@ -11,7 +13,7 @@ import type {
} from '../../types/runtime/agent-loop';
import { classifyModelTurnError, mergeUsage } from './runtime-helpers';
import type { ExecutionOptions, TokenUsage } from '../../types/sdk/agent';
import type { AgentMessage } from '../../types/sdk/message';
import type { AgentDbMessage, AgentMessage } from '../../types/sdk/message';
import { isAttachmentValidationError } from '../model/attachment-validation-error';
import { loadAi } from '../model/lazy-ai';
import { fromAiFinishReason, fromAiMessages } from '../model/messages';
@ -53,6 +55,25 @@ const STALL_RETRY_SAFE_CHUNK_TYPES = new Set<string>([
* chunks. Owns the smooth-stream transform option.
*/
export class StreamSink implements RunOutputSink<void> {
async inputBoundary(signal: AbortSignal): Promise<void> {
const acknowledged = createDeferredPromise();
await raceWithAbort(
async () => {
await this.guard.write({
type: 'input-boundary',
acknowledge: () => acknowledged.resolve(),
});
if (this.guard.isClosed) throw new Error('Agent stream closed before input consumption');
await acknowledged.promise;
},
AbortSignal.any([signal, this.guard.closedSignal]),
);
}
async emitInput(message: AgentDbMessage): Promise<void> {
await this.guard.write({ type: 'input', message });
}
private lastUsage: TokenUsage | undefined;
// Reads the in-flight turn's usage from the provider's raw stream events so an
// aborted run can still be billed (the SDK reports no usage on abort). The

4
packages/@n8n/agents/src/runtime/memory/memory-orchestrator.ts
View File

@ -305,10 +305,10 @@ export class MemoryOrchestrator {
async discardRejectedInput(
list: AgentMessageList,
messageIds: string[],
options: (RunOptions & ExecutionOptions) | undefined,
): Promise<void> {
const messageIds = list.inputDelta().map((message) => message.id);
list.removeInput();
list.removeInput(messageIds);
if (!this.config.memory || !options?.persistence || messageIds.length === 0) return;
try {
await this.config.memory.deleteMessages(messageIds);

10
packages/@n8n/agents/src/runtime/model/message-list.ts
View File

@ -426,9 +426,13 @@ export class AgentMessageList {
return this.all.filter((m) => this.inputSet.has(m));
}
removeInput(): void {
this.all = this.all.filter((message) => !this.inputSet.has(message));
this.inputSet.clear();
removeInput(messageIds: readonly string[]): void {
const ids = new Set(messageIds);
this.all = this.all.filter((message) => {
if (!this.inputSet.has(message) || !ids.has(message.id)) return true;
this.inputSet.delete(message);
return false;
});
}
/** All messages currently in the list, as live references. */

22
packages/@n8n/agents/src/runtime/streaming/__tests__/stream-writer-guard.test.ts Normal file
View File

@ -0,0 +1,22 @@
import { describe, expect, it } from 'vitest';
import type { StreamChunk } from '../../../types';
import { StreamWriterGuard } from '../stream-writer-guard';
describe('StreamWriterGuard', () => {
it.each(['close', 'abort'] as const)(
'marks the stream closed after writer.%s()',
async (action) => {
const writer = new WritableStream<StreamChunk>().getWriter();
const guard = new StreamWriterGuard(writer);
expect(guard.isClosed).toBe(false);
expect(guard.closedSignal.aborted).toBe(false);
await Promise.allSettled([writer[action](), writer.closed]);
expect(guard.isClosed).toBe(true);
expect(guard.closedSignal.aborted).toBe(true);
},
);
});

15
packages/@n8n/agents/src/runtime/streaming/stream-writer-guard.ts
View File

@ -11,15 +11,20 @@ import type { TokenUsage } from '../../types/sdk/agent';
*/
export class StreamWriterGuard {
private closed = false;
private readonly closure = new AbortController();
private errorWritten = false;
constructor(private readonly writer: WritableStreamDefaultWriter<StreamChunk>) {
writer.closed
.then(() => {
this.closed = true;
})
.catch(() => {});
const close = () => {
this.closed = true;
this.closure.abort(new Error('Agent stream closed'));
};
void writer.closed.then(close, close);
}
get closedSignal(): AbortSignal {
return this.closure.signal;
}
get isClosed(): boolean {

1
packages/@n8n/agents/src/types/index.ts
View File

@ -46,6 +46,7 @@ export type {
ModelConfig,
RunOptions,
ExecutionOptions,
AgentInputBoundary,
SmoothStreamOptions,
AgentExecutionCounter,
SideCallTask,

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

@ -11,7 +11,7 @@ import type {
import type { FinishReason, SerializableAgentState, TokenUsage } from '../index';
import type { ExecutionOptions, RunOptions } from '../sdk/agent';
import type { AgentMessage } from '../sdk/message';
import type { AgentDbMessage, AgentMessage } from '../sdk/message';
import type { JSONObject } from '../utils/json';
import type { AgentMessageList } from '../../runtime/model/message-list';
import type { ToolCallBatchResult, ToolCallInput, ToolCallSuspension } from './tool-execution';
@ -113,6 +113,9 @@ export interface RunServices {
* `StreamChunk`s through the stream writer guard.
*/
export interface RunOutputSink<TResult> {
/** Wait until the host has recorded preceding output before it commits input. */
inputBoundary?(signal: AbortSignal): Promise<void>;
emitInput?(message: AgentDbMessage): Promise<void>;
/** Run one LLM turn. Streaming implementations also emit text/tool chunks here. */
callModel(ctx: ModelCallContext): Promise<ModelTurnResult>;
/**

14
packages/@n8n/agents/src/types/sdk/agent.ts
View File

@ -7,7 +7,7 @@ import type {
} from 'ai';
import type { JsonSchema7Type } from 'zod-to-json-schema';
import type { AgentMessage, ContentMetadata } from './message';
import type { AgentDbMessage, AgentMessage, ContentMetadata } from './message';
import type { ProviderId, ProviderCredentials } from '../../runtime/model/provider-credentials';
import type {
AgentEvent,
@ -90,6 +90,8 @@ export interface AgentResult {
export type StreamChunk = ContentMetadata &
(
| { type: 'input-boundary'; acknowledge: () => void }
| { type: 'input'; message: AgentDbMessage }
| { type: 'start-step' }
| { type: 'finish-step' }
| { type: 'text-start'; id: string }
@ -207,7 +209,17 @@ export interface SideCallUsageReport {
reportId: string;
}
export interface AgentInputBoundary {
/** Current turn messages, ready for durable storage. */
messages: AgentDbMessage[];
lastCreatedAt: number;
completing: boolean;
canContinue: boolean;
}
export interface ExecutionOptions {
/** Commit additional input between model calls. Stream consumers must acknowledge input-boundary chunks. */
onInputBoundary?: (boundary: AgentInputBoundary) => Promise<AgentDbMessage[]>;
maxIterations?: number;
abortSignal?: AbortSignal;
providerOptions?: ProviderOptions;

57
packages/@n8n/db/src/migrations/common/1790683443934-AddAgentMessageSteering.ts Normal file
View File

@ -0,0 +1,57 @@
import type { MigrationContext, ReversibleMigration } from '../migration-types';
export class AddAgentMessageSteering1790683443934 implements ReversibleMigration {
async up({ schemaBuilder, escape, runQuery, isPostgres }: MigrationContext) {
const { addColumns, addForeignKey, createIndex, column } = schemaBuilder;
const executionTable = escape.tableName('agent_execution');
const acceptsSteering = escape.columnName('acceptsSteering');
// Native ALTER TABLE avoids copying execution history on SQLite.
await runQuery(
`ALTER TABLE ${executionTable} ADD COLUMN ${acceptsSteering} BOOLEAN NOT NULL DEFAULT FALSE`,
);
if (isPostgres) {
await runQuery(
`COMMENT ON COLUMN ${executionTable}.${acceptsSteering} IS 'Accept input until the runtime closes admission'`,
);
}
await addColumns(
'agent_message_queue',
[
column('steeringExecutionId')
.varchar(36)
.comment('Execution reserved to consume this input'),
column('steeringOrder').int.comment('Acceptance order among outstanding steers'),
],
{ recreatesOnSqlite: true },
);
await addForeignKey(
'agent_message_queue',
'steeringExecutionId',
['agent_execution', 'id'],
undefined,
'NO ACTION',
);
await createIndex(
'agent_message_queue',
['steeringExecutionId', 'steeringOrder'],
true,
undefined,
`${escape.columnName('steeringExecutionId')} IS NOT NULL`,
);
}
async down({
schemaBuilder: { dropIndex, dropForeignKey, dropColumns },
escape,
runQuery,
}: MigrationContext) {
await dropIndex('agent_message_queue', ['steeringExecutionId', 'steeringOrder']);
await dropForeignKey('agent_message_queue', 'steeringExecutionId', ['agent_execution', 'id']);
await dropColumns('agent_message_queue', ['steeringExecutionId', 'steeringOrder'], {
recreatesOnSqlite: true,
});
await runQuery(
`ALTER TABLE ${escape.tableName('agent_execution')} DROP COLUMN ${escape.columnName('acceptsSteering')}`,
);
}
}

28
packages/@n8n/db/src/migrations/sqlite/1790683443934-AddAgentMessageSteering.ts Normal file
View File

@ -0,0 +1,28 @@
import { AddAgentMessageSteering1790683443934 as BaseMigration } from '../common/1790683443934-AddAgentMessageSteering';
import type { MigrationContext } from '../migration-types';
export class AddAgentMessageSteering1790683443934 extends BaseMigration {
async up(context: MigrationContext) {
await this.preserveQueueSequence(context, async () => await super.up(context));
}
async down(context: MigrationContext) {
await this.preserveQueueSequence(context, async () => await super.down(context));
}
private async preserveQueueSequence(context: MigrationContext, migrate: () => Promise<void>) {
const name = `${context.tablePrefix}agent_message_queue`;
const [sequence] = await context.runQuery<Array<{ seq: string }>>(
'SELECT CAST("seq" AS TEXT) AS "seq" FROM "sqlite_sequence" WHERE "name" = :name',
{ name },
);
await migrate();
if (!sequence) return;
// Table recreation resets AUTOINCREMENT to the largest retained ID. Keep retired IDs reserved.
await context.runQuery('DELETE FROM "sqlite_sequence" WHERE "name" = :name', { name });
await context.runQuery('INSERT INTO "sqlite_sequence" ("name", "seq") VALUES (:name, :seq)', {
name,
seq: sequence.seq,
});
}
}

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

@ -497,48 +497,69 @@ describe('AgentExecutionService', () => {
}
});
it('serializes timeline snapshot updates', async () => {
const params = await startSnapshotExecution();
let releaseFirstWrite!: () => void;
agentExecutionRepository.updateTimelineIfRunning
.mockImplementationOnce(
async () =>
await new Promise<boolean>((resolve) => {
releaseFirstWrite = () => resolve(true);
}),
)
.mockResolvedValue(true);
const first: TimelineEvent[] = [{ type: 'text', content: 'First', timestamp: 1 }];
const second: TimelineEvent[] = [{ type: 'text', content: 'Second', timestamp: 1 }];
it.each([false, true])(
'drains paused snapshots after an input check (consumed: %s)',
async (consumed) => {
const params = await startSnapshotExecution();
let releaseFirstWrite!: () => void;
agentExecutionRepository.updateTimelineIfRunning
.mockImplementationOnce(
async () =>
await new Promise<boolean>((resolve) => {
releaseFirstWrite = () => resolve(true);
}),
)
.mockResolvedValue(true);
const first: TimelineEvent[] = [{ type: 'text', content: 'First', timestamp: 1 }];
const second: TimelineEvent[] = [{ type: 'text', content: 'Second', timestamp: 1 }];
const withInput: TimelineEvent[] = [
...second,
{
type: 'input',
timestamp: 2,
messageId: 'input-1',
},
];
service.recordTimelineSnapshot({
executionId: 'execution-1',
projectId: 'project-1',
agentId: 'agent-1',
threadId: 'thread-1',
timeline: first,
});
service.recordTimelineSnapshot({
executionId: 'execution-1',
projectId: 'project-1',
agentId: 'agent-1',
threadId: 'thread-1',
timeline: second,
});
await vi.waitFor(() =>
expect(agentExecutionRepository.updateTimelineIfRunning).toHaveBeenCalledTimes(1),
);
releaseFirstWrite();
await vi.waitFor(() =>
expect(agentExecutionRepository.updateTimelineIfRunning).toHaveBeenCalledTimes(2),
);
service.recordTimelineSnapshot({
executionId: 'execution-1',
projectId: 'project-1',
agentId: 'agent-1',
threadId: 'thread-1',
timeline: first,
});
service.recordTimelineSnapshot({
executionId: 'execution-1',
projectId: 'project-1',
agentId: 'agent-1',
threadId: 'thread-1',
timeline: second,
});
await vi.waitFor(() =>
expect(agentExecutionRepository.updateTimelineIfRunning).toHaveBeenCalledTimes(1),
);
const check = service.withTimelineWritesPaused('execution-1', async () => {
if (consumed) {
service.recordTimelineSnapshot({
...params,
executionId: 'execution-1',
timeline: withInput,
});
}
});
releaseFirstWrite();
await check;
await vi.waitFor(() =>
expect(agentExecutionRepository.updateTimelineIfRunning).toHaveBeenCalledTimes(2),
);
expect(agentExecutionRepository.updateTimelineIfRunning).toHaveBeenLastCalledWith(
'execution-1',
second,
);
await service.finalizeExecution('execution-1', { ...params, record: makeMessageRecord() });
});
expect(agentExecutionRepository.updateTimelineIfRunning).toHaveBeenLastCalledWith(
'execution-1',
consumed ? withInput : second,
);
await service.finalizeExecution('execution-1', { ...params, record: makeMessageRecord() });
},
);
it('notifies only after a timeline snapshot retry persists', async () => {
vi.useFakeTimers();

21
packages/cli/src/modules/agents/__tests__/agent-session-langsmith-export.service.test.ts
View File

@ -233,6 +233,27 @@ describe('AgentSessionLangSmithExportService', () => {
);
});
it('exports additional input from its canonical message reference', async () => {
const message = {
id: 'steered',
role: 'user' as const,
content: [{ type: 'text' as const, text: 'Use the newer file' }],
};
const { service } = setupSession(
makeExecution({
inputMessages: [message],
timeline: [{ type: 'input', messageId: message.id, timestamp: 100 }],
}),
);
await service.exportSession(input);
expect(submittedRuns()).toContainEqual(
expect.objectContaining({
name: 'Additional user input',
inputs: { message },
}),
);
});
it('checks the caller for each child before sending a trace', async () => {
const { service, agentExecutionService, threadRepository } = setup();
agentExecutionService.getThreadDetail

33
packages/cli/src/modules/agents/__tests__/execution-recorder.test.ts
View File

@ -291,6 +291,39 @@ describe('ExecutionRecorder', () => {
]);
});
it('keeps committed inputs in order when restored markers repeat', () => {
vi.useFakeTimers();
vi.setSystemTime(1_000);
const recorder = new ExecutionRecorder();
const inputC = {
type: 'input',
messageId: 'c',
timestamp: 2_000,
} satisfies TimelineEvent;
const inputD = {
type: 'input',
messageId: 'd',
timestamp: 3_000,
} satisfies TimelineEvent;
recorder.record({ type: 'text-delta', id: 't1', delta: 'Before C' });
vi.setSystemTime(2_000);
recorder.recordInputs([inputC]);
recorder.record({ type: 'text-delta', id: 't2', delta: 'After C' });
vi.setSystemTime(3_000);
recorder.recordInputs([structuredClone(inputC), inputD]);
recorder.record({ type: 'text-delta', id: 't3', delta: 'After D' });
recorder.record({ type: 'finish', finishReason: 'stop' });
expect(recorder.getMessageRecord().timeline).toEqual([
{ type: 'text', content: 'Before C', timestamp: 1_000, endTime: 2_000 },
inputC,
{ type: 'text', content: 'After C', timestamp: 2_000, endTime: 3_000 },
inputD,
{ type: 'text', content: 'After D', timestamp: 3_000, endTime: 3_000 },
]);
});
it('does not create empty text events for whitespace-only segments', () => {
const recorder = new ExecutionRecorder();

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

@ -152,6 +152,7 @@ export class AgentExecutionService {
>();
private readonly timelineSnapshotWrites = new Map<string, Promise<void>>();
private readonly pausedTimelineSnapshots = new Set<string>();
private readonly executionsNeedingTitleSync = new Set<string>();
@ -367,6 +368,21 @@ export class AgentExecutionService {
this.ensureTimelineSnapshotWrite(executionId);
}
/** Keep older snapshots behind the transaction that commits additional input. */
async withTimelineWritesPaused<T>(executionId: string, commit: () => Promise<T>): Promise<T> {
this.pausedTimelineSnapshots.add(executionId);
try {
await this.timelineSnapshotWrites.get(executionId);
return await commit();
} catch (error) {
this.pendingTimelineSnapshots.delete(executionId);
throw error;
} finally {
this.pausedTimelineSnapshots.delete(executionId);
this.ensureTimelineSnapshotWrite(executionId);
}
}
async finalizeExecution(executionId: string, params: RecordMessageParams): Promise<string> {
this.stopHeartbeat(executionId);
this.pendingTimelineSnapshots.delete(executionId);
@ -512,6 +528,7 @@ export class AgentExecutionService {
private ensureTimelineSnapshotWrite(executionId: string): void {
if (
this.pausedTimelineSnapshots.has(executionId) ||
this.timelineSnapshotWrites.has(executionId) ||
!this.pendingTimelineSnapshots.has(executionId)
) {
@ -525,7 +542,7 @@ export class AgentExecutionService {
}
private async drainTimelineSnapshots(executionId: string): Promise<void> {
while (true) {
while (!this.pausedTimelineSnapshots.has(executionId)) {
const snapshot = this.pendingTimelineSnapshots.get(executionId);
if (!snapshot) return;
this.pendingTimelineSnapshots.delete(executionId);

14
packages/cli/src/modules/agents/agent-session-langsmith-export.service.ts
View File

@ -315,6 +315,20 @@ function buildExecutionRun(
function buildEventRun(event: TimelineEvent, execution: AgentExecution, path: string): DraftRun {
switch (event.type) {
case 'input':
return {
path,
name: 'Additional user input',
runType: 'chain',
startTime: event.timestamp,
endTime: event.timestamp,
inputs: {
message: execution.inputMessages?.find(({ id }) => id === event.messageId),
},
outputs: {},
metadata: {},
children: [],
};
case 'background-task-signal':
return buildBackgroundTaskSignalRun(event, path);
case 'text':

24
packages/cli/src/modules/agents/builder/__tests__/format-preview-context.test.ts
View File

@ -72,10 +72,29 @@ describe('formatPreviewSessionContext', () => {
makeExecution({
id: 'exec-1',
userMessage: 'First question',
inputMessages: [
{
id: 'initial-input',
role: 'user',
content: [{ type: 'text', text: 'First question' }],
},
{
id: 'steered-input',
role: 'user',
content: [
{ type: 'file', fileId: 'file-1', fileName: 'orders.csv', mimeType: 'text/csv' },
],
},
],
timeline: [
{ type: 'reasoning', content: 'I should search the orders.', timestamp: 0, endTime: 1 },
toolCallEvent(),
{ type: 'text', content: 'Here are your orders.', timestamp: 1, endTime: 2 },
{
type: 'input',
timestamp: 3,
messageId: 'steered-input',
},
],
}),
makeExecution({ id: 'exec-2', userMessage: 'Second question' }),
@ -95,6 +114,11 @@ describe('formatPreviewSessionContext', () => {
expect(block).toContain('Input: {"query":"open orders"}');
expect(block).toContain('Output: {"count":2}');
expect(block).toContain('Assistant: Here are your orders.');
expect(block!.match(/User: First question/g)).toHaveLength(1);
expect(block!.match(/User: \[Attachment: orders\.csv\]/g)).toHaveLength(1);
expect(block!.indexOf('User: [Attachment: orders.csv]')).toBeGreaterThan(
block!.indexOf('Assistant: Here are your orders.'),
);
});
it('scopes a single turn to the anchor plus trailing resume continuations', () => {

31
packages/cli/src/modules/agents/builder/format-preview-context.ts
View File

@ -1,3 +1,5 @@
import type { AgentPersistedMessageDto } from '@n8n/api-types';
import type { AgentExecutionThread } from '../entities/agent-execution-thread.entity';
import type { AgentExecution } from '../entities/agent-execution.entity';
import type { TimelineEvent } from '../execution-recorder';
@ -33,7 +35,22 @@ function stringifyToolValue(value: unknown): string {
return truncate(serialized, MAX_TOOL_VALUE_CHARS);
}
function formatTimelineEvent(event: TimelineEvent): string {
function formatInput(message: AgentPersistedMessageDto): string {
const text = message.content
.flatMap((part) => {
if (part.type === 'text') return [part.text];
if (part.type === 'file') return [`[Attachment: ${part.fileName ?? part.fileId ?? 'file'}]`];
return [];
})
.join('\n');
return `User: ${truncate(text, MAX_TEXT_CHARS)}`;
}
function formatTimelineEvent(event: TimelineEvent, execution: AgentExecution): string {
if (event.type === 'input') {
const message = execution.inputMessages?.find(({ id }) => id === event.messageId);
return message ? formatInput(message) : '';
}
if (event.type === 'background-task-signal') {
return `Background task results received: ${stringifyToolValue(event.signal.tasks)}`;
}
@ -73,11 +90,19 @@ function formatExecution(execution: AgentExecution): string {
if (execution.error) headerParts.push(`error=${execution.error}`);
const lines = [`## Turn (${headerParts.join(', ')})`];
if (execution.userMessage !== null) {
if (execution.inputMessages !== undefined) {
const steeredIds = new Set(
execution.timeline?.filter((event) => event.type === 'input').map((event) => event.messageId),
);
for (const message of execution.inputMessages) {
if (!steeredIds.has(message.id)) lines.push(formatInput(message));
}
} else if (execution.userMessage !== null) {
lines.push(`User: ${truncate(execution.userMessage, MAX_TEXT_CHARS)}`);
}
for (const event of execution.timeline ?? []) {
lines.push(formatTimelineEvent(event));
const line = formatTimelineEvent(event, execution);
if (line) lines.push(line);
}
return lines.join('\n');
}

7
packages/cli/src/modules/agents/entities/agent-execution.entity.ts
View File

@ -43,6 +43,13 @@ export class AgentExecution extends WithTimestampsAndStringId {
@Column({ type: 'varchar', length: 16 })
status: AgentExecutionStatus;
@Column({
type: 'boolean',
default: false,
comment: 'Accept input until the runtime closes admission',
})
acceptsSteering: boolean;
@DateTimeColumn({ precision: 3, nullable: true })
startedAt: Date | null;

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

@ -18,6 +18,10 @@ import type { AgentQueueDispatch } from '../types/agent-queued-message';
@Entity({ name: 'agent_message_queue' })
@Index(['threadId', 'id'])
@Index(['threadId'], { unique: true, where: '"executionId" IS NOT NULL' })
@Index(['steeringExecutionId', 'steeringOrder'], {
unique: true,
where: '"steeringExecutionId" IS NOT NULL',
})
export class AgentMessageQueue extends WithTimestamps {
@Generated()
@PrimaryColumn({
@ -63,4 +67,19 @@ export class AgentMessageQueue extends WithTimestamps {
comment: 'Current execution; NULL means pending',
})
executionId: string | null;
@ManyToOne(() => AgentExecution, { nullable: true, onDelete: 'NO ACTION' })
@JoinColumn({ name: 'steeringExecutionId' })
steeringExecution: Relation<AgentExecution> | null;
@Column({
type: 'varchar',
length: 36,
nullable: true,
comment: 'Execution reserved to consume this input',
})
steeringExecutionId: string | null;
@Column({ type: 'int', nullable: true, comment: 'Acceptance order among outstanding steers' })
steeringOrder: number | null;
}

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

@ -262,6 +262,7 @@ export interface RecordedUsage {
}
export type TimelineEvent =
| { type: 'input'; messageId: string; timestamp: number }
| { type: 'background-task-signal'; signal: AgentBackgroundJobSignal; timestamp: number }
| { type: 'text'; content: string; timestamp: number; endTime?: number }
| { type: 'reasoning'; content: string; timestamp: number; endTime?: number }
@ -383,6 +384,17 @@ export class ExecutionRecorder {
private childTraceChars = new Map<string, number>();
/** Record additional input only after its transaction commits. */
recordInputs(events: Array<Extract<TimelineEvent, { type: 'input' }>>): void {
this.flushReasoningBuffer();
this.flushTextBuffer();
for (const event of events) {
if (this.timeline.some((item) => item.type === 'input' && item.messageId === event.messageId))
continue;
this.appendCompletedEvent(event);
}
}
/** Record the human response that caused a suspended tool call to resume. */
recordHitlResponse(toolCallId: string, response: unknown): void {
this.flushReasoningBuffer();

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

@ -77,14 +77,40 @@ export class AgentExecutionRepository extends BaseRepository<AgentExecution> {
async updateTimelineIfRunning(
executionId: string,
timeline: AgentExecution['timeline'],
ctx: OperationContext = {},
): Promise<boolean> {
const result = await this.update({ id: executionId, status: 'running' }, {
timeline,
updatedAt: new Date(),
} as QueryDeepPartialEntity<AgentExecution>);
const result = await this.managerFor(ctx).update(
AgentExecution,
{ id: executionId, status: 'running' },
{
timeline,
updatedAt: new Date(),
} as QueryDeepPartialEntity<AgentExecution>,
);
return result.affected === 1;
}
async findSteerable(threadId: string, ctx: OperationContext = {}) {
return await this.managerFor(ctx).findOne(AgentExecution, {
select: ['id'],
where: {
threadId,
status: 'running',
acceptsSteering: true,
},
});
}
async closeSteering(threadId: string, executionId: string, ctx: OperationContext) {
await this.managerFor(ctx).update(
AgentExecution,
{ id: executionId, threadId, acceptsSteering: true },
{
acceptsSteering: false,
},
);
}
async updateIfRunning(
executionId: string,
values: AgentExecutionFinalizationValues,
@ -92,11 +118,26 @@ export class AgentExecutionRepository extends BaseRepository<AgentExecution> {
ctx: OperationContext = {},
costIncrement?: number,
): Promise<boolean> {
const current = await this.managerFor(ctx).findOne(AgentExecution, {
select: ['timeline'],
where: { id: executionId, status: 'running' },
});
const inputIds = new Set(
values.timeline?.filter((event) => event.type === 'input').map((event) => event.messageId),
);
// A failed commit acknowledgement must not let terminal recording erase durable input.
if (
current?.timeline?.some((event) => event.type === 'input' && !inputIds.has(event.messageId))
)
return false;
// Build the SET values once. `cost` is never set as a literal here —
// the only way to move cost is the additive `costIncrement` fragment
// below, which preserves any in-flight side-call `incrementCost` calls
// (`COALESCE(cost, 0) + :costIncrement` instead of `cost = :value`).
const setValues = { ...values } as QueryDeepPartialEntity<AgentExecution>;
const setValues = {
...values,
acceptsSteering: false,
} as QueryDeepPartialEntity<AgentExecution>;
const params: Record<string, unknown> = { executionId, status: 'running' };
if (costIncrement !== undefined && costIncrement > 0) {
setValues.cost = () => 'COALESCE(cost, 0) + :costIncrement';

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

@ -1,6 +1,6 @@
import { BaseRepository, TransactionRunner, type OperationContext } from '@n8n/db';
import { Service } from '@n8n/di';
import { DataSource, IsNull, Not } from '@n8n/typeorm';
import { DataSource, In, IsNull, Not } from '@n8n/typeorm';
import { isDraftIntegration } from '@n8n/api-types';
import { AgentMessageQueue } from '../entities/agent-message-queue.entity';
@ -21,7 +21,14 @@ export class AgentMessageQueueRepository extends BaseRepository<AgentMessageQueu
) {
const repository = this.managerFor(ctx).getRepository(AgentMessageQueue);
return await repository.save(
repository.create({ threadId, messageId, payload, executionId: null }),
repository.create({
threadId,
messageId,
payload,
executionId: null,
steeringExecutionId: null,
steeringOrder: null,
}),
);
}
@ -45,6 +52,7 @@ export class AgentMessageQueueRepository extends BaseRepository<AgentMessageQueu
threadId,
id,
executionId: IsNull(),
steeringExecutionId: IsNull(),
});
return result.affected === 1;
}
@ -57,6 +65,63 @@ export class AgentMessageQueueRepository extends BaseRepository<AgentMessageQueu
});
}
async reserveSteering(threadId: string, id: string, executionId: string, ctx: OperationContext) {
const last = await this.managerFor(ctx).findOne(AgentMessageQueue, {
where: { threadId, steeringExecutionId: executionId },
order: { steeringOrder: 'DESC' },
});
const result = await this.managerFor(ctx).update(
AgentMessageQueue,
{ threadId, id, executionId: IsNull(), steeringExecutionId: IsNull() },
{ steeringExecutionId: executionId, steeringOrder: (last?.steeringOrder ?? 0) + 1 },
);
return result.affected === 1;
}
async findSteering(threadId: string, executionId: string, ctx: OperationContext) {
return await this.managerFor(ctx).find(AgentMessageQueue, {
where: { threadId, executionId: IsNull(), steeringExecutionId: executionId },
relations: { message: true },
order: { steeringOrder: 'ASC' },
});
}
async findSteeringExecutionIds(threadId: string, ctx: OperationContext): Promise<string[]> {
const items = await this.managerFor(ctx).find(AgentMessageQueue, {
select: ['steeringExecutionId'],
where: { threadId, steeringExecutionId: Not(IsNull()) },
});
return [
...new Set(
items.flatMap((item) => (item.steeringExecutionId ? [item.steeringExecutionId] : [])),
),
];
}
async releaseSteering(threadId: string, executionId: string, ctx: OperationContext) {
const result = await this.managerFor(ctx).update(
AgentMessageQueue,
{ threadId, steeringExecutionId: executionId },
{ steeringExecutionId: null, steeringOrder: null },
);
return (result.affected ?? 0) > 0;
}
async consumeSteering(
threadId: string,
executionId: string,
ids: string[],
ctx: OperationContext,
) {
if (ids.length === 0) return;
await this.managerFor(ctx).delete(AgentMessageQueue, {
threadId,
steeringExecutionId: executionId,
executionId: IsNull(),
id: In(ids),
});
}
async findActive(threadId: string, ctx: OperationContext) {
return await this.managerFor(ctx).findOneBy(AgentMessageQueue, {
threadId,

12
packages/cli/src/modules/agents/repositories/agent-message.repository.ts
View File

@ -92,6 +92,7 @@ export class AgentMessageRepository extends BaseRepository<AgentMessageEntity> {
return await manager.findOneByOrFail(AgentMessageEntity, { id: message.id });
}
/** The caller holds the session lock so concurrent input cannot take the same position. */
async linkExecutionInput(
executionId: string,
messageId: string,
@ -108,11 +109,17 @@ export class AgentMessageRepository extends BaseRepository<AgentMessageEntity> {
) {
throw new UnexpectedError('The input message does not belong to this session and resource');
}
const inputs = await manager.find(AgentExecutionMessageLink, {
select: ['messageId', 'position'],
where: { executionId, direction: 'input' },
order: { position: 'DESC' },
});
if (inputs.some((input) => input.messageId === messageId)) return;
await manager.insert(AgentExecutionMessageLink, {
executionId,
messageId,
direction: 'input',
position: 0,
position: (inputs[0]?.position ?? -1) + 1,
});
}
@ -183,9 +190,10 @@ export class AgentMessageRepository extends BaseRepository<AgentMessageEntity> {
messages: AgentDbMessage[];
executionId?: string;
},
ctx: OperationContext = {},
): Promise<void> {
if (params.messages.length === 0) return;
await this.runInTransaction({}, async (manager) => {
await this.runInTransaction(ctx, async (manager) => {
const now = new Date();
if (params.executionId) {
const ownership = await manager.update(

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

@ -2116,7 +2116,7 @@ describe('AgentExecutionRepository', () => {
expect(await memory.getMessages('builder-without-session')).toHaveLength(1);
});
it('keeps output membership and positions on repeated saves and hides internal input', async () => {
it('keeps input and output positions on repeated saves and hides internal input', async () => {
const threadId = uuid();
const resourceId = 'user-1';
const memory = Container.get(N8nMemory).getImplementation(agentId);
@ -2133,6 +2133,23 @@ describe('AgentExecutionRepository', () => {
};
const admission = await executionService.startExecutionRecording(params, new Date());
const [inputId] = admission.inputMessageIds;
const additional = await messageRepository.createInput(
{
threadId,
resourceId,
content: { role: 'user', content: [{ type: 'text', text: 'Additional internal result' }] },
origin: { source: null, hidden: true },
},
{},
);
for (const id of [additional.id, inputId, additional.id]) {
await messageRepository.linkExecutionInput(
admission.executionId,
id,
{ threadId, resourceId },
{},
);
}
const input = {
...buildInboundUserMessage(params.userMessage, [])[0],
id: inputId,
@ -2157,7 +2174,16 @@ describe('AgentExecutionRepository', () => {
hostMetadata: { [EXECUTION_METADATA_KEY]: admission.executionId },
};
await memory.saveMessages({ ...scope, messages: [input, output] });
await memory.saveMessages({ ...scope, messages: [later, latest, output, input] });
await memory.saveMessages({
...scope,
messages: [
later,
latest,
output,
input,
{ ...additional.content, id: additional.id, createdAt: additional.createdAt },
],
});
expect((await messageRepository.findOneByOrFail({ id: inputId })).modelContent).toBeNull();
const links = await repository.manager.find(AgentExecutionMessageLink, {
where: { executionId: admission.executionId },
@ -2167,13 +2193,14 @@ describe('AgentExecutionRepository', () => {
links.map(({ messageId, direction, position }) => ({ messageId, direction, position })),
).toEqual([
{ messageId: inputId, direction: 'input', position: 0 },
{ messageId: additional.id, direction: 'input', position: 1 },
{ messageId: output.id, direction: 'output', position: 0 },
{ messageId: later.id, direction: 'output', position: 1 },
{ messageId: latest.id, direction: 'output', position: 2 },
]);
const detail = await executionService.getThreadDetail(threadId, projectId, agentId, viewerId);
expect(detail?.executions[0]).toMatchObject({
inputMessageIds: [inputId],
inputMessageIds: [inputId, additional.id],
inputMessages: [],
userMessage: null,
});

166
packages/cli/test/migration/1790683443934-add-agent-message-steering.test.ts Normal file
View File

@ -0,0 +1,166 @@
import {
createTestMigrationContext,
initDbUpToMigration,
runSingleMigration,
} from '@n8n/backend-test-utils';
import { DbConnection } from '@n8n/db';
import { Container } from '@n8n/di';
import { DataSource } from '@n8n/typeorm';
import { randomUUID } from 'node:crypto';
const MIGRATION = 'AddAgentMessageSteering1790683443934';
describe('AddAgentMessageSteering migration', () => {
let dataSource: DataSource;
const projectId = randomUUID();
const agentId = randomUUID();
const threadId = randomUUID();
const executionId = randomUUID();
const steeringExecutionId = randomUUID();
const messageIds = [randomUUID(), randomUUID(), randomUUID()];
beforeAll(async () => {
await Container.get(DbConnection).init();
dataSource = Container.get(DataSource);
const ctx = createTestMigrationContext(dataSource);
await ctx.queryRunner.clearDatabase();
await ctx.queryRunner.release();
await initDbUpToMigration(MIGRATION);
const seed = createTestMigrationContext(dataSource);
const values = { projectId, agentId, threadId, executionId, now: new Date() };
try {
await seed.runQuery(
`INSERT INTO ${seed.escape.tableName('project')} ("id", "name", "type", "createdAt", "updatedAt") VALUES (:projectId, 'Project', 'team', :now, :now)`,
values,
);
await seed.runQuery(
`INSERT INTO ${seed.escape.tableName('agents')} ("id", "name", "projectId", "integrations", "tools", "skills") VALUES (:agentId, 'Agent', :projectId, '[]', '{}', '{}')`,
values,
);
await seed.runQuery(
`INSERT INTO ${seed.escape.tableName('agent_execution_threads')} ("id", "agentId", "agentName", "projectId") VALUES (:threadId, :agentId, 'Agent', :projectId)`,
values,
);
await seed.runQuery(
`INSERT INTO ${seed.escape.tableName('agent_execution')} ("id", "threadId", "status") VALUES (:executionId, :threadId, 'running')`,
values,
);
await seed.runQuery(
`INSERT INTO ${seed.escape.tableName('agent_execution')} ("id", "threadId", "status") VALUES (:steeringExecutionId, :threadId, 'running')`,
{ steeringExecutionId, threadId },
);
await seed.runQuery(
`INSERT INTO ${seed.escape.tableName('agents_resources')} ("id") VALUES (:threadId)`,
values,
);
await seed.runQuery(
`INSERT INTO ${seed.escape.tableName('agents_threads')} ("id", "resourceId") VALUES (:threadId, :threadId)`,
values,
);
for (const id of messageIds) {
await seed.runQuery(
`INSERT INTO ${seed.escape.tableName('agents_messages')} ("id", "threadId", "resourceId", "role", "content") VALUES (:id, :threadId, :threadId, 'user', :content)`,
{
id,
threadId,
content: JSON.stringify({ role: 'user', content: [{ type: 'text', text: 'Input' }] }),
},
);
}
await seed.runQuery(
`INSERT INTO ${seed.escape.tableName('agent_execution_message_links')} ("executionId", "messageId", "direction", "position") VALUES (:executionId, :messageId, 'input', 0)`,
{ executionId, messageId: messageIds[0] },
);
await seed.runQuery(
`INSERT INTO ${seed.escape.tableName('agent_message_queue')} ("threadId", "messageId", "payload", "executionId") VALUES (:threadId, :messageId, '{"kind":"preview"}', :executionId)`,
{ ...values, messageId: messageIds[0] },
);
await seed.runQuery(
`INSERT INTO ${seed.escape.tableName('agent_message_queue')} ("threadId", "messageId", "payload") VALUES (:threadId, :messageId, '{"kind":"preview"}')`,
{ ...values, messageId: messageIds[1] },
);
// Retired IDs must stay retired when SQLite rebuilds the table.
await seed.runQuery(
`DELETE FROM ${seed.escape.tableName('agent_message_queue')} WHERE "executionId" IS NULL`,
);
} finally {
await seed.queryRunner.release();
}
});
afterAll(async () => {
await Container.get(DbConnection).close();
});
it('keeps existing rows, foreign keys, and increasing queue IDs through up, down, and up', async () => {
await runSingleMigration(MIGRATION);
await dataSource.undoLastMigration({ transaction: 'none' });
await runSingleMigration(MIGRATION);
const ctx = createTestMigrationContext(dataSource);
try {
const executions = await ctx.runQuery<Array<{ acceptsSteering: boolean | number }>>(
`SELECT "acceptsSteering" FROM ${ctx.escape.tableName('agent_execution')} WHERE "id" = :executionId`,
{ executionId },
);
expect(Boolean(executions[0].acceptsSteering)).toBe(false);
const rows = await ctx.runQuery<
Array<{ executionId: string; steeringExecutionId: string | null }>
>(
`SELECT "executionId", "steeringExecutionId" FROM ${ctx.escape.tableName('agent_message_queue')}`,
);
expect(rows).toEqual([{ executionId, steeringExecutionId: null }]);
expect(
await ctx.runQuery(
`SELECT "messageId" FROM ${ctx.escape.tableName('agent_execution_message_links')}`,
),
).toEqual([{ messageId: messageIds[0] }]);
expect(
await ctx.runQuery(`SELECT "id" FROM ${ctx.escape.tableName('agents_messages')}`),
).toHaveLength(3);
await expect(
ctx.runQuery(
`INSERT INTO ${ctx.escape.tableName('agent_message_queue')} ("threadId", "messageId", "payload") VALUES (:threadId, :messageId, '{"kind":"preview"}')`,
{ threadId, messageId: messageIds[0] },
),
).rejects.toThrow();
await ctx.runQuery(
`INSERT INTO ${ctx.escape.tableName('agent_message_queue')} ("threadId", "messageId", "payload", "steeringExecutionId", "steeringOrder") VALUES (:threadId, :messageId, '{"kind":"preview"}', :steeringExecutionId, 1)`,
{ threadId, steeringExecutionId, messageId: messageIds[1] },
);
const ids = await ctx.runQuery<Array<{ id: string | number }>>(
`SELECT "id" FROM ${ctx.escape.tableName('agent_message_queue')} ORDER BY "id"`,
);
expect(Number(ids[1].id)).toBeGreaterThan(2);
await expect(
ctx.runQuery(
`INSERT INTO ${ctx.escape.tableName('agent_message_queue')} ("threadId", "messageId", "payload", "steeringExecutionId", "steeringOrder") VALUES (:threadId, :messageId, '{"kind":"preview"}', :steeringExecutionId, 1)`,
{ threadId, steeringExecutionId, messageId: messageIds[2] },
),
).rejects.toThrow();
await expect(
ctx.runQuery(
`DELETE FROM ${ctx.escape.tableName('agent_execution')} WHERE "id" = :steeringExecutionId`,
{ steeringExecutionId },
),
).rejects.toThrow();
await ctx.runQuery(
`DELETE FROM ${ctx.escape.tableName('agents_messages')} WHERE "id" = :messageId`,
{ messageId: messageIds[0] },
);
expect(
await ctx.runQuery(
`SELECT "messageId" FROM ${ctx.escape.tableName('agent_message_queue')}`,
),
).toEqual([{ messageId: messageIds[1] }]);
await ctx.runQuery(
`DELETE FROM ${ctx.escape.tableName('agent_execution_threads')} WHERE "id" = :threadId`,
{ threadId },
);
expect(
await ctx.runQuery(`SELECT "id" FROM ${ctx.escape.tableName('agent_message_queue')}`),
).toEqual([]);
} finally {
await ctx.queryRunner.release();
}
});
});