mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-17 05:29:44 +02:00
feat(editor): Allow reordering queued messages in Agent Preview (#40101)
This commit is contained in:
Robin Braumann
GitHub
parent
b581d16a82
commit
33505d7faa
3
docs/generated/postgres-schema/README.md
generated
3
docs/generated/postgres-schema/README.md
generated
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
3
docs/generated/sqlite-schema/README.md
generated
3
docs/generated/sqlite-schema/README.md
generated
|
|
@ -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
|
||||
|
|
|
|||
1
docs/generated/sqlite-schema/agent_execution.md
generated
1
docs/generated/sqlite-schema/agent_execution.md
generated
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@
|
|||
<summary><strong>Table Definition</strong></summary>
|
||||
|
||||
```sql
|
||||
CREATE TABLE "agent_message_queue" ("id" integer PRIMARY KEY AUTOINCREMENT NOT NULL, "threadId" varchar(128) NOT NULL, "payload" text NOT NULL, "executionId" varchar(36), "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "messageId" varchar(36) NOT NULL, "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)
|
||||
```
|
||||
|
||||
</details>
|
||||
|
|
@ -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
|
||||
|
|
|
|||
1
docs/generated/sqlite-schema/agents_messages.md
generated
1
docs/generated/sqlite-schema/agents_messages.md
generated
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
|
|
|
|||
|
|
@ -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',
|
||||
]),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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',
|
||||
|
|
|
|||
|
|
@ -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<void> {
|
||||
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(
|
||||
|
|
|
|||
|
|
@ -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<void> {
|
||||
while (!this.stopped) {
|
||||
if (!(await this.consumeNext(threadId))) return;
|
||||
|
|
|
|||
|
|
@ -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<void> {
|
||||
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(
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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<AgentExecutionThread>;
|
||||
|
|
|
|||
|
|
@ -20,10 +20,12 @@ export class AgentMessageQueueRepository extends BaseRepository<AgentMessageQueu
|
|||
ctx: OperationContext,
|
||||
) {
|
||||
const repository = this.managerFor(ctx).getRepository(AgentMessageQueue);
|
||||
const last = await repository.findOne({ where: { threadId }, order: { position: 'DESC' } });
|
||||
return await repository.save(
|
||||
repository.create({
|
||||
threadId,
|
||||
messageId,
|
||||
position: (last?.position ?? -1) + 1,
|
||||
payload,
|
||||
executionId: null,
|
||||
steeringExecutionId: null,
|
||||
|
|
@ -36,7 +38,7 @@ export class AgentMessageQueueRepository extends BaseRepository<AgentMessageQueu
|
|||
return await this.find({
|
||||
where: { threadId, executionId: IsNull() },
|
||||
relations: { message: true },
|
||||
order: { id: 'ASC' },
|
||||
order: { position: 'ASC', id: 'ASC' },
|
||||
});
|
||||
}
|
||||
|
||||
|
|
@ -47,6 +49,43 @@ export class AgentMessageQueueRepository extends BaseRepository<AgentMessageQueu
|
|||
});
|
||||
}
|
||||
|
||||
async movePending(
|
||||
threadId: string,
|
||||
id: string,
|
||||
targetId: string,
|
||||
expectedIds: string[],
|
||||
ctx: OperationContext,
|
||||
): Promise<boolean> {
|
||||
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<AgentMessageQueu
|
|||
return await this.managerFor(ctx).findOne(AgentMessageQueue, {
|
||||
where: { threadId },
|
||||
relations: { message: true },
|
||||
order: { id: 'ASC' },
|
||||
order: { position: 'ASC', id: 'ASC' },
|
||||
});
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -101,15 +101,18 @@ describe('Agent thread grants', () => {
|
|||
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());
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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<Array<{ id: string | number }>>(
|
||||
`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<Array<{ id: string | number }>>(
|
||||
`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();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
|
@ -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.",
|
||||
|
|
|
|||
|
|
@ -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<AgentChatQueueItem[]>([]);
|
||||
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<string>());
|
||||
|
|
@ -98,6 +101,10 @@ vi.mock('@n8n/design-system', async (importOriginal) => ({
|
|||
N8nApprovalCard: (await importOriginal<typeof import('@n8n/design-system')>()).N8nApprovalCard,
|
||||
N8nAiActivityStepGroup: (await importOriginal<typeof import('@n8n/design-system')>())
|
||||
.N8nAiActivityStepGroup,
|
||||
N8nAiActivityStepButton: (await importOriginal<typeof import('@n8n/design-system')>())
|
||||
.N8nAiActivityStepButton,
|
||||
N8nAiActivityStepChevron: (await importOriginal<typeof import('@n8n/design-system')>())
|
||||
.N8nAiActivityStepChevron,
|
||||
N8nLink: (await importOriginal<typeof import('@n8n/design-system')>()).N8nLink,
|
||||
useDropdownSearch: (await importOriginal<typeof import('@n8n/design-system')>())
|
||||
.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<void>();
|
||||
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();
|
||||
|
|
|
|||
|
|
@ -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<typeof effectScope>[] = [];
|
||||
|
|
@ -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<ReturnType<typeof makeControllableSseResponse>> = [];
|
||||
const signals: Array<AbortSignal | null> = [];
|
||||
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<ReturnType<typeof makeControllableSseResponse>> = [];
|
||||
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<AgentChatMessagesResponse>();
|
||||
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<ReturnType<typeof makeControllableSseResponse>> = [];
|
||||
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<AgentChatMessagesResponse>();
|
||||
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<AgentChatMessagesResponse>();
|
||||
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<typeof makeControllableSseResponse>;
|
||||
const signals: Array<AbortSignal | null> = [];
|
||||
|
|
@ -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<ReturnType<typeof makeControllableSseResponse>> = [];
|
||||
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<void>();
|
||||
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<AgentChatMessagesResponse>();
|
||||
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<AgentChatMessagesResponse>().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<void>();
|
||||
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] });
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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<HTMLDivElement>('messageQueue');
|
||||
const queueListId = useId();
|
||||
const queueExpanded = ref(false);
|
||||
const queueOrder = shallowRef<AgentChatQueueItem[]>();
|
||||
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<HTMLButtonElement>(
|
||||
`[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"
|
||||
>
|
||||
<template v-if="queueRows.length" #header>
|
||||
<div :class="$style.messageQueue" data-testid="agent-message-queue">
|
||||
<DefineQueueList v-slot="{ items }">
|
||||
<ul :class="[$style.backgroundJobList, $style.queueList]">
|
||||
<li v-for="item in items" :key="item.id" data-testid="agent-queued-message">
|
||||
<template v-if="displayedQueueRows.length" #header>
|
||||
<div ref="messageQueue" :class="$style.messageQueue" data-testid="agent-message-queue">
|
||||
<Draggable
|
||||
:id="queueListId"
|
||||
:model-value="visibleQueueRows"
|
||||
item-key="id"
|
||||
tag="ul"
|
||||
:class="[$style.backgroundJobList, $style.queueList]"
|
||||
:handle="`.${$style.queueDragHandle}:not(:disabled)`"
|
||||
:disabled="!!queueEdit || isReorderingQueue"
|
||||
:move="canDropQueueItem"
|
||||
:ghost-class="$style.queueGhost"
|
||||
:drag-class="$style.queueDragging"
|
||||
@start="startQueueDrag"
|
||||
@end="endQueueDrag"
|
||||
>
|
||||
<template #item="{ element: item, index }">
|
||||
<li :data-queue-id="item.id" data-testid="agent-queued-message">
|
||||
<N8nTooltip
|
||||
:content="locale.baseText('agents.chat.queue.reorderTooltip')"
|
||||
:disabled="!canDragQueueItem(index)"
|
||||
placement="top"
|
||||
>
|
||||
<N8nButton
|
||||
icon-only
|
||||
variant="ghost"
|
||||
size="xsmall"
|
||||
:class="$style.queueDragHandle"
|
||||
:disabled="!canDragQueueItem(index)"
|
||||
:aria-label="
|
||||
locale.baseText('agents.chat.queue.reorder', {
|
||||
interpolate: { position: index + 1, count: displayedQueueRows.length },
|
||||
})
|
||||
"
|
||||
aria-keyshortcuts="ArrowUp ArrowDown"
|
||||
data-testid="agent-queue-drag-handle"
|
||||
@keydown="onQueueHandleKeydown($event, index)"
|
||||
>
|
||||
<template #icon>
|
||||
<N8nIcon icon="grip-vertical" size="large" aria-hidden="true" />
|
||||
</template>
|
||||
</N8nButton>
|
||||
</N8nTooltip>
|
||||
<div :class="$style.queuePreview" :title="item.message">
|
||||
<N8nInput
|
||||
v-if="queueEdit?.item.id === item.id"
|
||||
v-if="queueEdit && queueEdit.item.id === item.id"
|
||||
v-model="queueEdit.text"
|
||||
type="textarea"
|
||||
size="small"
|
||||
|
|
@ -933,7 +1038,7 @@ onBeforeUnmount(() => {
|
|||
/>
|
||||
<span v-else-if="item.message">{{ item.message }}</span>
|
||||
<p
|
||||
v-if="queueEdit?.item.id === item.id && queueEdit.unavailable"
|
||||
v-if="queueEdit && queueEdit.item.id === item.id && queueEdit.unavailable"
|
||||
:class="$style.queueEditNotice"
|
||||
role="status"
|
||||
>
|
||||
|
|
@ -957,7 +1062,7 @@ onBeforeUnmount(() => {
|
|||
}}</span>
|
||||
</div>
|
||||
<div :class="$style.queueActions">
|
||||
<template v-if="queueEdit?.item.id === item.id">
|
||||
<template v-if="queueEdit && queueEdit.item.id === item.id">
|
||||
<N8nTooltip
|
||||
:content="locale.baseText('agents.chat.queue.save')"
|
||||
:disabled="!canSaveQueueEdit"
|
||||
|
|
@ -1053,25 +1158,28 @@ onBeforeUnmount(() => {
|
|||
</template>
|
||||
</div>
|
||||
</li>
|
||||
</ul>
|
||||
</DefineQueueList>
|
||||
<QueueList :items="queueRows.slice(0, 2)" />
|
||||
<N8nAiActivityStepGroup
|
||||
v-if="queueRows.length > 2"
|
||||
:key="continueSessionId"
|
||||
:label="
|
||||
</template>
|
||||
</Draggable>
|
||||
<N8nAiActivityStepButton
|
||||
v-if="displayedQueueRows.length > 2"
|
||||
:aria-expanded="queueExpanded"
|
||||
:aria-controls="queueListId"
|
||||
:disabled="!!queueOrder"
|
||||
full-width
|
||||
@click="queueExpanded = !queueExpanded"
|
||||
>
|
||||
{{
|
||||
queuedMessages.length > 2
|
||||
? locale.baseText('agents.chat.queue.title', {
|
||||
adjustToNumber: queuedMessages.length - 2,
|
||||
interpolate: { count: queuedMessages.length - 2 },
|
||||
})
|
||||
: locale.baseText('agents.chat.queue.edit')
|
||||
"
|
||||
full-width
|
||||
content-position="above"
|
||||
>
|
||||
<QueueList :items="queueRows.slice(2)" />
|
||||
</N8nAiActivityStepGroup>
|
||||
}}
|
||||
<template #suffix>
|
||||
<N8nAiActivityStepChevron :open="queueExpanded" direction="down" />
|
||||
</template>
|
||||
</N8nAiActivityStepButton>
|
||||
</div>
|
||||
</template>
|
||||
<template v-if="attachedFiles.length > 0" #attachments>
|
||||
|
|
@ -1142,16 +1250,18 @@ onBeforeUnmount(() => {
|
|||
}
|
||||
|
||||
.messageQueue {
|
||||
--text-color: var(--text-color--subtle);
|
||||
--text-color: light-dark(var(--color--neutral-600), var(--text-color--subtler));
|
||||
--icon-color: var(--color--neutral-400);
|
||||
|
||||
margin: calc(-1 * var(--spacing--2xs)) calc(-1 * var(--spacing--2xs)) 0;
|
||||
background: var(--background--subtle);
|
||||
border-radius: var(--radius--lg) var(--radius--lg) 0 0;
|
||||
border-bottom: var(--border);
|
||||
border-bottom-color: var(--border-color--subtle);
|
||||
}
|
||||
|
||||
.messageQueue :global(.n8n-icon) {
|
||||
color: light-dark(var(--color--neutral-600), var(--color--neutral-400));
|
||||
color: var(--icon-color);
|
||||
}
|
||||
|
||||
.backgroundJobDetails {
|
||||
|
|
@ -1195,11 +1305,16 @@ onBeforeUnmount(() => {
|
|||
font-size: var(--font-size--2xs);
|
||||
}
|
||||
|
||||
.queueList {
|
||||
max-height: calc(20vh + 2 * var(--height--xl));
|
||||
}
|
||||
|
||||
.queueList > li {
|
||||
align-items: center;
|
||||
padding-inline: var(--spacing--sm);
|
||||
color: var(--text-color);
|
||||
border-bottom: var(--border);
|
||||
border-bottom-color: var(--border-color--subtle);
|
||||
line-height: var(--line-height--md);
|
||||
}
|
||||
|
||||
|
|
@ -1221,6 +1336,30 @@ onBeforeUnmount(() => {
|
|||
flex-shrink: 0;
|
||||
}
|
||||
|
||||
.queueDragHandle {
|
||||
flex-shrink: 0;
|
||||
cursor: grab;
|
||||
touch-action: none;
|
||||
|
||||
&:active {
|
||||
cursor: grabbing;
|
||||
}
|
||||
|
||||
&:disabled {
|
||||
cursor: default;
|
||||
}
|
||||
}
|
||||
|
||||
.queueGhost {
|
||||
opacity: 0.4;
|
||||
}
|
||||
|
||||
.queueDragging {
|
||||
background: var(--background--subtle);
|
||||
box-shadow: var(--shadow--sm);
|
||||
cursor: grabbing;
|
||||
}
|
||||
|
||||
.queueEditNotice {
|
||||
margin: var(--spacing--3xs) 0;
|
||||
font-size: var(--font-size--2xs);
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ import type {
|
|||
AgentChatQueueResponse,
|
||||
AgentChatQueueUpdateDto,
|
||||
AgentChatQueueSteerDto,
|
||||
AgentChatQueueReorderDto,
|
||||
AgentChatResumeDto,
|
||||
AgentConfigMutationResponse,
|
||||
AgentConfigResponse,
|
||||
|
|
@ -592,6 +593,22 @@ export const updateAgentQueuedMessage = async (
|
|||
);
|
||||
};
|
||||
|
||||
export const reorderAgentQueuedMessage = async (
|
||||
context: IRestApiContext,
|
||||
projectId: string,
|
||||
agentId: string,
|
||||
threadId: string,
|
||||
queueId: string,
|
||||
payload: AgentChatQueueReorderDto,
|
||||
): Promise<void> => {
|
||||
await makeRestApiRequest(
|
||||
context,
|
||||
'POST',
|
||||
`/projects/${encodeURIComponent(projectId)}/agents/v2/${encodeURIComponent(agentId)}/chat/${encodeURIComponent(threadId)}/queue/${encodeURIComponent(queueId)}/reorder`,
|
||||
payload,
|
||||
);
|
||||
};
|
||||
|
||||
export const removeAgentQueuedMessage = async (
|
||||
context: IRestApiContext,
|
||||
projectId: string,
|
||||
|
|
|
|||
|
|
@ -24,6 +24,7 @@ import {
|
|||
removeAgentQueuedMessage,
|
||||
updateAgentQueuedMessage,
|
||||
steerAgentQueuedMessage,
|
||||
reorderAgentQueuedMessage,
|
||||
getTestChatMessages,
|
||||
} from './useAgentApi';
|
||||
|
||||
|
|
@ -87,6 +88,7 @@ type ResumePayload =
|
|||
|
||||
const STOP_ACCEPTANCE_TIMEOUT_MS = 30 * TIME.SECOND;
|
||||
const MAX_WAITING_STREAMS = 2;
|
||||
const MAX_START_VALIDATION_ATTEMPTS = 3;
|
||||
|
||||
function getApprovalDecision(value: unknown): boolean | undefined {
|
||||
if (!isRecord(value) || typeof value.approved !== 'boolean') return undefined;
|
||||
|
|
@ -108,6 +110,7 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
const queuedMessages = ref<AgentChatQueueItem[]>([]);
|
||||
const removingQueueIds = ref(new Set<string>());
|
||||
const steeringQueueIds = ref(new Set<string>());
|
||||
const isReorderingQueue = ref(false);
|
||||
const consumedQueueIds = new Set<string>();
|
||||
const steerableExecutionId = ref<string | null>(null);
|
||||
const streams = new Map<AbortController, StreamSession>();
|
||||
|
|
@ -138,6 +141,7 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
let disposed = false;
|
||||
let historyVersion = 0;
|
||||
let streamVersion = 0;
|
||||
let ownershipVersion = 0;
|
||||
let refreshAfterStream = false;
|
||||
let retryCount = 0;
|
||||
let retryTimer: ReturnType<typeof setTimeout> | undefined;
|
||||
|
|
@ -358,6 +362,41 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
}
|
||||
}
|
||||
|
||||
async function reorderQueuedMessage(
|
||||
queueId: string,
|
||||
targetQueueId: string,
|
||||
expectedQueueIds: string[],
|
||||
): Promise<void> {
|
||||
const threadId = params.continueSessionId?.value ?? acceptedSessionId.value;
|
||||
if (!threadId || disposed || isReorderingQueue.value) return;
|
||||
const target = targetKey();
|
||||
isReorderingQueue.value = true;
|
||||
try {
|
||||
await reorderAgentQueuedMessage(
|
||||
rootStore.restApiContext,
|
||||
params.projectId.value,
|
||||
params.agentId.value,
|
||||
threadId,
|
||||
queueId,
|
||||
{ targetQueueId, expectedQueueIds },
|
||||
);
|
||||
} catch (error) {
|
||||
if (!disposed && target === targetKey()) {
|
||||
const status = isRecord(error) ? error.httpStatusCode : undefined;
|
||||
const key =
|
||||
status === 404 || status === 409
|
||||
? 'agents.chat.queue.reorderUnavailable'
|
||||
: 'agents.chat.queue.reorderError';
|
||||
showError(error, locale.baseText(key));
|
||||
}
|
||||
} finally {
|
||||
if (!disposed && target === targetKey()) {
|
||||
await refreshQueue();
|
||||
if (!disposed && target === targetKey()) isReorderingQueue.value = false;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function steerQueuedMessage(queueId: string): Promise<void> {
|
||||
const threadId = params.continueSessionId?.value ?? acceptedSessionId.value;
|
||||
const executionId = steerableExecutionId.value;
|
||||
|
|
@ -437,6 +476,13 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
clearTimeout(retryTimer);
|
||||
refreshHistoryFromPush();
|
||||
}
|
||||
watch(
|
||||
[activeExecutionId, abortController],
|
||||
() => {
|
||||
ownershipVersion++;
|
||||
},
|
||||
{ flush: 'sync' },
|
||||
);
|
||||
watch(
|
||||
() => pushStore.isConnected,
|
||||
(connected) => {
|
||||
|
|
@ -452,6 +498,7 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
queuedMessages.value = [];
|
||||
removingQueueIds.value.clear();
|
||||
steeringQueueIds.value.clear();
|
||||
isReorderingQueue.value = false;
|
||||
consumedQueueIds.clear();
|
||||
steerableExecutionId.value = null;
|
||||
queueVersion++;
|
||||
|
|
@ -495,6 +542,7 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
queueId?: string;
|
||||
userMessage?: ChatMessage;
|
||||
executionId?: string;
|
||||
needsStartValidation?: boolean;
|
||||
busy?: boolean;
|
||||
onAccepted?: () => void;
|
||||
/**
|
||||
|
|
@ -754,10 +802,15 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
if (session.userMessage) {
|
||||
const inputMessageId = event.inputMessageIds?.[0];
|
||||
if (inputMessageId) session.userMessage.id = inputMessageId;
|
||||
// Local messages enter FIFO order. An earlier request cannot own a later turn.
|
||||
for (const [controller, earlier] of streams) {
|
||||
if (controller === session.controller) break;
|
||||
if (earlier.userMessage) controller.abort();
|
||||
}
|
||||
for (const [controller, other] of streams) {
|
||||
if (controller === session.controller) continue;
|
||||
if (other.executionId) {
|
||||
finalizeStream(other);
|
||||
controller.abort();
|
||||
} else {
|
||||
// A waiting stream can carry a buffered start from a completed execution.
|
||||
other.needsStartValidation = true;
|
||||
}
|
||||
}
|
||||
abortController.value = session.controller;
|
||||
|
|
@ -1045,6 +1098,34 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
}
|
||||
}
|
||||
|
||||
async function handleDelayedExecutionStart(
|
||||
event: Extract<AgentSseEvent, { type: 'execution-started' }>,
|
||||
session: StreamSession,
|
||||
): Promise<{ done?: boolean } | undefined> {
|
||||
const isCurrent = () =>
|
||||
!disposed && !session.controller.signal.aborted && session.target === targetKey();
|
||||
for (let attempt = 0; attempt < MAX_START_VALIDATION_ATTEMPTS && isCurrent(); attempt++) {
|
||||
const ownershipAtStart = ownershipVersion;
|
||||
const history = await getChatMessages(
|
||||
rootStore.restApiContext,
|
||||
params.projectId.value,
|
||||
params.agentId.value,
|
||||
event.sessionId,
|
||||
).catch(() => undefined);
|
||||
if (!isCurrent()) break;
|
||||
// Timeline updates do not invalidate execution ownership.
|
||||
if (ownershipAtStart !== ownershipVersion) continue;
|
||||
if (history?.activeExecutionId !== event.executionId) break;
|
||||
// The buffered stream replays this execution from the start.
|
||||
messages.value = messages.value.filter(
|
||||
(message) => message.executionId !== event.executionId,
|
||||
);
|
||||
return handleEvent(event, session);
|
||||
}
|
||||
session.controller.abort();
|
||||
return { done: true };
|
||||
}
|
||||
|
||||
async function consumeStream(
|
||||
response: Response,
|
||||
session: StreamSession,
|
||||
|
|
@ -1072,7 +1153,10 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
} catch {
|
||||
continue;
|
||||
}
|
||||
const result = handleEvent(event, session);
|
||||
const result =
|
||||
event.type === 'execution-started' && session.needsStartValidation
|
||||
? await handleDelayedExecutionStart(event, session)
|
||||
: handleEvent(event, session);
|
||||
if (result?.done) {
|
||||
break readerLoop;
|
||||
}
|
||||
|
|
@ -1557,6 +1641,8 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
removingQueueIds,
|
||||
removeQueuedMessage,
|
||||
updateQueuedMessage,
|
||||
isReorderingQueue,
|
||||
reorderQueuedMessage,
|
||||
isSubmitting,
|
||||
isLoadingHistory,
|
||||
messages,
|
||||
|
|
|
|||
|
|
@ -91,6 +91,20 @@ export class AgentApiHelper {
|
|||
throw new TestError(`Failed to remove queued message: ${await response.text()}`);
|
||||
}
|
||||
|
||||
async reorderQueuedMessage(
|
||||
projectId: string,
|
||||
agentId: string,
|
||||
threadId: string,
|
||||
queueId: string,
|
||||
targetQueueId: string,
|
||||
expectedQueueIds: string[],
|
||||
) {
|
||||
return await this.api.request.post(
|
||||
`/rest/projects/${projectId}/agents/v2/${agentId}/chat/${threadId}/queue/${queueId}/reorder`,
|
||||
{ data: { targetQueueId, expectedQueueIds } },
|
||||
);
|
||||
}
|
||||
|
||||
async steerQueuedMessage(
|
||||
projectId: string,
|
||||
agentId: string,
|
||||
|
|
|
|||
|
|
@ -126,7 +126,7 @@ test.describe(
|
|||
'Agent message queue @mode:multi-main',
|
||||
{ annotation: [{ type: 'owner', description: 'Agent' }] },
|
||||
() => {
|
||||
test('preserves FIFO, deduplication, and remote steering across main crashes', async ({
|
||||
test('preserves queue order, deduplication, and remote steering across main crashes', async ({
|
||||
n8nContainer,
|
||||
createApiForMain,
|
||||
}) => {
|
||||
|
|
@ -226,6 +226,51 @@ test.describe(
|
|||
({ message }) => message,
|
||||
),
|
||||
).toEqual(['fifo-remote', 'third']);
|
||||
const firstQueueId = first.events.find((event) => event.type === 'message-queued')?.queueId;
|
||||
assert(firstQueueId);
|
||||
expect(
|
||||
(
|
||||
await consumer.agents.reorderQueuedMessage(
|
||||
project.id,
|
||||
agent.id,
|
||||
threadId,
|
||||
firstQueueId,
|
||||
restored.items[0].id,
|
||||
[restored.items[0].id, restored.items[1].id],
|
||||
)
|
||||
).status(),
|
||||
).toBe(409);
|
||||
const swaps = await Promise.all(
|
||||
clients.map(
|
||||
async (client) =>
|
||||
await client.agents.reorderQueuedMessage(
|
||||
project.id,
|
||||
agent.id,
|
||||
threadId,
|
||||
restored.items[0].id,
|
||||
restored.items[1].id,
|
||||
[restored.items[0].id, restored.items[1].id],
|
||||
),
|
||||
),
|
||||
);
|
||||
expect(swaps.map((response) => response.status()).sort()).toEqual([200, 409]);
|
||||
const appended = await open(0, threadId, 'append after reordering');
|
||||
await expect
|
||||
.poll(() => appended.events.find((event) => event.type === 'message-queued'))
|
||||
.toBeTruthy();
|
||||
const reordered = await consumer.agents.queuedMessages(project.id, agent.id, threadId);
|
||||
expect(reordered.items.map(({ message }) => message)).toEqual([
|
||||
'third',
|
||||
'fifo-remote',
|
||||
'append after reordering',
|
||||
]);
|
||||
await consumer.agents.removeQueuedMessage(
|
||||
project.id,
|
||||
agent.id,
|
||||
threadId,
|
||||
reordered.items[2].id,
|
||||
);
|
||||
appended.disconnect();
|
||||
removable.disconnect();
|
||||
const independent = await open(1, randomUUID(), 'independent', true);
|
||||
expect(await independent.done).toBeUndefined();
|
||||
|
|
@ -247,9 +292,21 @@ test.describe(
|
|||
),
|
||||
{ timeout: 30_000 },
|
||||
)
|
||||
.toEqual(['cancelled', 'running']);
|
||||
.toEqual(['cancelled', 'success', 'running']);
|
||||
await signalMain(n8nContainer, 0, 'SIGCONT');
|
||||
await expect.poll(() => executionId(second.events)).toBeTruthy();
|
||||
expect(
|
||||
(
|
||||
await ingress.agents.reorderQueuedMessage(
|
||||
project.id,
|
||||
agent.id,
|
||||
threadId,
|
||||
restored.items[1].id,
|
||||
restored.items[0].id,
|
||||
[restored.items[1].id, restored.items[0].id],
|
||||
)
|
||||
).status(),
|
||||
).toBe(409);
|
||||
const steeringRequestId = randomUUID();
|
||||
const steered = await open(0, threadId, 'additional input', undefined, steeringRequestId);
|
||||
await expect
|
||||
|
|
@ -272,6 +329,26 @@ test.describe(
|
|||
({ id }) => id === steerId,
|
||||
)?.steeringExecutionId,
|
||||
).toBe(executionId(second.events));
|
||||
const later = await open(0, threadId, 'pending after steering');
|
||||
await expect
|
||||
.poll(() => later.events.find((event) => event.type === 'message-queued'))
|
||||
.toBeTruthy();
|
||||
const laterId = later.events.find((event) => event.type === 'message-queued')?.queueId;
|
||||
assert(laterId);
|
||||
expect(
|
||||
(
|
||||
await consumer.agents.reorderQueuedMessage(
|
||||
project.id,
|
||||
agent.id,
|
||||
threadId,
|
||||
steerId,
|
||||
laterId,
|
||||
[steerId, laterId],
|
||||
)
|
||||
).status(),
|
||||
).toBe(409);
|
||||
await consumer.agents.removeQueuedMessage(project.id, agent.id, threadId, laterId);
|
||||
later.disconnect();
|
||||
await expect
|
||||
.poll(
|
||||
async () =>
|
||||
|
|
@ -322,7 +399,7 @@ test.describe(
|
|||
(await ingress.agents.executions(project.id, agent.id, threadId)).map(
|
||||
({ userMessage }) => userMessage,
|
||||
),
|
||||
).toEqual(['fifo-blocked', 'fifo-remote\nadditional input', 'third']);
|
||||
).toEqual(['fifo-blocked', 'third', 'fifo-remote\nadditional input']);
|
||||
|
||||
const crashThread = randomUUID();
|
||||
const interruptedId = randomUUID();
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user