From 33505d7faade510f270f032a6f7aa44744e3ab8d Mon Sep 17 00:00:00 2001 From: Robin Braumann <50590409+bjorger@users.noreply.github.com> Date: Thu, 1 Oct 2026 15:44:14 +0000 Subject: [PATCH] feat(editor): Allow reordering queued messages in Agent Preview (#40101) --- docs/generated/postgres-schema/README.md | 3 +- .../postgres-schema/public.agent_execution.md | 1 + .../public.agent_execution_threads.md | 1 + .../public.agent_message_queue.md | 5 +- .../postgres-schema/public.agents_messages.md | 1 + docs/generated/sqlite-schema/README.md | 3 +- .../sqlite-schema/agent_execution.md | 1 + .../sqlite-schema/agent_execution_threads.md | 1 + .../sqlite-schema/agent_message_queue.md | 6 +- .../sqlite-schema/agents_messages.md | 1 + packages/@n8n/api-types/src/agents/dto.ts | 5 + ...0853539575-AddAgentMessageQueuePosition.ts | 74 +++++ .../modules/agents/PREVIEW_CHAT_RECOVERY.md | 8 +- .../__tests__/agent-chat.controller.test.ts | 6 + .../modules/agents/agent-chat.controller.ts | 24 ++ .../agent-message-queue-consumer.service.ts | 2 +- .../agents/agent-message-queue.service.ts | 31 +- .../agents/agent-message-steering.service.ts | 2 +- .../entities/agent-message-queue.entity.ts | 5 +- .../agent-message-queue.repository.ts | 43 ++- .../agent-thread-grant.repository.test.ts | 9 +- .../agent-execution.repository.test.ts | 61 ++++ ...5-add-agent-message-queue-position.test.ts | 151 +++++++++ .../frontend/@n8n/i18n/src/locales/en.json | 4 + .../agents/__tests__/AgentChatPanel.test.ts | 97 +++++- .../__tests__/useAgentChatStream.test.ts | 306 +++++++++++++++++- .../features/agents/agents-design-language.md | 11 +- .../agents/components/AgentChatPanel.vue | 193 +++++++++-- .../agents/composables/useAgentApi.ts | 17 + .../agents/composables/useAgentChatStream.ts | 96 +++++- .../playwright/services/agent-api-helper.ts | 14 + .../agents/message-queue.spec.ts | 83 ++++- 32 files changed, 1202 insertions(+), 63 deletions(-) create mode 100644 packages/@n8n/db/src/migrations/common/1790853539575-AddAgentMessageQueuePosition.ts create mode 100644 packages/cli/test/migration/1790853539575-add-agent-message-queue-position.test.ts diff --git a/docs/generated/postgres-schema/README.md b/docs/generated/postgres-schema/README.md index 9573ba62389..0d76e183f01 100644 --- a/docs/generated/postgres-schema/README.md +++ b/docs/generated/postgres-schema/README.md @@ -24,7 +24,7 @@ Auto-generated from the PostgreSQL migrations in @n8n/db. Do not edit by hand. | [public.agent_execution_threads](public.agent_execution_threads.md) | 19 | | BASE TABLE | | [public.agent_files](public.agent_files.md) | 10 | | BASE TABLE | | [public.agent_history](public.agent_history.md) | 9 | | BASE TABLE | -| [public.agent_message_queue](public.agent_message_queue.md) | 9 | | BASE TABLE | +| [public.agent_message_queue](public.agent_message_queue.md) | 10 | | 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 | @@ -618,6 +618,7 @@ erDiagram bigint id varchar_36_ messageId FK json payload + integer position varchar_36_ steeringExecutionId FK integer steeringOrder varchar_128_ threadId FK diff --git a/docs/generated/postgres-schema/public.agent_execution.md b/docs/generated/postgres-schema/public.agent_execution.md index 7e34d217382..7094f46732d 100644 --- a/docs/generated/postgres-schema/public.agent_execution.md +++ b/docs/generated/postgres-schema/public.agent_execution.md @@ -102,6 +102,7 @@ erDiagram bigint id varchar_36_ messageId FK json payload + integer position varchar_36_ steeringExecutionId FK integer steeringOrder varchar_128_ threadId FK diff --git a/docs/generated/postgres-schema/public.agent_execution_threads.md b/docs/generated/postgres-schema/public.agent_execution_threads.md index 2c2979f6e2a..81d1978e3a7 100644 --- a/docs/generated/postgres-schema/public.agent_execution_threads.md +++ b/docs/generated/postgres-schema/public.agent_execution_threads.md @@ -138,6 +138,7 @@ erDiagram bigint id varchar_36_ messageId FK json payload + integer position varchar_36_ steeringExecutionId FK integer steeringOrder varchar_128_ threadId FK diff --git a/docs/generated/postgres-schema/public.agent_message_queue.md b/docs/generated/postgres-schema/public.agent_message_queue.md index b7ef79e5217..6f1b58ac1e5 100644 --- a/docs/generated/postgres-schema/public.agent_message_queue.md +++ b/docs/generated/postgres-schema/public.agent_message_queue.md @@ -9,6 +9,7 @@ | 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 | +| position | integer | 0 | false | | | Pending turn order within the session | | 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) | | @@ -27,6 +28,7 @@ | agent_message_queue_id_not_null | n | NOT NULL id | | agent_message_queue_messageId_not_null | n | NOT NULL "messageId" | | agent_message_queue_payload_not_null | n | NOT NULL payload | +| agent_message_queue_position_not_null | n | NOT NULL "position" | | agent_message_queue_threadId_not_null | n | NOT NULL "threadId" | | agent_message_queue_updatedAt_not_null | n | NOT NULL "updatedAt" | @@ -38,7 +40,7 @@ | 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) | +| IDX_agent_message_queue_threadId_position | CREATE INDEX "IDX_agent_message_queue_threadId_position" ON public.agent_message_queue USING btree ("threadId", "position") | | PK_733d8c959a6057f04f4da5ab721 | CREATE UNIQUE INDEX "PK_733d8c959a6057f04f4da5ab721" ON public.agent_message_queue USING btree (id) | ## Relations @@ -57,6 +59,7 @@ erDiagram bigint id varchar_36_ messageId FK json payload + integer position varchar_36_ steeringExecutionId FK integer steeringOrder varchar_128_ threadId FK diff --git a/docs/generated/postgres-schema/public.agents_messages.md b/docs/generated/postgres-schema/public.agents_messages.md index e32db1edfec..ce663172b14 100644 --- a/docs/generated/postgres-schema/public.agents_messages.md +++ b/docs/generated/postgres-schema/public.agents_messages.md @@ -78,6 +78,7 @@ erDiagram bigint id varchar_36_ messageId FK json payload + integer position varchar_36_ steeringExecutionId FK integer steeringOrder varchar_128_ threadId FK diff --git a/docs/generated/sqlite-schema/README.md b/docs/generated/sqlite-schema/README.md index 3fe85efd3c7..3d46305a2d1 100644 --- a/docs/generated/sqlite-schema/README.md +++ b/docs/generated/sqlite-schema/README.md @@ -24,7 +24,7 @@ Auto-generated from the SQLite migrations in @n8n/db. Do not edit by hand. | [agent_execution_threads](agent_execution_threads.md) | 19 | | table | | [agent_files](agent_files.md) | 10 | | table | | [agent_history](agent_history.md) | 9 | | table | -| [agent_message_queue](agent_message_queue.md) | 9 | | table | +| [agent_message_queue](agent_message_queue.md) | 10 | | 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 | @@ -604,6 +604,7 @@ erDiagram INTEGER id varchar_36_ messageId FK TEXT payload + INTEGER position varchar_36_ steeringExecutionId FK INTEGER steeringOrder varchar_128_ threadId FK diff --git a/docs/generated/sqlite-schema/agent_execution.md b/docs/generated/sqlite-schema/agent_execution.md index be4a76accce..da5bd99afc3 100644 --- a/docs/generated/sqlite-schema/agent_execution.md +++ b/docs/generated/sqlite-schema/agent_execution.md @@ -106,6 +106,7 @@ erDiagram INTEGER id varchar_36_ messageId FK TEXT payload + INTEGER position varchar_36_ steeringExecutionId FK INTEGER steeringOrder varchar_128_ threadId FK diff --git a/docs/generated/sqlite-schema/agent_execution_threads.md b/docs/generated/sqlite-schema/agent_execution_threads.md index 2cb5710ee90..b4c3e3d5881 100644 --- a/docs/generated/sqlite-schema/agent_execution_threads.md +++ b/docs/generated/sqlite-schema/agent_execution_threads.md @@ -138,6 +138,7 @@ erDiagram INTEGER id varchar_36_ messageId FK TEXT payload + INTEGER position varchar_36_ steeringExecutionId FK INTEGER steeringOrder varchar_128_ threadId FK diff --git a/docs/generated/sqlite-schema/agent_message_queue.md b/docs/generated/sqlite-schema/agent_message_queue.md index 53112839dbc..b739c067eea 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, "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) +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, "position" integer NOT NULL DEFAULT 0, 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,7 @@ 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 | | | | +| position | INTEGER | 0 | false | | | | | steeringExecutionId | varchar(36) | | true | | [agent_execution](agent_execution.md) | | | steeringOrder | INTEGER | | true | | | | | threadId | varchar(128) | | false | | [agent_execution_threads](agent_execution_threads.md) | | @@ -43,7 +44,7 @@ CREATE TABLE "agent_message_queue" ("id" integer PRIMARY KEY AUTOINCREMENT NOT N | 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") | +| IDX_agent_message_queue_threadId_position | CREATE INDEX "IDX_agent_message_queue_threadId_position" ON "agent_message_queue" ("threadId", "position") | ## Relations @@ -61,6 +62,7 @@ erDiagram INTEGER id varchar_36_ messageId FK TEXT payload + INTEGER position varchar_36_ steeringExecutionId FK INTEGER steeringOrder varchar_128_ threadId FK diff --git a/docs/generated/sqlite-schema/agents_messages.md b/docs/generated/sqlite-schema/agents_messages.md index 5c116ce4a3d..18e06229630 100644 --- a/docs/generated/sqlite-schema/agents_messages.md +++ b/docs/generated/sqlite-schema/agents_messages.md @@ -83,6 +83,7 @@ erDiagram INTEGER id varchar_36_ messageId FK TEXT payload + INTEGER position varchar_36_ steeringExecutionId FK INTEGER steeringOrder varchar_128_ threadId FK diff --git a/packages/@n8n/api-types/src/agents/dto.ts b/packages/@n8n/api-types/src/agents/dto.ts index 29214502dda..67567f5ec80 100644 --- a/packages/@n8n/api-types/src/agents/dto.ts +++ b/packages/@n8n/api-types/src/agents/dto.ts @@ -277,6 +277,11 @@ export class AgentChatQueueSteerDto extends Z.class({ executionId: z.string().min(1).max(36), }) {} +export class AgentChatQueueReorderDto extends Z.class({ + targetQueueId: z.string().regex(/^[1-9]\d*$/), + expectedQueueIds: z.array(z.string().regex(/^[1-9]\d*$/)).min(2), +}) {} + export class AgentChatResumeDto extends Z.class({ runId: z.string().min(1), toolCallId: z.string().min(1), diff --git a/packages/@n8n/db/src/migrations/common/1790853539575-AddAgentMessageQueuePosition.ts b/packages/@n8n/db/src/migrations/common/1790853539575-AddAgentMessageQueuePosition.ts new file mode 100644 index 00000000000..50bfc54eecf --- /dev/null +++ b/packages/@n8n/db/src/migrations/common/1790853539575-AddAgentMessageQueuePosition.ts @@ -0,0 +1,74 @@ +import type { MigrationContext, ReversibleMigration } from '../migration-types'; + +export class AddAgentMessageQueuePosition1790853539575 implements ReversibleMigration { + async up({ + schemaBuilder, + escape, + runQuery, + isSqlite, + queryRunner, + tablePrefix, + }: MigrationContext) { + const { addColumns, column, createIndex, dropIndex } = schemaBuilder; + const table = escape.tableName('agent_message_queue'); + const id = escape.columnName('id'); + const threadId = escape.columnName('threadId'); + const position = escape.columnName('position'); + if (isSqlite) { + // Keep the AUTOINCREMENT sequence, including IDs of deleted rows. + await runQuery(`ALTER TABLE ${table} ADD COLUMN ${position} integer NOT NULL DEFAULT 0`); + } else { + await addColumns( + 'agent_message_queue', + [ + column('position') + .int.notNull.default(0) + .comment('Pending turn order within the session'), + ], + { recreatesOnSqlite: true }, + ); + } + await runQuery(` + UPDATE ${table} AS queue SET ${position} = ordered.${position} + FROM ( + SELECT ${id}, ROW_NUMBER() OVER (PARTITION BY ${threadId} ORDER BY ${id}) - 1 AS ${position} + FROM ${table} + ) AS ordered + WHERE queue.${id} = ordered.${id} + `); + await dropIndex('agent_message_queue', ['threadId', 'id'], { + customIndexName: queryRunner.connection.namingStrategy.indexName( + `${tablePrefix}agent_message_queue`, + ['threadId', 'id'], + ), + }); + await createIndex('agent_message_queue', ['threadId', 'position']); + } + + async down({ + schemaBuilder: { dropIndex, dropColumns, createIndex }, + escape, + runQuery, + isSqlite, + queryRunner, + tablePrefix, + }: MigrationContext) { + await dropIndex('agent_message_queue', ['threadId', 'position']); + if (isSqlite) { + await runQuery( + `ALTER TABLE ${escape.tableName('agent_message_queue')} DROP COLUMN ${escape.columnName('position')}`, + ); + } else { + await dropColumns('agent_message_queue', ['position'], { recreatesOnSqlite: true }); + } + await createIndex( + 'agent_message_queue', + ['threadId', 'id'], + false, + queryRunner.connection.namingStrategy.indexName(`${tablePrefix}agent_message_queue`, [ + 'threadId', + 'id', + ]), + ); + } +} diff --git a/packages/cli/src/modules/agents/PREVIEW_CHAT_RECOVERY.md b/packages/cli/src/modules/agents/PREVIEW_CHAT_RECOVERY.md index bfcf99a23d7..f3aa886e740 100644 --- a/packages/cli/src/modules/agents/PREVIEW_CHAT_RECOVERY.md +++ b/packages/cli/src/modules/agents/PREVIEW_CHAT_RECOVERY.md @@ -12,9 +12,11 @@ Stop uses `DELETE /projects/:projectId/agents/v2/:agentId/chat/:threadId/executi Recovery guarantees the latest retained timeline, followed by newer snapshots. Updates can arrive in batches. Snapshots do not contain original text-stream IDs or every transient live event. Child text and reasoning share the existing 4,000-character limit for each delegation. Text beyond that limit is not part of recovery. -`GET /projects/:projectId/agents/v2/:agentId/chat/:threadId/queue` restores pending messages in FIFO order. It requires `agent:read` and the Preview session owner. `DELETE` on the same path with `/:queueId` requires `agent:execute`. Removal locks the session and rejects an item that has started. Successful removal discards its attachments. Queue changes use `agentMessageQueueUpdated` notifications across mains. Execution updates also refresh the queue. Neither notification replays stream events. +`GET /projects/:projectId/agents/v2/:agentId/chat/:threadId/queue` restores pending messages in their saved order. It requires `agent:read` and the Preview session owner. `DELETE` on the same path with `/:queueId` requires `agent:execute`. Removal locks the session and rejects an item that has started. Successful removal discards its attachments. Queue changes use `agentMessageQueueUpdated` notifications across mains. Execution updates also refresh the queue. Neither notification replays stream events. -Accepted input survives backend process failure through the durable queue. Interrupted executions are not replayed. Valid approval checkpoints retain their place. The existing Agents enablement controls the feature. This contract does not include token replay, sender deduplication, or queue reordering. +Accepted input survives backend process failure through the durable queue. Interrupted executions are not replayed. Valid approval checkpoints retain their place. The existing Agents enablement controls the feature. This contract does not include token replay or sender deduplication. + +Users can drag future turns to reorder them. `POST /projects/:projectId/agents/v2/:agentId/chat/:threadId/queue/:queueId/reorder` takes `{ targetQueueId, expectedQueueIds }`. The expected IDs list pending messages in their current order. It excludes accepted steers. Under the session lock, the endpoint checks this order and moves the named message to the target message's position. It shifts the messages between them in the same transaction. It requires `agent:execute` and the Preview session owner. Both messages must still be pending and unreserved for steering. A changed queue returns 409. The client refreshes the queue after success or failure. Reordering preserves message IDs, content, attachments, and the current execution. New messages join the end. Saved positions survive refresh and recovery. Users can edit the text of a pending message. Edits retain its attachments and queue position. The update uses the session lock and fails if the consumer has claimed the item. Queue notifications refresh the edited text across mains. The execution-start event carries the accepted text so waiting streams do not show the original draft. If a message starts during editing, the editor keeps the unsaved text available to copy until the user dismisses it. @@ -24,4 +26,4 @@ A reserved message stays in the queue panel until the current model request and After consumption, `message-steered` carries the queue ID, execution ID, and persisted user message. The active execution stream relays this event across mains. The client removes the row, detaches its waiting stream, and inserts the user message between assistant segments. It keeps the execution and Stop target. Repeated events and late acceptance events cannot restore the row. Reload resolves the input markers through the execution's message links. The existing limit of two waiting streams remains in effect. -Stop closes steering admission before it signals cancellation. Failure, cancellation, suspension, and interrupted-execution recovery return unconsumed messages to their original FIFO positions. Approval checkpoints still block ordinary processing. A committed input remains in its execution after a crash and is never requeued. Preview continuations use the same boundary. Integrations and sub-agents do not accept steering. +Stop closes steering admission before it signals cancellation. Failure, cancellation, suspension, and interrupted-execution recovery return unconsumed messages to their saved queue positions. Approval checkpoints still block ordinary processing. A committed input remains in its execution after a crash and is never requeued. Preview continuations use the same boundary. Integrations and sub-agents do not accept steering. diff --git a/packages/cli/src/modules/agents/__tests__/agent-chat.controller.test.ts b/packages/cli/src/modules/agents/__tests__/agent-chat.controller.test.ts index 8f554293638..c580661ac07 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-chat.controller.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-chat.controller.test.ts @@ -174,6 +174,7 @@ describe('AgentChatController route access scopes', () => { ['getQueuedMessages', 'agent:read'], ['removeQueuedMessage', 'agent:execute'], ['updateQueuedMessage', 'agent:execute'], + ['reorderQueuedMessage', 'agent:execute'], ['steerQueuedMessage', 'agent:execute'], ['getBackgroundJobs', 'agent:read'], ['getTestChatMessages', 'agent:read'], @@ -186,6 +187,11 @@ describe('AgentChatController route access scopes', () => { describe('AgentChatController queue mutations', () => { it.each([ ['updateQueuedMessage', 'updatePending', { message: 'Edited message' }], + [ + 'reorderQueuedMessage', + 'reorderPending', + { targetQueueId: '2', expectedQueueIds: ['1', '2'] }, + ], ['steerQueuedMessage', 'steer', { executionId: 'execution-1' }], ] as const)( '%s reads the body after the request and response arguments', diff --git a/packages/cli/src/modules/agents/agent-chat.controller.ts b/packages/cli/src/modules/agents/agent-chat.controller.ts index 5382a48391b..31c8df2e5b9 100644 --- a/packages/cli/src/modules/agents/agent-chat.controller.ts +++ b/packages/cli/src/modules/agents/agent-chat.controller.ts @@ -4,6 +4,7 @@ import { AgentChatMessageDto, AgentChatQueueUpdateDto, AgentChatQueueSteerDto, + AgentChatQueueReorderDto, type AgentChatMessagesResponse, type AgentChatQueueResponse, AgentChatResumeDto, @@ -651,6 +652,29 @@ export class AgentChatController { }); } + @Post('/:agentId/chat/:threadId/queue/:queueId/reorder') + @ProjectScope('agent:execute') + async reorderQueuedMessage( + req: AuthenticatedRequest<{ + projectId: string; + agentId: string; + threadId: string; + queueId: string; + }>, + _res: Response, + @Body payload: AgentChatQueueReorderDto, + ): Promise { + if (!/^[1-9]\d*$/.test(req.params.queueId)) throw new BadRequestError('Invalid queue ID'); + const agent = await this.agentsService.findById(req.params.agentId, req.params.projectId); + if (!agent) throw new NotFoundError('Agent not found'); + await this.messageQueue.reorderPending({ + ...req.params, + userId: req.user.id, + targetQueueId: payload.targetQueueId, + expectedQueueIds: payload.expectedQueueIds, + }); + } + @Delete('/:agentId/chat/:threadId/queue/:queueId') @ProjectScope('agent:execute') async removeQueuedMessage( diff --git a/packages/cli/src/modules/agents/agent-message-queue-consumer.service.ts b/packages/cli/src/modules/agents/agent-message-queue-consumer.service.ts index ccba5b2c232..0c55a2292c9 100644 --- a/packages/cli/src/modules/agents/agent-message-queue-consumer.service.ts +++ b/packages/cli/src/modules/agents/agent-message-queue-consumer.service.ts @@ -82,7 +82,7 @@ export class AgentMessageQueueConsumer { .finally(() => this.activeThreads.delete(threadId)); } - /** Run messages in FIFO order until the session is empty, blocked, or unavailable on this main. */ + /** Run messages in queue order until the session is empty, blocked, or unavailable on this main. */ private async drain(threadId: string): Promise { while (!this.stopped) { if (!(await this.consumeNext(threadId))) return; diff --git a/packages/cli/src/modules/agents/agent-message-queue.service.ts b/packages/cli/src/modules/agents/agent-message-queue.service.ts index 731518e4f1c..670f913582c 100644 --- a/packages/cli/src/modules/agents/agent-message-queue.service.ts +++ b/packages/cli/src/modules/agents/agent-message-queue.service.ts @@ -177,7 +177,7 @@ export class AgentMessageQueueService { steerableExecutionId: steerable?.id ?? null, items: items .filter((item) => item.payload.kind === 'preview') - // Show the next inputs first. Keep unreserved messages in their original FIFO order. + // Show accepted steers first. Keep future turns in their saved queue order. .sort( (a, b) => (a.steeringOrder ?? Number.MAX_SAFE_INTEGER) - @@ -251,6 +251,33 @@ export class AgentMessageQueueService { this.updates.notifyQueueUpdated(input.threadId); } + async reorderPending(input: { + projectId: string; + agentId: string; + threadId: string; + userId: string; + queueId: string; + targetQueueId: string; + expectedQueueIds: string[]; + }): Promise { + await this.txRunner.run({}, async (ctx) => { + const thread = await this.threadRepository.lockById(input.threadId, ctx); + if (!thread) throw new NotFoundError('Session not found'); + await this.assertPreviewAccess(thread, input, ctx); + const moved = await this.repository.movePending( + thread.id, + input.queueId, + input.targetQueueId, + input.expectedQueueIds, + ctx, + ); + if (!moved) { + throw new ConflictError('The queue has changed. Refresh it and try again.'); + } + }); + this.updates.notifyQueueUpdated(input.threadId); + } + async steer(input: { projectId: string; agentId: string; @@ -297,7 +324,7 @@ export class AgentMessageQueueService { } /** - * Give the oldest pending message exclusive use of the session for its execution. + * Give the next pending message exclusive use of the session for its execution. * Running work and valid suspended checkpoints block a claim. */ async claimNext( diff --git a/packages/cli/src/modules/agents/agent-message-steering.service.ts b/packages/cli/src/modules/agents/agent-message-steering.service.ts index 8aac0f4973d..be6ecef28d7 100644 --- a/packages/cli/src/modules/agents/agent-message-steering.service.ts +++ b/packages/cli/src/modules/agents/agent-message-steering.service.ts @@ -54,7 +54,7 @@ export class AgentMessageSteeringService { this.updates.notifyQueueUpdated(threadId); } - /** Return unconsumed input to its original FIFO position. The caller holds the session lock. */ + /** Return unconsumed input to its saved queue position. The caller holds the session lock. */ async release(threadId: string, executionId: string, ctx: OperationContext) { await this.executions.closeSteering(threadId, executionId, ctx); return await this.queue.releaseSteering(threadId, executionId, ctx); 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 fce57672f88..af7f34b54c7 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 @@ -16,7 +16,7 @@ import { AgentMessageEntity } from './agent-message.entity'; import type { AgentQueueDispatch } from '../types/agent-queued-message'; @Entity({ name: 'agent_message_queue' }) -@Index(['threadId', 'id']) +@Index(['threadId', 'position']) @Index(['threadId'], { unique: true, where: '"executionId" IS NOT NULL' }) @Index(['steeringExecutionId', 'steeringOrder'], { unique: true, @@ -31,6 +31,9 @@ export class AgentMessageQueue extends WithTimestamps { }) id: string; + @Column({ type: 'int', default: 0, comment: 'Pending turn order within the session' }) + position: number; + @ManyToOne(() => AgentExecutionThread, { onDelete: 'CASCADE' }) @JoinColumn({ name: 'threadId' }) thread: Relation; 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 ef31a7ad420..7d5547dc329 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 @@ -20,10 +20,12 @@ export class AgentMessageQueueRepository extends BaseRepository { + const manager = this.managerFor(ctx); + const items = await manager.find(AgentMessageQueue, { + where: { threadId, executionId: IsNull(), steeringExecutionId: IsNull() }, + order: { position: 'ASC', id: 'ASC' }, + }); + if ( + items.length !== expectedIds.length || + items.some((item, index) => item.id !== expectedIds[index]) + ) { + return false; + } + const from = items.findIndex((item) => item.id === id); + const to = items.findIndex((item) => item.id === targetId); + const item = items[from]; + if (!item || to < 0 || from === to) return false; + const start = Math.min(from, to); + const end = Math.max(from, to); + if (items.slice(start, end + 1).some((entry) => entry.payload.kind !== 'preview')) return false; + + const reordered = [...items]; + reordered.splice(from, 1); + reordered.splice(to, 0, item); + for (let index = start; index <= end; index++) { + await manager.update(AgentMessageQueue, reordered[index].id, { + position: items[index].position, + }); + } + return true; + } + async removePending(threadId: string, id: string, ctx: OperationContext) { const result = await this.managerFor(ctx).delete(AgentMessageQueue, { threadId, @@ -61,7 +100,7 @@ export class AgentMessageQueueRepository extends BaseRepository { await peer.destroy(); // Template databases skip the migration wrapper setup. await Container.get(DbConnection).migrate(); - await source.undoLastMigration(); + const migration = source.migrations.find( + (migration) => migration.constructor.name === 'CreateAgentThreadGrantTable1790837340960', + )!; const runner = source.createQueryRunner(); try { + await migration.down(runner); expect(await runner.hasTable(grants.metadata.tablePath)).toBe(false); + expect(await runner.manager.existsBy(threads.target, { id: 'parent' })).toBe(true); + await migration.up(runner); } finally { await runner.release(); } - expect(await threads.existsBy({ id: 'parent' })).toBe(true); - await Container.get(DbConnection).migrate(); expect(await grants.findKeys('parent')).toEqual(new Set()); }); }); 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 6535594dfbc..b2521ab1c4f 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 @@ -1697,6 +1697,67 @@ describe('AgentExecutionRepository', () => { }, ); + it('moves pending messages across multiple positions and rejects a changed order', async () => { + const local = recordingServices(); + const remote = recordingServices(undefined, peer); + const threadId = uuid(); + const target = { projectId, agentId, threadId, userId: owner.id }; + await enqueue(local, input(threadId, 'A', 'new')); + const active = await claim(local, threadId); + const data = input(threadId, 'B'); + data.payload.attachments = [ + { id: 'file-b', fileName: 'notes.txt', mimeType: 'text/plain', sizeBytes: 3 }, + ]; + const b = await enqueue(local, data); + const c = await enqueue(local, input(threadId, 'C')); + const d = await enqueue(local, input(threadId, 'D')); + const steer = await enqueue(local, input(threadId, 'Steering input')); + await local.queue.steer({ + ...target, + queueId: steer.id, + executionId: active.admission.executionId, + }); + const original = await local.messageRepository.findOneByOrFail({ id: b.messageId }); + const move = { + ...target, + queueId: b.id, + targetQueueId: d.id, + expectedQueueIds: [b.id, c.id, d.id], + }; + const results = await Promise.allSettled([ + local.queue.reorderPending(move), + remote.queue.reorderPending(move), + ]); + expect(results.map(({ status }) => status).sort()).toEqual(['fulfilled', 'rejected']); + expect((await remote.queue.listPending(target)).items.map(({ id }) => id)).toEqual([ + steer.id, + c.id, + d.id, + b.id, + ]); + await expect(remote.queue.reorderPending(move)).rejects.toThrow('The queue has changed'); + await remote.queue.reorderPending({ + ...move, + targetQueueId: c.id, + expectedQueueIds: [c.id, d.id, b.id], + }); + for (const queueId of [active.item.id, steer.id]) { + await expect(local.queue.reorderPending({ ...move, queueId })).rejects.toThrow( + 'The queue has changed', + ); + } + expect((await local.queue.listPending(target)).items.map(({ id }) => id)).toEqual([ + steer.id, + b.id, + c.id, + d.id, + ]); + expect(await local.messageRepository.findOneByOrFail({ id: b.messageId })).toEqual(original); + expect(await repository.findOneByOrFail({ id: active.admission.executionId })).toMatchObject({ + status: 'running', + }); + }); + it('consumes reserved messages in acceptance order and retains their history after interruption', async () => { const local = recordingServices(); const remote = recordingServices(undefined, peer); diff --git a/packages/cli/test/migration/1790853539575-add-agent-message-queue-position.test.ts b/packages/cli/test/migration/1790853539575-add-agent-message-queue-position.test.ts new file mode 100644 index 00000000000..5fbebdaa49c --- /dev/null +++ b/packages/cli/test/migration/1790853539575-add-agent-message-queue-position.test.ts @@ -0,0 +1,151 @@ +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 = 'AddAgentMessageQueuePosition1790853539575'; + +describe('AddAgentMessageQueuePosition migration', () => { + let dataSource: DataSource; + const projectId = randomUUID(); + const agentId = randomUUID(); + const threadIds = [randomUUID(), randomUUID()]; + const executionId = randomUUID(); + const messageIds = Array.from({ length: 5 }, () => randomUUID()); + let retiredId: number; + + beforeAll(async () => { + await Container.get(DbConnection).init(); + dataSource = Container.get(DataSource); + const cleanup = createTestMigrationContext(dataSource); + try { + await cleanup.queryRunner.clearDatabase(); + } finally { + await cleanup.queryRunner.release(); + } + await initDbUpToMigration(MIGRATION); + const ctx = createTestMigrationContext(dataSource); + try { + const table = ctx.escape.tableName; + const column = ctx.escape.columnName; + await ctx.runQuery( + `INSERT INTO ${table('project')} (${column('id')}, ${column('name')}, ${column('type')}, ${column('createdAt')}, ${column('updatedAt')}) VALUES (:projectId, 'Project', 'team', :now, :now)`, + { projectId, now: new Date() }, + ); + await ctx.runQuery( + `INSERT INTO ${table('agents')} (${column('id')}, ${column('name')}, ${column('projectId')}, ${column('integrations')}, ${column('tools')}, ${column('skills')}) VALUES (:agentId, 'Agent', :projectId, '[]', '{}', '{}')`, + { agentId, projectId }, + ); + for (const threadId of threadIds) { + await ctx.runQuery( + `INSERT INTO ${table('agent_execution_threads')} (${column('id')}, ${column('agentId')}, ${column('agentName')}, ${column('projectId')}) VALUES (:threadId, :agentId, 'Agent', :projectId)`, + { threadId, agentId, projectId }, + ); + await ctx.runQuery( + `INSERT INTO ${table('agents_resources')} (${column('id')}) VALUES (:threadId)`, + { threadId }, + ); + await ctx.runQuery( + `INSERT INTO ${table('agents_threads')} (${column('id')}, ${column('resourceId')}) VALUES (:threadId, :threadId)`, + { threadId }, + ); + } + await ctx.runQuery( + `INSERT INTO ${table('agent_execution')} (${column('id')}, ${column('threadId')}, ${column('status')}) VALUES (:executionId, :threadId, 'running')`, + { executionId, threadId: threadIds[0] }, + ); + for (const [index, messageId] of messageIds.entries()) { + const threadId = threadIds[index === 1 || index === 4 ? 1 : 0]; + await ctx.runQuery( + `INSERT INTO ${table('agents_messages')} (${column('id')}, ${column('threadId')}, ${column('resourceId')}, ${column('role')}, ${column('content')}) VALUES (:messageId, :threadId, :threadId, 'user', :content)`, + { messageId, threadId, content: JSON.stringify({ role: 'user', content: 'Input' }) }, + ); + await ctx.runQuery( + `INSERT INTO ${table('agent_message_queue')} (${column('threadId')}, ${column('messageId')}, ${column('payload')}, ${column('executionId')}, ${column('steeringExecutionId')}, ${column('steeringOrder')}) VALUES (:threadId, :messageId, '{"kind":"preview"}', :executionId, :steeringExecutionId, :steeringOrder)`, + { + threadId, + messageId, + executionId: index === 0 ? executionId : null, + steeringExecutionId: index === 2 ? executionId : null, + steeringOrder: index === 2 ? 1 : null, + }, + ); + } + const [last] = await ctx.runQuery>( + `SELECT ${column('id')} FROM ${table('agent_message_queue')} WHERE ${column('messageId')} = :messageId`, + { messageId: messageIds[4] }, + ); + retiredId = Number(last.id); + await ctx.runQuery( + `DELETE FROM ${table('agent_message_queue')} WHERE ${column('messageId')} = :messageId`, + { messageId: messageIds[4] }, + ); + } finally { + await ctx.queryRunner.release(); + } + }); + + afterAll(async () => { + await Container.get(DbConnection).close(); + }); + + it('preserves each session order and active reservations through up, down, and up', async () => { + await runSingleMigration(MIGRATION); + await dataSource.undoLastMigration({ transaction: 'none' }); + await runSingleMigration(MIGRATION); + const ctx = createTestMigrationContext(dataSource); + const table = ctx.escape.tableName; + const column = ctx.escape.columnName; + try { + const rows = await ctx.runQuery( + `SELECT ${column('messageId')}, ${column('position')}, ${column('executionId')}, ${column('steeringExecutionId')}, ${column('steeringOrder')} FROM ${table('agent_message_queue')} ORDER BY ${column('id')}`, + ); + expect(rows).toEqual([ + { + messageId: messageIds[0], + position: 0, + executionId, + steeringExecutionId: null, + steeringOrder: null, + }, + { + messageId: messageIds[1], + position: 0, + executionId: null, + steeringExecutionId: null, + steeringOrder: null, + }, + { + messageId: messageIds[2], + position: 1, + executionId: null, + steeringExecutionId: executionId, + steeringOrder: 1, + }, + { + messageId: messageIds[3], + position: 2, + executionId: null, + steeringExecutionId: null, + steeringOrder: null, + }, + ]); + await ctx.runQuery( + `INSERT INTO ${table('agent_message_queue')} (${column('threadId')}, ${column('messageId')}, ${column('payload')}, ${column('position')}) VALUES (:threadId, :messageId, '{"kind":"preview"}', 1)`, + { threadId: threadIds[1], messageId: messageIds[4] }, + ); + const [inserted] = await ctx.runQuery>( + `SELECT ${column('id')} FROM ${table('agent_message_queue')} WHERE ${column('messageId')} = :messageId`, + { messageId: messageIds[4] }, + ); + expect(Number(inserted.id)).toBeGreaterThan(retiredId); + } finally { + await ctx.queryRunner.release(); + } + }); +}); diff --git a/packages/frontend/@n8n/i18n/src/locales/en.json b/packages/frontend/@n8n/i18n/src/locales/en.json index ebc11cbda68..6ef8459ca21 100644 --- a/packages/frontend/@n8n/i18n/src/locales/en.json +++ b/packages/frontend/@n8n/i18n/src/locales/en.json @@ -8048,6 +8048,10 @@ "agents.chat.queue.editSteeringUnavailable": "This message is waiting for the next step. Your changes weren't applied.", "agents.chat.queue.editError": "Couldn't save your changes. Try again.", "agents.chat.queue.remove": "Remove queued message", + "agents.chat.queue.reorder": "Reorder queued message {position} of {count}", + "agents.chat.queue.reorderTooltip": "Drag to reorder. Use the Up and Down arrow keys to move the message.", + "agents.chat.queue.reorderUnavailable": "The queue changed. Review the updated order and try again.", + "agents.chat.queue.reorderError": "Couldn't change the queue order. Try again.", "agents.chat.queue.loadError": "Couldn't load queued messages. Reopen the chat to try again.", "agents.chat.queue.removeError": "Couldn't remove the message. Check whether it has started and try again.", "agents.chat.queue.sendError": "Couldn't confirm the message was queued. Check the queue before sending again.", diff --git a/packages/frontend/editor-ui/src/features/agents/__tests__/AgentChatPanel.test.ts b/packages/frontend/editor-ui/src/features/agents/__tests__/AgentChatPanel.test.ts index 981fb37b22f..db042dc8070 100644 --- a/packages/frontend/editor-ui/src/features/agents/__tests__/AgentChatPanel.test.ts +++ b/packages/frontend/editor-ui/src/features/agents/__tests__/AgentChatPanel.test.ts @@ -1,6 +1,7 @@ import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; import { flushPromises, mount } from '@vue/test-utils'; import { computed, defineComponent, h, ref, nextTick } from 'vue'; +import Draggable from 'vuedraggable'; import { createMemoryHistory, createRouter } from 'vue-router'; import { AGENT_SESSION_DETAIL_VIEW } from '../constants'; import { APPROVAL_TOOL_NAME, N8N_CHAT_ACTION_TOOL_NAME, WAIT_TOOL_NAME } from '@n8n/api-types'; @@ -31,6 +32,8 @@ const trackSubmittedMessageMock = vi.fn(); const queuedMessagesMock = ref([]); const removeQueuedMessageMock = vi.fn(); const updateQueuedMessageMock = vi.fn(); +const reorderQueuedMessageMock = vi.fn(); +const isReorderingQueueMock = ref(false); const steerQueuedMessageMock = vi.fn(); const canSteerMock = ref(false); const steeringQueueIdsMock = ref(new Set()); @@ -98,6 +101,10 @@ vi.mock('@n8n/design-system', async (importOriginal) => ({ N8nApprovalCard: (await importOriginal()).N8nApprovalCard, N8nAiActivityStepGroup: (await importOriginal()) .N8nAiActivityStepGroup, + N8nAiActivityStepButton: (await importOriginal()) + .N8nAiActivityStepButton, + N8nAiActivityStepChevron: (await importOriginal()) + .N8nAiActivityStepChevron, N8nLink: (await importOriginal()).N8nLink, useDropdownSearch: (await importOriginal()) .useDropdownSearch, @@ -217,6 +224,8 @@ vi.mock('../composables/useAgentChatStream', () => ({ steerQueuedMessage: steerQueuedMessageMock, removeQueuedMessage: removeQueuedMessageMock, updateQueuedMessage: updateQueuedMessageMock, + reorderQueuedMessage: reorderQueuedMessageMock, + isReorderingQueue: isReorderingQueueMock, isCancelling: isCancellingMock, messagingState: computed(() => (isStreamingMock.value ? 'receiving' : 'idle')), fatalError: fatalErrorMock, @@ -253,6 +262,8 @@ describe('AgentChatPanel', () => { beforeEach(() => { vi.clearAllMocks(); updateQueuedMessageMock.mockReset().mockResolvedValue('updated'); + reorderQueuedMessageMock.mockReset().mockResolvedValue(undefined); + isReorderingQueueMock.value = false; messagesMock.value = []; isStreamingMock.value = false; isSubmittingMock.value = false; @@ -411,6 +422,90 @@ describe('AgentChatPanel', () => { wrapper.unmount(); }); + it('drags messages across the collapsed boundary and supports keyboard moves from the handle', async () => { + const items: AgentChatQueueItem[] = [1, 2, 3, 4].map((id) => ({ + id: String(id), + message: `Message ${id}`, + createdAt: '2026-09-24T12:00:00.000Z', + steeringExecutionId: id === 1 ? 'running' : null, + })); + queuedMessagesMock.value = items; + const container = document.createElement('div'); + document.body.append(container); + const wrapper = mountPanel({}, container); + const queue = wrapper.get('[data-testid="agent-message-queue"]'); + const draggable = wrapper.getComponent(Draggable); + expect( + queue + .get('[data-queue-id="1"] [data-testid="agent-queue-drag-handle"]') + .attributes('disabled'), + ).toBeDefined(); + const saved = createDeferredPromise(); + reorderQueuedMessageMock.mockImplementationOnce(async () => { + isReorderingQueueMock.value = true; + await saved.promise; + queuedMessagesMock.value = [items[0], items[2], items[3], items[1]]; + isReorderingQueueMock.value = false; + }); + draggable.vm.$emit('start'); + await nextTick(); + expect(queue.get('button[aria-expanded]').attributes('aria-expanded')).toBe('true'); + expect(queue.findAll('li').map((row) => row.attributes('data-queue-id'))).toEqual([ + '1', + '2', + '3', + '4', + ]); + draggable.vm.$emit('end', { oldIndex: 1, newIndex: 3 }); + await nextTick(); + expect(reorderQueuedMessageMock).toHaveBeenCalledWith('2', '4', ['2', '3', '4']); + for (const button of queue.findAll('[data-testid="agent-queue-drag-handle"]')) { + expect(button.attributes('disabled')).toBeDefined(); + } + expect(queue.findAll('li').map((row) => row.attributes('data-queue-id'))).toEqual([ + '1', + '3', + '4', + '2', + ]); + saved.resolve(); + await flushPromises(); + const handle = queue.get('[data-queue-id="2"] [data-testid="agent-queue-drag-handle"]'); + expect(document.activeElement).toBe(handle.element); + await handle.trigger('keydown', { key: 'ArrowDown' }); + await queue + .get('[data-queue-id="3"] [data-testid="agent-queue-drag-handle"]') + .trigger('keydown', { key: 'ArrowUp' }); + expect(reorderQueuedMessageMock).toHaveBeenCalledTimes(1); + await handle.trigger('keydown', { key: 'ArrowUp' }); + expect(reorderQueuedMessageMock).toHaveBeenLastCalledWith('2', '4', ['3', '4', '2']); + expect(stopGeneratingMock).not.toHaveBeenCalled(); + wrapper.unmount(); + container.remove(); + }); + + it('uses the order at drag start and restores the refreshed queue after the reorder settles', async () => { + const items = [1, 2, 3].map((id) => ({ + id: String(id), + message: `Message ${id}`, + createdAt: '2026-09-24T12:00:00.000Z', + steeringExecutionId: null, + })); + queuedMessagesMock.value = items; + const wrapper = mountPanel(); + const draggable = wrapper.getComponent(Draggable); + draggable.vm.$emit('start'); + queuedMessagesMock.value = items.slice(1); + await nextTick(); + draggable.vm.$emit('end', { oldIndex: 0, newIndex: 2 }); + await flushPromises(); + expect(reorderQueuedMessageMock).toHaveBeenCalledWith('1', '3', ['1', '2', '3']); + expect( + wrapper.findAll('[data-queue-id]').map((row) => row.attributes('data-queue-id')), + ).toEqual(['2', '3']); + wrapper.unmount(); + }); + it('edits queued text inline while preserving attachments and the composer draft', async () => { queuedMessagesMock.value = [ { @@ -472,7 +567,7 @@ describe('AgentChatPanel', () => { ); await nextTick(); expect(row.text()).toContain('agents.chat.queue.steering'); - for (const action of ['steer', 'edit', 'remove']) + for (const action of ['steer', 'edit', 'remove', 'reorder']) expect( row.get(`[aria-label="agents.chat.queue.${action}"]`).attributes('disabled'), ).toBeDefined(); diff --git a/packages/frontend/editor-ui/src/features/agents/__tests__/useAgentChatStream.test.ts b/packages/frontend/editor-ui/src/features/agents/__tests__/useAgentChatStream.test.ts index 716c2be0e3f..5a48cd3250a 100644 --- a/packages/frontend/editor-ui/src/features/agents/__tests__/useAgentChatStream.test.ts +++ b/packages/frontend/editor-ui/src/features/agents/__tests__/useAgentChatStream.test.ts @@ -19,13 +19,15 @@ vi.mock('@n8n/i18n', () => ({ })); vi.mock('@n8n/composables/useToast', () => ({ - useToast: () => ({ showError: vi.fn() }), + useToast: () => ({ showError: showQueueErrorMock }), })); +const showQueueErrorMock = vi.fn(); const getAgentChatQueueMock = vi.fn().mockResolvedValue({ items: [] }); const removeAgentQueuedMessageMock = vi.fn().mockResolvedValue({ removed: true }); const updateAgentQueuedMessageMock = vi.fn(); const steerAgentQueuedMessageMock = vi.fn(); +const reorderAgentQueuedMessageMock = vi.fn(); const getChatMessagesMock = vi.fn(); const getTestChatMessagesMock = vi.fn(); const cancelAgentChatRunMock = vi.fn(); @@ -59,6 +61,7 @@ vi.mock('../composables/useAgentApi', async (importOriginal) => { removeAgentQueuedMessage: (...args: unknown[]) => removeAgentQueuedMessageMock(...args), updateAgentQueuedMessage: (...args: unknown[]) => updateAgentQueuedMessageMock(...args), steerAgentQueuedMessage: (...args: unknown[]) => steerAgentQueuedMessageMock(...args), + reorderAgentQueuedMessage: (...args: unknown[]) => reorderAgentQueuedMessageMock(...args), getChatMessages: (...args: unknown[]) => getChatMessagesMock(...args), getTestChatMessages: (...args: unknown[]) => getTestChatMessagesMock(...args), cancelAgentChatRun: (...args: unknown[]) => cancelAgentChatRunMock(...args), @@ -202,6 +205,8 @@ beforeEach(() => { getAgentChatQueueMock.mockReset().mockResolvedValue({ items: [] }); removeAgentQueuedMessageMock.mockReset().mockResolvedValue({ removed: true }); steerAgentQueuedMessageMock.mockReset().mockResolvedValue(undefined); + reorderAgentQueuedMessageMock.mockReset().mockResolvedValue(undefined); + showQueueErrorMock.mockClear(); }); const hookScopes: ReturnType[] = []; @@ -3773,9 +3778,15 @@ describe('useAgentChatStream — queued submissions', () => { streams[1].close([{ type: 'done', executionId: 'B' }]); }); - it.each([false, true])( - 'ignores an older start after B starts (B settled: %s)', - async (settled) => { + it.each([ + { settled: false, unavailable: false }, + { settled: true, unavailable: false }, + { settled: false, unavailable: true }, + ])( + 'ignores an older start with a stale queue (B settled: $settled, history unavailable: $unavailable)', + async ({ settled, unavailable }) => { + const queued = { id: '1', message: 'A', createdAt: new Date().toISOString() }; + getAgentChatQueueMock.mockResolvedValue({ items: [queued] }); const streams: Array> = []; const signals: Array = []; vi.stubGlobal( @@ -3795,13 +3806,25 @@ describe('useAgentChatStream — queued submissions', () => { const hook = buildHook('thread-1'); await hook.sendMessage('A'); await hook.sendMessage('B'); + getChatMessagesMock.mockResolvedValue({ + messages: [], + openSuspensions: [], + activeExecutionId: 'B', + }); streams[1].emit([{ type: 'execution-started', executionId: 'B', sessionId: 'thread-1' }]); await flushPromises(); + expect(hook.queuedMessages.value).toEqual([queued]); if (settled) { + getChatMessagesMock.mockResolvedValue({ + messages: [], + openSuspensions: [], + activeExecutionId: null, + }); streams[1].close([{ type: 'done', executionId: 'B' }]); await flushPromises(); } - streams[0].close([{ type: 'execution-started', executionId: 'A', sessionId: 'thread-1' }]); + if (unavailable) getChatMessagesMock.mockRejectedValueOnce(new Error('History unavailable')); + streams[0].emit([{ type: 'execution-started', executionId: 'A', sessionId: 'thread-1' }]); await flushPromises(); expect(signals[0]?.aborted).toBe(true); expect(hook.activeExecutionId.value).toBe(settled ? null : 'B'); @@ -3812,10 +3835,118 @@ describe('useAgentChatStream — queued submissions', () => { await hook.stopGenerating(); expect(cancelAgentChatExecutionMock.mock.lastCall?.[4]).toBe('B'); } + streams[0].close(); hook.detachStream(); }, ); + it.each([false, true])( + 'ignores a delayed ownership response after C starts (history recovery: %s)', + async (recovered) => { + const streams: Array> = []; + const fetchMock = vi.fn(async (_url, init: RequestInit) => { + const stream = makeControllableSseResponse( + [{ type: 'message-queued', queueId: String(streams.length + 1), sessionId: 'thread-1' }], + init.signal ?? null, + ); + streams.push(stream); + return stream.response; + }); + vi.stubGlobal('fetch', fetchMock); + const hook = buildHook('thread-1'); + await hook.sendMessage('A'); + await hook.sendMessage('B'); + streams[1].emit([{ type: 'execution-started', executionId: 'B', sessionId: 'thread-1' }]); + await flushPromises(); + await hook.sendMessage('C'); + if (recovered) { + streams[1].close([{ type: 'done', executionId: 'B' }]); + await flushPromises(); + } + + const validation = Promise.withResolvers(); + getChatMessagesMock.mockReturnValueOnce(validation.promise); + streams[0].emit([{ type: 'execution-started', executionId: 'A', sessionId: 'thread-1' }]); + await flushPromises(); + expect(hook.activeExecutionId.value).toBe(recovered ? null : 'B'); + + getChatMessagesMock.mockResolvedValue({ + messages: ['B', 'C'].map((text) => ({ + id: text, + executionId: text, + role: 'user', + content: [{ type: 'text', text }], + })), + openSuspensions: [], + activeExecutionId: 'C', + }); + if (recovered) hook.refresh(); + else + streams[2].emit([{ type: 'execution-started', executionId: 'C', sessionId: 'thread-1' }]); + await flushPromises(); + validation.resolve({ messages: [], openSuspensions: [], activeExecutionId: 'A' }); + await flushPromises(); + expect(fetchMock.mock.calls[0][1].signal?.aborted).toBe(true); + expect(hook.activeExecutionId.value).toBe('C'); + if (recovered) + streams[2].emit([{ type: 'execution-started', executionId: 'C', sessionId: 'thread-1' }]); + streams[2].emit([{ type: 'text-delta', id: 'c-text', delta: 'C output' }]); + await flushPromises(); + expect(hook.messages.value.map(({ content }) => content)).toEqual(['B', 'C', 'C output']); + await hook.stopGenerating(); + expect(cancelAgentChatExecutionMock.mock.lastCall?.[4]).toBe('C'); + hook.detachStream(); + }, + ); + + it('stops validating a delayed start after repeated ownership changes', async () => { + const streams: Array> = []; + const fetchMock = vi.fn(async (_url, init: RequestInit) => { + const executionId = ['A', 'B', 'C', 'D', 'E'][streams.length]; + const event: AgentSseEvent = + streams.length < 2 + ? { type: 'message-queued', queueId: executionId, sessionId: 'thread-1' } + : { type: 'execution-started', executionId, sessionId: 'thread-1' }; + const stream = makeControllableSseResponse([event], init.signal ?? null); + streams.push(stream); + return stream.response; + }); + vi.stubGlobal('fetch', fetchMock); + const hook = buildHook('thread-1'); + await hook.sendMessage('A'); + await hook.sendMessage('B'); + streams[1].emit([{ type: 'execution-started', executionId: 'B', sessionId: 'thread-1' }]); + await flushPromises(); + + let validation = Promise.withResolvers(); + getChatMessagesMock.mockReturnValueOnce(validation.promise); + streams[0].emit([{ type: 'execution-started', executionId: 'A', sessionId: 'thread-1' }]); + await flushPromises(); + for (const executionId of ['C', 'D', 'E']) { + await hook.sendMessage(executionId); + const nextValidation = Promise.withResolvers(); + getChatMessagesMock.mockReturnValueOnce(nextValidation.promise); + validation.resolve({ messages: [], openSuspensions: [], activeExecutionId: 'A' }); + await flushPromises(); + expect(hook.activeExecutionId.value).toBe(executionId); + validation = nextValidation; + } + + expect(fetchMock.mock.calls[0][1].signal?.aborted).toBe(true); + streams[4].emit([{ type: 'text-delta', id: 'e-text', delta: 'E output' }]); + await flushPromises(); + expect(hook.messages.value.map(({ content }) => content)).toEqual([ + 'B', + 'C', + 'D', + 'E', + 'E output', + ]); + await hook.stopGenerating(); + expect(cancelAgentChatExecutionMock.mock.lastCall?.[4]).toBe('E'); + hook.detachStream(); + }); + it('handles start before acceptance and preserves the original stream after a removal conflict', async () => { let stream: ReturnType; const signals: Array = []; @@ -3989,6 +4120,171 @@ describe('useAgentChatStream — queued submissions', () => { }, ); + it('keeps live output once after reordering, history recovery, and timeline updates', async () => { + const items = ['A', 'B'].map((message, index) => ({ + id: String(index + 1), + message, + createdAt: new Date().toISOString(), + })); + getAgentChatQueueMock.mockResolvedValue({ items }); + const streams: Array> = []; + const fetchMock = vi.fn(async (_url, init: RequestInit) => { + const event: AgentSseEvent = + streams.length === 0 + ? { type: 'execution-started', executionId: 'active', sessionId: 'thread-1' } + : { type: 'message-queued', queueId: String(streams.length), sessionId: 'thread-1' }; + const stream = makeControllableSseResponse([event], init.signal ?? null); + streams.push(stream); + return stream.response; + }); + vi.stubGlobal('fetch', fetchMock); + const hook = buildHook('thread-1'); + await hook.sendMessage('Current turn'); + await hook.sendMessage('A'); + await hook.sendMessage('B'); + await flushPromises(); + const saved = Promise.withResolvers(); + reorderAgentQueuedMessageMock.mockReturnValueOnce(saved.promise); + const reordering = hook.reorderQueuedMessage('1', '2', ['1', '2']); + expect(reorderAgentQueuedMessageMock).toHaveBeenCalledWith( + expect.anything(), + 'p1', + 'a1', + 'thread-1', + '1', + { targetQueueId: '2', expectedQueueIds: ['1', '2'] }, + ); + expect(hook.isReorderingQueue.value).toBe(true); + expect(hook.queuedMessages.value.map(({ id }) => id)).toEqual(['1', '2']); + getAgentChatQueueMock.mockResolvedValue({ items: [items[1], items[0]] }); + saved.resolve(); + await reordering; + expect(hook.queuedMessages.value.map(({ id }) => id)).toEqual(['2', '1']); + expect(hook.isReorderingQueue.value).toBe(false); + expect(hook.activeExecutionId.value).toBe('active'); + expect(fetchMock.mock.calls[0][1].signal?.aborted).toBe(false); + + getAgentChatQueueMock.mockResolvedValue({ items: [items[0]] }); + streams[2].emit([ + { type: 'execution-started', executionId: 'B', sessionId: 'thread-1' }, + { type: 'text-delta', id: 'b-text', delta: 'B output' }, + ]); + await flushPromises(); + expect(fetchMock.mock.calls[1][1].signal?.aborted).toBe(false); + expect(hook.activeExecutionId.value).toBe('B'); + expect(hook.queuedMessages.value.map(({ id }) => id)).toEqual(['1']); + expect(hook.messages.value.map(({ content }) => content)).toEqual([ + 'Current turn', + 'B', + 'B output', + ]); + + const recovered: AgentChatMessagesResponse = { + messages: ['Current turn', 'B', 'B output', 'A'].map((text, index) => ({ + id: `saved-${index}`, + role: index === 2 ? 'assistant' : 'user', + executionId: ['active', 'B', 'B', 'A'][index], + content: [{ type: 'text', text }], + })), + openSuspensions: [], + activeExecutionId: 'A', + }; + getChatMessagesMock.mockResolvedValue(recovered); + getAgentChatQueueMock.mockResolvedValue({ items: [] }); + streams[2].close([{ type: 'done', executionId: 'B' }]); + await flushPromises(); + const validation = Promise.withResolvers(); + getChatMessagesMock.mockReturnValueOnce(validation.promise); + streams[1].emit([ + { + type: 'execution-started', + executionId: 'A', + sessionId: 'thread-1', + inputMessageIds: ['saved-3'], + }, + { type: 'text-delta', id: 'a-text', delta: 'A output' }, + ]); + await flushPromises(); + getChatMessagesMock.mockResolvedValue({ + ...recovered, + messages: [ + ...recovered.messages, + { + id: 'a-partial', + role: 'assistant', + executionId: 'A', + content: [{ type: 'text', text: 'A out' }], + }, + ], + }); + for (let update = 0; update < 3; update++) { + for (const listener of [...pushListeners]) + listener({ + type: 'agentExecutionUpdated', + data: { projectId: 'p1', agentId: 'a1', threadId: 'thread-1', executionId: 'A' }, + }); + await flushPromises(); + } + expect(hook.messages.value.at(-1)?.content).toBe('A out'); + getChatMessagesMock.mockReturnValue(Promise.withResolvers().promise); + validation.resolve(recovered); + await flushPromises(); + streams[1].emit([{ type: 'text-delta', id: 'a-text', delta: ' continues' }]); + await flushPromises(); + expect(hook.activeExecutionId.value).toBe('A'); + expect(hook.queuedMessages.value).toEqual([]); + expect(hook.messages.value.map(({ content }) => content)).toEqual([ + 'Current turn', + 'B', + 'B output', + 'A', + 'A output continues', + ]); + expect(hook.messages.value[3].id).toBe('saved-3'); + await hook.stopGenerating(); + expect(cancelAgentChatExecutionMock.mock.lastCall?.[4]).toBe('A'); + hook.detachStream(); + }); + + it('keeps a new session reorder busy when the previous session refresh completes', async () => { + const stale = Promise.withResolvers<{ items: [] }>(); + getAgentChatQueueMock.mockReturnValueOnce(stale.promise); + const scope = effectScope(); + hookScopes.push(scope); + const threadId = ref('thread-1'); + const hook = scope.run(() => + useAgentChatStream({ projectId: ref('p1'), agentId: ref('a1'), continueSessionId: threadId }), + )!; + const first = hook.reorderQueuedMessage('1', '2', ['1', '2']); + await flushPromises(); + threadId.value = 'thread-2'; + await flushPromises(); + const saved = Promise.withResolvers(); + reorderAgentQueuedMessageMock.mockReturnValueOnce(saved.promise); + const second = hook.reorderQueuedMessage('3', '4', ['3', '4']); + stale.resolve({ items: [] }); + await first; + expect(hook.isReorderingQueue.value).toBe(true); + saved.resolve(); + await second; + expect(hook.isReorderingQueue.value).toBe(false); + }); + + it.each([ + [409, 'agents.chat.queue.reorderUnavailable'], + [500, 'agents.chat.queue.reorderError'], + ])('refreshes the queue and explains reorder failure %i', async (status, errorKey) => { + const item = { id: '2', message: 'Still pending', createdAt: new Date().toISOString() }; + getAgentChatQueueMock.mockResolvedValue({ items: [item] }); + const error = { httpStatusCode: status }; + reorderAgentQueuedMessageMock.mockRejectedValueOnce(error); + const hook = buildHook('thread-1'); + await hook.reorderQueuedMessage('1', '2', ['1', '2']); + expect(hook.queuedMessages.value).toEqual([item]); + expect(hook.isReorderingQueue.value).toBe(false); + expect(showQueueErrorMock).toHaveBeenCalledWith(error, errorKey); + }); + it('restores pending inputs after reload and ignores a queue snapshot invalidated by removal', async () => { const item = { id: '1', message: 'Queued', createdAt: new Date().toISOString() }; getAgentChatQueueMock.mockResolvedValue({ items: [item] }); diff --git a/packages/frontend/editor-ui/src/features/agents/agents-design-language.md b/packages/frontend/editor-ui/src/features/agents/agents-design-language.md index bc2ee30d1af..d32976db1da 100644 --- a/packages/frontend/editor-ui/src/features/agents/agents-design-language.md +++ b/packages/frontend/editor-ui/src/features/agents/agents-design-language.md @@ -220,9 +220,10 @@ Use one action on the right. Show Stop when a turn can be stopped and the composer has no text or attachments. Otherwise, show Send. Keep file and voice input available during a turn. Stack the background task card above the composer. Attach pending messages to -the top of the composer. Use a subtle background and dividers between messages. -Use muted gray for queue text and icons. Use the same gray for all queue icons. -Give text more contrast than icons. Keep text contrast at least 4.5:1. +the top of the composer. Use `--background--subtle` for the queue background. +Use `--color--neutral-600` for queue text and the Steer label in light mode. +Use `--text-color--subtler` in dark mode. Use `--color--neutral-400` for all queue +icons. Use `--border-color--subtle` for the dividers. Use `2xs` text and `large` icons. Keep the action targets at least 24 by 24 pixels. Keep the first two messages visible. Put the third and later messages in a collapsed activity group. Show the number @@ -234,4 +235,6 @@ message. It does not restore the composer draft. Edit queued text in place. Use compact Save and Cancel icon actions. Enter saves, Shift+Enter adds a line, and Escape cancels. Keep attachments unchanged. Do not pause the queue during editing. If the message starts, disable Save and retain the draft until the user dismisses it. -Put the action to send a message to the current execution immediately before Edit. Use the existing corner-down-right icon with the Steer label on its right. Use the existing button and tooltip. Enable it only when the server reports an eligible execution. Keep an accepted steering request in the panel with a waiting status. Disable its actions until the runtime consumes it or returns it to ordinary FIFO processing. Preserve an open edit draft if another client reserves the message, and disable Save. When the runtime consumes the message, show it between the surrounding assistant output. Keep Stop bound to the same execution. +Put a six-dot drag handle on the left of each pending message. Drag the handle to move the message. Support the Up and Down arrow keys on the handle. Expand the queue when a drag starts or a message moves. Keep keyboard focus on the moved message. Show the new order during saving. Disable queue actions while the order saves. Messages reserved for steering cannot move. If the move no longer applies, refresh the queue and show an error. + +Put the action to send a message to the current execution immediately before Edit. Use the existing corner-down-right icon with the Steer label on its right. Use the existing button and tooltip. Enable it only when the server reports an eligible execution. Keep an accepted steering request in the panel with a waiting status. Disable its actions until the runtime consumes it or returns it to ordinary queue processing. Preserve an open edit draft if another client reserves the message, and disable Save. When the runtime consumes the message, show it between the surrounding assistant output. Keep Stop bound to the same execution. diff --git a/packages/frontend/editor-ui/src/features/agents/components/AgentChatPanel.vue b/packages/frontend/editor-ui/src/features/agents/components/AgentChatPanel.vue index 69a3936e271..459c7984edd 100644 --- a/packages/frontend/editor-ui/src/features/agents/components/AgentChatPanel.vue +++ b/packages/frontend/editor-ui/src/features/agents/components/AgentChatPanel.vue @@ -2,15 +2,20 @@ import { computed, ref, + shallowRef, toRef, watch, onMounted, onBeforeUnmount, useTemplateRef, nextTick, + useId, } from 'vue'; +import Draggable from 'vuedraggable'; import { N8nAiActivityStepGroup, + N8nAiActivityStepButton, + N8nAiActivityStepChevron, N8nButton, N8nCallout, N8nIcon, @@ -19,7 +24,7 @@ import { N8nText, N8nTooltip, } from '@n8n/design-system'; -import { createReusableTemplate, useDocumentVisibility, useIntervalFn } from '@vueuse/core'; +import { useDocumentVisibility, useIntervalFn } from '@vueuse/core'; import { useI18n } from '@n8n/i18n'; import { type AgentChatQueueItem, @@ -110,6 +115,8 @@ const { steerQueuedMessage, removeQueuedMessage, updateQueuedMessage, + reorderQueuedMessage, + isReorderingQueue, isSubmitting, isLoadingHistory, isStreaming, @@ -152,9 +159,14 @@ const queueRows = computed(() => { } return queuedMessages.value; }); -const [DefineQueueList, QueueList] = createReusableTemplate<{ items: AgentChatQueueItem[] }>({ - inheritAttrs: false, -}); +const queueElement = useTemplateRef('messageQueue'); +const queueListId = useId(); +const queueExpanded = ref(false); +const queueOrder = shallowRef(); +const displayedQueueRows = computed(() => queueOrder.value ?? queueRows.value); +const visibleQueueRows = computed(() => + queueExpanded.value ? displayedQueueRows.value : displayedQueueRows.value.slice(0, 2), +); const canSaveQueueEdit = computed(() => { const edit = queueEdit.value; return ( @@ -176,11 +188,64 @@ function startQueueEdit(item: AgentChatQueueItem) { } function isQueueItemBusy(item: AgentChatQueueItem) { return ( + isReorderingQueue.value || !!item.steeringExecutionId || steeringQueueIds.value.has(item.id) || removingQueueIds.value.has(item.id) ); } +function canMoveQueueItem(items: AgentChatQueueItem[], from: number, to: number) { + if (queueEdit.value || from === to || !items[from] || !items[to]) return false; + return !items.slice(Math.min(from, to), Math.max(from, to) + 1).some(isQueueItemBusy); +} +function canDragQueueItem(index: number) { + const items = displayedQueueRows.value; + return canMoveQueueItem(items, index, index - 1) || canMoveQueueItem(items, index, index + 1); +} +function startQueueDrag() { + queueOrder.value = [...queueRows.value]; + queueExpanded.value = true; +} +function canDropQueueItem(event: { draggedContext: { index: number; futureIndex: number } }) { + const { index, futureIndex } = event.draggedContext; + return canMoveQueueItem(displayedQueueRows.value, index, futureIndex); +} +function endQueueDrag(event: { oldIndex?: number; newIndex?: number }) { + const items = queueOrder.value; + queueOrder.value = undefined; + if (items && event.oldIndex !== undefined && event.newIndex !== undefined) { + void moveQueueItem(items, event.oldIndex, event.newIndex); + } +} +async function moveQueueItem(items: AgentChatQueueItem[], from: number, to: number) { + if (!canMoveQueueItem(items, from, to)) return; + const item = items[from]; + const reordered = [...items]; + reordered.splice(from, 1); + reordered.splice(to, 0, item); + queueOrder.value = reordered; + queueExpanded.value = true; + await reorderQueuedMessage( + item.id, + items[to].id, + items.filter((entry) => !entry.steeringExecutionId).map((entry) => entry.id), + ); + if (queueOrder.value !== reordered) return; + queueOrder.value = undefined; + await nextTick(); + queueElement.value + ?.querySelector( + `[data-queue-id="${item.id}"] [data-testid="agent-queue-drag-handle"]:not(:disabled)`, + ) + ?.focus(); +} +function onQueueHandleKeydown(event: KeyboardEvent, index: number) { + if (queueOrder.value || !['ArrowUp', 'ArrowDown'].includes(event.key)) return; + event.preventDefault(); + event.stopPropagation(); + const delta = event.key === 'ArrowUp' ? -1 : 1; + void moveQueueItem(queueRows.value, index, index + delta); +} async function saveQueueEdit() { const edit = queueEdit.value; if (!edit || !canSaveQueueEdit.value) return; @@ -604,6 +669,8 @@ watch( () => { queuedExternalMessage = undefined; queueEdit.value = undefined; + queueExpanded.value = false; + queueOrder.value = undefined; }, ); @@ -914,14 +981,52 @@ onBeforeUnmount(() => { @stop="stopGenerating" @files-selected="handleFilesSelected" > -