From b09fd0dc69d730a67385588b7f9bef1f838197a1 Mon Sep 17 00:00:00 2001
From: Robin Braumann <50590409+bjorger@users.noreply.github.com>
Date: Tue, 29 Sep 2026 12:41:34 +0000
Subject: [PATCH] feat(core): Add agent input boundaries and steering storage
(no-changelog) (#39529)
---
docs/generated/postgres-schema/README.md | 8 +-
.../postgres-schema/public.agent_execution.md | 6 +
.../public.agent_execution_message_links.md | 1 +
.../public.agent_execution_threads.md | 3 +
.../public.agent_message_queue.md | 8 +
.../postgres-schema/public.agents_messages.md | 2 +
docs/generated/sqlite-schema/README.md | 8 +-
.../sqlite-schema/agent_execution.md | 7 +-
.../agent_execution_message_links.md | 1 +
.../sqlite-schema/agent_execution_threads.md | 3 +
.../sqlite-schema/agent_message_queue.md | 16 +-
.../sqlite-schema/agents_messages.md | 2 +
packages/@n8n/agents/src/index.ts | 1 +
.../runtime/__tests__/agent-runtime.test.ts | 369 ++++++++++++++++--
.../src/runtime/__tests__/file-parts.test.ts | 21 +-
.../memory-orchestrator-persist-input.test.ts | 24 +-
.../agents/src/runtime/loop/agent-runtime.ts | 89 +++--
.../src/runtime/loop/hydrate-file-parts.ts | 20 +-
.../agents/src/runtime/loop/stream-sink.ts | 23 +-
.../src/runtime/memory/memory-orchestrator.ts | 4 +-
.../agents/src/runtime/model/message-list.ts | 10 +-
.../__tests__/stream-writer-guard.test.ts | 22 ++
.../runtime/streaming/stream-writer-guard.ts | 15 +-
packages/@n8n/agents/src/types/index.ts | 1 +
.../agents/src/types/runtime/agent-loop.ts | 5 +-
packages/@n8n/agents/src/types/sdk/agent.ts | 14 +-
.../1790683443934-AddAgentMessageSteering.ts | 57 +++
.../1790683443934-AddAgentMessageSteering.ts | 28 ++
.../__tests__/agent-execution.service.test.ts | 101 +++--
...t-session-langsmith-export.service.test.ts | 21 +
.../__tests__/execution-recorder.test.ts | 33 ++
.../modules/agents/agent-execution.service.ts | 19 +-
.../agent-session-langsmith-export.service.ts | 14 +
.../__tests__/format-preview-context.test.ts | 24 ++
.../agents/builder/format-preview-context.ts | 31 +-
.../agents/entities/agent-execution.entity.ts | 7 +
.../entities/agent-message-queue.entity.ts | 19 +
.../src/modules/agents/execution-recorder.ts | 12 +
.../agent-execution.repository.ts | 51 ++-
.../agent-message-queue.repository.ts | 69 +++-
.../repositories/agent-message.repository.ts | 12 +-
.../agent-execution.repository.test.ts | 33 +-
...3443934-add-agent-message-steering.test.ts | 166 ++++++++
43 files changed, 1220 insertions(+), 160 deletions(-)
create mode 100644 packages/@n8n/agents/src/runtime/streaming/__tests__/stream-writer-guard.test.ts
create mode 100644 packages/@n8n/db/src/migrations/common/1790683443934-AddAgentMessageSteering.ts
create mode 100644 packages/@n8n/db/src/migrations/sqlite/1790683443934-AddAgentMessageSteering.ts
create mode 100644 packages/cli/test/migration/1790683443934-add-agent-message-steering.test.ts
diff --git a/docs/generated/postgres-schema/README.md b/docs/generated/postgres-schema/README.md
index aa40c04c359..2d191e89431 100644
--- a/docs/generated/postgres-schema/README.md
+++ b/docs/generated/postgres-schema/README.md
@@ -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
}
diff --git a/docs/generated/postgres-schema/public.agent_execution.md b/docs/generated/postgres-schema/public.agent_execution.md
index 0b5a5bd3bc3..7e34d217382 100644
--- a/docs/generated/postgres-schema/public.agent_execution.md
+++ b/docs/generated/postgres-schema/public.agent_execution.md
@@ -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
}
diff --git a/docs/generated/postgres-schema/public.agent_execution_message_links.md b/docs/generated/postgres-schema/public.agent_execution_message_links.md
index fc563f5b028..eff53fc8d56 100644
--- a/docs/generated/postgres-schema/public.agent_execution_message_links.md
+++ b/docs/generated/postgres-schema/public.agent_execution_message_links.md
@@ -48,6 +48,7 @@ erDiagram
integer position
}
"public.agent_execution" {
+ boolean acceptsSteering
json attachments
json author
integer completionTokens
diff --git a/docs/generated/postgres-schema/public.agent_execution_threads.md b/docs/generated/postgres-schema/public.agent_execution_threads.md
index 08046a75c2c..07e8b12b2c5 100644
--- a/docs/generated/postgres-schema/public.agent_execution_threads.md
+++ b/docs/generated/postgres-schema/public.agent_execution_threads.md
@@ -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
}
diff --git a/docs/generated/postgres-schema/public.agent_message_queue.md b/docs/generated/postgres-schema/public.agent_message_queue.md
index 9f6064e82c1..b7ef79e5217 100644
--- a/docs/generated/postgres-schema/public.agent_message_queue.md
+++ b/docs/generated/postgres-schema/public.agent_message_queue.md
@@ -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
diff --git a/docs/generated/postgres-schema/public.agents_messages.md b/docs/generated/postgres-schema/public.agents_messages.md
index 7741ec2b276..e32db1edfec 100644
--- a/docs/generated/postgres-schema/public.agents_messages.md
+++ b/docs/generated/postgres-schema/public.agents_messages.md
@@ -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
}
diff --git a/docs/generated/sqlite-schema/README.md b/docs/generated/sqlite-schema/README.md
index dd118cbc112..37b341a3886 100644
--- a/docs/generated/sqlite-schema/README.md
+++ b/docs/generated/sqlite-schema/README.md
@@ -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
}
diff --git a/docs/generated/sqlite-schema/agent_execution.md b/docs/generated/sqlite-schema/agent_execution.md
index ee08257d4d9..be4a76accce 100644
--- a/docs/generated/sqlite-schema/agent_execution.md
+++ b/docs/generated/sqlite-schema/agent_execution.md
@@ -6,7 +6,7 @@
Table Definition
```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)
```
@@ -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
}
diff --git a/docs/generated/sqlite-schema/agent_execution_message_links.md b/docs/generated/sqlite-schema/agent_execution_message_links.md
index 8be37cce2e2..1c4b7e04013 100644
--- a/docs/generated/sqlite-schema/agent_execution_message_links.md
+++ b/docs/generated/sqlite-schema/agent_execution_message_links.md
@@ -56,6 +56,7 @@ erDiagram
INTEGER position
}
"agent_execution" {
+ BOOLEAN acceptsSteering
TEXT attachments
TEXT author
INTEGER completionTokens
diff --git a/docs/generated/sqlite-schema/agent_execution_threads.md b/docs/generated/sqlite-schema/agent_execution_threads.md
index c31f00dfcc2..7cc1f6a6361 100644
--- a/docs/generated/sqlite-schema/agent_execution_threads.md
+++ b/docs/generated/sqlite-schema/agent_execution_threads.md
@@ -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
}
diff --git a/docs/generated/sqlite-schema/agent_message_queue.md b/docs/generated/sqlite-schema/agent_message_queue.md
index e283d081cdc..53112839dbc 100644
--- a/docs/generated/sqlite-schema/agent_message_queue.md
+++ b/docs/generated/sqlite-schema/agent_message_queue.md
@@ -6,7 +6,7 @@
Table Definition
```sql
-CREATE TABLE "agent_message_queue" ("id" integer PRIMARY KEY AUTOINCREMENT NOT NULL, "threadId" varchar(128) NOT NULL, "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)
```
@@ -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
diff --git a/docs/generated/sqlite-schema/agents_messages.md b/docs/generated/sqlite-schema/agents_messages.md
index b55b3676787..5c116ce4a3d 100644
--- a/docs/generated/sqlite-schema/agents_messages.md
+++ b/docs/generated/sqlite-schema/agents_messages.md
@@ -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
}
diff --git a/packages/@n8n/agents/src/index.ts b/packages/@n8n/agents/src/index.ts
index 09019c5c08e..a7b3e9cf85a 100644
--- a/packages/@n8n/agents/src/index.ts
+++ b/packages/@n8n/agents/src/index.ts
@@ -76,6 +76,7 @@ export type {
McpVerifyResult,
ModelConfig,
ExecutionOptions,
+ AgentInputBoundary,
SmoothStreamOptions,
TokenUsage,
AgentExecutionCounter,
diff --git a/packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts b/packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts
index 00128c327ce..779fd1a0ea8 100644
--- a/packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts
+++ b/packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts
@@ -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): Promise {
runtime: AgentRuntime,
mode: 'generate' | 'stream',
messages: Message[] | string = [structuredClone(input)],
- abortSignal?: AbortSignal,
+ options: Pick = {},
) {
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();
diff --git a/packages/@n8n/agents/src/runtime/__tests__/file-parts.test.ts b/packages/@n8n/agents/src/runtime/__tests__/file-parts.test.ts
index 334c123b73c..cf0d74e5e2a 100644
--- a/packages/@n8n/agents/src/runtime/__tests__/file-parts.test.ts
+++ b/packages/@n8n/agents/src/runtime/__tests__/file-parts.test.ts
@@ -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({
diff --git a/packages/@n8n/agents/src/runtime/__tests__/memory-orchestrator-persist-input.test.ts b/packages/@n8n/agents/src/runtime/__tests__/memory-orchestrator-persist-input.test.ts
index 9187b783b9b..43a8f0b765f 100644
--- a/packages/@n8n/agents/src/runtime/__tests__/memory-orchestrator-persist-input.test.ts
+++ b/packages/@n8n/agents/src/runtime/__tests__/memory-orchestrator-persist-input.test.ts
@@ -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]);
});
});
diff --git a/packages/@n8n/agents/src/runtime/loop/agent-runtime.ts b/packages/@n8n/agents/src/runtime/loop/agent-runtime.ts
index 46cbbd9c834..cea39498640 100644
--- a/packages/@n8n/agents/src/runtime/loop/agent-runtime.ts
+++ b/packages/@n8n/agents/src/runtime/loop/agent-runtime.ts
@@ -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;
}
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(
+ ctx: PreparedLoopContext,
+ sink: RunOutputSink,
+ state: LoopState,
+ ): Promise {
+ 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> {
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() });
};
}
diff --git a/packages/@n8n/agents/src/runtime/loop/hydrate-file-parts.ts b/packages/@n8n/agents/src/runtime/loop/hydrate-file-parts.ts
index 44ec4b152cd..081600a5cbc 100644
--- a/packages/@n8n/agents/src/runtime/loop/hydrate-file-parts.ts
+++ b/packages/@n8n/agents/src/runtime/loop/hydrate-file-parts.ts
@@ -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.
}
diff --git a/packages/@n8n/agents/src/runtime/loop/stream-sink.ts b/packages/@n8n/agents/src/runtime/loop/stream-sink.ts
index 4d633fe04a2..ffef0313b46 100644
--- a/packages/@n8n/agents/src/runtime/loop/stream-sink.ts
+++ b/packages/@n8n/agents/src/runtime/loop/stream-sink.ts
@@ -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([
* chunks. Owns the smooth-stream transform option.
*/
export class StreamSink implements RunOutputSink {
+ async inputBoundary(signal: AbortSignal): Promise {
+ 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 {
+ 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
diff --git a/packages/@n8n/agents/src/runtime/memory/memory-orchestrator.ts b/packages/@n8n/agents/src/runtime/memory/memory-orchestrator.ts
index e1ba832b4ea..b9eca602920 100644
--- a/packages/@n8n/agents/src/runtime/memory/memory-orchestrator.ts
+++ b/packages/@n8n/agents/src/runtime/memory/memory-orchestrator.ts
@@ -305,10 +305,10 @@ export class MemoryOrchestrator {
async discardRejectedInput(
list: AgentMessageList,
+ messageIds: string[],
options: (RunOptions & ExecutionOptions) | undefined,
): Promise {
- 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);
diff --git a/packages/@n8n/agents/src/runtime/model/message-list.ts b/packages/@n8n/agents/src/runtime/model/message-list.ts
index 59b448a3346..093651e8c59 100644
--- a/packages/@n8n/agents/src/runtime/model/message-list.ts
+++ b/packages/@n8n/agents/src/runtime/model/message-list.ts
@@ -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. */
diff --git a/packages/@n8n/agents/src/runtime/streaming/__tests__/stream-writer-guard.test.ts b/packages/@n8n/agents/src/runtime/streaming/__tests__/stream-writer-guard.test.ts
new file mode 100644
index 00000000000..dfc9d477b32
--- /dev/null
+++ b/packages/@n8n/agents/src/runtime/streaming/__tests__/stream-writer-guard.test.ts
@@ -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().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);
+ },
+ );
+});
diff --git a/packages/@n8n/agents/src/runtime/streaming/stream-writer-guard.ts b/packages/@n8n/agents/src/runtime/streaming/stream-writer-guard.ts
index 54907389083..7353805ec67 100644
--- a/packages/@n8n/agents/src/runtime/streaming/stream-writer-guard.ts
+++ b/packages/@n8n/agents/src/runtime/streaming/stream-writer-guard.ts
@@ -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) {
- 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 {
diff --git a/packages/@n8n/agents/src/types/index.ts b/packages/@n8n/agents/src/types/index.ts
index 320ca7a24b7..575561eb1db 100644
--- a/packages/@n8n/agents/src/types/index.ts
+++ b/packages/@n8n/agents/src/types/index.ts
@@ -46,6 +46,7 @@ export type {
ModelConfig,
RunOptions,
ExecutionOptions,
+ AgentInputBoundary,
SmoothStreamOptions,
AgentExecutionCounter,
SideCallTask,
diff --git a/packages/@n8n/agents/src/types/runtime/agent-loop.ts b/packages/@n8n/agents/src/types/runtime/agent-loop.ts
index fdc85340232..4d28eadc57b 100644
--- a/packages/@n8n/agents/src/types/runtime/agent-loop.ts
+++ b/packages/@n8n/agents/src/types/runtime/agent-loop.ts
@@ -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 {
+ /** Wait until the host has recorded preceding output before it commits input. */
+ inputBoundary?(signal: AbortSignal): Promise;
+ emitInput?(message: AgentDbMessage): Promise;
/** Run one LLM turn. Streaming implementations also emit text/tool chunks here. */
callModel(ctx: ModelCallContext): Promise;
/**
diff --git a/packages/@n8n/agents/src/types/sdk/agent.ts b/packages/@n8n/agents/src/types/sdk/agent.ts
index 1392b32e850..5944f2fe7fb 100644
--- a/packages/@n8n/agents/src/types/sdk/agent.ts
+++ b/packages/@n8n/agents/src/types/sdk/agent.ts
@@ -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;
maxIterations?: number;
abortSignal?: AbortSignal;
providerOptions?: ProviderOptions;
diff --git a/packages/@n8n/db/src/migrations/common/1790683443934-AddAgentMessageSteering.ts b/packages/@n8n/db/src/migrations/common/1790683443934-AddAgentMessageSteering.ts
new file mode 100644
index 00000000000..f190e79df33
--- /dev/null
+++ b/packages/@n8n/db/src/migrations/common/1790683443934-AddAgentMessageSteering.ts
@@ -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')}`,
+ );
+ }
+}
diff --git a/packages/@n8n/db/src/migrations/sqlite/1790683443934-AddAgentMessageSteering.ts b/packages/@n8n/db/src/migrations/sqlite/1790683443934-AddAgentMessageSteering.ts
new file mode 100644
index 00000000000..58fcfe598d8
--- /dev/null
+++ b/packages/@n8n/db/src/migrations/sqlite/1790683443934-AddAgentMessageSteering.ts
@@ -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) {
+ const name = `${context.tablePrefix}agent_message_queue`;
+ const [sequence] = await context.runQuery>(
+ '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,
+ });
+ }
+}
diff --git a/packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts
index 639fc4cc775..cb83624f8f5 100644
--- a/packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts
+++ b/packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts
@@ -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((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((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();
diff --git a/packages/cli/src/modules/agents/__tests__/agent-session-langsmith-export.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-session-langsmith-export.service.test.ts
index 996dce23016..891844b2b4e 100644
--- a/packages/cli/src/modules/agents/__tests__/agent-session-langsmith-export.service.test.ts
+++ b/packages/cli/src/modules/agents/__tests__/agent-session-langsmith-export.service.test.ts
@@ -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
diff --git a/packages/cli/src/modules/agents/__tests__/execution-recorder.test.ts b/packages/cli/src/modules/agents/__tests__/execution-recorder.test.ts
index 23c87e288c5..cc22244277a 100644
--- a/packages/cli/src/modules/agents/__tests__/execution-recorder.test.ts
+++ b/packages/cli/src/modules/agents/__tests__/execution-recorder.test.ts
@@ -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();
diff --git a/packages/cli/src/modules/agents/agent-execution.service.ts b/packages/cli/src/modules/agents/agent-execution.service.ts
index 6d8385695e7..b895c807513 100644
--- a/packages/cli/src/modules/agents/agent-execution.service.ts
+++ b/packages/cli/src/modules/agents/agent-execution.service.ts
@@ -152,6 +152,7 @@ export class AgentExecutionService {
>();
private readonly timelineSnapshotWrites = new Map>();
+ private readonly pausedTimelineSnapshots = new Set();
private readonly executionsNeedingTitleSync = new Set();
@@ -367,6 +368,21 @@ export class AgentExecutionService {
this.ensureTimelineSnapshotWrite(executionId);
}
+ /** Keep older snapshots behind the transaction that commits additional input. */
+ async withTimelineWritesPaused(executionId: string, commit: () => Promise): Promise {
+ 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 {
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 {
- while (true) {
+ while (!this.pausedTimelineSnapshots.has(executionId)) {
const snapshot = this.pendingTimelineSnapshots.get(executionId);
if (!snapshot) return;
this.pendingTimelineSnapshots.delete(executionId);
diff --git a/packages/cli/src/modules/agents/agent-session-langsmith-export.service.ts b/packages/cli/src/modules/agents/agent-session-langsmith-export.service.ts
index ff4f6b57920..a7636003153 100644
--- a/packages/cli/src/modules/agents/agent-session-langsmith-export.service.ts
+++ b/packages/cli/src/modules/agents/agent-session-langsmith-export.service.ts
@@ -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':
diff --git a/packages/cli/src/modules/agents/builder/__tests__/format-preview-context.test.ts b/packages/cli/src/modules/agents/builder/__tests__/format-preview-context.test.ts
index d2f6bda3b43..0e9b8bfaa13 100644
--- a/packages/cli/src/modules/agents/builder/__tests__/format-preview-context.test.ts
+++ b/packages/cli/src/modules/agents/builder/__tests__/format-preview-context.test.ts
@@ -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', () => {
diff --git a/packages/cli/src/modules/agents/builder/format-preview-context.ts b/packages/cli/src/modules/agents/builder/format-preview-context.ts
index 905c3495f2e..18fd83861a4 100644
--- a/packages/cli/src/modules/agents/builder/format-preview-context.ts
+++ b/packages/cli/src/modules/agents/builder/format-preview-context.ts
@@ -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');
}
diff --git a/packages/cli/src/modules/agents/entities/agent-execution.entity.ts b/packages/cli/src/modules/agents/entities/agent-execution.entity.ts
index fcaf3cc4c17..60adc628356 100644
--- a/packages/cli/src/modules/agents/entities/agent-execution.entity.ts
+++ b/packages/cli/src/modules/agents/entities/agent-execution.entity.ts
@@ -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;
diff --git a/packages/cli/src/modules/agents/entities/agent-message-queue.entity.ts b/packages/cli/src/modules/agents/entities/agent-message-queue.entity.ts
index 4444fe11b96..fce57672f88 100644
--- a/packages/cli/src/modules/agents/entities/agent-message-queue.entity.ts
+++ b/packages/cli/src/modules/agents/entities/agent-message-queue.entity.ts
@@ -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 | 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;
}
diff --git a/packages/cli/src/modules/agents/execution-recorder.ts b/packages/cli/src/modules/agents/execution-recorder.ts
index 59bd5e7b09e..975660d598c 100644
--- a/packages/cli/src/modules/agents/execution-recorder.ts
+++ b/packages/cli/src/modules/agents/execution-recorder.ts
@@ -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();
+ /** Record additional input only after its transaction commits. */
+ recordInputs(events: Array>): 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();
diff --git a/packages/cli/src/modules/agents/repositories/agent-execution.repository.ts b/packages/cli/src/modules/agents/repositories/agent-execution.repository.ts
index 228cb757861..310bf36cb00 100644
--- a/packages/cli/src/modules/agents/repositories/agent-execution.repository.ts
+++ b/packages/cli/src/modules/agents/repositories/agent-execution.repository.ts
@@ -77,14 +77,40 @@ export class AgentExecutionRepository extends BaseRepository {
async updateTimelineIfRunning(
executionId: string,
timeline: AgentExecution['timeline'],
+ ctx: OperationContext = {},
): Promise {
- const result = await this.update({ id: executionId, status: 'running' }, {
- timeline,
- updatedAt: new Date(),
- } as QueryDeepPartialEntity);
+ const result = await this.managerFor(ctx).update(
+ AgentExecution,
+ { id: executionId, status: 'running' },
+ {
+ timeline,
+ updatedAt: new Date(),
+ } as QueryDeepPartialEntity,
+ );
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 {
ctx: OperationContext = {},
costIncrement?: number,
): Promise {
+ 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;
+ const setValues = {
+ ...values,
+ acceptsSteering: false,
+ } as QueryDeepPartialEntity;
const params: Record = { executionId, status: 'running' };
if (costIncrement !== undefined && costIncrement > 0) {
setValues.cost = () => 'COALESCE(cost, 0) + :costIncrement';
diff --git a/packages/cli/src/modules/agents/repositories/agent-message-queue.repository.ts b/packages/cli/src/modules/agents/repositories/agent-message-queue.repository.ts
index e18ab059c8f..ef31a7ad420 100644
--- a/packages/cli/src/modules/agents/repositories/agent-message-queue.repository.ts
+++ b/packages/cli/src/modules/agents/repositories/agent-message-queue.repository.ts
@@ -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 {
+ 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,
diff --git a/packages/cli/src/modules/agents/repositories/agent-message.repository.ts b/packages/cli/src/modules/agents/repositories/agent-message.repository.ts
index 4a0c419c2c1..4ea74d04b46 100644
--- a/packages/cli/src/modules/agents/repositories/agent-message.repository.ts
+++ b/packages/cli/src/modules/agents/repositories/agent-message.repository.ts
@@ -92,6 +92,7 @@ export class AgentMessageRepository extends BaseRepository {
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 {
) {
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 {
messages: AgentDbMessage[];
executionId?: string;
},
+ ctx: OperationContext = {},
): Promise {
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(
diff --git a/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts b/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts
index ff0c6cee58f..2d8352c4747 100644
--- a/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts
+++ b/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts
@@ -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,
});
diff --git a/packages/cli/test/migration/1790683443934-add-agent-message-steering.test.ts b/packages/cli/test/migration/1790683443934-add-agent-message-steering.test.ts
new file mode 100644
index 00000000000..e63ee5c71db
--- /dev/null
+++ b/packages/cli/test/migration/1790683443934-add-agent-message-steering.test.ts
@@ -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>(
+ `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>(
+ `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();
+ }
+ });
+});